本文技術方案支持.Net/.Net Core/.Net Framework 數據分頁,幾乎是任何應用系統的必備功能。但當數據量較大時,分頁操作的效率就會變得很低。大數據量分頁時,一個操作耗時5秒、10秒、甚至更長時間都是有可能的,但這在用戶使用的角度是不可接受的…… 數據分頁往往有三種常用方案。 第 ...
作者:
目錄目錄
一、緣由
在 上一篇文章,介紹了.NET的2種向量類型(Vector4、Vector<T>
)。本文來介紹第3種。
.NET Core 3.0 增加了對內在函數(Intrinsics Functions)的支持,並增加了第3種向量類型——
3. 總位寬固定的向量(Vector of fixed total bit width)。例如 只讀結構體 Vector64<T>
、Vector128<T>
、Vector256<T>
,及輔助的靜態類 Vector64、Vector128、Vector256。這些向量類型沒有Nuget包,只能在 .NET Core 3.0或更高版本的.NET環境中運行。
這些向量類型比較特殊,沒有直接提供數學運算的函數(到了.NET7,才增加了少量數學函數),而是需要通過內在函數來進行數學運算。
內在函數就是CPU的特殊指令集,其中有向量運算相關的。例如x86體系的向量指令集,有 SSE(Streaming SIMD Extensions,流式SIMD擴展)、AVX(Advanced Vector Extensions,高級矢量擴展)等;且 Arm體系的向量指令集,有 NEON(學名為“Advanced single instruction multiple data”,縮寫為“AdvSIMD”)等。
上一篇文章中,發現有硬體加速時,Vector<byte>.Count
的值為32,換算後是256位,表示它使用了AVX2指令集。可見現在絕大多數PC機的CPU,已支持了AVX2指令集。
由於本文是測試浮點求和,用AVX指令集就夠了,於是便演示了C#下如何使用AVX指令集來操作Vector256<T>
。且還編寫了C++程式,來做對比。
二、在C#中使用
2.1 文檔查看心得
與這種向量類型相關的,主要是這3個名稱空間——
- System.Runtime.Intrinsics:用於提供各種位寬的向量類型,如 只讀結構體
Vector64<T>
、Vector128<T>
、Vector256<T>
,及輔助的靜態類 Vector64、Vector128、Vector256。官方文檔說明:包含用於創建和傳遞各種大小和格式的寄存器狀態的類型,用於指令集擴展。有關操作這些寄存器的說明,請參閱 System.Runtime.Intrinsics.X86 和 System.Runtime.Intrinsics.Arm。 - System.Runtime.Intrinsics.X86:用於提供x86體系的內在函數類,如Avx等。官方文檔說明:公開 x86 和 x64 系統的 select 指令集擴展。 對於每個擴展,這些指令集表示為單獨的類。 可以通過查詢相應類型上的 IsSupported 屬性來確定是否支持當前環境中的任何擴展。
- System.Runtime.Intrinsics.Arm:用於提供Arm體系的內在函數類,如AdvSimd等。官方文檔說明:公開 ARM 系統的 select 指令集擴展。 對於每個擴展,這些指令集表示為單獨的類。 可以通過查詢相應類型上的 IsSupported 屬性來確定是否支持當前環境中的任何擴展。
簡單來說,“System.Runtime.Intrinsics”用於定義通用的向量類型,隨後它的各種子命名空間,以CPU體系來命名。子命名空間里,包含各個內在函數類,每個類對應一套指令集。類中的各個靜態方法就是內在函數,對應指令集內的各條指令。
對於每一個內在函數類,都提供靜態屬性 IsSupported,用於檢查當前運行環境是否支持該指令集。例如“Avx.IsSupported”,是用於檢測是否支持AVX指令集。
觀察子命名空間里的內在函數類,發現有些類的尾碼是“64”(如Avx.X64,及Arm里的AdvSimd.Arm64),這些是64位模式下特有的指令集,它們的指令一般比較少。平時應儘量使用尾碼不是“64”的類,因為這些它們是 32位或64位 環境都能工作的類。
由於本文是測試用AVX指令集做浮點求和,便去官方文檔的“Avx類”里找相關的靜態方法。會發現官方充分的利用了.NET平臺支持方法名重載(overload)的特征,方法名簡潔、易懂,不再像C語言版的內在函數那樣有很多奇怪縮寫規則。
很快就能能找到求和相關的靜態方法“Add”,且它利用了重載,有 Single、Double 這2種簽名的函數:
Add(Vector256<Double>, Vector256<Double>)
Add(Vector256<Single>, Vector256<Single>)
本文只需要處理單精度浮點數,於是只需使用後者。點擊它的鏈接,查看該方法的詳細文檔。內容如下。
__m256 _mm256_add_ps (__m256 a, __m256 b)
VADDPS ymm, ymm, ymm/m256
C#
public static System.Runtime.Intrinsics.Vector256<float> Add (System.Runtime.Intrinsics.Vector256<float> left, System.Runtime.Intrinsics.Vector256<float> right);
參數
left Vector256<Single>
right Vector256<Single>
返回 Vector256<Single>
此時發現內在函數的文檔說明,不如平常的.NET方法的文檔詳細,例如 參數、返回值 沒有說明,且方法簡介里是 2行奇怪的文字。
其實不用怕,內在函數的文檔說明雖然簡單,但其實關鍵內容已經說了,就是方法簡介里的“2行奇怪的文字”——
- 第1行是 對應C語言版的內在函數的申明。如
__m256 _mm256_add_ps (__m256 a, __m256 b)
,“_mm256_add_ps”是函數名. __m256是256位的向量類型,對應C#的Vector256<T>
. - 第2行是 對應CPU指令的申明。如
VADDPS ymm, ymm, ymm/m256
,“VADDPS”是指令名. ymm是256位寄存器,m256是“256位數據的記憶體地址”,對應C#的Vector256<T>
.
由於SIMD是同時處理多個數據,傳統的“參數、返回值”不容易將處理細節說清楚。於是.NET文檔里乾脆不說,而是提供了內在函數與指令的申明,讓使用者去查CPU廠商的文檔,因為CPU廠商的文檔很詳盡。
例如AVX是Intel提出來的指令集,故可去Intel的文檔。Intel已提供了便於線上查詢的文檔《Intel® Intrinsics Guide》, 地址是“https://www.intel.com/content/www/us/en/docs/intrinsics-guide/index.html”.
在瀏覽器打開該地址,然後在屏幕中間的查詢框里輸入“C語言版的內在函數名”,如“_mm256_add_ps”,查詢框的下麵便會列出查詢結果,隨後點擊想看的函數就行。由於1條指令對應多個內在函數,若選擇用指令名來查詢的話,會出現大量匹配,難挑選,故一般使用內在函數名來查詢。
摘錄一下Intel文檔對“_mm256_add_ps”的說明——
Synopsis
__m256 _mm256_add_ps (__m256 a, __m256 b)
#include <immintrin.h>
Instruction: vaddps ymm, ymm, ymm
CPUID Flags: AVX
Description
Add packed single-precision (32-bit) floating-point elements in a and b, and store the results in dst.
Operation
FOR j := 0 to 7
i := j*32
dst[i+31:i] := a[i+31:i] + b[i+31:i]
ENDFOR
dst[MAX:256] := 0
Latency and Throughput
Architecture Latency Throughput (CPI)
Alderlake 2 0.5
Icelake Intel Core 4 0.5
Icelake Xeon 4 0.5
Skylake 4 0.5
可見它除了常規的說明信息外,還提供了“Operation”(偽代碼)、“Latency and Throughput”(延遲和吞吐量)。其中對我們最有用的是 “Operation”(偽代碼),能清晰的瞭解該內在函數的操作細節。
例如對於“_mm256_add_ps”,就是將256位數據,分為8組 32位浮點數,分別進行加法運算。高於256位的內容會置零,這個是跟AVX-512有關的,本文不用理會它。
2.2 搭建測試項目(BenchmarkVectorCore30)及處理準備工作
首先需等搭建測試項目。由於.NET Core 3.0才支持這種向量類型,於是得使用VS2019來打開解決方案文件(BenchmarkVector.sln)。
然後建立新項目“BenchmarkVectorCore30”,它是 .NET Core 3.0 控制台程式的項目。並讓“BenchmarkVectorCore30”引用共用項目“BenchmarkVector”。
新增的測試函數,也準備放在BenchmarkVectorDemo類里。此時需考慮讓 BenchmarkVectorDemo類相容之前的運行環境(.NET Core 2.0、.NET Framework 4.5 等),於是可以利用條件編譯來處理。
由於需要在多個地方進行條件編譯判斷,故專門定義一個“Allow_Intrinsics”(允許內在函數)的條件編譯符號比較好。於是修改了“BenchmarkVectorDemo.cs”的頂部內容,摘錄如下。
#if NETCOREAPP3_0_OR_GREATER
#define Allow_Intrinsics
#endif
using System;
using System.Collections.Generic;
using System.IO;
using System.Numerics;
using System.Reflection;
using System.Text;
using System.Runtime.InteropServices;
#if Allow_Intrinsics
using System.Runtime.Intrinsics;
using System.Runtime.Intrinsics.X86;
#endif
using System.Runtime.CompilerServices;
namespace BenchmarkVector {
/// <summary>
/// Benchmark Vector Demo
/// </summary>
static class BenchmarkVectorDemo {
說明——
- 用“NETCOREAPP3_0_OR_GREATER”進行條件編譯檢查,檢查通過時定義“Allow_Intrinsics”條件編譯符號。因為本文測試的是.NET Core 3.0新增功能,於是使用 .NET Core 時代新增的條件編譯符號“NETCOREAPP3_0_OR_GREATER”就行了,不用使用“.NET Framework相容的條件編譯寫法”(因為 .NET Framework不支持“NETCOREAPP3_0_OR_GREATER”等內置符號,恰好它也不支持內在函數,故不會有“Allow_Intrinsics”符號,正好滿足了本文的條件編譯需求)。
- 在支持內在函數(Allow_Intrinsics)時,使用using指令導入“System.Runtime.Intrinsics”、“System.Runtime.Intrinsics.X86”這2個命名空間。
2.3 編寫基於AVX的浮點數組求和函數(SumVectorAvx)
Vector256<T>
很像Vector<T>
,也提供了Count屬性,能獲得元素個數。故可按Count分組分別進行求和(即Map階段),最後再將這些組的結果加起來(即Reduce階段)。
參考SumVectorT的經驗,我們可以寫出SumVectorAvx。代碼如下。
private static float SumVectorAvx(float[] src, int count, int loops) {
#if Allow_Intrinsics
float rt = 0; // Result.
//int VectorWidth = 32 / 4; // sizeof(__m256) / sizeof(float);
int VectorWidth = Vector256<float>.Count; // Block width.
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector256<float> vrt = Vector256<float>.Zero; // Vector result.
int p; // Index for src data.
int i;
// Load.
Vector256<float>[] vsrc = new Vector256<float>[cntBlock]; // Vector src.
p = 0;
for (i = 0; i < cntBlock; ++i) {
vsrc[i] = Vector256.Create(src[p], src[p + 1], src[p + 2], src[p + 3], src[p + 4], src[p + 5], src[p + 6], src[p + 7]); // Load.
p += VectorWidth;
}
// Body.
for (int j = 0; j < loops; ++j) {
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
vrt = Avx.Add(vrt, vsrc[i]); // Add. vrt += vsrc[i];
}
// Remainder processs.
p = cntBlock * nBlockWidth;
for (i = 0; i < cntRem; ++i) {
rt += src[p + i];
}
}
// Reduce.
for (i = 0; i < VectorWidth; ++i) {
rt += vrt.GetElement(i);
}
return rt;
#else
throw new NotSupportedException();
#endif
}
對比 SumVectorT,除了將 Vector<T>
類型換為 Vector256<T>
,因.NET Core 3.0的限制,還有這些變化——
Vector256<T>
未提供構造函數,且Vector256.Create
不支持數組參數(.NET 7才支持數組參數、Span參數),故只能使用最笨的逐個元素傳遞的辦法。Vector256<T>
不支持運算符重載(.NET 7才支持),需改為使用“Avx.Add”。Vector256<T>
不支持索引器(.NET 7才支持),需改為擴展方法 GetElement 來獲取每個元素的值。
2.4 使用Span改進數據載入(SumVectorAvxSpan)
剛纔的SumVectorAvx有個缺點,每次需要“將float[]轉為Vector256
在C/C++里,對於值類型的指針,是支持做 reinterpret_cast(重新解釋數據類型) 類型轉換的,這樣就能避免對數據做類型轉換的開銷。但是在C#里,只能在“非安全代碼”里使用指針與reinterpret_cast,但“非安全代碼”一般是儘量少用。
.NET Core 2.1 支持 Span(切片),可以用Span來實現 reinterpret_cast,便解決了這一難題。具體辦法是使用 “MemoryMarshal.Cast”來做 reinterpret_cast。
代碼如下。
private static float SumVectorAvxSpan(float[] src, int count, int loops) {
#if Allow_Intrinsics
float rt = 0; // Result.
int VectorWidth = Vector256<float>.Count; // Block width.
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector256<float> vrt = Vector256<float>.Zero; // Vector result.
int p; // Index for src data.
ReadOnlySpan<Vector256<float>> vsrc; // Vector src.
int i;
// Body.
for (int j = 0; j < loops; ++j) {
// Vector processs.
vsrc = System.Runtime.InteropServices.MemoryMarshal.Cast<float, Vector256<float> >(new Span<float>(src)); // Reinterpret cast. `float*` to `Vector256<float>*`.
for (i = 0; i < cntBlock; ++i) {
vrt = Avx.Add(vrt, vsrc[i]); // Add. vrt += vsrc[i];
}
// Remainder processs.
p = cntBlock * nBlockWidth;
for (i = 0; i < cntRem; ++i) {
rt += src[p + i];
}
}
// Reduce.
for (i = 0; i < VectorWidth; ++i) {
rt += vrt.GetElement(i);
}
return rt;
#else
throw new NotSupportedException();
#endif
}
2.5 使用指針改進數據載入(SumVectorAvxPtr)
查看一下Avx類,發現它提供了載入方法:
LoadAlignedVector256(Single*)
:__m256 _mm256_load_ps (float const * mem_addr)
。從已對齊的地址載入。LoadVector256(Single*)
:__m256 _mm256_loadu_ps (float const * mem_addr)
。從未對齊的地址載入。由於.NET中應由.NET自動管理記憶體地址,故一般情況下應使用它,保險一點。
但這些載入方法都是用指針參數的。故我們需要啟用“非安全代碼”,才能編寫使用了指針的函數。修改項目屬性,切換到“Build”頁面,Configuration 下拉框選擇“All Configurations”,然後勾選“Allow unsafe code”(允許非安全代碼),保存,這便允許了“非安全代碼”。
隨後使用fixed語句可以得到數組起始數據的指針,並可用指針地址計算,來代替數組索引計算。代碼如下。
private static float SumVectorAvxPtr(float[] src, int count, int loops) {
#if Allow_Intrinsics && UNSAFE
unsafe {
float rt = 0; // Result.
int VectorWidth = Vector256<float>.Count; // Block width.
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector256<float> vrt = Vector256<float>.Zero; // Vector result.
Vector256<float> vload;
float* p; // Pointer for src data.
int i;
// Body.
fixed(float* p0 = &src[0]) {
for (int j = 0; j < loops; ++j) {
p = p0;
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
vload = Avx.LoadVector256(p); // Load. vload = *(*__m256)p;
vrt = Avx.Add(vrt, vload); // Add. vrt += vsrc[i];
p += nBlockWidth;
}
// Remainder processs.
for (i = 0; i < cntRem; ++i) {
rt += p[i];
}
}
}
// Reduce.
for (i = 0; i < VectorWidth; ++i) {
rt += vrt.GetElement(i);
}
return rt;
}
#else
throw new NotSupportedException();
#endif
}
2.6 完整的BenchmarkVector類
在測試方法(Benchmark)里,增加這些函數的測試。
此時,完整的BenchmarkVector類的代碼如下。
#if NETCOREAPP3_0_OR_GREATER
#define Allow_Intrinsics
#endif
using System;
using System.Collections.Generic;
using System.IO;
using System.Numerics;
using System.Reflection;
using System.Text;
using System.Runtime.InteropServices;
#if Allow_Intrinsics
using System.Runtime.Intrinsics;
using System.Runtime.Intrinsics.X86;
#endif
using System.Runtime.CompilerServices;
namespace BenchmarkVector {
/// <summary>
/// Benchmark Vector Demo
/// </summary>
static class BenchmarkVectorDemo {
/// <summary>
/// Is release make.
/// </summary>
public static readonly bool IsRelease =
#if DEBUG
false
#else
true
#endif
;
/// <summary>
/// Output Environment.
/// </summary>
/// <param name="tw">Output <see cref="TextWriter"/>.</param>
/// <param name="indent">The indent.</param>
public static void OutputEnvironment(TextWriter tw, string indent) {
if (null == tw) return;
if (null == indent) indent="";
//string indentNext = indent + "\t";
tw.WriteLine(indent + string.Format("IsRelease:\t{0}", IsRelease));
tw.WriteLine(indent + string.Format("EnvironmentVariable(PROCESSOR_IDENTIFIER):\t{0}", Environment.GetEnvironmentVariable("PROCESSOR_IDENTIFIER")));
tw.WriteLine(indent + string.Format("Environment.ProcessorCount:\t{0}", Environment.ProcessorCount));
tw.WriteLine(indent + string.Format("Environment.Is64BitOperatingSystem:\t{0}", Environment.Is64BitOperatingSystem));
tw.WriteLine(indent + string.Format("Environment.Is64BitProcess:\t{0}", Environment.Is64BitProcess));
tw.WriteLine(indent + string.Format("Environment.OSVersion:\t{0}", Environment.OSVersion));
tw.WriteLine(indent + string.Format("Environment.Version:\t{0}", Environment.Version));
//tw.WriteLine(indent + string.Format("RuntimeEnvironment.GetSystemVersion:\t{0}", System.Runtime.InteropServices.RuntimeEnvironment.GetSystemVersion())); // Same Environment.Version
tw.WriteLine(indent + string.Format("RuntimeEnvironment.GetRuntimeDirectory:\t{0}", System.Runtime.InteropServices.RuntimeEnvironment.GetRuntimeDirectory()));
#if (NET47 || NET462 || NET461 || NET46 || NET452 || NET451 || NET45 || NET40 || NET35 || NET20) || (NETSTANDARD1_0)
#else
tw.WriteLine(indent + string.Format("RuntimeInformation.FrameworkDescription:\t{0}", System.Runtime.InteropServices.RuntimeInformation.FrameworkDescription));
#endif
tw.WriteLine(indent + string.Format("BitConverter.IsLittleEndian:\t{0}", BitConverter.IsLittleEndian));
tw.WriteLine(indent + string.Format("IntPtr.Size:\t{0}", IntPtr.Size));
tw.WriteLine(indent + string.Format("Vector.IsHardwareAccelerated:\t{0}", Vector.IsHardwareAccelerated));
tw.WriteLine(indent + string.Format("Vector<byte>.Count:\t{0}\t# {1}bit", Vector<byte>.Count, Vector<byte>.Count * sizeof(byte) * 8));
tw.WriteLine(indent + string.Format("Vector<float>.Count:\t{0}\t# {1}bit", Vector<float>.Count, Vector<float>.Count*sizeof(float)*8));
tw.WriteLine(indent + string.Format("Vector<double>.Count:\t{0}\t# {1}bit", Vector<double>.Count, Vector<double>.Count * sizeof(double) * 8));
Assembly assembly = typeof(Vector4).GetTypeInfo().Assembly;
//tw.WriteLine(string.Format("Vector4.Assembly:\t{0}", assembly));
tw.WriteLine(string.Format("Vector4.Assembly.CodeBase:\t{0}", assembly.CodeBase));
assembly = typeof(Vector<float>).GetTypeInfo().Assembly;
tw.WriteLine(string.Format("Vector<T>.Assembly.CodeBase:\t{0}", assembly.CodeBase));
}
/// <summary>
/// Do Benchmark.
/// </summary>
/// <param name="tw">Output <see cref="TextWriter"/>.</param>
/// <param name="indent">The indent.</param>
public static void Benchmark(TextWriter tw, string indent) {
if (null == tw) return;
if (null == indent) indent = "";
//string indentNext = indent + "\t";
// init.
int tickBegin, msUsed;
double mFlops; // MFLOPS/s .
double scale;
float rt;
const int count = 1024*4;
const int loops = 1000 * 1000;
//const int loops = 1;
const double countMFlops = count * (double)loops / (1000.0 * 1000);
float[] src = new float[count];
for(int i=0; i< count; ++i) {
src[i] = i;
}
tw.WriteLine(indent + string.Format("Benchmark: \tcount={0}, loops={1}, countMFlops={2}", count, loops, countMFlops));
// SumBase.
tickBegin = Environment.TickCount;
rt = SumBase(src, count, loops);
msUsed = Environment.TickCount - tickBegin;
mFlops = countMFlops * 1000 / msUsed;
tw.WriteLine(indent + string.Format("SumBase:\t{0}\t# msUsed={1}, MFLOPS/s={2}", rt, msUsed, mFlops));
double mFlopsBase = mFlops;
// SumVector4.
tickBegin = Environment.TickCount;
rt = SumVector4(src, count, loops);
msUsed = Environment.TickCount - tickBegin;
mFlops = countMFlops * 1000 / msUsed;
scale = mFlops / mFlopsBase;
tw.WriteLine(indent + string.Format("SumVector4:\t{0}\t# msUsed={1}, MFLOPS/s={2}, scale={3}", rt, msUsed, mFlops, scale));
// SumVectorT.
tickBegin = Environment.TickCount;
rt = SumVectorT(src, count, loops);
msUsed = Environment.TickCount - tickBegin;
mFlops = countMFlops * 1000 / msUsed;
scale = mFlops / mFlopsBase;
tw.WriteLine(indent + string.Format("SumVectorT:\t{0}\t# msUsed={1}, MFLOPS/s={2}, scale={3}", rt, msUsed, mFlops, scale));
// SumVectorAvx.
#if Allow_Intrinsics
if (Avx.IsSupported) {
try {
tickBegin = Environment.TickCount;
rt = SumVectorAvx(src, count, loops);
msUsed = Environment.TickCount - tickBegin;
mFlops = countMFlops * 1000 / msUsed;
scale = mFlops / mFlopsBase;
tw.WriteLine(indent + string.Format("SumVectorAvx:\t{0}\t# msUsed={1}, MFLOPS/s={2}, scale={3}", rt, msUsed, mFlops, scale));
// SumVectorAvxSpan.
tickBegin = Environment.TickCount;
rt = SumVectorAvxSpan(src, count, loops);
msUsed = Environment.TickCount - tickBegin;
mFlops = countMFlops * 1000 / msUsed;
scale = mFlops / mFlopsBase;
tw.WriteLine(indent + string.Format("SumVectorAvxSpan:\t{0}\t# msUsed={1}, MFLOPS/s={2}, scale={3}", rt, msUsed, mFlops, scale));
// SumVectorAvxPtr.
tickBegin = Environment.TickCount;
rt = SumVectorAvxPtr(src, count, loops);
msUsed = Environment.TickCount - tickBegin;
mFlops = countMFlops * 1000 / msUsed;
scale = mFlops / mFlopsBase;
tw.WriteLine(indent + string.Format("SumVectorAvxPtr:\t{0}\t# msUsed={1}, MFLOPS/s={2}, scale={3}", rt, msUsed, mFlops, scale));
} catch (Exception ex) {
tw.WriteLine("Run SumVectorAvx fail!");
tw.WriteLine(ex);
}
}
#endif
}
/// <summary>
/// Sum - base.
/// </summary>
/// <param name="src">Soure array.</param>
/// <param name="count">Soure array count.</param>
/// <param name="loops">Benchmark loops.</param>
/// <returns>Return the sum value.</returns>
private static float SumBase(float[] src, int count, int loops) {
float rt = 0; // Result.
for (int j=0; j< loops; ++j) {
for(int i=0; i< count; ++i) {
rt += src[i];
}
}
return rt;
}
/// <summary>
/// Sum - Vector4.
/// </summary>
/// <param name="src">Soure array.</param>
/// <param name="count">Soure array count.</param>
/// <param name="loops">Benchmark loops.</param>
/// <returns>Return the sum value.</returns>
private static float SumVector4(float[] src, int count, int loops) {
float rt = 0; // Result.
const int VectorWidth = 4;
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector4 vrt = Vector4.Zero; // Vector result.
int p; // Index for src data.
int i;
// Load.
Vector4[] vsrc = new Vector4[cntBlock]; // Vector src.
p = 0;
for (i = 0; i < vsrc.Length; ++i) {
vsrc[i] = new Vector4(src[p], src[p + 1], src[p + 2], src[p + 3]);
p += VectorWidth;
}
// Body.
for (int j = 0; j < loops; ++j) {
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
// Equivalent to scalar model: rt += src[i];
vrt += vsrc[i]; // Add.
}
// Remainder processs.
p = cntBlock * nBlockWidth;
for (i = 0; i < cntRem; ++i) {
rt += src[p + i];
}
}
// Reduce.
rt += vrt.X + vrt.Y + vrt.Z + vrt.W;
return rt;
}
/// <summary>
/// Sum - Vector<T>.
/// </summary>
/// <param name="src">Soure array.</param>
/// <param name="count">Soure array count.</param>
/// <param name="loops">Benchmark loops.</param>
/// <returns>Return the sum value.</returns>
private static float SumVectorT(float[] src, int count, int loops) {
float rt = 0; // Result.
int VectorWidth = Vector<float>.Count; // Block width.
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector<float> vrt = Vector<float>.Zero; // Vector result.
int p; // Index for src data.
int i;
// Load.
Vector<float>[] vsrc = new Vector<float>[cntBlock]; // Vector src.
p = 0;
for (i = 0; i < vsrc.Length; ++i) {
vsrc[i] = new Vector<float>(src, p);
p += VectorWidth;
}
// Body.
for (int j = 0; j < loops; ++j) {
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
vrt += vsrc[i]; // Add.
}
// Remainder processs.
p = cntBlock * nBlockWidth;
for (i = 0; i < cntRem; ++i) {
rt += src[p + i];
}
}
// Reduce.
for (i = 0; i < VectorWidth; ++i) {
rt += vrt[i];
}
return rt;
}
/// <summary>
/// Sum - Vector AVX.
/// </summary>
/// <param name="src">Soure array.</param>
/// <param name="count">Soure array count.</param>
/// <param name="loops">Benchmark loops.</param>
/// <returns>Return the sum value.</returns>
private static float SumVectorAvx(float[] src, int count, int loops) {
#if Allow_Intrinsics
float rt = 0; // Result.
//int VectorWidth = 32 / 4; // sizeof(__m256) / sizeof(float);
int VectorWidth = Vector256<float>.Count; // Block width.
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector256<float> vrt = Vector256<float>.Zero; // Vector result.
int p; // Index for src data.
int i;
// Load.
Vector256<float>[] vsrc = new Vector256<float>[cntBlock]; // Vector src.
p = 0;
for (i = 0; i < cntBlock; ++i) {
vsrc[i] = Vector256.Create(src[p], src[p + 1], src[p + 2], src[p + 3], src[p + 4], src[p + 5], src[p + 6], src[p + 7]); // Load.
p += VectorWidth;
}
// Body.
for (int j = 0; j < loops; ++j) {
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
vrt = Avx.Add(vrt, vsrc[i]); // Add. vrt += vsrc[i];
}
// Remainder processs.
p = cntBlock * nBlockWidth;
for (i = 0; i < cntRem; ++i) {
rt += src[p + i];
}
}
// Reduce.
for (i = 0; i < VectorWidth; ++i) {
rt += vrt.GetElement(i);
}
return rt;
#else
throw new NotSupportedException();
#endif
}
/// <summary>
/// Sum - Vector AVX - Span.
/// </summary>
/// <param name="src">Soure array.</param>
/// <param name="count">Soure array count.</param>
/// <param name="loops">Benchmark loops.</param>
/// <returns>Return the sum value.</returns>
private static float SumVectorAvxSpan(float[] src, int count, int loops) {
#if Allow_Intrinsics
float rt = 0; // Result.
int VectorWidth = Vector256<float>.Count; // Block width.
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector256<float> vrt = Vector256<float>.Zero; // Vector result.
int p; // Index for src data.
ReadOnlySpan<Vector256<float>> vsrc; // Vector src.
int i;
// Body.
for (int j = 0; j < loops; ++j) {
// Vector processs.
vsrc = System.Runtime.InteropServices.MemoryMarshal.Cast<float, Vector256<float> >(new Span<float>(src)); // Reinterpret cast. `float*` to `Vector256<float>*`.
for (i = 0; i < cntBlock; ++i) {
vrt = Avx.Add(vrt, vsrc[i]); // Add. vrt += vsrc[i];
}
// Remainder processs.
p = cntBlock * nBlockWidth;
for (i = 0; i < cntRem; ++i) {
rt += src[p + i];
}
}
// Reduce.
for (i = 0; i < VectorWidth; ++i) {
rt += vrt.GetElement(i);
}
return rt;
#else
throw new NotSupportedException();
#endif
}
/// <summary>
/// Sum - Vector AVX - Ptr.
/// </summary>
/// <param name="src">Soure array.</param>
/// <param name="count">Soure array count.</param>
/// <param name="loops">Benchmark loops.</param>
/// <returns>Return the sum value.</returns>
private static float SumVectorAvxPtr(float[] src, int count, int loops) {
#if Allow_Intrinsics && UNSAFE
unsafe {
float rt = 0; // Result.
int VectorWidth = Vector256<float>.Count; // Block width.
int nBlockWidth = VectorWidth; // Block width.
int cntBlock = count / nBlockWidth; // Block count.
int cntRem = count % nBlockWidth; // Remainder count.
Vector256<float> vrt = Vector256<float>.Zero; // Vector result.
Vector256<float> vload;
float* p; // Pointer for src data.
int i;
// Body.
fixed(float* p0 = &src[0]) {
for (int j = 0; j < loops; ++j) {
p = p0;
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
vload = Avx.LoadVector256(p); // Load. vload = *(*__m256)p;
vrt = Avx.Add(vrt, vload); // Add. vrt += vsrc[i];
p += nBlockWidth;
}
// Remainder processs.
for (i = 0; i < cntRem; ++i) {
rt += p[i];
}
}
}
// Reduce.
for (i = 0; i < VectorWidth; ++i) {
rt += vrt.GetElement(i);
}
return rt;
}
#else
throw new NotSupportedException();
#endif
}
}
}
2.7 測試結果
在我的電腦(lntel(R) Core(TM) i5-8250U CPU @ 1.60GHz
、Windows 10)上運行時,x64、Release版程式的輸出信息為:
BenchmarkVectorCore30
IsRelease: True
EnvironmentVariable(PROCESSOR_IDENTIFIER): Intel64 Family 6 Model 142 Stepping 10, GenuineIntel
Environment.ProcessorCount: 8
Environment.Is64BitOperatingSystem: True
Environment.Is64BitProcess: True
Environment.OSVersion: Microsoft Windows NT 6.2.9200.0
Environment.Version: 3.1.26
RuntimeEnvironment.GetRuntimeDirectory: C:\Program Files\dotnet\shared\Microsoft.NETCore.App\3.1.26\
RuntimeInformation.FrameworkDescription: .NET Core 3.1.26
BitConverter.IsLittleEndian: True
IntPtr.Size: 8
Vector.IsHardwareAccelerated: True
Vector<byte>.Count: 32 # 256bit
Vector<float>.Count: 8 # 256bit
Vector<double>.Count: 4 # 256bit
Vector4.Assembly.CodeBase: file:///C:/Program Files/dotnet/shared/Microsoft.NETCore.App/3.1.26/System.Numerics.Vectors.dll
Vector<T>.Assembly.CodeBase: file:///C:/Program Files/dotnet/shared/Microsoft.NETCore.App/3.1.26/System.Private.CoreLib.dll
Benchmark: count=4096, loops=1000000, countMFlops=4096
SumBase: 6.871948E+10 # msUsed=4938, MFLOPS/s=829.485621709194
SumVector4: 2.748779E+11 # msUsed=1218, MFLOPS/s=3362.8899835796387, scale=4.054187192118227
SumVectorT: 5.497558E+11 # msUsed=609, MFLOPS/s=6725.7799671592775, scale=8.108374384236454
SumVectorAvx: 5.497558E+11 # msUsed=609, MFLOPS/s=6725.7799671592775, scale=8.108374384236454
SumVectorAvxSpan: 5.497558E+11 # msUsed=625, MFLOPS/s=6553.6, scale=7.9008
SumVectorAvxPtr: 5.497558E+11 # msUsed=610, MFLOPS/s=6714.754098360656, scale=8.095081967213115
從中可以看出,SumVectorAvx這3個函數的性能,與SumVectorT差不多。這是因為SumVectorT在該電腦上是256bit,表示它內部使用AVX指令集來硬體加速運算,故性能與手工開發AVX的函數差不多。故應儘可能的使用 Vector<T>
,這樣能適應各種硬體,而不是每一種硬體都開發一套。除非是需要使用內在函數時,才使用 Vector256<T>
等類型,但它需要針對不同的硬體平臺分別去開發。
由於預先轉換了數據類型,導致 SumVectorAvx與另外2個函數的性能差不多。但在實際使用時,類型轉換帶來的記憶體分配等開銷很大,應儘量避免。於是當不允許“不安全代碼”時,應該用Span來避免類型轉換;而在允許“不安全代碼”時,可以用指針來編寫。
Span
(切片):優點是無需啟用“不安全代碼”,能實現reinterpret_cast等操作來避免多餘的開銷,自帶數據越界檢查。缺點是部分內在函數不支持Span(內建函數到了.NET7,才全面改善Span的支持),性能比指針稍低(因多了數據越界檢查)。Unsafe code
(不安全代碼):優點是能使用所有內在函數,能充分利用指針的reinterpret_cast等特點來減少多餘的開銷,便於移植 C/C++ 的SIMD代碼,性能比Span高(因沒有數據越界檢查)。缺點是需要啟用“不安全代碼”,且程式員疏忽時會遇到 數據越界 等問題。
三、在C++中使用
3.1 搭建測試項目(BenchmarkVectorCpp)
Visual Studio支持在同一個解決方案文件(*.sln)里建立不同編程語言的項目。
於是用VS2017打開本文的解決方案文件(BenchmarkVector.sln),添加一個C++的“Console App”項目,命名為“BenchmarkVectorCpp”。
隨後建立一個源碼文件“BenchmarkVectorCpp.cpp”。
3.2 基本演算法(SumBase)
基本演算法就是直接寫個迴圈,進行數組求和。
可參考先前C#的SumBase,來編寫它的C++版函數。代碼如下。
// Sum - base.
float SumBase(const float* src, size_t count, int loops) {
float rt = 0; // Result.
size_t i;
for (int j = 0; j < loops; ++j) {
for (i = 0; i < count; ++i) {
rt += src[i];
}
}
return rt;
}
3.3 Avx版演算法(SumVectorAvx)
VC++雖然沒有提供SIMD向量類型,但它很早就支持了AVX的內在函數。引用“immintrin.h”,便可使用AVX的內在函數。
可參考先前C#的SumVectorAvxPtr,來編寫它的C++版函數。代碼如下。
// Sum - Vector AVX.
float SumVectorAvx(const float* src, size_t count, int loops) {
float rt = 0; // Result.
size_t VectorWidth = sizeof(__m256) / sizeof(float); // Block width.
size_t nBlockWidth = VectorWidth; // Block width.
size_t cntBlock = count / nBlockWidth; // Block count.
size_t cntRem = count % nBlockWidth; // Remainder count.
__m256 vrt = _mm256_setzero_ps(); // Vector result. [AVX] Set zero.
__m256 vload; // Vector load.
const float* p; // Pointer for src data.
size_t i;
// Body.
for (int j = 0; j < loops; ++j) {
p = src;
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
vload = _mm256_load_ps(p); // Load. vload = *(*__m256)p;
vrt = _mm256_add_ps(vrt, vload); // Add. vrt += vload;
p += nBlockWidth;
}
// Remainder processs.
for (i = 0; i < cntRem; ++i) {
rt += p[i];
}
}
// Reduce.
p = (const float*)&vrt;
for (i = 0; i < VectorWidth; ++i) {
rt += p[i];
}
return rt;
}
_mm256_load_ps、_mm256_add_ps等函數名,在C#的內在函數的文檔里可以看到,且可在Intel文檔里查看詳細說明。詳見“ 2.1 文檔查看心得”。
3.4 測試方法(Benchmark)
Benchmark是測試方法,代碼如下。
// Do Benchmark.
void Benchmark() {
const size_t alignment = 256 / 8; // sizeof(__m256) / sizeof(BYTE);
// init.
clock_t tickBegin, msUsed;
double mFlops; // MFLOPS/s .
double scale;
float rt;
const int count = 1024 * 4;
const int loops = 1000 * 1000;
//const int loops = 1;
const double countMFlops = count * (double)loops / (1000.0 * 1000);
float* src = (float*)_aligned_malloc(sizeof(float)*count, alignment); // new float[count];
if (NULL == src) {
printf("Memory alloc fail!");
return;
}
for (int i = 0; i < count; ++i) {
src[i] = (float)i;
}
printf("Benchmark: \tcount=%d, loops=%d, countMFlops=%f\n", count, loops, countMFlops);
// SumBase.
tickBegin = clock();
rt = SumBase(src, count, loops);
msUsed = clock() - tickBegin;
mFlops = countMFlops * CLOCKS_PER_SEC / msUsed;
printf("SumBase:\t%g\t# msUsed=%d, MFLOPS/s=%f\n", rt, (int)msUsed, mFlops);
double mFlopsBase = mFlops;
// SumVectorAvx.
__try {
tickBegin = clock();
rt = SumVectorAvx(src, count, loops);
msUsed = clock() - tickBegin;
mFlops = countMFlops * CLOCKS_PER_SEC / msUsed;
scale = mFlops / mFlopsBase;
printf("SumVectorAvx:\t%g\t# msUsed=%d, MFLOPS/s=%f, scale=%f\n", rt, (int)msUsed, mFlops, scale);
}
__except (EXCEPTION_EXECUTE_HANDLER) {
printf("Run SumVectorAvx fail!");
}
// done.
_aligned_free(src);
}
因為AVX對於地址對齊的數據,性能最好。於是使用了 _aligned_malloc 分配記憶體,用 _aligned_free 釋放記憶體。
C語言標準庫里提供了clock函數來計時,CLOCKS_PER_SEC常量是它在每秒的間隔值。於是便能計算出耗時時間。
因C++或VC++官方庫里未提供檢測AVX指令集的辦法,而手工寫一個的話,太影響篇幅。於是本文用了一個簡單的辦法,利用VC++的SEH(Structured Exception Handling,結構化異常處理)來做異常處理,當“__try”塊運行時發現不支持AVX指令集時,會進入“__except”塊。
3.5 BenchmarkVectorCpp.cpp的完整代碼
BenchmarkVectorCpp.cpp的完整代碼如下。
// BenchmarkVectorCpp.cpp : This file contains the 'main' function. Program execution begins and ends there.
//
#include <immintrin.h>
#include <malloc.h>
#include <stdio.h>
#include <time.h>
#ifndef EXCEPTION_EXECUTE_HANDLER
#define EXCEPTION_EXECUTE_HANDLER (1)
#endif // !EXCEPTION_EXECUTE_HANDLER
// Sum - base.
float SumBase(const float* src, size_t count, int loops) {
float rt = 0; // Result.
size_t i;
for (int j = 0; j < loops; ++j) {
for (i = 0; i < count; ++i) {
rt += src[i];
}
}
return rt;
}
// Sum - Vector AVX.
float SumVectorAvx(const float* src, size_t count, int loops) {
float rt = 0; // Result.
size_t VectorWidth = sizeof(__m256) / sizeof(float); // Block width.
size_t nBlockWidth = VectorWidth; // Block width.
size_t cntBlock = count / nBlockWidth; // Block count.
size_t cntRem = count % nBlockWidth; // Remainder count.
__m256 vrt = _mm256_setzero_ps(); // Vector result. [AVX] Set zero.
__m256 vload; // Vector load.
const float* p; // Pointer for src data.
size_t i;
// Body.
for (int j = 0; j < loops; ++j) {
p = src;
// Vector processs.
for (i = 0; i < cntBlock; ++i) {
vload = _mm256_load_ps(p); // Load. vload = *(*__m256)p;
vrt = _mm256_add_ps(vrt, vload); // Add. vrt += vload;
p += nBlockWidth;
}
// Remainder processs.
for (i = 0; i < cntRem; ++i) {
rt += p[i];
}
}
// Reduce.
p = (const float*)&vrt;
for (i = 0; i < VectorWidth; ++i) {
rt += p[i];
}
return rt;
}
// Do Benchmark.
void Benchmark() {
const size_t alignment = 256 / 8; // sizeof(__m256) / sizeof(BYTE);
// init.
clock_t tickBegin, msUsed;
double mFlops; // MFLOPS/s .
double scale;
float rt;
const int count = 1024 * 4;
const int loops = 1000 * 1000;
//const int loops = 1;
const double countMFlops = count * (double)loops / (1000.0 * 1000);
float* src = (float*)_aligned_malloc(sizeof(float)*count, alignment); // new float[count];
if (NULL == src) {
printf("Memory alloc fail!");
return;
}
for (int i = 0; i < count; ++i) {
src[i] = (float)i;
}
printf("Benchmark: \tcount=%d, loops=%d, countMFlops=%f\n", count, loops, countMFlops);
// SumBase.
tickBegin = clock();
rt = SumBase(src, count, loops);
msUsed = clock() - tickBegin;
mFlops = countMFlops * CLOCKS_PER_SEC / msUsed;
printf("SumBase:\t%g\t# msUsed=%d, MFLOPS/s=%f\n", rt, (int)msUsed, mFlops);
double mFlopsBase = mFlops;
// SumVectorAvx.
__try {
tickBegin = clock();
rt = SumVectorAvx(src, count, loops);
msUsed = clock() - tickBegin;
mFlops = countMFlops * CLOCKS_PER_SEC / msUsed;
scale = mFlops / mFlopsBase;
printf("SumVectorAvx:\t%g\t# msUsed=%d, MFLOPS/s=%f, scale=%f\n", rt, (int)msUsed, mFlops, scale);
}
__except (EXCEPTION_EXECUTE_HANDLER) {
printf("Run SumVectorAvx fail!");
}
// done.
_aligned_free(src);
}
int main() {
printf("BenchmarkVectorCpp\n");
printf("\n");
printf("Pointer size:\t%d\n", (int)(sizeof(void*)));
#ifdef _DEBUG
printf("IsRelease:\tFalse\n");
#else
printf("IsRelease:\tTrue\n");
#endif // _DEBUG
#ifdef _MSC_VER
printf("_MSC_VER:\t%d\n", _MSC_VER);
#endif // _MSC_VER
#ifdef __AVX__
printf("__AVX__:\t%d\n", __AVX__);
#endif // __AVX__
printf("\n");
// Benchmark.
Benchmark();
}
3.6 測試結果
在我的電腦(lntel(R) Core(TM) i5-8250U CPU @ 1.60GHz
、Windows 10)上運行時,x64、Release版程式的輸出信息為:
Pointer size: 8
IsRelease: True
_MSC_VER: 1916
__AVX__: 1
Benchmark: count=4096, loops=1000000, countMFlops=4096.000000
SumBase: 6.87195e+10 # msUsed=4938, MFLOPS/s=829.485622
SumVectorAvx: 5.49756e+11 # msUsed=616, MFLOPS/s=6649.350649, scale=8.016234
從中可以看出——
- SumBase:C++版(MFLOPS/s=829.485622),與C#版(MFLOPS/s=829.485621709194)的性能相同。
- SumVectorAvx:C++版(MFLOPS/s=6649.350649),與C#版(MFLOPS/s=6714.754098360656)的性能幾乎相同。
也就說,對於使用內在函數來做SIMD,C++與C#的性能是相同。故可以根據項目需要,選擇最合適的開發語言就行。
四、小結
C#使用向量類型的最佳實踐——
- 若僅需要使用單精度浮點類型(float),且是開發數學上的向量運算相關的功能,可根據業務上對向量運算的要求,使用維度匹配的向量類(例如 2維向量處理時用Vector2、3維向量處理時用Vector3、3維齊次向量處理時用Vector4)。其他情況下,至少應編寫一套傳統的、不使用向量類型的代碼。
- 若某項計算任務需要進一步做性能優化、且它的工作比較適合SIMD處理時,可以再開發一套基於
Vector<T>
的向量代碼。在使用時別忘了檢查是否支持硬體加速,若不支持,應退回到使用傳統代碼。 - 若發現內在函數能帶來很大的提升時,可考慮使用內在函數,開發基於總位寬固定的向量的代碼(如
Vector256<T>
)。在使用時別忘了檢查當前平臺是否支持該內在函數,若不支持,應退回到使用前面的方案(第1~2條)。
源碼地址——
https://github.com/zyl910/BenchmarkVector/tree/main/BenchmarkVector2
參考文獻
- MSDN《
Vector256<T>
結構》. https://docs.microsoft.com/zh-cn/dotnet/api/system.runtime.intrinsics.vector256-1?view=netcore-3.0 - MSDN《Vector256 類》. https://docs.microsoft.com/zh-cn/dotnet/api/system.runtime.intrinsics.vector256?view=netcore-3.0
- MSDN《Avx 類》. https://docs.microsoft.com/zh-cn/dotnet/api/system.runtime.intrinsics.x86.avx?view=netcore-3.0
- MSDN《MemoryMarshal 類》. https://docs.microsoft.com/zh-cn/dotnet/api/system.runtime.interopservices.memorymarshal?view=netcore-3.0
- Tanner Gooding《Hardware Intrinsics in .NET Core》. https://devblogs.microsoft.com/dotnet/hardware-intrinsics-in-net-core/
- Intel《Intel® Intrinsics Guide》. https://www.intel.com/content/www/us/en/docs/intrinsics-guide/index.html
- zyl910《[C] 跨平臺使用Intrinsic函數範例1——使用SSE、AVX指令集 處理 單精度浮點數組求和(支持vc、gcc,相容Windows、Linux、Mac)》. https://www.cnblogs.com/zyl910/archive/2012/10/22/simdsumfloat.html
- zyl910《[x86]SIMD指令集發展歷程表(MMX、SSE、AVX等)》. https://www.cnblogs.com/zyl910/archive/2012/02/26/x86_simd_table.html
- zyl910《
C# 使用SIMD向量類型加速浮點數組求和運算(1):使用Vector4、Vector<T>
》. https://www.cnblogs.com/zyl910/p/dotnet_simd_BenchmarkVector1.html - wikipedia《Advanced Vector Extensions 2 (AVX2)》. https://en.wikipedia.org/wiki/Advanced_Vector_Extensions#Advanced_Vector_Extensions_2
- wikipedia《SSE2 (Streaming SIMD Extensions 2)》. https://en.wikipedia.org/wiki/SSE2