我需要最快的方法來處理數值陣列資料的數學運算

如果這是在不正確的論壇中，我深表歉意。盡管在這個站點上發現了很多陣列操作，但其中大多數是平均/求和……使用 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回圈（就像 a List）處理佇列中的值，而不必在我準備好之前出隊。如果有人知道這是否會影響性能，我會很感興趣，因為我沒有對其進行基準測驗。請記住，它必須是執行緒安全的。如果您有以執行緒安全的方式處理多組陣列資料的替代方法，我會“全神貫注”。

性能要求的原因是這不是唯一發生的程序，我有多個設備以大約每 1-5 毫秒 1 的速率“流式傳輸”陣列結果，對于來自不同執行緒/行程的每個設備/connections，還有其他幾個更密集的演算法要處理，所以這不是瓶頸。

感謝您對優化和性能的任何見解。

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

uj5u.com熱心網友回復：

在處理大量資料的簡單操作時，您會對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，它可能4在大多數支持AVX2 的現代 CPU 上。

如果您能夠接受失去精度，可以去float，你會得到一個很大再次加倍在一個單一的CPU運算處理的資料量更大的勝利。所有這一切甚至無需更改您的演算法。

uj5u.com熱心網友回復：

您可以通過減少記憶體分配來進一步優化代碼。如果該方法被頻繁呼叫，花費的時間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();

uj5u.com熱心網友回復：

在我看來，這將是相當好的優化代碼。第二個選項更快的一個主要原因是它線性訪問記憶體，而不是在多個不同的陣列之間跳轉。另一個因素是foreach回圈有一些開銷，因此將其放在外回圈中也會有所幫助。

通過將佇列和 foreach 回圈切換為串列/陣列和 for 回圈，您可能會獲得一點性能，但由于PixelSize比CountToAverage我預期的要大得多，因此收益相當小。

展開回圈以一次處理 4 個值可能會有所幫助。c# 編譯器可以自動應用此類優化，但通常很難判斷應用了哪些優化，因此僅進行測驗可能更容易。

下一步是研究并行化。像這樣的簡單求和代碼可能會受益于SIMD 一次處理多個值。但是該鏈接顯示，使用特定于處理器的內在函式比更通用的具有更大的好處Vector<T>，但可能需要針對您所針對的每個平臺使用單獨的代碼路徑。該鏈接還提供了在各種優化級別上求和值的性能示例，以及示例代碼，因此非常值得一讀。

另一種選擇是使用多個執行緒Parallel.For/Foreach，但在 6μs 時，開銷可能會大于任何增益，除非資料的大小明顯更大。

標籤：C＃ 数组 多线程 表现 优化

上一篇：.NET應用程式的性能改程序序是什么？

下一篇：使用data.table在j中使用順序任務加速for回圈