docs(vol5): 修正 fence 性能结论并附基准测试

code/volumn_codes/vol5/ch03-atomic-memory-model/CMakeLists.txt

@@ -11,6 +11,15 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
 # 可运行的可执行文件（含 main 函数）
 add_executable(atomic_default_ctor_verify atomic_default_ctor_verify.cpp)
 
+if(CMAKE_SYSTEM_NAME STREQUAL "Linux" AND CMAKE_SYSTEM_PROCESSOR MATCHES "^(x86_64|AMD64)$")
+    find_package(Threads REQUIRED)
+    add_executable(fence_vs_ordered_benchmark fence_vs_ordered_benchmark.cpp)
+    target_link_libraries(fence_vs_ordered_benchmark PRIVATE Threads::Threads)
+    if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
+        target_compile_options(fence_vs_ordered_benchmark PRIVATE -O3 -march=native)
+    endif()
+endif()
+
 # 汇编验证文件（无 main 函数，仅编译为 .s 文件用于指令检查）
 # 使用方式：cmake --build build --target asm_check
 set(ASM_OUTPUT_DIR ${CMAKE_BINARY_DIR}/asm)

code/volumn_codes/vol5/ch03-atomic-memory-model/README.md

@@ -1,19 +1,29 @@
 # ch03: Atomic 操作与内存模型
 
-本目录包含 4 个示例，分为两类。
+本目录包含 5 个示例，分为可运行示例与汇编验证示例。
 
 ## 可运行示例
 
 | 文件 | 说明 | 运行预期 |
 |------|------|---------|
 | `atomic_default_ctor_verify.cpp` | 验证 `std::atomic<T>` 不要求 T 可默认构造 | 输出 `42`、`100`、`200` 三行 |
+| `fence_vs_ordered_benchmark.cpp` | 比较逐字段 release/acquire 与 fence + relaxed 的完整交接耗时（Linux x86-64） | 输出每轮耗时和两种写法的中位数 |
 
 ```bash
 cmake -B build -DCMAKE_BUILD_TYPE=Release
 cmake --build build
 ./build/atomic_default_ctor_verify
 ```
 
+benchmark 需要 CMake 3.20 或更高版本。请在仓库根目录构建；示例会将两个线程固定到 CPU 0 和 CPU 2，如果这些 CPU 不可用，先修改源码开头的 CPU 编号。
+
+```bash
+cmake -S code/volumn_codes/vol5/ch03-atomic-memory-model \
+  -B /tmp/vol5-fence-build -DCMAKE_BUILD_TYPE=Release
+cmake --build /tmp/vol5-fence-build --target fence_vs_ordered_benchmark
+/tmp/vol5-fence-build/fence_vs_ordered_benchmark
+```
+
 ## 汇编验证示例
 
 以下 3 个文件**没有 `main()` 函数**，不能作为可执行文件运行。它们的设计目的是通过 `-S` 生成汇编文件，验证文章中关于 x86 指令的技术断言。

code/volumn_codes/vol5/ch03-atomic-memory-model/fence_vs_ordered_benchmark.cpp

@@ -0,0 +1,149 @@
+// Standard: C++20
+// Platform: Linux x86-64; CPU 0 and CPU 2 are separate cores on the test host.
+
+#include <algorithm>
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <iomanip>
+#include <iostream>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+#include <immintrin.h>
+#include <pthread.h>
+#include <sched.h>
+
+constexpr std::uint64_t kIterations = 2'000'000;
+constexpr int kTrials = 7;
+constexpr int kProducerCpu = 0;
+constexpr int kConsumerCpu = 2;
+
+bool pin_to_cpu(int cpu) {
+    cpu_set_t set;
+    CPU_ZERO(&set);
+    CPU_SET(cpu, &set);
+    return pthread_setaffinity_np(pthread_self(), sizeof(set), &set) == 0;
+}
+
+struct SharedState {
+    alignas(64) std::array<std::atomic<std::uint64_t>, 3> fields{};
+    alignas(64) std::atomic<std::uint64_t> acknowledged{0};
+    alignas(64) std::array<std::uint64_t, 3> payload{};
+};
+
+struct Sample {
+    double ns_per_round;
+    std::uint64_t checksum;
+};
+
+template <bool UseFences> Sample run_once(std::uint64_t iterations) {
+    SharedState state;
+    std::uint64_t checksum = 0;
+    std::atomic<bool> affinity_failed{false};
+    constexpr auto kStoreOrder = UseFences ? std::memory_order_relaxed : std::memory_order_release;
+    constexpr auto kLoadOrder = UseFences ? std::memory_order_relaxed : std::memory_order_acquire;
+
+    const auto start = std::chrono::steady_clock::now();
+    std::thread producer([&] {
+        if (!pin_to_cpu(kProducerCpu)) {
+            affinity_failed.store(true, std::memory_order_relaxed);
+        }
+        for (std::uint64_t i = 1; i <= iterations; ++i) {
+            // 确认上一轮读取已完成，再改写非原子 payload。
+            while (state.acknowledged.load(std::memory_order_acquire) != i - 1) {
+                _mm_pause();
+            }
+
+            state.payload[0] = i;
+            state.payload[1] = i + 7;
+            state.payload[2] = i * 3;
+
+            if constexpr (UseFences) {
+                std::atomic_thread_fence(std::memory_order_release);
+            }
+            for (auto& field : state.fields) {
+                field.store(i, kStoreOrder);
+            }
+        }
+    });
+
+    std::thread consumer([&] {
+        if (!pin_to_cpu(kConsumerCpu)) {
+            affinity_failed.store(true, std::memory_order_relaxed);
+        }
+        for (std::uint64_t i = 1; i <= iterations; ++i) {
+            // 两种写法每次轮询都读取三个字段，只改变内存序。
+            while (true) {
+                const auto a = state.fields[0].load(kLoadOrder);
+                const auto b = state.fields[1].load(kLoadOrder);
+                const auto c = state.fields[2].load(kLoadOrder);
+                if (a == i && b == i && c == i) {
+                    break;
+                }
+                _mm_pause();
+            }
+
+            if constexpr (UseFences) {
+                std::atomic_thread_fence(std::memory_order_acquire);
+            }
+            checksum += state.payload[0] + state.payload[1] + state.payload[2];
+            state.acknowledged.store(i, std::memory_order_release);
+        }
+    });
+
+    producer.join();
+    consumer.join();
+    const auto elapsed = std::chrono::steady_clock::now() - start;
+    if (affinity_failed.load(std::memory_order_relaxed)) {
+        throw std::runtime_error("could not pin worker threads to CPUs 0 and 2");
+    }
+
+    const std::uint64_t expected = 5 * iterations * (iterations + 1) / 2 + 7 * iterations;
+    if (checksum != expected) {
+        throw std::runtime_error("incorrect checksum");
+    }
+
+    const double nanoseconds = std::chrono::duration<double, std::nano>(elapsed).count();
+    return {nanoseconds / static_cast<double>(iterations), checksum};
+}
+
+double median(std::vector<double> samples) {
+    std::sort(samples.begin(), samples.end());
+    return samples[samples.size() / 2];
+}
+
+int main() {
+    run_once<false>(20'000);
+    run_once<true>(20'000);
+
+    std::vector<double> ordered;
+    std::vector<double> fenced;
+    ordered.reserve(kTrials);
+    fenced.reserve(kTrials);
+
+    std::cout << std::fixed << std::setprecision(1);
+    std::cout << "Iterations per trial: " << kIterations << '\n';
+    std::cout << "Producer CPU: " << kProducerCpu << ", consumer CPU: " << kConsumerCpu << '\n';
+    std::cout << "Round-trip time includes both threads and the acknowledgement.\n";
+    for (int trial = 0; trial < kTrials; ++trial) {
+        // 交替测试顺序，避免一种写法始终先运行。
+        if (trial % 2 == 0) {
+            ordered.push_back(run_once<false>(kIterations).ns_per_round);
+            fenced.push_back(run_once<true>(kIterations).ns_per_round);
+        } else {
+            fenced.push_back(run_once<true>(kIterations).ns_per_round);
+            ordered.push_back(run_once<false>(kIterations).ns_per_round);
+        }
+        std::cout << "Trial " << trial + 1 << ": per-field release/acquire = " << ordered.back()
+                  << " ns, fences + relaxed = " << fenced.back() << " ns\n";
+    }
+
+    const double ordered_median = median(ordered);
+    const double fenced_median = median(fenced);
+    std::cout << "Median: per-field release/acquire = " << ordered_median
+              << " ns, fences + relaxed = " << fenced_median << " ns\n";
+    std::cout << "Fence / per-field ratio: " << fenced_median / ordered_median << '\n';
+}

documents/vol5-concurrency/ch03-atomic-memory-model/03-fence-and-barrier.md

@@ -82,7 +82,7 @@ void consumer()
 
 第二种模式是第一种的反转：线程 A 用普通的 release store，线程 B 用独立的 acquire fence。条件是线程 B 中有一个 atomic load 排在 fence 之前，并且这个 load 读到了线程 A 的 store 写入的值。
 
-这种模式的一个典型应用场景是"邮箱扫描"：我们有多个邮箱（每个邮箱由一个原子标志位标识），读者需要扫描所有邮箱，但只需要跟其中写入了自己的数据的那个邮箱建立同步。如果用带 acquire 的 load 去读每一个邮箱标志位，那即使标志位不是自己的，也会引入不必要的屏障开销。更好的做法是用 relaxed load 扫描，发现自己关心的邮箱有数据之后，只对那一个邮箱做一次 acquire fence：
+这种模式的一个典型应用场景是"邮箱扫描"：我们有多个邮箱（每个邮箱由一个原子标志位标识），读者需要扫描所有邮箱，但只在读取属于自己的数据时才需要建立同步。一种写法是用 relaxed load 扫描，发现匹配的邮箱后，再执行 acquire fence：
 
 ```cpp
 #include <atomic>
@@ -117,13 +117,13 @@ void read_my_mail(int my_id)
 }
 ```
 
-这里的关键洞察是：acquire fence 只在我们确认需要同步的时候才执行。前面的 31 次 relaxed load 不引入任何屏障，性能代价极低。这就是 fence 相比"带排序的原子操作"的灵活性——它可以把"决策"和"同步"分开，在确认需要同步之后再施加屏障。
+这里的关键是把"是否读取数据"与"为读取数据建立同步"分开：没有匹配的邮箱就不会执行 acquire fence。relaxed load 仍有原子读取和缓存访问的成本；acquire load 也不一定生成单独的硬件屏障指令。因此，这种写法提供了按需同步的灵活性，但不能仅凭源码断定它比逐个使用 acquire load 更快。
 
 ### fence-fence 同步
 
 第三种模式是两端都用 fence。线程 A 用 release fence + relaxed store，线程 B 用 relaxed load + acquire fence。条件是线程 A 的 fence 排在 store 之前，线程 B 的 load 读到了 store 的值，且 load 排在 fence 之前。
 
-这种模式的适用场景是"批量发布"：线程 A 准备好一组数据后，用一次 release fence 同时发布多个 relaxed store。对应的消费者用一次 acquire fence 读取多个 relaxed load。比起对每个原子操作都设置 release/acquire，一次 fence 覆盖多个操作显然更高效：
+这种模式可以用于发布一组预先写好的数据：线程 A 先写数据，再执行 release fence 和多个 relaxed store；线程 B 先读取原子标志，再执行 acquire fence。如果其中一个 load 读到了对应的 store，两端的 fence 就能建立同步，使线程 A 在 release fence 之前写好的数据可供线程 B 在 acquire fence 之后读取：
 
 ```cpp
 #include <atomic>
@@ -162,13 +162,29 @@ void thread_b()
 }
 ```
 
-这种模式在无锁数据结构里很常见——当你需要同时发布多个字段，但不希望每个字段都带一个 release store 时，一次 release fence + 多个 relaxed store 是更优雅的选择。
+这里的"一组数据"指 release fence 之前写好的 `data`，不是说三个 `arr` 元素作为整体同时可见。读者可能只看到其中部分新标志；acquire fence 也不会改变此前 load 返回的值。只有在对应原子读写连接了两端的 fence，且数据在 fence 两侧按要求访问时，才能依赖这条同步关系。
 
 ### fence 与原子操作的比较：何时用 fence
 
-到这里我们可以总结一下 fence 相比"带排序的原子操作"的优势和劣势了。fence 的优势在于灵活性：一次 fence 可以覆盖多个原子操作，可以把同步延迟到真正需要的时候才施加，可以避免不必要的屏障开销。劣势在于可读性和易错性——fence 的语义比带排序的原子操作更难推理，因为它跟具体原子操作之间的 sequenced-before 关系必须被程序员自己保证，编译器不会帮你检查。
+我们回头看，fence 真正多给我们的是同步位置的选择：它可以借助符合条件的原子读写建立同步，也可以等运行时确认要读取数据，再执行 acquire fence。不过，写法灵活不等于机器指令更少或延迟更低；独立 fence 和带排序原子操作的开销还要看目标平台。代价则是推理更费劲：我们得核对原子读写与 fence 的先后关系，还要确认 load 究竟读到了哪次 store。
 
-笔者的建议是：在大多数场景下，优先使用带排序的原子操作（比如 `store(..., release)` + `load(..., acquire)`），只有在确认性能敏感且能从 fence 的灵活性中获益时，才考虑用 fence 替代。记住，fence 不是"更高级"的写法，它是一种"更手动"的写法——手动意味着更大的自由度，也意味着更容易出错。
+那实际跑起来呢？我们在 WSL2 上的 Intel Core i5-12400（x86-64）试了一组双线程基准：使用 CMake 3.20.5 和 GCC 16.2 构建，编译选项为 `-O3 -march=native`，生产者和消费者分别固定在 CPU 0、CPU 2。生产者每轮写三个数据并发布三个原子标志，消费者等三个标志都更新后读取数据，再用另一个原子变量确认，生产者才进入下一轮。两版只改变发布和读取的内存序：一版逐字段使用 release store / acquire load，另一版使用一对 release / acquire fence 配合三个 relaxed store / load。测试源码在 `code/volumn_codes/vol5/ch03-atomic-memory-model/fence_vs_ordered_benchmark.cpp`，每次测试 200 万轮，每种写法运行 7 次并取中位数。示例要求 Linux x86-64、CMake 3.20 或更高版本；请在仓库根目录运行以下命令。如果你的环境没有 CPU 0、CPU 2，可以在源码开头修改固定的 CPU 编号：
+
+```bash
+cmake -S code/volumn_codes/vol5/ch03-atomic-memory-model \
+  -B /tmp/vol5-fence-build -DCMAKE_BUILD_TYPE=Release
+cmake --build /tmp/vol5-fence-build --target fence_vs_ordered_benchmark
+/tmp/vol5-fence-build/fence_vs_ordered_benchmark
+```
+
+| 写法 | 7 次中位数 |
+| --- | ---: |
+| 逐字段 release / acquire | 124.8 ns/轮 |
+| fence + relaxed | 117.4 ns/轮 |
+
+这次运行中 fence 版的中位数略低，但单次结果有明显波动。这里测量的是完整交接过程，包含轮询、缓存通信和确认操作，不是单条 fence 指令的延迟。这一组结果不能证明 fence 稳定更快；换个平台、换种访问模式，还得重新看生成代码和实测结果。
+
+笔者的建议是：在大多数场景下，优先使用带排序的原子操作（比如 `store(..., release)` + `load(..., acquire)`）；只有同步位置确实需要与具体原子操作分开，或在目标平台上测到了收益时，再考虑 fence。记住，fence 不是"更高级"的写法，它是一种"更手动"的写法——手动意味着更大的自由度，也意味着更容易出错。
 
 ## std::atomic_signal_fence：线程内的信号屏障
 
@@ -250,7 +266,7 @@ ARM 提供三种屏障指令。`DMB`（Data Memory Barrier）确保在它之前
 
 DMB 还有选项后缀：`DMB ST` 只管 store 屏障，`DMB LD` 只管 load 屏障，`DMB ISH` 是 inner shareable 域的全向屏障（多核之间最常见的使用场景）。当 C++ 代码中调用 `std::atomic_thread_fence(memory_order_release)` 时，在 ARM 上编译器通常会生成 `DMB ISH` 指令。而对于 `memory_order_acquire`，GCC 和 Clang 会生成更轻量的 `DMB ISHLD` 指令，只对 load 操作施加屏障。
 
-关于这些 CPU 屏障指令，我们通常不需要直接使用——标准库的 `atomic_thread_fence` 和带排序的原子操作已经帮我们封装好了。但理解底层机制有助于我们做出更好的性能决策：在 x86 上，`seq_cst` 的额外开销是一次 `mfence`；在 ARM 上，每次 `acquire`/`release` 都是一次 `DMB`，开销大得多。
+关于这些 CPU 屏障指令，我们通常不需要直接使用——标准库的 `atomic_thread_fence` 和带排序的原子操作已经帮我们封装好了。不过别把内存序直接换算成固定的指令成本：前面看到，x86 上的 `seq_cst` fence 也可能用带 `LOCK` 前缀的指令实现；ARM 上带排序的原子操作与独立 fence 具体生成什么指令，同样取决于目标架构和编译器。要比较性能，先看生成的汇编，再在目标平台上测量。
 
 ## volatile 不是线程安全机制