zyl910

優化技巧、硬件體系、圖像處理、圖形學、游戲編程、國際化與文本信息處理。

:: :: :: :: :: ::

:: ::

公告

[C#] 對圖像進行垂直翻轉(FlipY)的跨平臺SIMD硬件加速向量算法，兼談并行處理收益極少的原因

將位圖進行水平翻轉或垂直翻轉，這些是圖像處理的常用算法。本文介紹最簡單的垂直翻轉，便于后續文章來探討復雜的水平翻轉算法。
本文除了會給出標量算法外，還會給出向量算法。且這些算法是跨平臺的，同一份源代碼，能在 X86及Arm架構上運行，且均享有SIMD硬件加速。

一、標量算法

1.1 算法實現

垂直翻轉的算法原理是很簡單的，就是將位圖的每一行重新排列——

第0行放到第(height-1)行；
第1行放到第(height-1-1)行；
第2行放到第(height-1-2)行；

用i表示當前行的話，那么垂直翻轉的規律是——

第i行放到第(height-1-i)行。

由于此時是整行數據的復制，所以無需關心像素格式。算好每行的字節數后，便可以用字節復制的辦法來復制一行數據了。而水平翻轉需區分不同的像素格式，比垂直翻轉要復雜的多，后面的文章會詳細講解水平翻轉。
下面就是標量算法的源代碼。

public static unsafe void ScalarDoBatch(byte* pSrc, int strideSrc, int width, int height, byte* pDst, int strideDst) {
    int strideCommon = Math.Min(Math.Abs(strideSrc), Math.Abs(strideDst));
    byte* pRow = pSrc;
    byte* qRow = pDst + (height - 1) * (long)strideDst; // Set to last row.
    for (int i = 0; i < height; i++) {
        byte* p = pRow;
        byte* q = qRow;
        for (int j = 0; j < strideCommon; j++) {
            *q++ = *p++;
        }
        pRow += strideSrc;
        qRow -= strideDst;
    }
}

為了支持任意的像素格式，可以根據 stride（跨距）來計算每一行的字節數。注意stride有可能是負數，故需要用Abs函數來計算絕對值。計算出strideCommon后，內循環便根據它，對當前行的數據進行逐個字節復制處理，直至復制完一整行。

外循環采用了“順序讀取、逆序寫入”的策略。具體來說——

讀取是從第0行開始的，每次循環后移動到下一行。于是在上面的源代碼中，pRow的初值就是 pSrc，每次循環后pRow會加上跨距strideSrc。
寫入是從最后一行開始的，每次循環后移動到前一行。于是在上面的源代碼中，qRow的初值是 pDst + (height - 1) * (long)strideDst（目標位圖最后一行的地址），每次循環后qRow會減去跨距strideDst。

1.2 基準測試代碼

使用 BenchmarkDotNet 進行基準測試。
可以使用上一篇文章的公共函數，寫好標量算法的基準測試代碼。源代碼如下。

[Benchmark(Baseline = true)]
public void Scalar() {
    ScalarDo(_sourceBitmapData, _destinationBitmapData, false);
}

[Benchmark]
public void ScalarParallel() {
    ScalarDo(_sourceBitmapData, _destinationBitmapData, true);
}

public static unsafe void ScalarDo(BitmapData src, BitmapData dst, bool useParallel = false) {
    int width = src.Width;
    int height = src.Height;
    int strideSrc = src.Stride;
    int strideDst = dst.Stride;
    byte* pSrc = (byte*)src.Scan0.ToPointer();
    byte* pDst = (byte*)dst.Scan0.ToPointer();
    bool allowParallel = useParallel && (height > 16) && (Environment.ProcessorCount > 1);
    if (allowParallel) {
        Parallel.For(0, height, i => {
            int start = i;
            int len = 1;
            byte* pSrc2 = pSrc + start * (long)strideSrc;
            byte* pDst2 = pDst + (height - 1 - start) * (long)strideDst;
            ScalarDoBatch(pSrc2, strideSrc, width, len, pDst2, strideDst);
        });
    } else {
        ScalarDoBatch(pSrc, strideSrc, width, height, pDst, strideDst);
    }
}

由于現在是圖像垂直翻轉，于是并行算法需按照同樣的規則來計算數據的地址。即上面提到的“順序讀取、逆序寫入”。
具體來說——

pSrc2 是源數據的指針。由于是順序讀取，故它的計算辦法是 pSrc + start * (long)strideSrc.
pDst2 是目標數據的指針。由于是逆序寫入，故它的計算辦法是 pDst + (height - 1 - start) * (long)strideDst. 編寫代碼時有時可能會漏掉了其中的"- 1"，這是一種常見錯誤，容易導致指針訪問非法地址等問題。請大家一定要細心。

二、向量算法

2.1 算法思路

上面的標量算法是每次復制1個字節，而向量算法可以每次復制1個向量。

若一行數據的字節數，恰好是向量大小的整數倍的話，那可以直接寫個簡單循環來處理。非常簡單。

但在實際使用中，字節數大多數時候并不是向量大小的整數倍，此時處理起來會復雜一些。

2.1.1 解決非整數倍帶來的難點

最保守的辦法是——先用向量算法將整數倍的數據給處理完，隨后對于末尾的數據，退回為標量算法來處理。

該辦法確實有效，但存在以下缺點：

需要編寫2種算法——向量的和標量的。會帶來更多開發成本與維護成本。
末尾的數據是標量處理的，向量硬件加速不起作用。

有沒有辦法只需寫1種算法，且對末尾的數據也能向量化處理呢？

對于“一邊讀取、一邊寫入”這種情況，由于讀、寫的一般是不同的數據區域，故有一種辦法來實現上面的期望。該辦法的關鍵思路是——先計算好“末尾指針”（向量處理時最后一筆數據的指針地址），然后在循環內每次均檢查指針地址，當發現當前指針已經越過“末尾指針”時便強制將當前指針設置為“末尾指針”，以完成剩余數據的處理。

使用這種辦法，能使末尾的數據也能向量化處理。代價就是——部分末尾的數據會被重復處理。但由于現在是“一邊讀取、一邊寫入”，故重復處理并不會帶來負面影響。

在向量運算的循環中，增加指針地址判斷的分支代碼，的確會影響性能。但是由于現在CPU都擁有強大的分支預測能力，對于“末尾指針”判斷這種在最后一次循環時才生效的分支，分支預測絕大多數是成功的，從而掩蓋了該分支判斷的花費。

其實本系列的前面幾篇文章，也是使用這一辦法，使末尾的數據也能向量化處理。所以代碼中有“The last block is also use vector”的注釋。

（另注：在C語言中，C99標準新增了restrict關鍵字來標記“指針指向不同的數據區域”情況。restrict用于限定和約束指針，表示這個指針只訪問這塊內存的唯一方式，也就是告訴編譯器，這塊內存中的內容的操作都只會通過這個指針，而不會通過其他變量或者指針。C# 目前還沒有這樣的關鍵字）

2.2 算法實現

根據上面的思路，編寫代碼。源代碼如下。

public static unsafe void UseVectorsDoBatch(byte* pSrc, int strideSrc, int width, int height, byte* pDst, int strideDst) {
    int strideCommon = Math.Min(Math.Abs(strideSrc), Math.Abs(strideDst));
    int vectorWidth = Vector<byte>.Count;
    int maxX = strideCommon - vectorWidth;
    byte* pRow = pSrc;
    byte* qRow = pDst + (height - 1) * (long)strideDst; // Set to last row.
    for (int i = 0; i < height; i++) {
        Vector<byte>* pLast = (Vector<byte>*)(pRow + maxX);
        Vector<byte>* qLast = (Vector<byte>*)(qRow + maxX);
        Vector<byte>* p = (Vector<byte>*)pRow;
        Vector<byte>* q = (Vector<byte>*)qRow;
        for (; ; ) {
            Vector<byte> data;
            // Load.
            data = *p;
            // Store.
            *q = data;
            // Next.
            if (p >= pLast) break;
            ++p;
            ++q;
            if (p > pLast) p = pLast; // The last block is also use vector.
            if (q > qLast) q = qLast;
        }
        pRow += strideSrc;
        qRow -= strideDst;
    }
}

由于只需要利用向量類型來做內存復制，于是可以不用調用 Vector靜態類中的方法。甚至連Load、Store 方法也無需使用，因為對向量類型類型進行指針讀寫操作時，編譯器會自動生成“未對齊加載”、“未對齊存儲”的指令。

2.3 基準測試代碼

隨后為該算法編寫基準測試代碼。

[Benchmark]
public void UseVectors() {
    UseVectorsDo(_sourceBitmapData, _destinationBitmapData, false);
}

[Benchmark]
public void UseVectorsParallel() {
    UseVectorsDo(_sourceBitmapData, _destinationBitmapData, true);
}

public static unsafe void UseVectorsDo(BitmapData src, BitmapData dst, bool useParallel = false) {
    int vectorWidth = Vector<byte>.Count;
    int width = src.Width;
    int height = src.Height;
    if (width <= vectorWidth) {
        ScalarDo(src, dst, useParallel);
        return;
    }
    int strideSrc = src.Stride;
    int strideDst = dst.Stride;
    byte* pSrc = (byte*)src.Scan0.ToPointer();
    byte* pDst = (byte*)dst.Scan0.ToPointer();
    bool allowParallel = useParallel && (height > 16) && (Environment.ProcessorCount > 1);
    if (allowParallel) {
        Parallel.For(0, height, i => {
            int start = i;
            int len = 1;
            byte* pSrc2 = pSrc + start * (long)strideSrc;
            byte* pDst2 = pDst + (height - 1 - start) * (long)strideDst;
            UseVectorsDoBatch(pSrc2, strideSrc, width, len, pDst2, strideDst);
        });
    } else {
        UseVectorsDoBatch(pSrc, strideSrc, width, height, pDst, strideDst);
    }
}

也是需要細心處理并行時的地址計算。

三、使用系統所提供的MemoryCopy方法進行復制

內循環只是做復制一整行數據的操作，恰好 .NET 系統提供了內存復制的方法“Buffer.MemoryCopy”。于是，利用它也可以完成任務。源代碼如下。

public static unsafe void UseCopyDoBatch(byte* pSrc, int strideSrc, int width, int height, byte* pDst, int strideDst) {
    int strideCommon = Math.Min(Math.Abs(strideSrc), Math.Abs(strideDst));
    byte* pRow = pSrc;
    byte* qRow = pDst + (height - 1) * (long)strideDst; // Set to last row.
    for (int i = 0; i < height; i++) {
        Buffer.MemoryCopy(pRow, qRow, strideCommon, strideCommon);
        pRow += strideSrc;
        qRow -= strideDst;
    }
}

代碼變得更簡潔了。

四、對算法進行檢查

可按照上一篇的辦法，使用SumDifference函數來對各種算法進行檢查。
且由于現在是做圖像的垂直翻轉運算，翻轉2次后便能變為原樣。于是，現在還可以對 Scalar 算法進行檢查。源代碼如下。

bool allowCheck = true;
if (allowCheck) {
    try {
        TextWriter writer = Console.Out;
        long totalDifference, countByteDifference;
        int maxDifference;
        double averageDifference;
        long totalByte = Width * Height * 3;
        double percentDifference;
        // Baseline
        ScalarDo(_sourceBitmapData, _expectedBitmapData);
        // Scalar
        ScalarDo(_expectedBitmapData, _destinationBitmapData);
        totalDifference = SumDifference(_sourceBitmapData, _destinationBitmapData, out countByteDifference, out maxDifference);
        averageDifference = (countByteDifference > 0) ? (double)totalDifference / countByteDifference : 0;
        percentDifference = 100.0 * countByteDifference / totalByte;
        writer.WriteLine(string.Format("Difference of Scalar: {0}/{1}={2}, max={3}, percentDifference={4:0.000000}%", totalDifference, countByteDifference, averageDifference, maxDifference, percentDifference));
        // ScalarParallel
        ScalarParallel();
        totalDifference = SumDifference(_expectedBitmapData, _destinationBitmapData, out countByteDifference, out maxDifference);
        averageDifference = (countByteDifference > 0) ? (double)totalDifference / countByteDifference : 0;
        percentDifference = 100.0 * countByteDifference / totalByte;
        writer.WriteLine(string.Format("Difference of ScalarParallel: {0}/{1}={2}, max={3}, percentDifference={4:0.000000}%", totalDifference, countByteDifference, averageDifference, maxDifference, percentDifference));
        // UseVectors
        UseVectors();
        totalDifference = SumDifference(_expectedBitmapData, _destinationBitmapData, out countByteDifference, out maxDifference);
        averageDifference = (countByteDifference > 0) ? (double)totalDifference / countByteDifference : 0;
        percentDifference = 100.0 * countByteDifference / totalByte;
        writer.WriteLine(string.Format("Difference of UseVectors: {0}/{1}={2}, max={3}, percentDifference={4:0.000000}%", totalDifference, countByteDifference, averageDifference, maxDifference, percentDifference));
        // UseVectorsParallel
        UseVectorsParallel();
        totalDifference = SumDifference(_expectedBitmapData, _destinationBitmapData, out countByteDifference, out maxDifference);
        averageDifference = (countByteDifference > 0) ? (double)totalDifference / countByteDifference : 0;
        percentDifference = 100.0 * countByteDifference / totalByte;
        writer.WriteLine(string.Format("Difference of UseVectorsParallel: {0}/{1}={2}, max={3}, percentDifference={4:0.000000}%", totalDifference, countByteDifference, averageDifference, maxDifference, percentDifference));
        // UseCopy
        UseCopy();
        totalDifference = SumDifference(_expectedBitmapData, _destinationBitmapData, out countByteDifference, out maxDifference);
        averageDifference = (countByteDifference > 0) ? (double)totalDifference / countByteDifference : 0;
        percentDifference = 100.0 * countByteDifference / totalByte;
        writer.WriteLine(string.Format("Difference of UseCopy: {0}/{1}={2}, max={3}, percentDifference={4:0.000000}%", totalDifference, countByteDifference, averageDifference, maxDifference, percentDifference));
        // UseCopyParallel
        UseCopyParallel();
        totalDifference = SumDifference(_expectedBitmapData, _destinationBitmapData, out countByteDifference, out maxDifference);
        averageDifference = (countByteDifference > 0) ? (double)totalDifference / countByteDifference : 0;
        percentDifference = 100.0 * countByteDifference / totalByte;
        writer.WriteLine(string.Format("Difference of UseCopyParallel: {0}/{1}={2}, max={3}, percentDifference={4:0.000000}%", totalDifference, countByteDifference, averageDifference, maxDifference, percentDifference));
    } catch (Exception ex) {
        Debug.WriteLine(ex.ToString());
    }
}

五、基準測試結果

5.1 X86 架構

X86架構下的基準測試結果如下。

BenchmarkDotNet v0.14.0, Windows 11 (10.0.22631.4541/23H2/2023Update/SunValley3)
AMD Ryzen 7 7840H w/ Radeon 780M Graphics, 1 CPU, 16 logical and 8 physical cores
.NET SDK 8.0.403
  [Host]     : .NET 8.0.10 (8.0.1024.46610), X64 RyuJIT AVX-512F+CD+BW+DQ+VL+VBMI
  DefaultJob : .NET 8.0.10 (8.0.1024.46610), X64 RyuJIT AVX-512F+CD+BW+DQ+VL+VBMI


| Method             | Width | Mean         | Error      | StdDev     | Ratio | RatioSD |
|------------------- |------ |-------------:|-----------:|-----------:|------:|--------:|
| Scalar             | 1024  |  1,077.72 us |  20.704 us |  24.647 us |  1.00 |    0.03 |
| ScalarParallel     | 1024  |    177.58 us |   3.489 us |   3.263 us |  0.16 |    0.00 |
| UseVectors         | 1024  |     79.40 us |   1.549 us |   2.713 us |  0.07 |    0.00 |
| UseVectorsParallel | 1024  |     19.54 us |   0.373 us |   0.547 us |  0.02 |    0.00 |
| UseCopy            | 1024  |     81.88 us |   1.608 us |   2.034 us |  0.08 |    0.00 |
| UseCopyParallel    | 1024  |     18.28 us |   0.357 us |   0.351 us |  0.02 |    0.00 |
|                    |       |              |            |            |       |         |
| Scalar             | 2048  |  4,360.82 us |  52.264 us |  48.888 us |  1.00 |    0.02 |
| ScalarParallel     | 2048  |    717.40 us |  13.745 us |  13.499 us |  0.16 |    0.00 |
| UseVectors         | 2048  |    992.42 us |  19.805 us |  57.457 us |  0.23 |    0.01 |
| UseVectorsParallel | 2048  |    409.04 us |   8.070 us |  19.022 us |  0.09 |    0.00 |
| UseCopy            | 2048  |  1,002.18 us |  19.600 us |  27.476 us |  0.23 |    0.01 |
| UseCopyParallel    | 2048  |    418.30 us |   6.980 us |   5.449 us |  0.10 |    0.00 |
|                    |       |              |            |            |       |         |
| Scalar             | 4096  | 16,913.07 us | 244.574 us | 216.808 us |  1.00 |    0.02 |
| ScalarParallel     | 4096  |  3,844.09 us |  46.626 us |  43.614 us |  0.23 |    0.00 |
| UseVectors         | 4096  |  4,419.30 us |  84.049 us |  78.620 us |  0.26 |    0.01 |
| UseVectorsParallel | 4096  |  4,000.12 us |  44.611 us |  39.546 us |  0.24 |    0.00 |
| UseCopy            | 4096  |  4,608.49 us |  33.594 us |  31.424 us |  0.27 |    0.00 |
| UseCopyParallel    | 4096  |  3,960.86 us |  47.334 us |  44.276 us |  0.23 |    0.00 |

Scalar: 標量算法。
ScalarParallel: 并發的標量算法。
UseVectors: 向量算法。
UseVectorsParallel: 并發的向量算法。
UseCopy: 使用“Buffer.MemoryCopy”的算法。
UseCopyParallel: 并發的使用“Buffer.MemoryCopy”的算法。

5.2 Arm 架構

同樣的源代碼可以在 Arm 架構上運行?；鶞蕼y試結果如下。

BenchmarkDotNet v0.14.0, macOS Sequoia 15.0.1 (24A348) [Darwin 24.0.0]
Apple M2, 1 CPU, 8 logical and 8 physical cores
.NET SDK 8.0.204
  [Host]     : .NET 8.0.4 (8.0.424.16909), Arm64 RyuJIT AdvSIMD [AttachedDebugger]
  DefaultJob : .NET 8.0.4 (8.0.424.16909), Arm64 RyuJIT AdvSIMD


| Method             | Width | Mean         | Error      | StdDev    | Ratio | RatioSD |
|------------------- |------ |-------------:|-----------:|----------:|------:|--------:|
| Scalar             | 1024  |  1,227.25 us |   0.694 us |  0.649 us |  1.00 |    0.00 |
| ScalarParallel     | 1024  |    261.38 us |   0.739 us |  0.617 us |  0.21 |    0.00 |
| UseVectors         | 1024  |    117.96 us |   0.105 us |  0.098 us |  0.10 |    0.00 |
| UseVectorsParallel | 1024  |     39.46 us |   0.297 us |  0.263 us |  0.03 |    0.00 |
| UseCopy            | 1024  |     92.95 us |   0.081 us |  0.063 us |  0.08 |    0.00 |
| UseCopyParallel    | 1024  |     34.90 us |   0.170 us |  0.159 us |  0.03 |    0.00 |
|                    |       |              |            |           |       |         |
| Scalar             | 2048  |  5,236.47 us |  69.941 us | 62.001 us |  1.00 |    0.02 |
| ScalarParallel     | 2048  |    952.35 us |   3.270 us |  3.059 us |  0.18 |    0.00 |
| UseVectors         | 2048  |    700.91 us |   4.339 us |  4.058 us |  0.13 |    0.00 |
| UseVectorsParallel | 2048  |    254.35 us |   1.183 us |  1.107 us |  0.05 |    0.00 |
| UseCopy            | 2048  |    757.75 us |  14.775 us | 25.485 us |  0.14 |    0.01 |
| UseCopyParallel    | 2048  |    252.87 us |   1.078 us |  1.009 us |  0.05 |    0.00 |
|                    |       |              |            |           |       |         |
| Scalar             | 4096  | 20,257.16 us | 100.815 us | 84.185 us |  1.00 |    0.01 |
| ScalarParallel     | 4096  |  3,728.60 us |  12.672 us | 11.233 us |  0.18 |    0.00 |
| UseVectors         | 4096  |  2,788.68 us |   2.712 us |  2.404 us |  0.14 |    0.00 |
| UseVectorsParallel | 4096  |  1,776.71 us |   1.510 us |  1.412 us |  0.09 |    0.00 |
| UseCopy            | 4096  |  2,448.65 us |   4.232 us |  3.959 us |  0.12 |    0.00 |
| UseCopyParallel    | 4096  |  1,796.17 us |   5.197 us |  4.861 us |  0.09 |    0.00 |

5.3 .NET Framework

同樣的源代碼可以在 .NET Framework 上運行?；鶞蕼y試結果如下。

BenchmarkDotNet v0.14.0, Windows 11 (10.0.22631.4541/23H2/2023Update/SunValley3)
AMD Ryzen 7 7840H w/ Radeon 780M Graphics, 1 CPU, 16 logical and 8 physical cores
  [Host]     : .NET Framework 4.8.1 (4.8.9282.0), X64 RyuJIT VectorSize=256
  DefaultJob : .NET Framework 4.8.1 (4.8.9282.0), X64 RyuJIT VectorSize=256


| Method             | Width | Mean         | Error      | StdDev     | Ratio | RatioSD | Code Size |
|------------------- |------ |-------------:|-----------:|-----------:|------:|--------:|----------:|
| Scalar             | 1024  |  1,062.91 us |  14.426 us |  12.788 us |  1.00 |    0.02 |   2,891 B |
| ScalarParallel     | 1024  |    183.82 us |   3.609 us |   4.296 us |  0.17 |    0.00 |   2,894 B |
| UseVectors         | 1024  |     71.65 us |   1.420 us |   1.328 us |  0.07 |    0.00 |   3,602 B |
| UseVectorsParallel | 1024  |     24.67 us |   0.471 us |   0.579 us |  0.02 |    0.00 |   3,605 B |
| UseCopy            | 1024  |     82.86 us |   1.653 us |   2.262 us |  0.08 |    0.00 |   3,280 B |
| UseCopyParallel    | 1024  |     24.16 us |   0.481 us |   0.659 us |  0.02 |    0.00 |   3,283 B |
|                    |       |              |            |            |       |         |           |
| Scalar             | 2048  |  4,344.08 us |  68.246 us |  60.498 us |  1.00 |    0.02 |   2,891 B |
| ScalarParallel     | 2048  |    681.94 us |  12.532 us |  11.722 us |  0.16 |    0.00 |   2,894 B |
| UseVectors         | 2048  |    981.58 us |  14.816 us |  13.134 us |  0.23 |    0.00 |   3,602 B |
| UseVectorsParallel | 2048  |    429.28 us |   8.360 us |  16.106 us |  0.10 |    0.00 |   3,605 B |
| UseCopy            | 2048  |    978.79 us |  15.720 us |  13.127 us |  0.23 |    0.00 |   3,280 B |
| UseCopyParallel    | 2048  |    438.06 us |   8.691 us |  15.672 us |  0.10 |    0.00 |   3,283 B |
|                    |       |              |            |            |       |         |           |
| Scalar             | 4096  | 17,306.43 us | 343.417 us | 352.664 us |  1.00 |    0.03 |   2,891 B |
| ScalarParallel     | 4096  |  3,717.65 us |  18.424 us |  17.233 us |  0.21 |    0.00 |   2,894 B |
| UseVectors         | 4096  |  4,451.39 us |  84.848 us |  87.132 us |  0.26 |    0.01 |   3,602 B |
| UseVectorsParallel | 4096  |  3,818.66 us |  24.223 us |  22.658 us |  0.22 |    0.00 |   3,605 B |
| UseCopy            | 4096  |  4,721.90 us |  88.960 us |  83.214 us |  0.27 |    0.01 |   3,280 B |
| UseCopyParallel    | 4096  |  3,820.63 us |  19.312 us |  18.065 us |  0.22 |    0.00 |   3,283 B |

六、并行處理收益極少的原因分析

6.1 觀察測試結果，產生疑問

觀察測試結果，首先能發現這2點情況——

UseVectors與UseCopy的測試結果非常接近，且 UseVectorsParallel的測試結果非常接近。這是因為“Buffer.MemoryCopy”內部也使用了向量算法來進行優化。
對于向量算法（UseVectors、UseCopy等）來說，.NET Framework 與 .NET 8.0 的測試結果非常接近。這是因為垂直翻轉僅需要做復制操作，這是非常簡單的操作。于是早在 .NET Framework 時代，“Buffer.MemoryCopy”已做好了向量化優化。

隨后會發現一個奇怪的地方——并行版算法（Parallel）并沒有比單線程算法快多少。尤其是對于4096寬度的圖片，16個線程并行處理時，仍需花費 90%左右的時間，并行處理的收益極少。

這很奇怪，因為垂直翻轉是可以分解為各行去處理的，非常適合并行化。且僅需做簡單的內存復制操作，沒有復雜的算術運算。16個線程并行處理時，理論上僅需要 1/16 的時間。但測試結果卻是 ——仍需花費 90%左右的時間。

再來看看一下Arm架構的測試結果，會發現也存在并行處理收益極少的現象。但是要稍微好一點，例如。

UseCopy：2,448.65 us
UseCopyParallel：1,796.17 us

1796.17 / 2448.65 ≈ 0.7335。即只需 73.35 % 的時間，但這也比理論值差了不少。

6.2 數據分析

在Width為 4096時，圖像高度也是 4096，且使用的是 24位圖像。那么一個圖像的總字節數是 4096*4096*3, 即 48MB 左右。

根據UseCopyParallel的結果，來計算一下每秒的處理速率——

X86: 48 / 3.96086 ≈ 12.1185 (GB/s)
Arm: 48 / 1.79617 ≈ 26.7235 (GB/s)

從上面的數據來看，Arm的吞吐率是相對于 X86的倍數是：26.7235/12.1185 ≈ 2.2052。

6.3 找到原因

我這臺Arm計算機，CPU是“Appla M2”。查了資料，發現M2具有“100GB/s統一內存帶寬”。

而我的X86電腦，用的是DDR5-5600內存條。DDR每次傳輸64位數據，故內存帶寬為 5600 * 64 / 8 = 44800 (MB/s)，即 44.8 GB/s。

我們再來計算一下倍數：100 / 44.8 ≈ 2.2321。

“2.2321”與“2.2052”非常接近。這表示現在遇到的是——內存帶寬的瓶頸。

這表示在4096時，我們的算法已經基本占滿了內存帶寬。此時雖然CPU有10多個線程，理論上很多算力還沒發揮出來，但由于內存帶寬已占滿了，故止步于這個值。

由于是“一讀一寫”，程序需要2次訪問內存——

X86: 12.1185*2 = 24.237 (GB/s) （理論上限是 44.8）
Arm: 26.7235*2 = 53.447 (GB/s) （理論上限是 100）

可以看出，程序已經達到內存帶寬理論上限的60%左右了。即我們的垂直翻轉算法，已經是足夠優秀了。

6.4 進一步優化的空間

其實程序是還可以進一步優化的，例如使用地址對齊、預取、非時間性（NonTemporal）寫入、循環展開等手段。但這些手段比較復雜，編碼難度大，更會使程序的可讀性大為降低。而且這些手段是特定硬件型號敏感的，例如更換了CPU型號后，很多細節得重新調校。

更關鍵的是，進一步優化的空間已經非常少了，離理論上限只有40%左右差距。就算達到理論上限，性能也僅提高這 40%左右，沒法翻倍。僅當硬件型號固定，且有充足理由時，此時才可考慮做進一步的優化處理。

這也給了我們一個提醒：在編寫向量化算法時，為了避免遇到內存帶寬瓶頸，程序應盡量少的讀寫內存。例如使用向量類型進行批量讀取，隨后盡量將數據在向量類型里處理好，最后使用向量類型進行批量寫入。

因為在使用向量類型時，是在CPU內的向量寄存器（Register）中處理的。寄存器與CPU同頻工作，吞吐率比內存高了幾個數量級。寄存器不夠用時，一般會挪到Cache（高速緩存）上進行周轉，Cache的速度也比內存快很多。

附錄

完整源代碼: https://github.com/zyl910/VectorTraits.Sample.Benchmarks/blob/main/VectorTraits.Sample.Benchmarks.Inc/Image/ImageFlipYBenchmark.cs
VectorTraits 的NuGet包: https://www.nuget.org/packages/VectorTraits
VectorTraits 的在線文檔: https://zyl910.github.io/VectorTraits_doc/
VectorTraits 源代碼: https://github.com/zyl910/VectorTraits

posted on 2024-11-30 03:08 zyl910 閱讀(162) 評論(0) 收藏舉報

刷新頁面返回頂部