C++20 内存模型与无锁编程深度实战：从 Memory Ordering 到无锁数据结构的生产级实现

在多核时代，无锁（lock-free）编程是构建极致性能系统的终极武器。然而，无锁编程的难点不在于"不加锁"本身，而在于正确理解和使用内存序（memory ordering）。C++20 标准进一步强化了内存模型，引入了 atomic_ref、atomic<shared_ptr>、latch/barrier 等新特性。本文将从硬件层面的内存乱序出发，系统性地拆解 C++ 内存模型的每一个 ordering 语义，并通过完整的可编译代码案例，展示如何在生产环境中构建高效且正确的无锁数据结构。

一、为什么需要内存模型？从硬件乱序说起

现代 CPU 为了追求性能，会对内存访问进行重排序（reordering）。在单线程程序中，这种重排序是不可见的（as-if rule），但在多线程环境下，一个线程的写入可能以不同顺序被另一个线程观察到，从而导致微妙的、极难复现的并发 Bug。

以经典的 IRIW（Independent Readers, Independent Writers）测试为例：

// 初始状态: x = 0, y = 0
// Thread 1: x = 1
// Thread 2: y = 1
// Thread 3: r1 = x; r2 = y
// Thread 4: r3 = y; r4 = x
// 可能结果: r1=1, r2=0, r3=1, r4=0（两个线程看到不同顺序的写入）

在 ARM/Power 等弱内存模型架构上，上述结果完全合法。这意味着：如果没有显式内存屏障，多线程程序的行为是未定义的（在不同架构上表现不一致）。C++ 内存模型的作用，就是为程序员提供一套标准化的工具，在"正确性"和"性能"之间做出精确权衡。

关键的硬件背景：Store Buffer 导致写入对其他核心不可见；Invalidate Queue 导致核心的缓存行状态不能即时响应；Store-to-Load Forwarding 让核心能读到自己的写入却可能错过其他核心的写入。这些硬件优化正是内存屏障和 ordering 约束要解决的根源问题。

二、C++ 六种 Memory Ordering 完整解析

C++11 引入了六种 memory ordering，C++20 对其语义做了更精确的定义。我们将从最宽松到最严格逐一剖析。

2.1 memory_order_relaxed — 最宽松保证

memory_order_relaxed 只保证原子操作本身的原子性和修改顺序（modification order）的一致性，不提供任何跨线程的同步或排序保证。它唯一保证的是：同一线程中，对同一原子变量的 relaxed 操作之间的相对顺序不会被破坏。

#include <atomic>
#include <thread>
#include <cassert>

std::atomic<int> x{0}, y{0};
int r1 = 0, r2 = 0;

void thread1() {
    x.store(1, std::memory_order_relaxed);
    y.store(1, std::memory_order_relaxed);
}

void thread3() {
    r1 = y.load(std::memory_order_relaxed);
    r2 = x.load(std::memory_order_relaxed);
}
// r1=1, r2=0 在 relaxed 下是合法的！

适用场景：引用计数增减（仅计数本身无依赖）、性能计数器、统计累加器。这类场景关心"最终计数值正确"，不关心中间过程的时序。

2.2 memory_order_consume — 数据依赖排序（理论上的优雅，实践中的鸡肋）

memory_order_atomic::memory_order_consume 保证：当前线程中，所有依赖于该 load 操作结果的读写操作不会被重排到 load 之前。这是唯一利用数据依赖关系的 ordering。

// 经典的依赖链 publish-consumner 模式
std::atomic<Data*> ptr{nullptr};
Data data;

void producer() {
    data.value = 42;
    data.name = "example";
    // 将 data 发布的 ptr 上
    ptr.store(&data, std::memory_order_release);
}

void consumer() {
    Data* p = nullptr;
    // 等待 ptr 非空
    while (!(p = ptr.load(std::memory_order_consume))) {}
    // consume 保证 p 指向的数据可见
    assert(p->value == 42);  // 不会触发
    assert(p->name == "example");  // 不会触发
}

然而，由于编译器追踪依赖链的复杂性，主流编译实现（GCC/Clang）目前将 consume 实现为 acquire。标准要求"尽可能实现但不容错"的语义，使其在实践中基本退化为 acquire。C++23 正在讨论用 memory_order_dependency 来更精确地解决这个问题。

2.3 memory_order_acquire / memory_order_release — 核心基石

这对 ordering 是无锁编程最重要的工具。它们建立 synchronizes-with 关系（当 acquire load 读到 release store 写入的值时，release 之前的所有写入对 acquire 线程可见）。

核心语义可类比为"单向屏障"：

• release store：确保该 store 之前的所有读写不会被重排到此 store 之后（"写屏障"的上半部分）
• acquire load：确保该 load 之后的所有读写不会被重排到此 load 之前（"读屏障"的下半部分）

// 经典的 acquire-release 同步模式（类似互斥锁的效果）
std::atomic<bool> ready{false};
int data = 0;

void producer_thread() {
    data = 42;
    ready.store(true, std::memory_order_release);  // data 写入必须先完成
}

void consumer_thread() {
    while (!ready.load(std::memory_order_acquire)) {}  // 等待 ready
    assert(data == 42);  // 保证能看到 data = 42
}

其底层硬件实现：在 x86 上，由于 TSO（Total Store Order）的天然保证，acquire/release 不需要额外屏障指令，只需阻止编译期重排。在 ARM 上，compilers 会生成 dmb ishld (acquire) 和 dmb ish (release) 指令。这使得 acquire-release 在 x86 上几乎零开销，在 ARM 上只需单条 barrier 指令。

2.4 memory_order_acq_rel — 读-改-写操作的桥梁

fetch_add、compare_exchange 等 Read-Modify-Write 操作同时涉及 load 和 store，可以用 acq_rel 表示"同时需要 acquire 和 release 语义"，但不建立 total order：

// 无锁引用计数递减
std::atomic<int> ref_count{1};

void release_reference() {
    if (ref_count.fetch_sub(1, std::memory_order_acq_rel) == 1) {
        // 最后一个引用释放——acquire 确保看到完整构造的对象
        delete this;
    }
}

为什么用 acq_rel 而不是 seq_cst？因为：release 确保之前的所有写入对后续 acquire 线程可见；acquire 确保在计数降为 0 时，能看到此前所有线程的完整操作。这比 seq_cst 节省了跨线程全序同步的开销。

2.5 memory_order_seq_cst — 全序一致性（默认且最常用）

这是 std::atomic 的默认 ordering，它保证所有线程对所有 seq_cst 操作看到完全一致的修改顺序（single total order）。这正是大多数人直觉中的"happens-before"——但代价也是最高的。

// seq_cst 保证全局一致的全序
std::atomic<int> x{0}, y{0};
int r1 = 0, r2 = 0;

void t1() { x.store(1, std::memory_order_seq_cst); }
void t2() { y.store(1, std::memory_order_seq_cst); }
void t3() { r1 = x.load(std::memory_order_seq_cst); }
void t4() { r2 = y.load(std::memory_order_seq_cst); }

// 不可能所有线程看到的 x,y 修改顺序不一致
// 不会出现 t3 看到 x=1 而 t4 看到 y=0，同时另一个线程相反的情况

在 ARM64 上，compilers 会生成 dmb ish（全屏障）配合 store-load 的额外屏障指令，使 seq_cst 的 store 操作在 ARM 上比 release store 慢 2-3 倍。这也是为什么正确选择 ordering 对性能至关重要的原因。

2.6 Ordering 选择决策树

┌────────────────────────────────────────────────────────┐
│              Memory Ordering 决策树                     │
├────────────────────────────────────────────────────────┤
│                                                        │
│  需要跨线程同步？                                       │
│  ├── 否 → relaxed                                      │
│  └── 是 → 读还是写？                                   │
│      ├── 纯读 → acquire（读取由别人写入的数据）           │
│      ├── 纯写 → release（发布数据供别人读取）             │
│      ├── 先读后写 → acq_rel（RMW 操作）                  │
│      └── 全序约束 → seq_cst（默认，安全但最慢）           │
│                                                        │
│  经验法则：默认用 seq_cst 写出正确代码，                 │
│  性能热点分析后再逐步降级到 acquire/release/relaxed       │
└────────────────────────────────────────────────────────┘

三、happens-before 与 Synchronizes-with 的形式化关系

C++ 内存模型通过两个核心关系来定义正确的并发语义：

3.1 happens-before（跨线程操作的可观测顺序）

如果操作 A happens-before 操作 B，则 A 的所有写入对 B 可见。它由以下规则传递组成：

• 单线程 sequenced-before：在同一线程中，语句 A 在语句 B 之前（如果不在同一线程中，则需要使用同步原语）
• 跨线程 synchronizes-with：线程 A 的 release store 同步到线程 B 的 acquire load 读到了该值
• 传递性：若 A → B 且 B → C，则 A → C

C++20 进一步强化了 inter-thread happens-before 的语义，整合了 latch、barrier 等库类型，将 synchronizes-with 机制扩展到更细粒度的线程间同步。

3.2 Coherence-Reads-Write（读写一致性保证）

C++ 标准保证了三个关键性质：

• Coherence-Write-Write：对同一变量的两个写入不能重排，所有线程看到相同的修改顺序
• Coherence-Read-Read：对同一变量的两次读取之间，读取的值变化不能出现回退
• Coherence-Read-Write / Write-Read：读到某值后，不能再读到该值之前的写入

这三个保证了：即使使用 relaxed ordering，所有线程对同一原子变量的修改顺序（modification order）也是一致的。这是实现正确无锁算法的基础。

四、无锁数据结构实战：Michael-Scott Queue

4.1 基本设计思路

Michael-Scott Queue 是最经典的无锁队列算法，由 M. Michael 和 M. Scott 于 1996 年提出（因此得名 MS-Queue）。其核心思想：用原子操作替代锁，通过 CAS（Compare-And-Swap）实现并发安全的入队/出队。

#pragma once
#include <atomic>
#include <memory>
#include <optional>

template<typename T>
class MichaelScottQueue {
    struct Node {
        T data;
        std::atomic<Node*> next{nullptr};

        template<typename U>
        explicit Node(U&& val) : data(std::forward<U>(val)) {}
    };

    std::atomic<Node*> head_;
    std::atomic<Node*> tail_;

public:
    MichaelScottQueue() {
        Node* dummy = new Node(T{});
        head_.store(dummy, std::memory_order_relaxed);
        tail_.store(dummy, std::memory_order_relaxed);
    }

    ~MichaelScottQueue() {
        while (Node* node = head_.load(std::memory_order_relaxed)) {
            head_.store(node->next.load(std::memory_order_relaxed),
                        std::memory_order_relaxed);
            delete node;
        }
    }

    // 禁止拷贝
    MichaelScottQueue(const MichaelScottQueue&) = delete;
    MichaelScottQueue& operator=(const MichaelScottQueue&) = delete;

    template<typename U>
    void enqueue(U&& value) {
        Node* new_node = new Node(std::forward<U>(value));
        Node* cur_tail = nullptr;
        Node* cur_tail_next = nullptr;

        while (true) {
            cur_tail = tail_.load(std::memory_order_acquire);
            cur_tail_next = cur_tail->next.load(std::memory_order_acquire);

            // 验证 tail 仍然是最新的
            if (cur_tail != tail_.load(std::memory_order_acquire))
                continue;

            if (cur_tail_next == nullptr) {
                // 尝试链接新节点：CAS 必须是 weak，因为这是循环内的 CAS
                if (cur_tail->next.compare_exchange_weak(
                        cur_tail_next, new_node,
                        std::memory_order_release,
                        std::memory_order_relaxed)) {
                    break;  // 链接成功
                }
                // CAS 失败说明有其他线程已经链接了新节点，继续重试
            } else {
                // tail 已落后，帮助推进 tail
                tail_.compare_exchange_weak(cur_tail, cur_tail_next,
                    std::memory_order_release, std::memory_order_relaxed);
            }
        }

        // 尝试更新 tail 指针（失败也无妨，后续入队会帮助推进）
        tail_.compare_exchange_strong(cur_tail, new_node,
            std::memory_order_release, std::memory_order_relaxed);
    }

    std::optional<T> dequeue() {
        Node* cur_head = nullptr;
        Node* cur_tail = nullptr;
        Node* cur_head_next = nullptr;
        T value{};

        while (true) {
            cur_head = head_.load(std::memory_order_acquire);
            cur_tail = tail_.load(std::memory_order_acquire);
            cur_head_next = cur_head->next.load(std::memory_order_acquire);

            // 验证快照一致性
            if (cur_head != head_.load(std::memory_order_acquire))
                continue;

            if (cur_head == cur_tail) {
                if (cur_head_next == nullptr) {
                    return std::nullopt;  // 队列为空
                }
                // tail 已落后，帮助推进
                tail_.compare_exchange_weak(cur_tail, cur_head_next,
                    std::memory_order_release, std::memory_order_relaxed);
            } else {
                // 先读出值，再 CAS 移动 head
                value = cur_head_next->data;
                if (head_.compare_exchange_weak(
                        cur_head, cur_head_next,
                        std::memory_order_release,
                        std::memory_order_relaxed)) {
                    break;  // 出队成功
                }
                // CAS 失败：其他线程先移动了 head，重试
            }
        }

        delete cur_head;  // 安全释放旧 dummy node
        return value;
    }

    bool empty() const {
        Node* h = head_.load(std::memory_order_acquire);
        Node* t = tail_.load(std::memory_order_acquire);
        Node* n = h->next.load(std::memory_order_acquire);
        return (h == t) && (n == nullptr);
    }
};

4.2 Ordering 选择的深度分析

在这段代码中，每处 ordering 的选取都有明确的技术原因：

• tail_.load 和 head_.load → acquire：确保看到其他线程通过 release store 更新的最新指针，以及该指针关联的数据（节点的 next 指针和 data 字段）
• cur_head->next.load → acquire：必须看到 enqueuer 在 release store 之前对节点 data 的完整写入
• node::next CAS → release（成功时）：确保新节点的 data 写入被后续 acquire load 可见。这是新数据点 visible 的关键 hapens-before 链接
• tail_.compare_exchange_weak → release + relaxed（失败）：成功时需要 semantically 语义，失败时无需同步，用 relaxed 节省开销量
• head_.compare_exchange_weak → release（成功时）：dequeue 成功后通知 tail 的推进以及可能的 destroy，需要 release 确保自己的修改对后续操作者可见
• 队列为空判断和 tail 帮助推进 → acquire + release：先 acquire 读取最新状态，再用 release 语义完成跨线程同步

关键点：我们用 CAS 的 release ordering .publish了新节点的 next 指针，而 dequeuer 通过 acquire s看到它——这就是 happens-before 的完整链条。

五、ABA 问题与解决方案

ABA 问题：线程 1 CAS 时期望 A，但此时变量实际经历了 A→B→A 的变化，CAS 会错误地成功。在无锁数据结构中，这会导致引用已被释放的内存，造成 use-before-free。

5.1 Tagged Pointer（标签指针）方案

利用对齐后指针未使用的低位作为版本计数器。在 64 位系统上，由于对齐要求（通常 8 字节或 16 字节对齐），指针的低位恒为 0，可以用来存储 tag。

#include <cstdint>
#include <cassert>

template<typename T>
class TaggedPointer {
    static constexpr uint64_t PTR_MASK = 0x0000FFFFFFFFFFFFULL;
    static constexpr uint64_t TAG_MASK = 0xFFFF000000000000ULL;
    static constexpr int      TAG_SHIFT = 48;

    // 假设 48 位虚拟地址空间（x86-64 架构标准）
    std::atomic<uint64_t> combined_{0};

public:
    explicit TaggedPointer(T* ptr = nullptr, uint16_t tag = 0) {
        combined_.store(combine(ptr, tag), std::memory_order_relaxed);
    }

    static uint64_t combine(T* ptr, uint16_t tag) {
        return (reinterpret_cast<uint64_t>(ptr) & PTR_MASK) |
               (static_cast<uint64_t>(tag) << TAG_SHIFT);
    }

    T* pointer() const {
        return reinterpret_cast<T*>(combined_.load(std::memory_order_acquire) & PTR_MASK);
    }

    uint16_t tag() const {
        return static_cast<uint16_t>(
            combined_.load(std::memory_order_acquire) >> TAG_SHIFT);
    }

    bool compare_and_swap(T* expected_ptr, uint16_t expected_tag,
                          T* desired_ptr, uint16_t desired_tag) {
        uint64_t expected = combine(expected_ptr, expected_tag);
        uint64_t desired  = combine(desired_ptr,  desired_tag);
        return combined_.compare_exchange_strong(expected, desired,
            std::memory_order_acq_rel, std::memory_order_acquire);
    }
};

这个方案巧妙地扩展了 CAS 的粒度：即使指针值回到 A，tag 也已递增，CAS 就不会误匹配。当然，64 位系统地址可能超过 48 位（Intel 5-level paging 用了 57 位），此时可以只用 32 位 tag，配合 16 字节对齐来借用 4 个低位。

5.2 Hazard Pointer（生产级可选方案）

Hazard Pointer 是工业标准方案（Facebook Folly、p0f::concurrent 等均提供）。每个线程将自己的"危险指针"写入全局表，表示该线程正在使用该出。其他线程释放节点前必须检查所有线程的 hazard pointer，看是否有正在引用的——只有没有线程引用才真正释放。

#include <atomic>
#include <array>
#include <thread>
#include <mutex>

class EpochBasedReclamation {
    static constexpr int NUM_EPOCHS = 3;

    alignas(64) std::atomic<uint64_t> global_epoch_{0};
    alignas(64) std::atomic<int> reader_count_{0};

    // 每 epoch 一个退休列表
    std::array<std::vector<void(*)(void*)>, NUM_EPOCHS> retired_lists_;
    std::mutex retire_mutex_;

    // TLS: 每个线程自己的活跃 epoch
    static thread_local uint64_t local_epoch_;

public:
    void enter_critical() {
        local_epoch_ = global_epoch_.load(std::memory_order_acquire);
        reader_count_.fetch_add(1, std::memory_order_acq_rel);
    }

    void exit_critical() {
        reader_count_.fetch_sub(1, std::memory_order_release);
    }

    template<typename T>
    void retire(T* ptr) {
        std::lock_guard lock(retire_mutex_);
        auto current_epoch = global_epoch_.load(std::memory_order_relaxed);
        retired_lists_[current_epoch % NUM_EPOCHS].push_back(
            [ptr](void* p){ delete static_cast<T*>(p); }
        );
        // 偶尔尝试推进 epoch
        try_advance_epoch();
    }

private:
    void try_advance_epoch() {
        auto current = global_epoch_.load(std::memory_order_relaxed);
        if (reader_count_.load(std::memory_order_acquire) == 0) {
            global_epoch_.compare_exchange_strong(current, current + 1,
                std::memory_order_acq_rel, std::memory_order_relaxed);
        }
        // 释放 epoch (current - 2) 对应的退休列表
        auto safe_to_free = (current + 1) % NUM_EPOCHS;
        std::lock_guard lock(retire_mutex_);
        retired_lists_[safe_to_free].clear();
    }
};

#include <atomic>

struct alignas(64) alignas(std::hardware_destructive_interference_size) Slot {
    int value = 0;
    int flag = 0; // 非 atomic 字段
};

Slot slot;

void atomic_update() {
    // 对 slot.flag 做原子操作，而不需要将整个 Slot 设为 atomic
    std::atomic_ref<int> atomic_flag(slot.flag);
    atomic_flag.store(1, std::memory_order_release);

    // 对 slot.value 做 RMW
    std::atomic_ref<int> atomic_value(slot.value);
    atomic_value.fetch_add(42, std::memory_order_acq_rel);
}

#include <memory>
#include <atomic>

std::atomic<std::shared_ptr<Config>> global_config{std::make_shared<Config>()};

// 线程安全地读 Config（lock-free）
std::shared_ptr<Config> read_config() {
    return atomic_load_explicit(&global_config, std::memory_order_acquire);
}

// 线程安全地热更新 Config（lock-free）
void update_config(std::shared_ptr<Config> new_config) {
    atomic_store_explicit(&global_config, std::move(new_config),
                         std::memory_order_release);
}

#include <latch>
#include <barrier>
#include <vector>
#include <thread>

constexpr int NUM_THREADS = 8;

void parallel_work() {
    std::latch ready{1};
    std::latch go{1};
    std::barrier<> sync_point(NUM_THREADS,
        [total = NUM_THREADS]() noexcept { /* 每个围栏阶段完成后 */ });
    std::vector<std::jthread> workers;

    for (int i = 0; i < NUM_THREADS; ++i) {
        workers.emplace_back([&, i] {
            // 1. 各自初始化
            init_local_state(i);

            // 2. 主线程确保所有人初始化
            ready.arrive_and_wait();
            go.arrive_and_wait();

            // 3. 并行工作
            do_work(i);

            // 4. 等待所有人完成
            sync_point.arrive_and_wait();

            // 5. 合并结果
            merge_results(i);
        });
    }
}

#include <semaphore>

template<typename T, size_t Capacity>
class BoundedSPSCQueue {
    std::array<T, Capacity> buffer_;
    alignas(64) std::atomic<size_t> head_{0};
    alignas(64) std::atomic<size_t> tail_{0};
    std::counting_semaphore<> nempty_{0};
    std::counting_semaphore<> nfull_{Capacity};

public:
    void produce(T value) {
        nfull_.acquire();  // 等待有空位

        size_t t = tail_.load(std::memory_order_relaxed);
        buffer_[t % Capacity] = std::move(value);
        tail_.store(t + 1, std::memory_order_release);

        nempty_.release();  // 通知有数据
    }

    T consume() {
        nempty_.acquire();  // 等待有数据

        size_t h = head_.load(std::memory_order_acquire);
        T val = std::move(buffer_[h % Capacity]);
        head_.store(h + 1, std::memory_order_release);

        nfull_.release();  // 通知有空位
        return val;
    }
};