SIMD 入门:从标量循环到 AVX2

标签:编译原理首次发布:2026-08-10最近修改:2026-08-10

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 的概念更简单,支持范围也更广。

学习目标

完成这份教程后,应当能够:

  1. 解释标量执行和 SIMD 执行的区别。
  2. 看懂“向量宽度”“lane”“掩码”“尾部处理”等基本术语。
  3. 判断一个循环是否适合 SIMD。
  4. 使用 GCC 的自动向量化并查看优化报告。
  5. 使用 AVX2 intrinsic 编写数组加法和条件过滤。
  6. 正确验证 SIMD 实现,而不是只看性能数字。

从一个普通循环开始

先看最熟悉的 C 循环:

c
for (size_t i = 0; i < n; i++) {    c[i] = a[i] + b[i];}

从语言语义看,每次循环只处理一个元素:

bash
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 个格子的容器:

text
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

先用“数组加法”作为例子。目标仍然是这个普通循环:

c
for (size_t i = 0; i < n; i++) {    c[i] = a[i] + b[i];}

普通写法一次处理一个元素。AVX2 写法一次处理 8 个 int32_t

完整 AVX2 函数

先不要急着理解每个名字,先看完整结构:

c
#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 函数代码分成两部分:

  1. 第一个 for 是 SIMD 主循环,每次处理 8 个元素。
  2. 第二个 for 是尾部处理,处理剩下不足 8 个的元素。

如果 n = 19,执行过程大概是:

text
第 1 次 SIMD: 处理 c[0] 到 c[7]第 2 次 SIMD: 处理 c[8] 到 c[15]尾部标量循环: 处理 c[16] 到 c[18]

所以完整 SIMD 代码通常不是只有一条“神奇指令”,而是:SIMD 主循环 + 标量尾部处理

如何编译运行

本目录已经有完整示例 02_avx2_add.c。可以这样运行:

bash
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

正常输出类似:

c
AVX2 加法验证通过:共验证 19 个元素,包括尾部标量处理

数据流:load、add、store

核心代码只有四行:

c
__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);

它们对应三个动作:

text
从内存加载数据(读取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)

  1. a:是 int32_t *,32 位,指向数组第一个元素
  2. a + i:还是 int32_t *,指向 a[i]
  3. _mm256_loadu_si256 是从某个地址开始一次读取256 位。对于 int32_t 数组来说,刚好是 8 个元素
  4. (const void *)(a + i):先把它转成通用地址,不关心原来是什么元素类型
  5. (const __m256i *):再告诉 intrinsic 从这个地址开始,当成一个 256 位向量来读

这个转换不会复制数据,也不会改变数组内容,只是改变“编译器如何看待这个地址”。

store 那一行也是同样的逻辑,但区别是:

text
load  只读取 a/b,所以使用 const 指针store 要写入 c,所以不能使用 const 指针

最后再注意一点:这些 API 不会自动检查数组边界。loadu 一次就是读取 8 个 int32_t

__m256i 向量是什么

__m256i 可以先理解成“一个 256 位的整数向量容器”。它通常对应 CPU 里的一个 256 位向量寄存器。

int32_t 是 32 位,因此一个 __m256i 可以放下 8 个 int32_t。可以把它想象成这样:

c
__m256i va+-----+-----+-----+-----+-----+-----+-----+-----+| a0  | a1  | a2  | a3  | a4  | a5  | a6  | a7  |+-----+-----+-----+-----+-----+-----+-----+-----+ 3232323232323232

每个小格子叫一个 lane。上面这个向量一共有 8 个 lane。如果:

c
va = [1, 2, 3, 4, 5, 6, 7, 8]vb = [10, 20, 30, 40, 50, 60, 70, 80]

执行:

c
__m256i vc = _mm256_add_epi32(va, vb);

得到:

c
vc = [11, 22, 33, 44, 55, 66, 77, 88]

这就是 SIMD 的核心:一条加法指令,同时完成 8 次整数加法。

intrinsic 名字怎么拆

intrinsic 的名字看起来很长,但多数名字都有规律。把 _mm256_add_epi32 拆开:

text
_mm256 _ add _ epi32

可以理解为:

片段 含义
_mm256 使用 256 位向量,通常对应 AVX/AVX2
add 做加法
epi32 packed 32-bit integer,也就是“打包在一个向量里的 32 位整数”,所谓 packed,就是多个小整数挤在一个大寄存器里。

所以含义就是:对 256 位向量中的多个 32 位整数做加法。先不要急着背所有名字,重点是看到名字时能猜出大概方向:

bash
_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 汇编指令:

asm
vpaddd ymm0, ymm1, ymm2

这里的 vpaddd 是真正的汇编指令,_mm256_add_epi32 是你在 C 代码里使用的 intrinsic。入门阶段先写 intrinsic,比直接写汇编更容易控制复杂度。

最后只需要记住一条主线:C 数组 -> load 到 __m256i -> 用 intrinsic 计算 -> store 回 C 数组

让编译器自动生成 SIMD

在手写 intrinsic 之前,应该先学会让编译器自动向量化。原因很简单:很多规则清晰的循环,编译器已经可以识别并自动生成 SIMD 指令。这一节的目标是:

  • 写一个普通 C 循环
  • 用 -O3 -march=native 编译
  • 确认程序结果正确
  • 查看编译器是否自动向量化
  • 查看最终汇编里是否出现向量指令

完整代码和运行结果

c
#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;}

先编译运行:

bash
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -march=native auto_vectorization.c -o auto_vectorization

正常输出:自动向量化示例验证通过:共验证 19 个元素。这一步只回答一个问题:程序结果是否正确。性能和汇编都放到后面看。

如何查看编译器是否自动向量化?

GCC 可以输出向量化报告:

bash
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -march=native -fopt-info-vec-optimized auto_vectorization.c -o auto_vectorization

如果成功,可能看到类似信息:

bash
auto_vectorization.c:8:5: note: loop vectorizedauto_vectorization.c:18:5: note: loop vectorized

查看最终汇编

生成汇编文件:

bash
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -march=native -S -masm=intel auto_vectorization.c -o auto_vectorization.s

搜索常见的向量加法和向量加载/存储指令:vpaddd|vmovdqu|vmovdqa,可以发现如下内容:

asm
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

编译器生成的汇编代码本来就不是按“人类阅读顺序”写的。这段数组加法代码逻辑虽然很简单,但汇编顺序看起来乱,是因为编译器在做指令调度。它不是严格按下面这样生成:

  1. 先处理 c[0…7]
  2. 再处理 c[8…15]
  3. 最后处理 c[16…18]

而是会穿插执行,这么做是为了减少等待时间。CPU 从内存读数据有延迟,编译器会把一些标量尾部操作插进 SIMD 操作中间,让 CPU 执行指令集的流水线尽量忙起来。

restrict 是什么

先看这个函数参数:

c
static void add_arrays(const int32_t *restrict a, const int32_t *restrict b, int32_t *restrict c, size_t n)

restrict 解决的是一个问题:这几个指针会不会指向同一块内存。

先对比两种调用:

c
add_arrays(a, b, c, n);                  // 常见情况:三个独立数组add_arrays(data, other, data + 1, n);    // 特殊情况:a 和 c 指向同一块数组的不同位置

第二种情况里:

bash
a[0] -> data[0]      c[0] -> data[1]a[1] -> data[1]      c[1] -> data[2]a[2] -> data[2]      c[2] -> data[3]

这时 ac 的内存是重叠的。普通 C 循环是逐轮执行的:

bash
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]

如果 ac 可能重叠,SIMD 这种“先批量读取,再批量写回”的顺序就可能改变程序结果。所以没有 restrict 时,编译器必须保守。

加上 restrict 后,就是在告诉编译器:a、b、c 是三块互不重叠的内存。不用考虑 add_arrays(data, other, data + 1, n) 这种调用。

最后对比一下 constrestrict

写法 重点
const int32_t *a 不能通过 a 修改数据
int32_t *restrict c c 指向的内存不会和其他相关指针重叠

哪些循环适合 SIMD

适合 SIMD 的典型特征:

  • 大量元素执行相同操作。
  • 每次迭代彼此独立。
  • 数据在内存中连续。
  • 分支可以转换为掩码。
  • 数据类型和精度要求明确。

常见场景:

  • 图片、音频、视频处理
  • 矩阵和向量计算
  • 压缩、校验、编码和加密
  • JSON 字符扫描
  • 数据库列扫描、聚合和过滤
  • 日志批量过滤
  • 向量距离计算

不容易直接 SIMD 化的情况:

  • 本次迭代依赖上一次结果
  • 指针链表遍历
  • 每个元素执行完全不同的逻辑
  • 数据随机分散在内存中
  • 数据规模很小,准备工作占主要成本

数据布局:AoS 与 SoA

假设日志记录采用结构体数组:

c
struct Record {    int16_t status;    int16_t latency_ms;};struct Record records[n];

这种布局叫 Array of Structures,简称 AoS:

bash
status0 latency0 status1 latency1 status2 latency2 ...

其实还可以使用两个独立数组来表示:

c
int16_t status[n];int16_t latency_ms[n];

这种布局叫 Structure of Arrays,简称 SoA:

bash
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 的机器上运行

先检查:

bash
lscpu | rg -o 'avx2|avx512f'

生产代码通常需要运行时分发:

c
if 支持 AVX2 {    调用 AVX2 实现} else {    调用标量实现 }

GCC 还提供 __builtin_cpu_supports("avx2")。注意:检测逻辑本身必须运行在不强制要求 AVX2 的通用代码路径中。

5. 认为 intrinsic 一定比编译器快

编译器可能已经自动向量化,手写版本也可能因为额外加载、寄存器压力或不理想的数据布局而更慢。必须通过报告、汇编和基准共同判断。

6. 只优化指令,不优化数据布局

连续数据通常比复杂指令技巧更重要。先解决随机访问、指针追逐和不必要的数据搬运,再考虑手写 SIMD。

实战性能对比

场景如下:用户在运行时提交一条查询规则(status >= 500 && latency_ms > 100),执行器把这条规则编译成过滤函数。

数据结构如下:

c
int16_t status[n];int16_t latency_ms[n];

这个优化成立的前提是字段范围确实放得进 int16_t。本例里 status 大约是 200..799latency_ms 大约是 0..499,不会超过 16 位有符号整数范围。

这里先暂时不使用 AVX-512。AVX2 是 256 位,int16_t 是 16 位,所以一条 AVX2 指令已经可以同时处理:256 / 16 = 16 条记录

当前示例代码不直接生成机器码,而是在 filter_bench.c 里写出“JIT 最终可能生成的 AVX2 代码形态”。等后面接真实 JIT 时,目标就是生成类似这段 intrinsic 对应的机器码。

标量规则解释器目标代码

一个通用过滤器通常会先解释规则:

c
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 个列:

c
__m256i vstatus = _mm256_loadu_si256((const __m256i *)(const void *)(status + i));__m256i vlatency = _mm256_loadu_si256((const __m256i *)(const void *)(latency_ms + i));

然后把两个条件分别变成两个向量掩码:

c
__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)。最后把两个条件合并:

c
__m256i matched = _mm256_and_si256(status_ok, latency_ok);

这一步相当于:status_ok && latency_ok。只不过它是在 16 个 lane 上同时完成的。

从向量掩码得到匹配数量

SIMD 16 位整数比较的每个 lane 通常不是返回普通的 01,而是:

  • 条件为假: 0x0000
  • 条件为真: 0xffff

如果 16 条记录的最终匹配结果里有若干个真值,可以用下面的代码把每个字节的最高位提取成掩码:

c
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 个二进制位。

完整代码

c
#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;}

测试结果

编译命令如下:

bash
gcc -std=c11 -O3 -Wall -Wextra -Wpedantic -mavx2 filter_bench.c -o filter_bench

程序会:

  1. 生成 statuslatency_ms 两列测试数据。
  2. 用标量规则解释器计算匹配数量。
  3. 用 JIT 目标形态的 AVX2 函数计算匹配数量。
  4. 先检查两种实现的结果是否一致。
  5. 报告各自的最佳耗时。
bash
# ./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 ms4.8 ms 之间,整体能快一倍以上。

这次提速主要来自三点:

  1. 只保留查询真正需要的 statuslatency_ms 两列,减少了内存读取。
  2. 把数据类型从 int32_t 压缩成 int16_t,让 AVX2 一次可以处理 16 条记录。
  3. 过滤条件固定后,AVX2 版本可以直接比较阈值,不需要在循环里解释规则。

不同次运行的百分比会有波动是正常的。基准测试会受到 CPU 调度、缓存状态、频率变化和虚拟机环境影响,所以不要只盯着某一次的数字。更重要的是看两个结论:结果必须一致,AVX2 版本在多次运行中稳定快于标量版本。

虽然 AVX2 一次处理 16 个 int16_t lane,但程序仍然要从内存读取两列数据,还要做比较、按位与、生成掩码和统计匹配数量。真实加速通常会小于理论 lane 数。

练习

练习 1:向量乘法

把数组加法改为:

c
c[i] = a[i] * b[i];

查找并使用 32 位整数低位乘法 intrinsic。提示:_mm256_mullo_epi32

练习 2:范围过滤

把过滤规则改为:

c
status >= 400 && status < 600

分别生成两个比较掩码,再用 AND 合并。

练习 3:输出匹配位置

当前程序只统计匹配数量。尝试把匹配元素的索引写入输出数组。

这个问题比计数更难,因为每 8 个元素的匹配数量不固定。先使用提取 mask 后逐位检查的简单实现,不必立刻追求完全无分支。

练习 4:运行时分发

构建两个函数:

c
filter_scalar(...)filter_avx2(...)

启动时使用 __builtin_cpu_supports("avx2") 选择实现,并通过函数指针保存。

练习 5:连接 JIT

让 JIT 生成的函数接收数组指针和元素数量。第一版可以只调用已经编译好的filter_avx2,随后再尝试真正生成 AVX2 指令。

知识图谱

text
SIMD├── 数据│   ├── lane│   ├── 连续内存│   ├── AoS / SoA│   └── 对齐├── 指令│   ├── load / store│   ├── 算术│   ├── 比较│   ├── mask│   └── reduction├── 编译方式│   ├── 自动向量化│   ├── intrinsic│   ├── 汇编│   └── JIT 生成├── 正确性│   ├── 尾部│   ├── 越界│   ├── 对齐│   ├── 溢出│   └── CPU 特性检测└── 性能    ├── 缓存    ├── 内存带宽    ├── 分支    ├── 数据布局    └── 基准测试

参考资料

Comments

评论区将在滚动到这里时加载。