SIMD:全称是 Single Instruction, Multiple Data,中文通常译为“单指令多数据”。
以下示例运行环境是 x86-64 Linux,主要使用 C、GCC 和 AVX2。本机当前环境:
- CPU: Intel Xeon Platinum 8260
- 架构: x86-64
- GCC: 8.3.0
- Go: 1.24.4
- 可用指令集: SSE2、AVX、AVX2、AVX-512
教程先使用 AVX2,而不是直接使用 AVX-512。AVX2 的概念更简单,支持范围也更广。
完成这份教程后,应当能够:
- 解释标量执行和 SIMD 执行的区别。
- 看懂“向量宽度”“lane”“掩码”“尾部处理”等基本术语。
- 判断一个循环是否适合 SIMD。
- 使用 GCC 的自动向量化并查看优化报告。
- 使用 AVX2 intrinsic 编写数组加法和条件过滤。
- 正确验证 SIMD 实现,而不是只看性能数字。
从一个普通循环开始
先看最熟悉的 C 循环:
for (size_t i = 0; i < n; i++) { c[i] = a[i] + b[i];}从语言语义看,每次循环只处理一个元素:
c[0] = a[0] + b[0]c[1] = a[1] + b[1]c[2] = a[2] + b[2]...这叫标量处理。这里的“标量”可以理解为单个数值。
AVX2 提供 256 位向量寄存器。一个 C int32_t 是 32 位,因此一个 256 位寄存器可以同时容纳:256 / 32 = 8 个 int32_t
把寄存器想象成有 8 个格子的容器:
a: [a0 a1 a2 a3 a4 a5 a6 a7]b: [b0 b1 b2 b3 b4 b5 b6 b7] | | 一条向量加法指令 vc: [c0 c1 c2 c3 c4 c5 c6 c7]每个格子叫一个 lane。一条向量加法指令对 8 个 lane 分别做加法,这就是“单指令,多数据”。
需要注意:这不等于整个循环一定快 8 倍。实际性能还受以下因素影响:
- 从内存读取和写回数据的速度
- CPU 执行端口和指令延迟
- 数据是否连续
- 分支是否容易处理
- 数组是否足够大
- 编译器是否已经自动生成 SIMD 指令
SIMD 不是多线程,多线程通常让多个 CPU 核心同时执行不同的指令流。SIMD 通常发生在一个核心内部,让一条指令同时处理多个数据。
从一段 AVX2 代码理解 intrinsic
先用“数组加法”作为例子。目标仍然是这个普通循环:
for (size_t i = 0; i < n; i++) { c[i] = a[i] + b[i];}普通写法一次处理一个元素。AVX2 写法一次处理 8 个 int32_t。
完整 AVX2 函数
先不要急着理解每个名字,先看完整结构:
#include <immintrin.h>#include <stdint.h>#include <stdio.h>#define N 19#define AVX2_INT32_LANES 8__attribute__((noinline))static void avx2_add(const int32_t *a, const int32_t *b, int32_t *c, size_t n) { size_t i = 0; for (; i + AVX2_INT32_LANES <= n; i += AVX2_INT32_LANES) { __m256i va = _mm256_loadu_si256((const __m256i *)(const void *)(a + i)); __m256i vb = _mm256_loadu_si256((const __m256i *)(const void *)(b + i)); __m256i vc = _mm256_add_epi32(va, vb); _mm256_storeu_si256((__m256i *)(void *)(c + i), vc); } for (; i < n; i++) { c[i] = a[i] + b[i]; }}int main(void) { int32_t a[N]; int32_t b[N]; int32_t c[N]; // 初始化输入数组 for (size_t i = 0; i < N; i++) { a[i] = (int32_t)i - 5; b[i] = (int32_t)(i * 3); } // 执行AVX2加法 add_arrays_avx2(a, b, c, N); // 验证结果 for (size_t i = 0; i < N; i++) { int32_t expected = a[i] + b[i]; if (c[i] != expected) { fprintf(stderr, "验证失败:下标 %zu,实际结果 %d,期望结果 %d\n", i, c[i], expected); return 1; } } printf("AVX2 加法验证通过:共验证 %d 个元素,包括尾部标量处理\n", N); return 0;}这段 avx2_add 函数代码分成两部分:
- 第一个
for是 SIMD 主循环,每次处理 8 个元素。 - 第二个
for是尾部处理,处理剩下不足 8 个的元素。
如果 n = 19,执行过程大概是:
第 1 次 SIMD: 处理 c[0] 到 c[7]第 2 次 SIMD: 处理 c[8] 到 c[15]尾部标量循环: 处理 c[16] 到 c[18]所以完整 SIMD 代码通常不是只有一条“神奇指令”,而是:SIMD 主循环 + 标量尾部处理。
如何编译运行
本目录已经有完整示例 02_avx2_add.c。可以这样运行:
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -mavx2 avx2_add.c -o avx2_add这条编译命令可以拆开理解:
| 部分 | 含义 |
|---|---|
gcc |
使用 GCC 编译器 |
-std=c11 |
使用 C11 标准编译 |
-O3 |
开启较高级别优化,方便生成性能更好的机器码 |
-Wall |
开启常见警告 |
-Wextra |
开启更多额外警告 |
-Wpedantic |
更严格地检查代码是否符合 C 标准 |
-mavx2 |
允许编译器使用 AVX2 指令;本例使用了 _mm256_* intrinsic,所以需要它 |
avx2_add.c |
输入源文件 |
-o avx2_add |
指定输出的可执行文件名为 avx2_add |
正常输出类似:
AVX2 加法验证通过:共验证 19 个元素,包括尾部标量处理数据流:load、add、store
核心代码只有四行:
__m256i va = _mm256_loadu_si256((const __m256i *)(const void *)(a + i));__m256i vb = _mm256_loadu_si256((const __m256i *)(const void *)(b + i));__m256i vc = _mm256_add_epi32(va, vb);_mm256_storeu_si256((__m256i *)(void *)(c + i), vc);它们对应三个动作:
从内存加载数据(读取va和vb) -> 在向量寄存器中计算(计算vc) -> 写回内存把这四行按 API 拆开看:
| API | 作用 | 输入 | 输出 |
|---|---|---|---|
_mm256_loadu_si256 |
从内存读取 256 位数据,放入一个 __m256i 向量 |
一个内存地址 | 返回 __m256i向量 |
_mm256_add_epi32 |
把两个 __m256i 向量按 32 位整数 lane 分别相加 |
两个 __m256i 向量 |
返回新的 __m256i向量 |
_mm256_storeu_si256 |
把一个 __m256i 向量的 256 位数据写回内存 |
一个目标内存地址和一个 __m256i向量 |
没有返回值 |
知道 API 的作用以后,剩下需要理解的就是类型转换,比如 (const __m256i *)(const void *)(a + i):
- a:是 int32_t *,32 位,指向数组第一个元素
- a + i:还是 int32_t *,指向 a[i]
- 但
_mm256_loadu_si256是从某个地址开始一次读取256位。对于int32_t数组来说,刚好是 8 个元素 - (const void *)(a + i):先把它转成通用地址,不关心原来是什么元素类型
- (const __m256i *):再告诉 intrinsic 从这个地址开始,当成一个 256 位向量来读
这个转换不会复制数据,也不会改变数组内容,只是改变“编译器如何看待这个地址”。
store 那一行也是同样的逻辑,但区别是:
load 只读取 a/b,所以使用 const 指针store 要写入 c,所以不能使用 const 指针最后再注意一点:这些 API 不会自动检查数组边界。loadu 一次就是读取 8 个 int32_t,
__m256i 向量是什么
__m256i 可以先理解成“一个 256 位的整数向量容器”。它通常对应 CPU 里的一个 256 位向量寄存器。
int32_t 是 32 位,因此一个 __m256i 可以放下 8 个 int32_t。可以把它想象成这样:
__m256i va+-----+-----+-----+-----+-----+-----+-----+-----+| a0 | a1 | a2 | a3 | a4 | a5 | a6 | a7 |+-----+-----+-----+-----+-----+-----+-----+-----+ 32位 32位 32位 32位 32位 32位 32位 32位每个小格子叫一个 lane。上面这个向量一共有 8 个 lane。如果:
va = [1, 2, 3, 4, 5, 6, 7, 8]vb = [10, 20, 30, 40, 50, 60, 70, 80]执行:
__m256i vc = _mm256_add_epi32(va, vb);得到:
vc = [11, 22, 33, 44, 55, 66, 77, 88]这就是 SIMD 的核心:一条加法指令,同时完成 8 次整数加法。
intrinsic 名字怎么拆
intrinsic 的名字看起来很长,但多数名字都有规律。把 _mm256_add_epi32 拆开:
_mm256 _ add _ epi32可以理解为:
| 片段 | 含义 |
|---|---|
_mm256 |
使用 256 位向量,通常对应 AVX/AVX2 |
add |
做加法 |
epi32 |
packed 32-bit integer,也就是“打包在一个向量里的 32 位整数”,所谓 packed,就是多个小整数挤在一个大寄存器里。 |
所以含义就是:对 256 位向量中的多个 32 位整数做加法。先不要急着背所有名字,重点是看到名字时能猜出大概方向:
_mm 128 位_mm256 256 位_mm512 512 位add 加法sub 减法mul 乘法load 从内存加载到向量寄存器store 从向量寄存器写回内存cmp 比较epi8 8 位整数 laneepi16 16 位整数 laneepi32 32 位整数 laneepi64 64 位整数 laneps 单精度浮点数,也就是 floatpd 双精度浮点数,也就是 double指令集和汇编
现在再看指令集表会更容易:
| 平台 | 指令集 | 常见向量宽度 |
|---|---|---|
| x86-64 | SSE/SSE2 | 128 位 |
| x86-64 | AVX/AVX2 | 256 位 |
| x86-64 | AVX-512 | 512 位 |
| ARM | NEON | 128 位 |
| ARM | SVE/SVE2 | 可变向量长度 |
这里我们选择 AVX2,是因为它在 x86-64 机器上比较常见,并且可以一次处理 256 位数据。
intrinsic 和手写汇编的关系可以这样理解:
- 手写汇编:你要自己决定用哪个寄存器、怎么写指令、怎么配合 ABI。
- intrinsic:你表达想做哪种向量操作,寄存器分配和最终编码交给编译器。
例如:__m256i vc = _mm256_add_epi32(va, vb);
常见情况下会被编译成类似这样的 AVX2 汇编指令:
vpaddd ymm0, ymm1, ymm2这里的 vpaddd 是真正的汇编指令,_mm256_add_epi32 是你在 C 代码里使用的 intrinsic。入门阶段先写 intrinsic,比直接写汇编更容易控制复杂度。
最后只需要记住一条主线:C 数组 -> load 到 __m256i -> 用 intrinsic 计算 -> store 回 C 数组
让编译器自动生成 SIMD
在手写 intrinsic 之前,应该先学会让编译器自动向量化。原因很简单:很多规则清晰的循环,编译器已经可以识别并自动生成 SIMD 指令。这一节的目标是:
- 写一个普通 C 循环
- 用 -O3 -march=native 编译
- 确认程序结果正确
- 查看编译器是否自动向量化
- 查看最终汇编里是否出现向量指令
完整代码和运行结果
#include <stdint.h>#include <stdio.h>#define N 19__attribute__((noinline))static void add_arrays(const int32_t *restrict a, const int32_t *restrict b, int32_t *restrict c, size_t n) { for (size_t i = 0; i < n; i++) { c[i] = a[i] + b[i]; }}int main(void) { int32_t a[N]; int32_t b[N]; int32_t c[N]; for (size_t i = 0; i < N; i++) { a[i] = (int32_t)i; b[i] = (int32_t)(i * 10); } add_arrays(a, b, c, N); for (size_t i = 0; i < N; i++) { int32_t expected = a[i] + b[i]; if (c[i] != expected) { fprintf(stderr, "验证失败:下标 %zu,实际结果 %d,期望结果 %d\n", i, c[i], expected); return 1; } } printf("自动向量化示例验证通过:共验证 %d 个元素\n", N); return 0;}先编译运行:
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -march=native auto_vectorization.c -o auto_vectorization正常输出:自动向量化示例验证通过:共验证 19 个元素。这一步只回答一个问题:程序结果是否正确。性能和汇编都放到后面看。
GCC 可以输出向量化报告:
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -march=native -fopt-info-vec-optimized auto_vectorization.c -o auto_vectorization如果成功,可能看到类似信息:
auto_vectorization.c:8:5: note: loop vectorizedauto_vectorization.c:18:5: note: loop vectorized查看最终汇编
生成汇编文件:
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -march=native -S -masm=intel auto_vectorization.c -o auto_vectorization.s搜索常见的向量加法和向量加载/存储指令:vpaddd|vmovdqu|vmovdqa,可以发现如下内容:
add_arrays.constprop.0:.LFB5: .cfi_startproc # 参数约定:rdi = a,rsi = b,rdx = c # 逻辑上:前 16 个元素用 SIMD,最后 3 个元素用标量处理 # 实际汇编顺序被编译器重排过,所以标量尾部会穿插在 SIMD 指令之间 mov eax, DWORD PTR 64[rsi] # 标量尾部:先读取 b[16],编译器把尾部处理提前了 vmovdqu32 ymm1, YMMWORD PTR [rdi] # 向量加载:读取 a[0..7] add eax, DWORD PTR 64[rdi] # 标量尾部:eax = b[16] + a[16] mov DWORD PTR 64[rdx], eax # 标量尾部:写回 c[16] mov eax, DWORD PTR 68[rsi] # 标量尾部:读取 b[17] vpaddd ymm0, ymm1, YMMWORD PTR [rsi] # 向量加法:读取 b[0..7],和 ymm1 里的 a[0..7] 相加,结果放到 ymm0 add eax, DWORD PTR 68[rdi] # 标量尾部:eax = b[17] + a[17] vmovdqu32 ymm2, YMMWORD PTR 32[rdi] # 向量加载:读取 a[8..15] mov DWORD PTR 68[rdx], eax # 标量尾部:写回 c[17] mov eax, DWORD PTR 72[rdi] # 标量尾部:读取 a[18] vmovdqu32 YMMWORD PTR [rdx], ymm0 # 向量存储:写回 c[0..7] vpaddd ymm0, ymm2, YMMWORD PTR 32[rsi] # 向量加法:读取 b[8..15],和 ymm2 里的 a[8..15] 相加,结果放到 ymm0 add eax, DWORD PTR 72[rsi] # 标量尾部:eax = a[18] + b[18] vmovdqu32 YMMWORD PTR 32[rdx], ymm0 # 向量存储:写回 c[8..15] mov DWORD PTR 72[rdx], eax # 标量尾部:写回 c[18] vzeroupper # 清理 AVX 上半部分寄存器状态 ret .cfi_endproc编译器生成的汇编代码本来就不是按“人类阅读顺序”写的。这段数组加法代码逻辑虽然很简单,但汇编顺序看起来乱,是因为编译器在做指令调度。它不是严格按下面这样生成:
- 先处理 c[0…7]
- 再处理 c[8…15]
- 最后处理 c[16…18]
而是会穿插执行,这么做是为了减少等待时间。CPU 从内存读数据有延迟,编译器会把一些标量尾部操作插进 SIMD 操作中间,让 CPU 执行指令集的流水线尽量忙起来。
restrict 是什么
先看这个函数参数:
static void add_arrays(const int32_t *restrict a, const int32_t *restrict b, int32_t *restrict c, size_t n)restrict 解决的是一个问题:这几个指针会不会指向同一块内存。
先对比两种调用:
add_arrays(a, b, c, n); // 常见情况:三个独立数组add_arrays(data, other, data + 1, n); // 特殊情况:a 和 c 指向同一块数组的不同位置第二种情况里:
a[0] -> data[0] c[0] -> data[1]a[1] -> data[1] c[1] -> data[2]a[2] -> data[2] c[2] -> data[3]这时 a 和 c 的内存是重叠的。普通 C 循环是逐轮执行的:
i = 0: 写 c[0],也就是写 data[1]i = 1: 读 a[1],也就是读 data[1]也就是说,上一轮写入的结果,可能会影响下一轮读取的值。但 SIMD 想做的是批量处理,例如一次处理 8 个元素:
- 普通循环:读 a[0] -> 写 c[0] -> 读 a[1] -> 写 c[1] -> …
- SIMD:先读 a[0…7] 和 b[0…7] -> 一起计算 -> 写 c[0…7]
如果 a 和 c 可能重叠,SIMD 这种“先批量读取,再批量写回”的顺序就可能改变程序结果。所以没有 restrict 时,编译器必须保守。
加上 restrict 后,就是在告诉编译器:a、b、c 是三块互不重叠的内存。不用考虑 add_arrays(data, other, data + 1, n) 这种调用。
最后对比一下 const 和 restrict:
| 写法 | 重点 |
|---|---|
const int32_t *a |
不能通过 a 修改数据 |
int32_t *restrict c |
c 指向的内存不会和其他相关指针重叠 |
哪些循环适合 SIMD
适合 SIMD 的典型特征:
- 大量元素执行相同操作。
- 每次迭代彼此独立。
- 数据在内存中连续。
- 分支可以转换为掩码。
- 数据类型和精度要求明确。
常见场景:
- 图片、音频、视频处理
- 矩阵和向量计算
- 压缩、校验、编码和加密
- JSON 字符扫描
- 数据库列扫描、聚合和过滤
- 日志批量过滤
- 向量距离计算
不容易直接 SIMD 化的情况:
- 本次迭代依赖上一次结果
- 指针链表遍历
- 每个元素执行完全不同的逻辑
- 数据随机分散在内存中
- 数据规模很小,准备工作占主要成本
数据布局:AoS 与 SoA
假设日志记录采用结构体数组:
struct Record { int16_t status; int16_t latency_ms;};struct Record records[n];这种布局叫 Array of Structures,简称 AoS:
status0 latency0 status1 latency1 status2 latency2 ...其实还可以使用两个独立数组来表示:
int16_t status[n];int16_t latency_ms[n];这种布局叫 Structure of Arrays,简称 SoA:
status0 status1 status2 status3 ...latency0 latency1 latency2 latency3 ...SoA 更容易一次连续加载 16 个 status 或 16 个 latency_ms,因此通常更适合 SIMD。数据库的列式存储也有类似优势:查询只读取需要的列,并对连续的同类数据批量计算。
常见错误
1. 忘记尾部元素
数组长度不一定刚好是向量 lane 数的倍数。比如 AVX2 处理 int32_t 时一轮是 8 个元素,处理 int16_t 时一轮是 16 个元素。必须使用标量循环、掩码加载或填充处理尾部。
2. 使用错误的有符号性
_mm256_cmpgt_epi16 做有符号 16 位整数比较,_mm256_cmpgt_epi32 做有符号 32 位整数比较。无符号比较的规则不同。选择 intrinsic 时必须先确认:
- 元素位宽
- 整数还是浮点数
- 有符号还是无符号
3. 假定地址天然对齐
只有明确控制分配和偏移时才能依赖对齐。否则先使用 loadu/storeu。
4. 在不支持 AVX2 的机器上运行
先检查:
lscpu | rg -o 'avx2|avx512f'生产代码通常需要运行时分发:
if 支持 AVX2 { 调用 AVX2 实现} else { 调用标量实现 }GCC 还提供 __builtin_cpu_supports("avx2")。注意:检测逻辑本身必须运行在不强制要求 AVX2 的通用代码路径中。
5. 认为 intrinsic 一定比编译器快
编译器可能已经自动向量化,手写版本也可能因为额外加载、寄存器压力或不理想的数据布局而更慢。必须通过报告、汇编和基准共同判断。
6. 只优化指令,不优化数据布局
连续数据通常比复杂指令技巧更重要。先解决随机访问、指针追逐和不必要的数据搬运,再考虑手写 SIMD。
实战性能对比
场景如下:用户在运行时提交一条查询规则(status >= 500 && latency_ms > 100),执行器把这条规则编译成过滤函数。
数据结构如下:
int16_t status[n];int16_t latency_ms[n];这个优化成立的前提是字段范围确实放得进 int16_t。本例里 status 大约是 200..799,latency_ms 大约是 0..499,不会超过 16 位有符号整数范围。
这里先暂时不使用 AVX-512。AVX2 是 256 位,int16_t 是 16 位,所以一条 AVX2 指令已经可以同时处理:256 / 16 = 16 条记录
当前示例代码不直接生成机器码,而是在 filter_bench.c 里写出“JIT 最终可能生成的 AVX2 代码形态”。等后面接真实 JIT 时,目标就是生成类似这段 intrinsic 对应的机器码。
标量规则解释器目标代码
一个通用过滤器通常会先解释规则:
if (rule->has_status_min && status[i] < rule->status_min) { continue;}if (rule->has_latency_min && latency_ms[i] <= rule->latency_min) { continue;}count++;这种写法很灵活:规则里有没有 status 条件、有没有 latency_ms 条件,都可以运行时决定。问题是每条记录都要检查这些 has_xxx,并且一次只处理一条记录。
AVX2 目标代码
假设当规则已经确定为:status >= 500 && latency_ms > 100。AVX2 版本每轮读取 16 条记录的 2 个列:
__m256i vstatus = _mm256_loadu_si256((const __m256i *)(const void *)(status + i));__m256i vlatency = _mm256_loadu_si256((const __m256i *)(const void *)(latency_ms + i));然后把两个条件分别变成两个向量掩码:
__m256i status_ok = _mm256_cmpgt_epi16(vstatus, v_status_min_minus_1);__m256i latency_ok = _mm256_cmpgt_epi16(vlatency, v_latency_min);这里 status >= 500 写成了 status > 499,因为 AVX2 提供的是 16 位整数“大于”比较:_mm256_cmpgt_epi16(a, b)。最后把两个条件合并:
__m256i matched = _mm256_and_si256(status_ok, latency_ok);这一步相当于:status_ok && latency_ok。只不过它是在 16 个 lane 上同时完成的。
从向量掩码得到匹配数量
SIMD 16 位整数比较的每个 lane 通常不是返回普通的 0 或 1,而是:
- 条件为假: 0x0000
- 条件为真: 0xffff
如果 16 条记录的最终匹配结果里有若干个真值,可以用下面的代码把每个字节的最高位提取成掩码:
unsigned int byte_mask = (unsigned int)_mm256_movemask_epi8(matched);count += (size_t)(__builtin_popcount(byte_mask) / 2);为什么要除以 2,因为一个 int16_t lane 有 2 个字节。条件为真时,这个 lane 是 0xffff,两个字节的最高位都会被 movemask 提取出来,所以一个匹配 lane 会贡献 2 个二进制位。
完整代码
#define _POSIX_C_SOURCE 200809L#include <immintrin.h>#include <inttypes.h>#include <stdint.h>#include <stdio.h>#include <time.h>// 测试数据量。故意不是 16 的倍数,用来验证 AVX2 主循环之后的尾部处理#define N 10000019// 每个版本重复测试多轮,最后取最快的一次作为测试结果,减少偶发抖动的影响#define ROUNDS 5// AVX2 向量寄存器是 256 位,int16_t 是 16 位,所以一次可以处理 16 个 lane#define AVX2_INT16_LANES 16/** 这个结构体模拟“运行时传进来的查询规则参数”。** 当前示例固定启用两个条件:* status >= status_min && latency_ms > latency_min* 所以这里只需要保存两个阈值*/typedef struct { int16_t status_min; int16_t latency_min;} query_rule_t;// 基准测试函数统一使用这个函数指针类型,方便 run_best 同时测试两个实现typedef size_t (*filter_fn_t)(const query_rule_t *rule, const int16_t *status, const int16_t *latency_ms, size_t n);/* * 使用 SoA 数据布局:每个字段单独放一个连续数组。 * * SIMD 更偏向这种布局,因为可以连续读取 16 个 status 或 16 个 latency_ms。 * 如果使用结构体数组,status 和 latency_ms 会交错排列,向量加载会更麻烦。 */static int16_t status_data[N];static int16_t latency_ms_data[N];/* 防止编译器认为基准结果没人使用,从而把计算整体优化掉。 */static volatile size_t result_sink;// 标量版本:返回符合条件的日志数量// 这是 GCC 的函数属性,用来控制编译器优化行为。这里主要是为了让这个函数保持“普通标量版本”,方便和手写 AVX2 版本做对比。__attribute__((noinline, optimize("no-tree-vectorize")))static size_t filter_scalar_specialized(const query_rule_t *rule, const int16_t *status, const int16_t *latency_ms, size_t n) { size_t count = 0; for (size_t i = 0; i < n; i++) { if (status[i] < rule->status_min) { continue; } if (latency_ms[i] <= rule->latency_min) { continue; } count++; } return count;}// status >= rule->status_min && latency_ms > rule->latency_min__attribute__((noinline))static size_t filter_target_avx2(const query_rule_t *rule, const int16_t *status, const int16_t *latency_ms, size_t n) { // 把一个标量值(rule->status_min - 1)复制到 16 个 lane 当中 // status >= 500 被写成 status > 499,是因为 AVX2 对 int16_t 提供的是“大于”比较(没有大于等于指令)。 const __m256i v_status_min_minus_1 = _mm256_set1_epi16((int16_t)(rule->status_min - 1)); // 把一个标量值(rule->latency_min)复制到 16 个 lane 当中 const __m256i v_latency_min = _mm256_set1_epi16(rule->latency_min); size_t count = 0; size_t i = 0; // 每次处理 16 条记录 // 循环条件 i + 16 <= n 用来保证后面至少还有 16 个元素,否则 _mm256_loadu_si256 会越界读取 for (; i + AVX2_INT16_LANES <= n; i += AVX2_INT16_LANES) { // 从 status 数组中读取 16 个元素,复制到 vstatus 中 // loadu 里的 u 表示 unaligned,不要求地址按 32 字节对齐 __m256i vstatus = _mm256_loadu_si256((const __m256i *)(const void *)(status + i)); // 从 latency_ms 数组中读取 16 个元素,复制到 vlatency 中 __m256i vlatency = _mm256_loadu_si256((const __m256i *)(const void *)(latency_ms + i)); // 每个 lane 的比较结果不是 0 或 1,而是: // 条件为假: 0x0000 // 条件为真: 0xffff // 计算 status >= status_min - 1 的结果 __m256i status_ok = _mm256_cmpgt_epi16(vstatus, v_status_min_minus_1); // 计算 latency_ms > latency_min 的结果 __m256i latency_ok = _mm256_cmpgt_epi16(vlatency, v_latency_min); // 两个条件都为真时,matched 对应 lane 才会是 0xffff __m256i matched = _mm256_and_si256(status_ok, latency_ok); /* * matched 里有 16 个 int16_t lane 的判断结果。 * 每个 lane 不是普通的 1 或 0,而是: * 匹配: 0xffff * 不匹配: 0x0000 * * _mm256_movemask_epi8 会把 matched 当成 32 个 byte, * 取每个 byte 的最高位,组成一个 32 位整数。 * * 例如只看前 4 个 int16_t lane: * lane 结果: [匹配, 不匹配, 匹配, 不匹配] * 实际数值: [0xffff, 0x0000, 0xffff, 0x0000] * 按 byte 看:[ff ff, 00 00, ff ff, 00 00] * 取最高位: [1 1, 0 0, 1 1, 0 0] * * 也就是说,1 个匹配的 int16_t lane 会贡献 2 个 bit。 * 所以 popcount(byte_mask) 数出来的是匹配 byte 数再除以 2,才是匹配的 int16_t lane 数量。 */ unsigned int byte_mask = (unsigned int)_mm256_movemask_epi8(matched); count += (size_t)(__builtin_popcount(byte_mask) / 2); } // 标量尾部处理 for (; i < n; i++) { count += (size_t)(status[i] >= rule->status_min && latency_ms[i] > rule->latency_min); } return count;}// 获取当前时间戳,单位纳秒static uint64_t now_ns(void) { struct timespec ts; if (clock_gettime(CLOCK_MONOTONIC, &ts) != 0) { fprintf(stderr, "获取时间失败:clock_gettime 调用失败\n"); return 0; } return (uint64_t)ts.tv_sec * UINT64_C(1000000000) + (uint64_t)ts.tv_nsec;}// 运行同一个函数多轮,返回最快耗时。static uint64_t run_best(filter_fn_t fn, const query_rule_t *rule, size_t expected) { uint64_t best = UINT64_MAX; for (int round = 0; round < ROUNDS; round++) { uint64_t start = now_ns(); size_t count = fn(rule, status_data, latency_ms_data, N); uint64_t elapsed = now_ns() - start; result_sink = count; // 每轮都检查结果,避免错误实现跑得很快但结果不对。 if (count != expected) { fprintf(stderr, "基准验证失败:实际结果 %zu,期望结果 %zu\n", count, expected); return UINT64_MAX; } if (elapsed < best) { best = elapsed; } } return best;}// 构造稳定、可重复的测试数据// status 范围约为 200..799; latency_ms 范围约为 0..499// 都能安全放进 int16_tstatic void init_data(void) { for (size_t i = 0; i < N; i++) { status_data[i] = (int16_t)(200 + (int16_t)(i % 600)); latency_ms_data[i] = (int16_t)((i * 17) % 500); }}int main(void) { // 模拟业务运行时收到的查询规则(条件):status >= 500 && latency_ms > 100 const query_rule_t rule = {500, 100}; // 初始化测试数据 init_data(); // 先分别运行标量版本和 AVX2 版本。 // 后面只有在两者结果完全一致时,才继续做性能测试。 size_t scalar_count = filter_scalar_specialized(&rule, status_data, latency_ms_data, N); size_t avx2_count = filter_target_avx2(&rule, status_data, latency_ms_data, N); if (scalar_count != avx2_count) { fprintf(stderr, "验证失败:标量结果 %zu,AVX2 结果 %zu\n", scalar_count, avx2_count); return 1; } uint64_t scalar_ns = run_best(filter_scalar_specialized, &rule, scalar_count); uint64_t avx2_ns = run_best(filter_target_avx2, &rule, scalar_count); if (scalar_ns == UINT64_MAX || avx2_ns == UINT64_MAX || avx2_ns == 0) { return 1; } printf("过滤规则: status >= %d && latency_ms > %d\n", rule.status_min, rule.latency_min); printf("验证匹配数量: %zu\n", scalar_count); printf("标量最佳耗时: %" PRIu64 " ns\n", scalar_ns); printf("AVX2 最佳耗时: %" PRIu64 " ns\n", avx2_ns); printf("AVX2 比标量快: %.2f%%\n", ((double)scalar_ns / (double)avx2_ns - 1.0) * 100.0); return 0;}测试结果
编译命令如下:
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -mavx2 filter_bench.c -o filter_bench程序会:
- 生成
status、latency_ms两列测试数据。 - 用标量规则解释器计算匹配数量。
- 用 JIT 目标形态的 AVX2 函数计算匹配数量。
- 先检查两种实现的结果是否一致。
- 报告各自的最佳耗时。
# ./filter_bench过滤规则: status >= 500 && latency_ms > 100验证匹配数量: 3989935标量最佳耗时: 10635810 nsAVX2 最佳耗时: 4354679 nsAVX2 比标量快: 144.24%# ./filter_bench过滤规则: status >= 500 && latency_ms > 100验证匹配数量: 3989935标量最佳耗时: 10251549 nsAVX2 最佳耗时: 4832338 nsAVX2 比标量快: 112.14%# ./filter_bench过滤规则: status >= 500 && latency_ms > 100验证匹配数量: 3989935标量最佳耗时: 10187503 nsAVX2 最佳耗时: 4463787 nsAVX2 比标量快: 128.23%从多次运行结果看,匹配数量一直是 3989935,说明标量版本和 AVX2 版本的计算结果一致。性能上,标量版本大约在 10 ms 左右,AVX2 版本大约在 4.3 ms 到 4.8 ms 之间,整体能快一倍以上。
这次提速主要来自三点:
- 只保留查询真正需要的
status和latency_ms两列,减少了内存读取。 - 把数据类型从
int32_t压缩成int16_t,让 AVX2 一次可以处理 16 条记录。 - 过滤条件固定后,AVX2 版本可以直接比较阈值,不需要在循环里解释规则。
不同次运行的百分比会有波动是正常的。基准测试会受到 CPU 调度、缓存状态、频率变化和虚拟机环境影响,所以不要只盯着某一次的数字。更重要的是看两个结论:结果必须一致,AVX2 版本在多次运行中稳定快于标量版本。
虽然 AVX2 一次处理 16 个 int16_t lane,但程序仍然要从内存读取两列数据,还要做比较、按位与、生成掩码和统计匹配数量。真实加速通常会小于理论 lane 数。
练习
练习 1:向量乘法
把数组加法改为:
c[i] = a[i] * b[i];查找并使用 32 位整数低位乘法 intrinsic。提示:_mm256_mullo_epi32。
练习 2:范围过滤
把过滤规则改为:
status >= 400 && status < 600分别生成两个比较掩码,再用 AND 合并。
练习 3:输出匹配位置
当前程序只统计匹配数量。尝试把匹配元素的索引写入输出数组。
这个问题比计数更难,因为每 8 个元素的匹配数量不固定。先使用提取 mask 后逐位检查的简单实现,不必立刻追求完全无分支。
练习 4:运行时分发
构建两个函数:
filter_scalar(...)filter_avx2(...)启动时使用 __builtin_cpu_supports("avx2") 选择实现,并通过函数指针保存。
练习 5:连接 JIT
让 JIT 生成的函数接收数组指针和元素数量。第一版可以只调用已经编译好的filter_avx2,随后再尝试真正生成 AVX2 指令。
知识图谱
SIMD├── 数据│ ├── lane│ ├── 连续内存│ ├── AoS / SoA│ └── 对齐├── 指令│ ├── load / store│ ├── 算术│ ├── 比较│ ├── mask│ └── reduction├── 编译方式│ ├── 自动向量化│ ├── intrinsic│ ├── 汇编│ └── JIT 生成├── 正确性│ ├── 尾部│ ├── 越界│ ├── 对齐│ ├── 溢出│ └── CPU 特性检测└── 性能 ├── 缓存 ├── 内存带宽 ├── 分支 ├── 数据布局 └── 基准测试参考资料
- Intel Intrinsics Guide: https://www.intel.com/content/www/us/en/docs/intrinsics-guide/index.html
查询每个 intrinsic 的参数、返回值、对应指令和支持的指令集。 - GCC Optimize Options: https://gcc.gnu.org/onlinedocs/gcc/Optimize-Options.html
查询-O3、向量化报告、自动优化相关的 GCC 编译选项。 - GCC x86 Options: https://gcc.gnu.org/onlinedocs/gcc/x86-Options.html
查询-march=native、-mavx2、-mavx512*等 x86 指令集开关。 - Go Assembly Guide: https://go.dev/doc/asm
后续如果想在 Go 里接手写汇编,可以先看这份 Go 汇编规则说明。 - Agner Fog Optimization Manuals: https://www.agner.org/optimize/
更深入的 CPU 优化资料,适合后面研究指令延迟、吞吐、缓存和分支预测。
Comments