命令速查表

• ib_ → Benchmark：性能测试工具
• ibv_ → Verbs：编程与设备信息
• ib → Fabric：网络拓扑与诊断
• opensm → 管理器：子网控制核心

网络拓扑与状态诊断（ib 前缀）

RDMA 编程与设备信息（ibv_ 前缀）

性能测试与基准评估（ib_ 前缀）

管理与配置辅助工具

物理硬件

• HCA 可能有多个物理端口 (port)：每个 port 有一个 LID 。
• 每个 port 可能有多个通道 (lane) ：用 1x, 2x, 4x, 8x 等表示。
• 总速率：总速率 = 每个 port 的 lane 数 × 每个 lane 的速率 × port 数量。
• port 和 lane 是有物理硬件决定的。

部署

• 通道是物理层的能力

  • 每个 InfiniBand 或 PCIe 接口由若干个 lane 组成，每个 lane 是一对差分信号线（发送 + 接收）。
  • 比如：
    • x1 表示 1 个 lane → 最低带宽
    • x4 表示 4 个 lane → 常见于 EDR（100G）
    • x8 表示 8 个 lane → HDR/NDR（200G/400G）
    • x12 是更高端的聚合方式 → 超大带宽（如 600G）

• “部署”决定你能否用满这些通道

  • 如果你部署的是 x4 的 HCA + x4 的交换机 + x4 的线缆，你就能跑满 4 个 lane。
  • 如果你部署的是 x8 的 HCA，但只用了 x4 的线缆，你只能跑 4 个 lane，剩下的通道闲置。
  • 如果你部署的是 x12 的交换机，但 HCA 只有 x4，你也只能用其中一部分。

举个例子：HDR 200G 的部署选择

只有在部署中三者都匹配时，才能真正跑满所有通道。

程序员关注的层面

带宽测试

理论带宽

1

lspci | grep -i infiniband

然后通过阅读产品说明书，确定理论带宽。

协商带宽

实际部署的速率，可能与 HCA 、线缆、交换机都相关，所以实际会有一个协商速率。

1

ibstat

实际带宽测试

发送速率：

1
2
3
4
5

# Server 端（等待连接）
ib_send_bw -d mlx5_0 -a

# Client 端（发起连接）
ib_send_bw -d mlx5_0 -a <server_ip>

令所有 C/C++ 程序员头疼的事，莫过于环境的搭建和兼容。这往往发生在链接阶段，分为两种：

• 编译时链接
• 运行时链接

读完本文，你将能解决大部分的环境问题。

运行时 so 的加载顺序

动态链接器（Dynamic Linker）

以上的 so 的加载顺序，是由 动态链接器（Dynamic Linker） 完成的。

每个 ELF 可执行文件在头部指定了它的动态链接器路径：

1
2

$ readelf -l your_program | grep interpreter
[Requesting program interpreter: /lib64/ld-linux-x86-64.so.2]

这个链接器负责在程序启动时加载所需的动态库，包括 libc.so.6（glibc 的主库）。

注意：ld.so（这是一个简称）和 ld 完全不是一个东西，ld 是一个可执行程序，在编译期使用，见下
文编译时链接器（ld）。

• ld.so 既是 so 也是可执行文件：

1
2

$ file /lib/x86_64-linux-gnu/ld-linux-x86-64.so.2
/lib/x86_64-linux-gnu/ld-linux-x86-64.so.2: ELF 64-bit LSB shared object, x86-64, version 1 (GNU/Linux), dynamically linked, BuildID[sha1]=e4de036b19e4768e7591b596c4be9f9015f2d28a, stripped

普通程序的 ELF 文件中有一个 PT_INTERP 段，指定了这个链接器的路径。

1
2
3
4
5

$ file /usr/bin/ls
/usr/bin/ls: ELF 64-bit LSB pie executable, x86-64, version 1 (SYSV), dynamically linked, interpreter /lib64/ld-linux-x86-64.so.2, BuildID[sha1]=36b86f957a1be53733633d184c3a3354f3fc7b12, for GNU/Linux 3.2.0, stripped

$ file /lib/x86_64-linux-gnu/libc.so.6
/lib/x86_64-linux-gnu/libc.so.6: ELF 64-bit LSB shared object, x86-64, version 1 (GNU/Linux), dynamically linked, interpreter /lib64/ld-linux-x86-64.so.2, BuildID[sha1]=cd410b710f0f094c6832edd95931006d883af48e, for GNU/Linux 3.2.0, stripped

• 动态链接器 ld.so 本身也依赖共享库（如 libc.so.6），但它又是负责加载共享库的工具

动态链接器是一个“自举”程序：

• 它是由 Linux 内核直接加载并执行，不依赖其他库来启动。
• 它的启动代码是 静态编译的，也就是说，它的最小启动逻辑不依赖 libc.so.6。
• 在启动后，它才会去加载 libc.so.6 和其他 .so 文件，完成符号解析和初始化。
• 有趣的是，libc.so.6 又依赖 ld.so 来加载。

1
2
3
4
5
6
7
8
9

你运行 ./myapp
↓
内核读取 ELF → 找到 PT_INTERP 段 → 加载 ld.so
↓
ld.so 启动（靠自身静态代码）
↓
ld.so 加载 libc.so.6、libstdc++.so.6 等共享库
↓
ld.so 跳转到 myapp 的入口地址（main）

如果 libc.so.6 丢失怎么办？

有开发者分享过真实案例，当系统误删了 libc.so.6 ，几乎所有命令都无法运行。但你仍然可以：

1

/lib64/ld-linux-x86-64.so.2 /bin/ln -s /lib64/libc-2.33.so /lib64/libc.so.6

这利用了 ld.so 的自举能力，手动加载 libc 并执行命令，从而恢复系统。

ldd 检查

ldd (List Dynamic Dependencies) 是一个 shell 脚本，用于查看一个可执行文件或共享库所依赖的动态链接
库的命令。

它的思路是：模拟 运行时链接器 的行为。

1. 设置环境变量来触发动态链接器 (ld-linux.so) 输出依赖信息，但是不真正执行程序。如：

   • LD_TRACE_LOADED_OBJECTS=1
   • LD_WARN, LD_BIND_NOW, LD_VERBOSE 等

2. 调用动态链接器查找 so .

局限性：

• 使用 dlopen() 动态加载库，ldd 是无法检测到的

补充工具

• readelf -d binary | grep -i rpath：查看 ELF 文件中的 RPATH 或 RUNPATH
• ldconfig -p：查看系统缓存中有哪些库
• strace -e openat ./your_program：查看运行时实际打开了哪些库文件
• LD_DEBUG=libs ./your_program：调试动态库加载过程

编译时 so 的查找顺序

GCC 查找 glibc 的路径顺序：

🧠 编译阶段（查找头文件）

• -I 和 -isystem 都是命令行参数，但 -isystem 会将路径标记为“系统路径”，避免某些警告。
• 环境变量如 CPATH 和 C_INCLUDE_PATH 是在没有显式参数时的补充手段。
• 默认路径 /usr/include 是 GCC 安装时配置的，可以通过 gcc -xc -E -v - 查看完整搜索路径。

🔗 链接阶段（查找库文件）

说明：

在编译安装 GCC 时，通常会使用 configure 脚本来指定安装路径：

1

./configure --prefix=/custom/gcc --libdir=/custom/gcc/lib64

当你使用这个 GCC 编译器时，它会自动使用这些路径来查找头文件和库文件。例如：

• 查找头文件时，会优先使用 /custom/gcc/include
• 查找库文件时，会优先使用 /custom/gcc/lib64
• 链接器（如 ld）也会被 GCC 告知去这些路径找库

这些路径是 编译器内部写死的默认值，你可以通过以下命令查看：

1
2
3
4
5
6

gcc -print-search-dirs
# 或
ld --verbose | grep SEARCH_DIR
# 查看：g++ 实际链接的是哪个版本的库；库文件位于哪个目录；系统是否安装了对应库
g++ -print-file-name=libxxx.so
g++ -print-file-name=libyyy.a

输出示例：

1
2
3

install: /custom/gcc/lib/gcc/x86_64-linux-gnu/12.2.0/
programs: /custom/gcc/libexec/gcc/x86_64-linux-gnu/12.2.0/:...
libraries: /custom/gcc/lib/gcc/x86_64-linux-gnu/12.2.0/:/custom/gcc/lib64/:..

gcc -v 会显示详细的编译和链接过程，包括实际使用了哪些路径和库文件。

1

gcc -v hello.c -o hello

使用 --sysroot 指定根路径

1

gcc --sysroot=/path/to/custom/root hello.c -o hello

编译时链接器（ld）

编译时链接器（ld） 和 运行时动态链接器 不是同一个东西：

glibc

glibc 是特殊的 so ，它封装了系统调用，所以 Linux 系统本身也依赖它。

C 语言允许编译时链接和运行时链接分开，所以常常遇到这样奇怪的问题：

• 在本机编译、链接，在本机不能运行：

1

./myapp: /lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.33' not found (required by ./myapp)

这是因为 gcc 在编译时指定的 glibc 和系统运行时的 glibc 不一样：

1
2
3
4
5
6
7
8
9

$ strings /lib/x86_64-linux-gnu/libc.so.6 | grep GLIBC_
GLIBC_2.2.5
...
GLIBC_2.31

$ strings ./myapp | grep GLIBCXX_
GLIBC_2.3
...
GLIBCXX_2.33

/lib/x86_64-linux-gnu/libc.so.6 中确实没有 GLIBC_2.33 的符号。

如果你的系统中有多个 glibc ，那么这种问题可以通过上面“运行时 so 的加载顺序”，使用环境变量调整。

但是，这种对 glibc 的依赖可能是间接的：

1
2
3
4
5
6
7
8
9
10

$ ls /path/to/gcc/v14.2.0/lib64 | grep libstdc++.so
libstdc++.so
libstdc++.so.6

$ ldd /path/to/gcc/v14.2.0/lib64/libstdc++.so.6
        linux-vdso.so.1 (0x00007ffd965f4000)
        libm.so.6 => /usr/lib64/libm.so.6 (0x0000147a213a9000)
        libc.so.6 => /usr/lib64/libc.so.6 (0x0000147a20fe4000)
        /lib64/ld-linux-x86-64.so.2 (0x0000147a219ea000)
        libgcc_s.so.1 => /path/to/gcc/v14.2.0/lib64/libgcc_s.so.1 (0x0000147a21be4000)

此时，如果你的编译和运行的机器不是同一台（意味着运行时 /usr/lib64/libc.so.6 文件和编译期不一致），这
时候就基本没有没有办法了。理论上，你可以下载一个与编译期相同的 libc.so.6 ，但是你无法保证系统的动态
链接器 /lib64/ld-linux-x86-64.so.2 与所下载的 libc.so.6 兼容。而 /lib64/ld-linux-x86-64.so.2 路
径是无法通过环境变量修改的，因为它是内核在程序启动时直接加载的，它的路径是硬编码在 ELF 可执行文件中
的。

所以最好的方式，是在编译时就确认 GCC 的 glibc 和系统运行时一致。

检查当前 GCC 使用的 glibc 路径

1
2
3

$ gcc --print-file-name=libc.so.6
$ g++ --print-file-name=libstdc++.so.6
$ gcc -print-search-dirs

示例：

1
2

$ /path/to/gcc/v14.2.0/bin/g++ --print-file-name=libstdc++.so.6
/path/toData/gcc-v14.2.0/x86_64/lib64/libstdc++.so.6

确认当前系统的 glibc 和 libstdc++ 版本

✅ 查看 glibc 版本（即 libc.so）

1
2
3

ldd --version
# 或者
getconf GNU_LIBC_VERSION

✅ 查看 libstdc++ 版本（C++ 标准库）

1

strings /usr/lib*/libstdc++.so.6 | grep GLIBCXX

你会看到类似：

1
2

GLIBCXX_3.4.21
GLIBCXX_3.4.26

这些是你系统支持的 C++ ABI 版本。

系统默认路径

由 /etc/ld.so.conf 和 ldconfig 决定

通常包括：

• /lib64/
• /usr/lib64/
• /lib/x86_64-linux-gnu/（Debian/Ubuntu）

你可以运行：

1

ldconfig -p | grep libc.so.6

来查看系统当前可用的 libc.so.6 版本和路径。

Hazard Pointers（危险指针）

✅ 作用：保护正在访问的指针，防止它被其他线程释放。

每个线程维护一组“hazard slots”，用于声明当前正在使用的节点。

在释放节点前，必须检查是否有线程将其标记为 hazard。如果有，就延迟释放。

📌 优点：精细控制：每个指针都可以独立保护。

快速释放：一旦没有线程标记该节点，就可以立即释放。

⚠️ 缺点：每个线程需要维护额外的指针集合。

实现复杂度较高，尤其在高并发场景下。

Epoch-Based Reclamation（基于世代的回收）

✅ 作用：将所有线程的操作划分到一个“epoch”（时间段）中。

每个线程在进入临界区时记录当前 epoch，退出时清除。

被删除的节点会挂到当前 epoch 的垃圾列表中，只有当所有线程都离开该 epoch，才安全释放这些节点。

📌 优点：实现简单，性能好，适合高吞吐量场景。

不需要每个指针都单独保护。

⚠️ 缺点：回收延迟：必须等待所有线程退出当前 epoch。

对长时间运行的线程敏感，可能阻止垃圾回收。

🧠 为什么需要它们？

在 lock-free 数据结构中，节点可能被多个线程同时访问或修改。没有互斥锁保护时，直接释放内存可能导致：

• 悬挂指针：其他线程还在访问被释放的节点。
• ABA 问题：节点被释放后地址复用，导致逻辑错误。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

// EBR: Epoch-Based Reclamation
#include <atomic>
#include <iostream>
#include <mutex>
#include <thread>
#include <unordered_map>
#include <vector>

constexpr int MAX_THREADS = 4;

std::atomic<int> global_epoch{0};
std::atomic<bool> active[MAX_THREADS];
std::atomic<int> local_epoch[MAX_THREADS];

std::mutex retire_mutex;
std::unordered_map<int, std::vector<int*>> retire_list[3];  // 每个 epoch 的垃圾列表

void enter_critical(int tid) {
    active[tid].store(true, std::memory_order_relaxed);
    local_epoch[tid].store(global_epoch.load(std::memory_order_relaxed),
                           std::memory_order_relaxed);
}

void exit_critical(int tid) {
    active[tid].store(false, std::memory_order_relaxed);
}

void retire_node(int tid, int* node) {
    std::lock_guard<std::mutex> lock(retire_mutex);
    int epoch = global_epoch.load(std::memory_order_relaxed);
    retire_list[epoch][tid].push_back(node);
}

void try_advance_epoch() {
    int current = global_epoch.load(std::memory_order_relaxed);
    for (int i = 0; i < MAX_THREADS; ++i) {
        if (active[i].load(std::memory_order_relaxed) &&
            local_epoch[i].load(std::memory_order_relaxed) != current) {
            return;  // 有线程还在旧 epoch，不能推进
        }
    }

    int next_epoch = (current + 1) % 3;
    global_epoch.store(next_epoch, std::memory_order_relaxed);

    // 回收两代前的垃圾
    int reclaim_epoch = (next_epoch + 1) % 3;
    std::lock_guard<std::mutex> lock(retire_mutex);
    for (auto& [tid, nodes] : retire_list[reclaim_epoch]) {
        for (auto node : nodes) {
            delete node;
        }
        nodes.clear();
    }
}

前言

源码：https://github.com/cameron314/concurrentqueue

原文
：Solving the ABA Problem for Lock-Free Free Lists

共识

• CAS 足以保证线程安全；
• 我们要做的是消除 CAS 的 ABA 问题。

常见的解决方案

Tagged / versioned 指针

每次修改 head 的同时也变更一个版本号（tag/version），把 (pointer, version) 当作 CAS 的对象。即使指针
回到同一个 A，只要版本号不同，CAS 就会失败，防止 ABA。

缺点包括：在某些机器上需要双字宽度（两个 word）CAS，如果 硬件 / 编译器 / 平台 不支持就只能 模拟 / 锁
住；或者要压缩 pointer 或 tag 大小，可能限制可管理的节点数量 / 地址空间。

LL/SC（Load-Linked / Store-Conditional）原语

在支持 LL/SC 的架构上，它自然就可以阻止 ABA：因为 store-conditional 会检测在 load 和 store 之间地址
是否被“写过”，哪怕写了后来改回原来的值也不行。缺点是很多主流架构（尤其 x86）不支持或者支持有限。

作者的方法：以引用计数＋“should be on freelist”标志（标志 + 引用计数）

作者提出了一个适用于 free list 的通用方法来避免 ABA，且保持 lock‐free 特性。

提示：该代码应该从 try_get() 开始阅读，然后回到 add() ，这样才能理解设计引用计数的意图，否则很容
易迷惑。

• 引用计数是状态标志：
  • 表示当前有多少线程正在操作节点；
  • 空闲链表本身也持有一个对节点的引用计数。
• 任何线程尝试摘取空闲链表的节点（总是从头部开始）时（try_get()）：
  • 必须先增加 head 的引用计数（不同线程可能并发地将节点的引用计数增加到某个值）；→ (3)
  • next = head->next；
  • 然后 CAS 竞争，尝试将 freeListHead 调整为 next；
    • 如果 CAS 竞争成功，则成功将 freeListHead 调整为 next，该节点被摘下；引用计数减 2（自己增加 的
      引用计数 + 链表本身持有的引用计数），我们称该线程为线程 1。→ (1)
    • 如果 CAS 竞争失败，则回退：将自己所增加的引用计数减去；用 CAS 返回的新的 head 重试（下一次获取
      新的 head 节点），这是线程 2；→ (2)
    • 注意：这两步之间以及每一步自身内，对引用计数的操作都有中间状态，但是此时引用计数必然大于 0 （
      重点！ ）。
• ABA 问题：如果成功摘取节点的线程 1完成节点的使用，将节点又添加回空闲链表（add()）：
  • 它会发现某个线程正在第 (2) 步的回退过程中（还没有将自身增加的引用计数减回去），所以此时引用计
    数还是大于 0（也就是说线程 1使用该节点的整个过程，该节点的引用计数都是大于 0 的，但是不影响）
    。
  • 此时它可以选择自旋地等待引用计数降为 0 （即所有之前在 try_get() 中 CAS 竞争失败的线程（
    如线程 2）回退成功）。因为只有引用计数降为 0 了，才能说明自己是唯一一个尝试在链表上操作该节点
    的线程。但是这从技术上说，就是一个 lock。
  • 作者设计了一个更聪明的办法：add() 设置一个 SHOULD_BE_ON_FREELIST 标志，然后直接放弃
    （add()的任务已经完成了）。
  • 让处于回退中的 线程 2，在回退结束的时候，检查 SHOULD_BE_ON_FREELIST 标志，如果有该标志，并且
    自己就是最后一个操作该节点的线程的话（即此时的引用计数已经降为 0 了），就帮助 add() 把该节点添
    加到 free list。

引用计数

现在我们回过头来看引用计数的作用：

• 如果 refs > 0：

  • (1) 要么该节点在链表上；
  • (2) 要么有线程正在操作该节点（准确来讲，是在 try_get()）；
  • (3) 要么以上两种情况同时存在。
  • 注：(1)(2) 可以只存在一种，见上面的 ABA 问题的描述。

• 如果 refs == 0：

  • 节点已经被从空闲链表上取下；
  • 并且其他线程都已经从 try_get() 成功回退了。

如果第三个线程，与线程 1、线程 2一起进入上面的 try_get() 竞争，但是当线程 3即将执行步
骤(3)时，发现 refs == 0，它就不能够再增加 head 的引用计数了，因为节点已经被成功取下。否则当它成
功增加引用计数，再去拿取 head->next 的时候，是未定义行为。

正是基于此，在步骤(3)增加引用计数的时候，我们需要判断 refs == 0 ? 和使用 CAS 增加引用计数，如果
不符合预期，则放弃增加，进入下一轮重试。

1
2
3
4

if ((refs & REFS_MASK) == 0 || !head->freeListRefs.compare_exchange_strong(refs, refs + 1, std::memory_order_acquire)) {
	head = freeListHead.load(std::memory_order_acquire);
	continue;
}

ABA 问题

在典型的无锁栈/队列里，指针会被反复复用（比如 pop 出再 push 回去）。如果只用 compare_exchange 比较指
针，那么：

• T1 读到 A
• T2 把 A → B → A 改一圈
• T1 CAS 时看到还是 A，以为结构没变，但其实已经发生变化

这就是 ABA 问题。

解决方案

版本标记指针方案

‌ 核心思想 ‌：为每个指针附加一个版本号标记，每次修改时递增版本号。

‌ 实现要点 ‌：

• 需要双字 CAS(DCAS)硬件支持
• 每次修改头指针时递增标记值
• 即使指针地址相同，标记值不同也会使 CAS 失败
• 适用于支持 DCAS 的平台(如某些 ARM 架构)
• 在非 DCAS 平台也可用，前提是将 data + version 放入一个机器字中，以维持操作的原子性

‌ 限制 ‌：

• 在某些机器上需要双字宽度（两个 word）CAS；
• 如果硬件／编译器／平台不支持就只能模拟／锁住；或者要压缩 pointer 或 tag 大小，可能限制可管理的节点数量／地址空间。

引用计数方案

‌ 核心思想 ‌：为每个节点维护引用计数，防止节点在被使用时被重新添加到列表。

‌ 关键实现细节 ‌：

每个节点包含 freeListRefs(引用计数)和 freeListNext(下一节点指针) 引用计数高位用作 “应返回自由列表”
标志位 try_get()操作前确保引用计数不为零 add()操作使用原子操作管理引用计数和标志位 ‌ 优势 ‌：

完全通用的解决方案，不依赖特定硬件特性保持真正的锁无关性质正确处理并发场景下的各种竞争条件

Hazard Pointer（危险指针）

思路：每个线程在访问共享指针之前，把自己正在访问的指针写到一个全局可见的“hazard pointer”里。

作用：其他线程在想要回收这个节点内存时，必须检查所有线程的 hazard pointers，如果发现有人还在用这个节
点，就不能释放。

优点：简单直接，内存可以安全回收。

缺点：维护 hazard pointers 有一定开销，每次回收都要检查所有线程。

Epoch-Based Reclamation（基于世代的回收，简称 Epoch GC）

思路：把时间切分成 epoch（世代）。线程进入临界区时声明自己在某个 epoch。当一个节点被删除后，先放到一
个“延迟回收队列”，等到所有线程都离开这个 epoch 之后，才能真正释放这些节点。

作用：保证没有线程会在旧 epoch 中访问到已经释放的节点。

优点：比 hazard pointer 更高效（不用逐个检查指针）。

缺点：需要所有线程都周期性地报告自己活跃的 epoch，否则内存可能迟迟回收不了。

🚩 为什么会和 ABA 有关？

像 Michael-Scott 队列这种链表结构，节点被 pop 出队后地址可能被重用。如果没有安全的内存回收，另一个线
程可能 CAS 成功指向了一个“已经被释放并重用的地址”，这就是 ABA 的根源。所以 hazard pointer 或 epoch
GC 是在链表队列里用来避免这种 悬空引用 + ABA 的。

而 moodycamel:: ConcurrentQueue 因为用的是 环形 buffer + sequence number，节点不会反复 malloc/free，
所以根本就不需要 hazard pointer 或 epoch GC。

TBB 的特殊性

本文的法则是以线程为调度单位的。

如果你使用的是 TBB ，那么请将 “线程” 对应为 “任务”，而将本文的“任务” 对应为“载荷”（ payloads）。

因为 TBB 是以任务为调度单位的：

• 每个 “任务” 是并发运行的最小单位，必须保证数据独立或线程安全。
• TBB 采用 “任务窃取” 算法来保证线程的复杂均衡，所以若干任务可能被同一线程或不同线程运行。

经验法则

以 tasksPerThread 按需分配线程

• tasksPerThread：均匀性。如果每个线程分配的任务数不均匀，那么任务数最多的线程就会成为瓶颈。
• 按需分配线程：如果任务可以很快完成，那么没有必要开启过多的线程，否则调度开销也不可小觑。

以空间换时间

为了任务能并发运行，进行任务分隔时，必须尽可能减少数据共享。
所以不要吝惜空间，为每个任务单独开辟内存（不论是为输入还是输出目的）都是值得且必要的。

长临界区使用 lock + 条件变量

如果一个任务的临界区比较大，意味着该任务执行时，其他线程在短时间内无法进入临界区。
与其让这些线程忙等，不如释放 CPU 进入阻塞 / 休眠。

前言

这是阅读 Cameron Desrochers 的
A Fast General Purpose Lock-Free Queue for C++
源码的笔记。

• Detailed Design of a Lock-Free Queue
• Solving the ABA Problem for Lock-Free Free Lists

系统概览

MPMC 队列由一系列 SPMC 队列组成。消费者使用启发式 (heuristic) 来决定消费哪个 SPMC 队列。允许批量入列
和出列，只需要很小的额外开销。

producer 需要一些 thread-local 数据； consumer 也可以用一些可选的 thread-local 数据来加速；这些
thread-local 数据可以与用户分配的 tokens 关联；如果用户没有为生产者提供 tokens ，则使用无锁哈希表（
以当前线程 ID 为键）来查找线程本地生产者队列：每个 SPMC 队列都使用一个预分配的 token （或隐式分配的
token，如果没有提供的话）来创建。由于 token 包含相当于线程特定的数据，因此它们不应该同时在多个线程中
使用（尽管可以将 token 的所有权转移给另一个线程；特别是，这允许在线程池任务中使用令牌，即使运行任务
的线程在中途发生变化）。

所有生产者队列都以无锁链表的形式连接在一起。当显式生产者不再有元素被添加时（即其令牌被销毁），它会被
标记为与任何生产者都无关联，但它会保留在链表中，且其内存不会被释放；下一个新生产者会重用旧生产者的内
存（这样，无锁生产者列表就只能添加）。隐式生产者永远不会被销毁（直到高层队列本身被销毁），因为无法知
道给定线程是否已完成对数据结构的使用。需要注意的是，最坏情况下的出队速度取决于生产者队列的数量，即使
它们都为空。

显式生产者队列和隐式生产者队列的生命周期存在根本区别：显式生产者队列的生产生命周期有限，与令牌的生命
周期绑定；而隐式生产者队列的生产生命周期不受限制，且与高级队列本身的生命周期相同。因此，为了最大化速
度和内存利用率，我们使用了两种略有不同的 SPMC 算法。通常，显式生产者队列设计得更快，占用的内存也更多
；而隐式生产者队列设计得更慢，但会将更多内存回收到高级队列的全局池中。为了获得最佳速度，请始终使用显
式令牌（除非您觉得它太不方便）。

任何分配的内存只有在高级队列被销毁时才会释放（尽管存在一些重用机制）。内存分配可以预先完成，如果内存
不足，操作就会失败（而不是分配更多内存）。如果需要，用户可以覆盖各种默认大小参数（以及队列使用的内存
分配函数）。

Full API (pseudocode)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# Allocates more memory if necessary
enqueue(item) : bool
enqueue(prod_token, item) : bool
enqueue_bulk(item_first, count) : bool
enqueue_bulk(prod_token, item_first, count) : bool

# Fails if not enough memory to enqueue
try_enqueue(item) : bool
try_enqueue(prod_token, item) : bool
try_enqueue_bulk(item_first, count) : bool
try_enqueue_bulk(prod_token, item_first, count) : bool

# Attempts to dequeue from the queue (never allocates)
try_dequeue(item&) : bool
try_dequeue(cons_token, item&) : bool
try_dequeue_bulk(item_first, max) : size_t
try_dequeue_bulk(cons_token, item_first, max) : size_t

# If you happen to know which producer you want to dequeue from
try_dequeue_from_producer(prod_token, item&) : bool
try_dequeue_bulk_from_producer(prod_token, item_first, max) : size_t

# A not-necessarily-accurate count of the total number of elements
size_approx() : size_t

Producer Queue (SPMC) Design

隐式和显式版本的共享设计

生产者队列由块组成（显式和隐式生产者队列使用相同的块对象，以实现更好的内存共享）。初始状态下，它没有
块。每个块可以容纳固定数量的元素（所有块的容量相同，均为 2 的幂）。此外，块包含一个标志，指示已填充
的槽位是否已被完全消耗（显式版本使用此标志来判断块何时为空），以及一个原子计数器，用于计数已完全出队
的元素数量（隐式版本使用此标志来判断块何时为空）。

为了实现无锁操作，生产者队列可以被认为是一个抽象的无限数组。尾部索引指示生产者下一个可用的槽位；它同
时也是已入队元素数量的两倍（ 入队计数 (enqueue count) ）。尾部索引仅由生产者写入，并且始终递增（除
非溢出并回绕，但就我们的目的而言，这种情况仍被视为“递增”）。由于只有一个线程在更新相关变量，因此生产
一个元素的过程非常简单。头索引指示下一个可以被消费的元素。头索引由消费者原子地递增，可能并发进行。为
了防止头索引达到/超过感知到的尾部索引，我们使用了一个额外的原子计数器： 出队计数 (dequeue count)
。出队计数是乐观的，即当消费者推测有元素需要出队时，它会递增。如果出队计数在递增后的值小于入队计数（
尾部），则保证至少有一个元素要出队（即使考虑到并发性），并且可以安全地递增头部索引，因为知道之后它会
小于尾部索引。另一方面，如果出队计数在递增后超过（或等于）尾部，则出队操作失败，并且出队计数在逻辑上
会递减（以使其最终与入队计数保持一致）：这可以通过直接递减出队计数来实现，但是（为了增加并行性并使所
有相关变量单调递增），改为递增出队过量提交计数器 (dequeue overcommit counter)。

1

出队计数的逻辑值 = 出队计数变量 - 出队过量提交值

在消费时，一旦如上所述确定了有效索引，仍然需要将其映射到一个块以及该块中的偏移量；为此会使用某种索引
数据结构（具体使用哪种结构取决于它是隐式队列还是显式队列）。最后，可以将元素移出，并更新某种状态，以
便最终知道该块何时完全消费。下文将分别在隐式和显式队列的各个部分中对这些机制进行完整描述。

如前所述，尾部和头部的索引/计数最终会溢出。这是预料之中的，并且已被考虑在内。因此，索引/计数被视为存
在于一个与最大整数值大小相同的圆上（类似于 360 度的圆，其中 359 在 1 之前）。为了检查一个索引/计数（
例如 a）是否位于另一个索引/计数（例如 b）之前（即逻辑小于），我们必须确定 a 是否沿着圆上的顺时针圆弧
更接近 b。使用以下"环形小于"算法（32 位版本）：a < b 变为 a - b > (1U << 31U)。a <= b 变为
a - b - 1ULL > (1ULL << 31ULL)。请注意，环形减法“仅适用于”普通无符号整数（假设为二进制补码）。需要
注意的是，尾部索引的增量不会超过头部索引（这会破坏队列）。请注意，尽管如此，从技术上讲仍然存在竞争条
件，即消费者（或生产者）看到的索引值过于陈旧，几乎比当前值落后一整圈（甚至更多！），从而导致队列的内
部状态损坏。但在实践中，这不是问题，因为遍历 2^31 个值（对于 32 位索引类型）需要一段时间，而其他核心
到那时会看到更新的值。实际上，许多无锁算法都基于相关的标签指针习语（tag-pointer idiom），其中前 16
位用于重复递增的标签，后 16 位用于指针值；这依赖于类似的假设，即一个核心不能将标签递增超过 2^15 次，
而其他核心却不知道。尽管如此，队列的默认索引类型是 64 位宽（如果 16 位看起来就足够了，那么理论上应该
可以避免任何潜在的竞争）。

内存分配失败也会得到妥善处理，不会损坏队列（只会报告失败）。此外，队列元素本身在操作时也应确保不会抛
出异常。

Block Pools

有两种不同的块池可供使用：首先，有一个初始的预分配块数组。一旦使用完毕，该池将永远保持为空。这简化了
其无等待（wait-free）实现，只需一条 fetch-and-add 原子指令（用于获取空闲块的下一个索引）并进行检查（
以确保该索引在范围内）。其次，有一个无锁（但非无等待）的全局空闲列表（“全局”是指对高级队列而言是全局
的），其中包含已用完且可重复使用的块，该列表实现为一个无锁单链表：头指针最初指向空（null）。要将块添
加到空闲列表，需要将块的下一个指针设置为头指针，然后使用比较并交换 (CAS) 更新头指针，使其指向该块，
前提是头指针未发生更改；如果发生更改，则重复该过程（这是一个经典的无锁 CAS 循环设计模式）。要从空闲
列表中移除一个块，可以使用类似的算法：读取头部块的下一个指针，然后将头部设置为该下一个指针（使用
CAS），前提是在此期间头部块没有发生变化。为了避免 ABA 问题，每个块都有一个引用计数，在执行 CAS 移除
块之前会递增，之后会递减；如果在块的引用计数大于 0 的情况下尝试将其重新添加到空闲列表中，则会设置一
个标志，指示该块应该在空闲列表中，并且下一个线程在完成最后一个引用的持有后会检查此标志，并将该块添加
到列表中（这种方法有效，因为我们不关心顺序）。我
在另一篇博文中更
详细地描述了这个无锁空闲列表的具体设计和实现。当生产者队列需要新块时，它首先检查初始块池，然后检查全
局空闲列表，只有当它在那里找不到空闲块时，它才会在堆上分配一个新块（如果不允许内存分配，则失败）。

基准测试

Ticket System

BlockQueue（只用分块）：使用分块内存布局，但不使用 ticket 分发机制。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

constexpr size_t BLOCK_SIZE = 64;
constexpr int N = 1'000'000;
constexpr int NUM_PRODUCERS = 8;
constexpr int NUM_CONSUMERS = 8;
constexpr int ITEMS_PER_PRODUCER = N / NUM_PRODUCERS;

struct Slot {
    std::atomic<bool> ready;
    int data;
};

struct Block {
    Slot slots[BLOCK_SIZE];
};

// BlockQueue: 分块但无 ticket
class BlockQueue {
public:
    BlockQueue() : head(0), tail(0) {
    }

    void enqueue(int value) {
        size_t index = head.fetch_add(1) % BLOCK_SIZE;
        block.slots[index].data = value;
        block.slots[index].ready.store(true, std::memory_order_release);
    }

    bool try_dequeue(int& value) {
        size_t index = tail.fetch_add(1) % BLOCK_SIZE;
        if (!block.slots[index].ready.load(std::memory_order_acquire))
            return false;
        value = block.slots[index].data;
        return true;
    }

private:
    std::atomic<size_t> head;
    std::atomic<size_t> tail;
    Block block;
};

TicketQueue（分块 + ticket）：模拟 moodycamel 的 ticket 分发方式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// TicketQueue: 分块 + ticket
class TicketQueue {
public:
    TicketQueue() : head(0), tail(0) {
    }

    void enqueue(int value) {
        size_t ticket = head.fetch_add(1);
        size_t index = ticket % BLOCK_SIZE;
        block.slots[index].data = value;
        block.slots[index].ready.store(true, std::memory_order_release);
    }

    bool try_dequeue(int& value) {
        size_t ticket = tail.fetch_add(1);
        size_t index = ticket % BLOCK_SIZE;
        while (!block.slots[index].ready.load(std::memory_order_acquire)) {
            // 自旋等待
        }
        value = block.slots[index].data;
        return true;
    }

private:
    std::atomic<size_t> head;
    std::atomic<size_t> tail;
    Block block;
};

Benchmark 代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

// 基准测试函数
template <typename QueueType>
double benchmark(const std::string& name, double& opsPerSec) {
    QueueType queue;
    std::atomic<int> totalConsumed{0};
    std::map<int, double> threadWaitTimes;

    auto start = std::chrono::high_resolution_clock::now();

    // 启动生产者线程
    std::vector<std::thread> producers;
    for (int p = 0; p < NUM_PRODUCERS; ++p) {
        producers.emplace_back([&queue, p]() {
            for (int i = 0; i < ITEMS_PER_PRODUCER; ++i) {
                queue.enqueue(i + p * ITEMS_PER_PRODUCER);
            }
        });
    }

    // 启动消费者线程
    std::vector<std::thread> consumers;
    for (int c = 0; c < NUM_CONSUMERS; ++c) {
        consumers.emplace_back([&queue, &totalConsumed, c, &threadWaitTimes]() {
            int item;
            auto localStart = std::chrono::high_resolution_clock::now();
            while (true) {
                auto t0 = std::chrono::high_resolution_clock::now();
                while (!queue.try_dequeue(item)) {
                    // busy wait
                }
                auto t1 = std::chrono::high_resolution_clock::now();
                threadWaitTimes[c] +=
                    std::chrono::duration<double>(t1 - t0).count();

                if (++totalConsumed >= N)
                    break;
            }
            auto localEnd = std::chrono::high_resolution_clock::now();
            double threadTime =
                std::chrono::duration<double>(localEnd - localStart).count();
            std::cout << "Consumer " << c << " finished in " << threadTime
                      << "s, wait time: " << threadWaitTimes[c] << "s\n";
        });
    }

    for (auto& t : producers)
        t.join();
    for (auto& t : consumers)
        t.join();

    auto end = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> elapsed = end - start;
    opsPerSec = N / elapsed.count();

    std::cout << "\n"
              << name << " completed in " << elapsed.count()
              << "s, throughput: " << opsPerSec << " ops/sec\n";

    return elapsed.count();
}

int main() {
    double opsBlock = 0.0, opsTicket = 0.0;
    double timeBlock = benchmark<BlockQueue>("BlockQueue", opsBlock);
    double timeTicket = benchmark<TicketQueue>("TicketQueue", opsTicket);
    double speedup = opsTicket / opsBlock;

    std::cout << std::left << "| " << std::setw(14) << "Queue Type"
              << "| " << std::setw(12) << "Time (s)"
              << "| " << std::setw(20) << "Throughput (ops/s)"
              << "| " << std::setw(10) << "Speedup"
              << "|\n";

    std::cout << std::string(70, '-') << "\n";

    // BlockQueue row
    std::cout << std::left << "| " << std::setw(14) << "BlockQueue"
              << "| " << std::setw(12) << std::fixed << std::setprecision(6)
              << timeBlock << "| " << std::setw(20) << std::fixed
              << std::setprecision(0) << opsBlock << "| " << std::setw(10)
              << "1.00×"
              << "|\n";

    // TicketQueue row
    std::ostringstream speedupStream;
    speedupStream << std::fixed << std::setprecision(2) << speedup << "×";

    std::cout << std::left << "| " << std::setw(14) << "TicketQueue"
              << "| " << std::setw(12) << std::fixed << std::setprecision(6)
              << timeTicket << "| " << std::setw(20) << std::fixed
              << std::setprecision(0) << opsTicket << "| " << std::setw(10)
              << speedupStream.str() << "|\n";

    return 0;
}

前言

这是阅读 Cameron Desrochers 的 A Fast Lock-Free Queue for C++ 源码的笔记。

仓库地址：https://github.com/cameron314/readerwriterqueue

其他参考文献：

An Introduction to Lock-Free Programming
C++ and Beyond 2012: Herb Sutter - atomic Weapons 1 of 2
C++ and Beyond 2012: Herb Sutter - atomic Weapons 2 of 2

内存屏障

约束 memory loads/stores 的顺序。

• releaase 内存屏障：告诉 CPU，如果屏障之后的任何写入变得可见，那么屏障之前的任何写入都应该在其他核心中可见，前提是其他核心在读取 写屏障之后写入的数据 后执行读屏障。
  换句话说，如果线程 B 可以看到在另一个线程 A 上的写屏障之后写入的新值，那么在执行读屏障（在线程 B 上）之后，可以保证在线程 A 上的写屏障之前发生的所有写入在线程 B 上可见。

实现细节

1. block: 一个连续的环形缓冲区，用来存储元素。这样可以预分配内存。
2. 块过小（这不利于无锁）时，无需将所有现有元素复制到新的块中；多个块（大小独立）以循环链表的形式链接在一起。
3. 当前插入的块称为 “尾块”，当前消费的块称为 “头块”。
4. 头索引指向下一个要读取的满槽；尾索引指向下一个要插入的空槽。如果两个索引相等，则块为空（确切地说，当队列已满时，恰好有一个插槽为空，以避免在具有相同头和尾索引的满块和空块之间产生歧义）。
5. 为了允许队列对任意线程创建 / 析构（独立于生产 / 消费线程），全内存屏障（memory_order_acq_cst）被用在析构函数地最后、析构函数的开头（这会强制所有的 CPU cores 同步 outstanding changes）。显然，在析构函数可以被安全地调用之前，生产者和消费者必须已经停止使用该队列。

Give me the codes

1. 用户不需要管理内存。
2. 预分配内存，在连续的块中。
3. try_enqueue: 保证不会分配内存（队列有初始容量）；
4. enqueue: 会根据需要动态扩容。
5. 没有使用 CAS loop；这意味者 enqueue 和 dequeue 是 O(1) 的（没有计入内存分配的时间）。
6. 因为在 x86 平台，内存屏障是空操作，所以 enqueue 和 dequeue 是一系列简单的 loads 和 stores (and branches) 。

此代码仅仅适用于以原子方式处理 自然对齐的整型（aligned integer） 和 原生指针大小（native-pointer-size） 的 loads/stores 的 CPU 上；
幸运的是，这包括了所有的现代处理器（包括 ARM, x86/x86_64 和 PowerPC）。
它不是为在 DEC Alpha 上运行而设计的（DEC Alpha 似乎具有有史以来最弱的内存排序保证）。

注：在 x86 上，memory_order_acquire/release 通常不需要额外指令就能实现语义，但仍然能限制编译器的重排。
fetch_add 不是一个原子操作，而是三个：load, add, store. 所以不适用上述说的 “自然对齐的整型” 或“原生指针大小”的 load/store.

性能优化点

1. 平凡析构：跳过析构，直接释放内存。
2. MCRingBuffer paper
   1. cache line padding
   2. local control variables
      1. 减少对全局 read/write 指针的读取
   3. local block

正确性测试

1. 定义不可预测性延时函数，用于模拟线程调度。
2. 写线程塞入 32 M 个数据；读线程读取 32 M 次。读写线程中使用 unpredDelay() 模拟调度延迟。
3. 测试能否顺序读取，失败则打印日志，不退出。
4. 测试程序无限运行，每次使用一个写线程和读线程。直至手动 Ctrl C 关闭。

查看磁盘类型

1
2
3

$ lsblk -d -o name,rota,type,size,model
NAME ROTA TYPE  SIZE MODEL
sda     1 disk  1.8T PERC H740P Mini

ROTA=1：这是旋转磁盘。

测试方法

顺序写吞吐测试（逼近最大写入速度）

1

fio --name=seqwrite --rw=write --bs=1M --size=5G --numjobs=4 --iodepth=32 --direct=1 --runtime=60 --group_reporting

随机读 IOPS 测试（逼近最大并发处理能力）

1

fio --name=randread --rw=randread --bs=4k --size=5G --numjobs=4 --iodepth=64 --direct=1 --runtime=60 --group_reporting

混合读写测试（模拟数据库负载）

1

fio --name=mixrw --rw=randrw --rwmixread=70 --bs=4k --size=5G --numjobs=4 --iodepth=32 --direct=1 --runtime=60 --group_reporting

磁盘的测试结果

由于是旋转磁盘，iodepth 总是 1（设成其他值不会生效）

单线程读写文件：

    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
#!/bin/bash

# 测试参数
DEVICE="./testfile"   # 修改为你要测试的文件或设备路径
RUNTIME=30                       # 每个测试运行时间（秒）
# BLOCK_SIZES=("4k" "16k" "64k" "256k" "1M" "4M" "16M" "32M" "64M" "128M")  # 测试块大小列表
BLOCK_SIZES=("128M" "64M" "32M" "16M" "4M" "1M" "256k" "64k" "16k" "4k")  # 测试块大小列表
LOG_FILE="fio_bs_output.log"        # 原始输出日志文件
PERFORMANCE_LOG="fio_bs_performance.log"  # 性能结果日志文件

# 输出表头
printf "%-8s | %-10s | %-8s | %-10s | %-10s\n" "RW" "BlockSize" "IOPS" "BW(MiB/s)" "AvgLat(ms)" | tee "$PERFORMANCE_LOG"
echo "-------------------------------------------------------------" | tee -a "$PERFORMANCE_LOG"

echo "" > "$LOG_FILE"  # 清空日志文件

# 循环测试不同块大小
for RW in read write; do
  for BS in "${BLOCK_SIZES[@]}"; do
    OUTPUT=$(fio --name=bs_test \
                 --filename="$DEVICE" \
                 --rw=$RW \
                 --bs=$BS \
                 --size=1G \
                 --time_based \
                 --runtime=$RUNTIME \
                 --numjobs=1 \
                 --direct=1 \
                 --ioengine=psync \
                 --group_reporting)
    echo "----------------------------------------------" >> "$LOG_FILE"
    echo "$OUTPUT" >> "$LOG_FILE"
    echo "" >> "$LOG_FILE"

    # 提取关键指标
    read IOPS BW BWUNIT LAT LAT_UNIT <<< $(echo "$OUTPUT" | awk '
        /IOPS=/ {match($0, /IOPS= *([0-9.]+)/, iops)}
        /BW=/ {
            match($0, /BW= *([0-9.]+)([KMG]iB)\/s/, bwinfo)
            bwval=bwinfo[1]; bwunit=bwinfo[2]
        }
        /clat \(/ {match($0, /avg= *([0-9.]+),/, lat); match($0, /\(([^)]+)\)/, lat_unit)}
        END {print iops[1], bwval, bwunit, lat[1], lat_unit[1]}
    ')

    # 延迟单位换算
    if [ "$LAT_UNIT" = "usec" ]; then
        LAT_MS=$(awk "BEGIN {printf \"%.2f\", $LAT/1000}")
    elif [ "$LAT_UNIT" = "msec" ]; then
        LAT_MS=$LAT
    else
        LAT_MS="Unknown"
    fi

    # 带宽单位换算为 MiB/s
    case "$BWUNIT" in
        "KiB") BW_MIB=$(awk "BEGIN {printf \"%.2f\", $BW/1024}") ;;
        "MiB") BW_MIB=$BW ;;
        "GiB") BW_MIB=$(awk "BEGIN {printf \"%.2f\", $BW*1024}") ;;
        *) BW_MIB="Unknown" ;;
    esac

    # 输出结果行
    printf "%-8s | %-10s | %-8s | %-10s | %-10s\n" "$RW" "$BS" "$IOPS" "$BW_MIB" "$LAT_MS" | tee -a "$PERFORMANCE_LOG"
  done
done

# 删除 fio 创建的测试文件
rm -f "$DEVICE"
echo "测试完成，结果已保存到 $LOG_FILE 和 $PERFORMANCE_LOG"

  

1
2
3
4
5
6
7
8
9
10
11
12

RW       | BlockSize  | IOPS     | BW(MiB/s)  | AvgLat(ms)
-----------------------------------------------------------
read     | 4k         | 194      | 0.76       | 5.14
read     | 16k        | 239      | 3.74       | 4.17
read     | 64k        | 347      | 21.7       | 2.88
read     | 256k       | 271      | 67.9       | 3.68
read     | 1M         | 94       | 94.9       | 10.53
read     | 4M         | 14       | 57.2       | 69.74
read     | 16M        | 6        | 97.6       | 163.96
read     | 32M        | 3        | 111        | 288.99
read     | 64M        | 1        | 112        | 573.70
read     | 128M       | 0        | 112        | 1147.00

1
2
3
4
5
6
7
8
9
10
11
12

RW       | BlockSize  | IOPS     | BW(MiB/s)  | AvgLat(ms)
-----------------------------------------------------------
write    | 4k         | 1884     | 7.36       | 0.53
write    | 16k        | 1452     | 22.7       | 0.69
write    | 64k        | 793      | 49.6       | 1.26
write    | 256k       | 307      | 76.8       | 3.24
write    | 1M         | 100      | 100        | 9.96
write    | 4M         | 18       | 73.7       | 53.99
write    | 16M        | 5        | 93.8       | 169.43
write    | 32M        | 3        | 101        | 313.24
write    | 64M        | 1        | 107        | 594.46
write    | 128M       | 0        | 102        | 1249.35

当 BlockSize=32M 以后，写入性能基本达到顶峰（110 MiB/s），和旋转磁盘的参数基本一致。

多线程读写同一个文件，BS=64KiB：

    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
#!/bin/bash

DEVICE="./mt_testfile"   # 测试文件路径
RUNTIME=30            # 每个测试运行时间（秒）
THREADS=(1 2 4 8 16 32 64 72 120)  # 测试线程数列表
BLOCK_SIZE="64k"      # 块大小，可根据需要修改
LOG_FILE="fio_thread_output.log"
PERFORMANCE_LOG="fio_thread_performance.log"

# 输出表头
printf "%-8s | %-8s | %-10s | %-10s | %-10s\n" "RW" "Threads" "IOPS" "BW(MiB/s)" "AvgLat(ms)" | tee "$PERFORMANCE_LOG"
echo "---------------------------------------------------------------" | tee -a "$PERFORMANCE_LOG"

echo "" > "$LOG_FILE"  # 清空日志文件

for RW in read write; do
  for THREAD in "${THREADS[@]}"; do
    OUTPUT=$(fio --name=thread_test \
                 --filename="$DEVICE" \
                 --rw=$RW \
                 --bs=$BLOCK_SIZE \
                 --size=5G \
                 --time_based \
                 --runtime=$RUNTIME \
                 --numjobs=$THREAD \
                 --direct=1 \
                 --ioengine=psync \
                 --group_reporting)
    echo "---------------------------------------------------------------" >> "$LOG_FILE"
    echo "$OUTPUT" >> "$LOG_FILE"
    echo "" >> "$LOG_FILE"

    # 提取关键指标
    read IOPS BW BWUNIT LAT LAT_UNIT <<< $(echo "$OUTPUT" | awk '
        /IOPS=/ {match($0, /IOPS= *([0-9.]+)/, iops)}
        /BW=/ {
            match($0, /BW= *([0-9.]+)([KMG]iB)\/s/, bwinfo)
            bwval=bwinfo[1]; bwunit=bwinfo[2]
        }
        /clat \(/ {match($0, /avg= *([0-9.]+),/, lat); match($0, /\(([^)]+)\)/, lat_unit)}
        END {print iops[1], bwval, bwunit, lat[1], lat_unit[1]}
    ')

    # 延迟单位换算
    if [ "$LAT_UNIT" = "usec" ]; then
        LAT_MS=$(awk "BEGIN {printf \"%.2f\", $LAT/1000}")
    elif [ "$LAT_UNIT" = "msec" ]; then
        LAT_MS=$LAT
    else
        LAT_MS="Unknown"
    fi

    # 带宽单位换算为 MiB/s
    case "$BWUNIT" in
        "KiB") BW_MIB=$(awk "BEGIN {printf \"%.2f\", $BW/1024}") ;;
        "MiB") BW_MIB=$BW ;;
        "GiB") BW_MIB=$(awk "BEGIN {printf \"%.2f\", $BW*1024}") ;;
        *) BW_MIB="Unknown" ;;
    esac

    # 输出结果行
  printf "%-8s | %-8s | %-10s | %-10s | %-10s\n" "$RW" "$THREAD" "$IOPS" "$BW_MIB" "$LAT_MS" | tee -a "$PERFORMANCE_LOG"
  done
done

rm -f "$DEVICE"
echo "测试完成，结果已保存到 $LOG_FILE 和 $PERFORMANCE_LOG"

  

1
2
3
4
5
6
7
8
9
10
11

RW       | Threads  | IOPS       | BW(MiB/s)  | AvgLat(ms)
------------------------------------------------------------
read     | 1        | 826        | 51.6       | 1.21
read     | 2        | 1300       | 81.3       | 1.54
read     | 4        | 1681       | 105        | 2.38
read     | 8        | 1778       | 111        | 4.49
read     | 16       | 1789       | 112        | 8.93
read     | 32       | 1790       | 112        | 17.86
read     | 64       | 1789       | 112        | 35.73
read     | 72       | 1790       | 112        | 40.18
read     | 120      | 1789       | 112        | 66.98

1
2
3
4
5
6
7
8
9
10
11

RW       | Threads  | IOPS       | BW(MiB/s)  | AvgLat(ms)
------------------------------------------------------------
write    | 1        | 847        | 52.9       | 1.18
write    | 2        | 1367       | 85.5       | 1.46
write    | 4        | 1757       | 110        | 2.27
write    | 8        | 1786       | 112        | 4.47
write    | 16       | 1788       | 112        | 8.92
write    | 32       | 1788       | 112        | 17.79
write    | 64       | 1788       | 112        | 35.09
write    | 72       | 1788       | 112        | 39.54
write    | 120      | 1784       | 112        | 64.50

当线程数增加，IO 性能随之提高，可能原因是 64KiB 小块数据大量提交到 I/O 队列，操作系统能更好地完成读写路径优化。
但达到8线程的时候，就基本到达性能顶峰了。

注意：fio 多线程写入同一个文件是没有加锁的，如果超过 page cache (一般是 4 KB)，那么可能乱序写入。

概念（以 fio 为例）

ioengine

ioengine（I/O 引擎）是 fio 提供以执行读写任务的底层接口。不同的引擎代表不同的 I/O 模型，比如同步、异步、内存映射、零拷贝等。

每种 ioengine 都依赖操作系统提供的底层 I/O 接口。例如：

你可以指定任意 ioengine （默认值是 sync / psync），但它是否能运行，必须得到操作系统的支持。这包括内核版本、系统接口、库文件等。如果系统不支持，fio 会报错或自动回退。

查看支持列表：

1

fio --enghelp

这会列出当前系统上可用的 ioengine，但注意：列出来 ≠ 能用，还要看运行时是否报错。

实际测试：

1

fio --name=test --ioengine=io_uring --rw=write --size=1G --bs=1M

如果不支持，会报错，例如：

1

fio: pid=132756, err=38/file:engines/io_uring.c:1351, func=io_queue_init, error=Function not implemented

iodepth

--iodepth 是传递给内核的参数。

如果你不显式设置 --iodepth，那么 fio 会根据所选的 I/O 引擎（--ioengine） 来决定默认值

fio 并不会主动维护队列，队列是内核的特性。

• 对于支持异步 I/O 的引擎（如 libaio 或 io_uring），你可以设置更高的 iodepth（如 32、64、128）来模拟高并发负载；
• 对 SSD 或 NVMe 设备，高 iodepth 能显著提升 IOPS 和吞吐量；
• 对机械硬盘，提升有限，但仍可用于测试调度策略和队列行为。

但如果你的 ioengine 是 sync 或 psync，这些是同步阻塞 I/O，根本不支持高并发，所以 iodepth 实际上不会生效。

可能的代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

#include <libaio.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>

#define FILE_PATH "testfile.bin"
#define BLOCK_SIZE 4096
#define IODEPTH 4  // 控制并发请求数量

int main() {
    int fd = open(FILE_PATH, O_CREAT | O_WRONLY | O_DIRECT, 0644);
    if (fd < 0) {
        perror("open");
        return 1;
    }

    // io_setup 是 libaio 的函数
    io_context_t ctx = 0;
    if (io_setup(IODEPTH, &ctx) < 0) {
        perror("io_setup");
        return 1;
    }

    struct iocb *iocbs[IODEPTH];
    struct iocb iocb_array[IODEPTH];
    char *buffers[IODEPTH];

    for (int i = 0; i < IODEPTH; i++) {
        // 分配对齐内存
        posix_memalign((void**)&buffers[i], BLOCK_SIZE, BLOCK_SIZE);
        memset(buffers[i], 'A' + i, BLOCK_SIZE);

        // 初始化 iocb
        io_prep_pwrite(&iocb_array[i], fd, buffers[i], BLOCK_SIZE, i * BLOCK_SIZE);
        iocbs[i] = &iocb_array[i];
    }

    // 提交所有请求
    int ret = io_submit(ctx, IODEPTH, iocbs);
    if (ret < 0) {
        perror("io_submit");
        return 1;
    }

    // 等待所有请求完成
    struct io_event events[IODEPTH];
    io_getevents(ctx, IODEPTH, IODEPTH, events, NULL);

    // 清理
    for (int i = 0; i < IODEPTH; i++) {
        free(buffers[i]);
    }
    io_destroy(ctx);
    close(fd);

    printf("All %d I/O requests completed.\n", IODEPTH);
    return 0;
}

延迟

结果分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

$ fio --name=seqwrite --rw=write --bs=1M --size=5G --numjobs=2 --direct=1 --runtime=60 --group_reporting
seqwrite: (g=0): rw=write, bs=(R) 1024KiB-1024KiB, (W) 1024KiB-1024KiB, (T) 1024KiB-1024KiB, ioengine=psync, iodepth=1
...
fio-3.41
Starting 2 processes
seqwrite: Laying out IO file (1 file / 5120MiB)
seqwrite: Laying out IO file (1 file / 5120MiB)
Jobs: 2 (f=2): [W(2)][100.0%][w=112MiB/s][w=112 IOPS][eta 00m:00s]
seqwrite: (groupid=0, jobs=2): err= 0: pid=70688: Sun Sep  7 22:37:47 2025
  write: IOPS=111, BW=111MiB/s (117MB/s)(6685MiB/60016msec); 0 zone resets
    clat (usec): min=9606, max=74763, avg=17922.82, stdev=1446.89
     lat (usec): min=9628, max=74791, avg=17951.49, stdev=1446.80
    clat percentiles (usec):
     |  1.00th=[15008],  5.00th=[16319], 10.00th=[16909], 20.00th=[17433],
     | 30.00th=[17695], 40.00th=[17695], 50.00th=[17957], 60.00th=[17957],
     | 70.00th=[18220], 80.00th=[18482], 90.00th=[19006], 95.00th=[19268],
     | 99.00th=[21365], 99.50th=[22414], 99.90th=[32375], 99.95th=[40109],
     | 99.99th=[74974]
   bw (KiB/s): min=96062, max=116736, per=100.00%, avg=114150.20, stdev=1061.35, samples=238
   iops        : min=   92, max=  114, avg=110.77, stdev= 1.18, samples=238
  lat (msec)   : 10=0.03%, 20=97.43%, 50=2.53%, 100=0.01%
  cpu          : usr=0.22%, sys=0.75%, ctx=6717, majf=0, minf=67
  IO depths    : 1=100.0%, 2=0.0%, 4=0.0%, 8=0.0%, 16=0.0%, 32=0.0%, >=64=0.0%
     submit    : 0=0.0%, 4=100.0%, 8=0.0%, 16=0.0%, 32=0.0%, 64=0.0%, >=64=0.0%
     complete  : 0=0.0%, 4=100.0%, 8=0.0%, 16=0.0%, 32=0.0%, 64=0.0%, >=64=0.0%
     issued rwts: total=0,6685,0,0 short=0,0,0,0 dropped=0,0,0,0
     latency   : target=0, window=0, percentile=100.00%, depth=1

Run status group 0 (all jobs):
  WRITE: bw=111MiB/s (117MB/s), 111MiB/s-111MiB/s (117MB/s-117MB/s), io=6685MiB (7010MB), run=60016-60016mse

🧾 测试配置解析
bash
fio --name=seqwrite --rw=write --bs=1M --size=10G --numjobs=1 --direct=1 --runtime=60 --group_reporting
参数 含义
rw=write 顺序写入
bs=1M 每次写入块大小为 1MiB
numjobs=1 单线程写入
direct=1 使用 Direct I/O，绕过页缓存
ioengine=psync 使用同步 I/O（每次 pwrite()）
iodepth=1 每次只挂起一个 I/O 请求（同步模式下默认如此）
📊 性能结果概览
指标 数值 说明
IOPS 96 每秒执行 96 次写入操作
带宽 96.2 MiB/s（101 MB/s） 每秒写入约 96 MiB 数据
总写入量 5772 MiB 在 60 秒内完成的写入总量
延迟（avg clat） 10.36 ms 每次写入的平均完成时间
CPU 使用率 usr=0.46%, sys=1.23% CPU 负载极低，瓶颈不在 CPU
⏱ 延迟分布分析
50% 的写入延迟低于 9.9 ms

95% 的写入低于 12.5 ms

99.95% 的写入延迟达到了 22.9 ms

最慢的写入高达 28.2 ms

尾部延迟略高，说明偶尔会有磁盘响应变慢的情况，可能是设备内部缓存刷新或寻址造成。

📈 带宽波动情况
平均带宽：约 96 MiB/s

最小带宽：60 MiB/s

最大带宽：104 MiB/s

标准差：6.2 MiB/s → 表明带宽相对稳定，但仍有轻微波动

🧠 深层解读
✅ 为什么 IOPS ≈ 带宽（MiB/s）？
因为你设置了 bs=1M，每次写入 1MiB 数据，所以：

Code
IOPS × Block Size = Bandwidth
96 IOPS × 1 MiB = 96 MiB/s
✅ 为什么 Direct I/O？
绕过页缓存，测试的是磁盘的真实物理性能，避免被内存加速 “欺骗”。

✅ 为什么使用 psync？
psync 是同步写入，每次调用 pwrite()，适合模拟数据库或日志系统的写入行为。但它无法并发挂起多个请求，限制了吞吐。

📌 性能瓶颈分析
磁盘类型：如果是 HDD，这个结果（96 MiB/s）非常合理；如果是 SSD，则偏低，可能受限于同步 I/O 或单线程。

IO 引擎限制：psync 是阻塞式，无法发挥磁盘的并发能力。

线程数限制：只有一个线程，磁盘可能未被充分利用。

使用共享内存优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

/**
 * Matrix multiplication (CUDA Kernel) on the device: C = A * B
 * wA is A's width and wB is B's width
 */
template <int BLOCK_SIZE> __global__ void MatrixMulCUDA(float *C, float *A, float *B, int wA, int wB)
{
    // Block index
    int bx = blockIdx.x;
    int by = blockIdx.y;

    // Thread index
    int tx = threadIdx.x;
    int ty = threadIdx.y;

    // Index of the first sub-matrix of A processed by the block
    int aBegin = wA * BLOCK_SIZE * by;

    // Index of the last sub-matrix of A processed by the block
    int aEnd = aBegin + wA - 1;

    // Step size used to iterate through the sub-matrices of A
    int aStep = BLOCK_SIZE;

    // Index of the first sub-matrix of B processed by the block
    int bBegin = BLOCK_SIZE * bx;

    // Step size used to iterate through the sub-matrices of B
    int bStep = BLOCK_SIZE * wB;

    // Csub is used to store the element of the block sub-matrix
    // that is computed by the thread
    float Csub = 0;

    // Loop over all the sub-matrices of A and B
    // required to compute the block sub-matrix
    for (int a = aBegin, b = bBegin; a <= aEnd; a += aStep, b += bStep) {
        // Declaration of the shared memory array As used to
        // store the sub-matrix of A
        __shared__ float As[BLOCK_SIZE][BLOCK_SIZE];

        // Declaration of the shared memory array Bs used to
        // store the sub-matrix of B
        __shared__ float Bs[BLOCK_SIZE][BLOCK_SIZE];

        // Load the matrices from device memory
        // to shared memory; each thread loads
        // one element of each matrix
        As[ty][tx] = A[a + wA * ty + tx];
        Bs[ty][tx] = B[b + wB * ty + tx];

        // Synchronize to make sure the matrices are loaded
        __syncthreads();

        // Multiply the two matrices together;
        // each thread computes one element
        // of the block sub-matrix
#pragma unroll

        for (int k = 0; k < BLOCK_SIZE; ++k) {
            Csub += As[ty][k] * Bs[k][tx];
        }

        // Synchronize to make sure that the preceding
        // computation is done before loading two new
        // sub-matrices of A and B in the next iteration
        __syncthreads();
    }

    // Write the block sub-matrix to device memory;
    // each thread writes one element
    int c               = wB * BLOCK_SIZE * by + BLOCK_SIZE * bx;
    C[c + wB * ty + tx] = Csub;
}