我需要最快的方法来处理数值数组数据的数学

Question

如果这是在错误的论坛中，我深表歉意。尽管在此站点上发现了很多数组操作，但其中大多数都是平均/求和...使用 LINQ 将数字数组作为一个集合，它可以很好地处理数组中的所有值。但我需要处理多个数组（大小相同）上的每个索引。

我的例程从设备接收数组数据，通常是double[512] 或ushort[512]；单个设备本身将始终具有相同大小的数组数据，但数组大小的范围可以从 256 到 2048，具体取决于设备。我需要保持CountToAverage 数组的数量来平均。每次收到一个数组，都要从队列中push和pop，以保证平均进程中数组的个数是一致的（这部分进程固定在Setup()中，用于本次基准测试。为了比较的目的，基准测试结果显示在代码之后。

1. 我正在寻找的是平均所有数组的每个索引的值的最快最有效的方法，并返回一个新数组（大小相同），其中每个索引是从数组集中平均的。要平均的数组数量可以在 3 到 25 之间（下面的代码将基准参数设置为 10）。我在测试中有 2 种不同的平均方法，第 2 种明显快，比第一种快 6-7 倍。我的第一个问题是；有什么方法可以更快地实现这一点，可以在 O(1) 或 O(log n) 时间复杂度下完成？

2. 其次，我使用一个队列（可以更改为ConcurrentQueue 来实现）作为要处理的数组的持有者。我使用队列的主要原因是因为我可以保证对数组的馈送进行 FIFO 处理，这很关键。此外，我可以通过foreach 循环（就像List）处理队列中的值，而无需在我准备好之前出列。如果有人知道这是否会阻碍性能，我会很感兴趣，因为我没有对其进行基准测试。请记住，它必须是线程安全的。如果您有另一种方法来以线程安全的方式处理多组数组数据，我会“全神贯注”。

性能要求的原因是这不是唯一发生的过程，我有多个设备以大约每 1-5 毫秒 1 的速率“流式传输”发送阵列结果，每个设备来自不同的线程/进程/连接，还有其他几个更密集的算法需要处理，所以这不是瓶颈。

感谢您对优化和性能的任何见解。

using System;
using System.Collections.Generic;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using BenchmarkDotNet.Running;
using Microsoft.Diagnostics.Tracing.Parsers.MicrosoftAntimalwareEngine;

namespace ArrayAverage
{
    public class ArrayAverage
    {
        [Params(10)]
        public int CountToAverage;

        [Params(512, 2048)]
        public int PixelSize;

        static Queue<double[]> calcRepo = new Queue<double[]>();
        static List<double[]> spectra = new();
        
        [Benchmark]
        public double[] CalculateIndexAverages()
        {
            // This is too slow
            var avg = new double[PixelSize];
            for (int i = 0; i < PixelSize; i++)
            {
                foreach (var arrayData in calcRepo)
                {
                    avg[i] += arrayData[i];
                }
                avg[i] /= calcRepo.Count;
            }
            return avg;
        }
        
        [Benchmark]
        public double[] CalculateIndexAverages2()
        {
            // this is faster, but is it the fastest?
            var sum = new double[PixelSize];
            int cnt = calcRepo.Count;
            foreach (var arrayData in calcRepo)
            {
                for (int i = 0; i < PixelSize; i++)
                {
                    sum[i] += arrayData[i];
                }
            }

            var avg = new double[PixelSize];
            for (int i = 0; i < PixelSize; i++)
            {
                avg[i] = sum[i] / cnt;
            }

            return avg;
        }
        
        [GlobalSetup]
        public void Setup()
        {
            // Just generating some data as simple Triangular curve simulating a range of spectra
            for (double offset = 0; offset < CountToAverage; offset++)
            {
                var values = new double[PixelSize];
                var decrement = 0;
                for (int i = 0; i < PixelSize; i++)
                {
                    if (i > (PixelSize / 2))
                        decrement--;
                    values[i] = (offset / 7) + i + (decrement * 2);
                }
                calcRepo.Enqueue(values);
            }
        }        
    }
    
    public class App
    {
        public static void Main()
        {
            BenchmarkRunner.Run<ArrayAverage>();
        }
    }
}

基准测试结果：

BenchmarkDotNet=v0.13.1, OS=Windows 10.0.19043.1348 (21H1/May2021Update)
Intel Core i7-6700HQ CPU 2.60GHz (Skylake), 1 CPU, 8 logical and 4 physical cores
.NET SDK=6.0.100-preview.7.21379.14
  [Host]     : .NET 5.0.12 (5.0.1221.52207), X64 RyuJIT  [AttachedDebugger]
  DefaultJob : .NET 5.0.12 (5.0.1221.52207), X64 RyuJIT

Method	Arrays To Average	Array Size	Mean	Error	StdDev
CalculateIndexAverages	10	512	32.164 μs	0.5485 μs	0.5130 μs
CalculateIndexAverages2	10	512	5.792 μs	0.1135 μs	0.2241 μs
CalculateIndexAverages	10	2048	123.628 μs	2.3394 μs	1.9535 μs
CalculateIndexAverages2	10	2048	22.311 μs	0.4366 μs	0.8093 μs

Answer 1

在处理大量数据的简单操作时，你会对SIMD很感兴趣：

SIMD 代表“单指令多数据”。它是一组处理器指令......允许数学运算在一组值上并行执行。

在您的特定情况下，使用Vector<T> 示例可以让您快速获胜。天真地将您最快的方法转换为使用 Vectors 已经在我的 PC 上提供了约 2 倍的速度。

public double[] CalculateIndexAverages4() {
    // Assumption: PixelSize is a round multiple of Vector<>.Count
    // If not, you'll have to add in the 'remainder' from the example.
    var batch = Vector<double>.Count;
    
    var sum = new double[PixelSize];
    foreach (var arrayData in calcRepo) {
        // Vectorised summing:
        for (int i = 0; i <= PixelSize - batch; i += batch) {
            var vSum = new Vector<double>(sum, i);
            var vData = new Vector<double>(arrayData, i);
            (vSum + vData).CopyTo(sum, i);
        }
    }

    var vCnt = Vector<double>.One * calcRepo.Count;
    // Reuse sum[] for averaging, so we don't incur memory allocation cost
    for (int i = 0; i <= PixelSize - batch; i += batch) {
        var vSum = new Vector<double>(sum, i);
        (vSum / vCnt).CopyTo(sum, i);
    }
    return sum;
}

Vector<T>.Count 告诉您有多少项被并行化到一条指令中。对于double，在大多数支持AVX2 的现代CPU 上可能是4。

如果您不介意丢失精度并且可以转到float，那么通过再次将单个 CPU 操作中处理的数据量翻倍，您将获得更多更大的胜利。所有这些甚至不需要改变你的算法。

Answer 2

您可以通过减少内存分配来进一步优化代码。如果该方法被频繁调用，则花费在GC 上的时间将完全占主导地位。

// Assuming the data fits on the stack. Some 100k pixels should be safe.
Span<double> sum = stackalloc double[PixelSize];
// ...
Span<double> avg = stackalloc double[PixelSize];

还可能删除avg 的额外堆栈分配并简单地重用sum：

for (int i = 0; i < sum.Length; i++)
{
    sum[i] /= cnt;
}

// TODO: Avoid array allocation! Maybe use a pre-allocated array and fill it here.
return sum.ToArray();

Answer 3

在我看来，这将是相当优化的代码。第二个选项更快的一个主要原因是它线性地访问内存，而不是在多个不同的数组之间跳转。另一个因素是foreach 循环有一些开销，因此将它放在外部循环中也会有所帮助。

通过将队列和 foreach 循环切换为列表/数组和 for 循环，您可能会获得一点性能，但由于 PixelSize 比 CountToAverage 大得多，我预计收益会相当小。

展开循环以一次处理 4 个值可能会有所帮助。 c# 编译器可以自动应用此类优化，但通常很难判断应用了哪些优化，因此仅进行测试可能更容易。

下一步是研究并行化。像这样的简单求和代码可能会受益于 SIMD 一次处理多个值。但是该链接显示，与更通用的Vector<T> 相比，使用特定于处理器的内在函数具有更大的优势，但可能需要针对您所针对的每个平台使用单独的代码路径。该链接还包含在各种优化级别上求和值的性能示例，以及示例代码，因此非常值得一读。

另一种选择是使用Parallel.For/Foreach 的多个线程，但在 6μs 时，开销可能会大于任何增益，除非数据的大小明显更大。