C++ 内存模型与无锁编程

引言：为什么内存模型至关重要

在多核处理器架构下，编译器优化和CPU缓存重排序会导致内存操作在不同线程中呈现不一致的视图。C++11首次在语言标准中定义了正式的内存模型（Memory Model），为并发编程提供了可移植的保证。对于架构师而言，深入理解内存模型不仅是正确编写无锁数据结构的前提，更是设计高性能并发系统的理论基础。

本文将从内存序的语义出发，逐步建立对happens-before关系的理解，最终推导出生产级MPMC无锁队列的完整实现。

内存序模型详解

C++标准定义了六种内存序（memory_order），从最弱到最强依次为：

#include 
#include 

// memory_order_relaxed：无序，仅保证原子性
std::atomic counter\{0\};
void relaxed_increment() {
    counter.fetch_add(1, std::memory_order_relaxed);
    // 仅保证读-改-写是原子的，不保证与其他操作的顺序
}

// memory_order_acquire/release：同步对
// 线程A（生产者）
std::atomic ready\{false\};
int shared_data = 0;

void producer() {
    shared_data = 42;  // 普通写
    ready.store(true, std::memory_order_release);
    // release保证：所有之前的写操作对acquire线程可见
}

// 线程B（消费者）
void consumer() {
    while (!ready.load(std::memory_order_acquire))
        ;  // 自旋等待
    // acquire保证：能看到release之前的所有写
    assert(shared_data == 42);  // 安全
}

// memory_order_seq_cst：顺序一致性（默认值）
std::atomic seq_counter\{0\};
void seq_cst_example() {
    seq_counter.fetch_add(1, std::memory_order_seq_cst);
    // 最强保证：全局全序，但性能最差
}

// memory_order_acq_rel：读-改-写操作同时获取和释放
void acq_rel_example(std::atomic& flag) {
    int expected = 0;
    flag.compare_exchange_strong(expected, 1,
        std::memory_order_acq_rel,
        std::memory_order_acquire);
    // 成功时acq_rel，失败时acquire
}

内存序的选择直接影响硬件层面的指令生成。relaxed在x86上几乎零开销，在ARM上可能省略DMB屏障；release/acquire在x86上利用已有的Store Buffer隐式保证，在ARM上需要DMB/ISB指令；seq_cst可能需要MFENCE或更重的屏障。选择恰当的内存序是性能优化的关键。

happens-before 关系建立

happens-before是C++内存模型的核心概念，它定义了操作之间的可见性关系。如果操作A happens-before 操作B，则A对内存的修改对B可见。

// happens-before关系的建立方式：
// 1. 顺序一致性（sequenced-before）：同一线程内的执行顺序
// 2. 传递性（transitivity）：A hb B, B hb C => A hb C
// 3. 原子操作的release-acquire同步
// 4. mutex的lock-unlock同步
// 5. 线程创建/汇合同步

// 同步关系链示例
std::atomic flag1\{0\}, flag2\{0\};
int data = 0;

// 线程1
void thread1() {
    data = 1;                              // (1)
    flag1.store(1, std::memory_order_release); // (2) release
}

// 线程2
void thread2() {
    while (flag1.load(std::memory_order_acquire) != 1) \{\} // (3) acquire
    // (2) synchronizes-with (3) => (1) hb (3) => data==1
    flag2.store(1, std::memory_order_release);              // (4)
}

// 线程3
void thread3() {
    while (flag2.load(std::memory_order_acquire) != 1) \{\} // (5) acquire
    // (4) sw (5) => (1) hb (5) => data==1 保证可见
    assert(data == 1);                                       // (6)
}

// 传递性确保线程3能看到线程1的所有写操作
// 即使线程1和线程3之间没有直接的同步关系

理解happens-before的关键在于"同步关系"（synchronizes-with）。Release-acquire对是建立跨线程同步关系的主要手段。一旦同步关系建立，happens-before的传递性确保整个修改链对后续线程可见。

原子操作与内存屏障

原子操作是构建无锁数据结构的基石。C++提供了丰富的原子操作API，从简单的load/store到复杂的CAS（Compare-And-Swap），每一种操作都可以指定内存序。

#include 
#include 

// 原子标志：无锁自旋锁
class SpinLock {
    std::atomic locked_\{false\};
public:
    void lock() {
        // CAS操作：期望false，设为true
        while (locked_.exchange(true, std::memory_order_acquire))
            ;  // 自旋
    }
    
    void unlock() {
        locked_.store(false, std::memory_order_release);
    }
};

// 带退避策略的自旋锁
class BackoffSpinLock {
    std::atomic locked_\{false\};
    static constexpr int max_backoff = 16;
public:
    void lock() {
        int backoff = 1;
        while (locked_.exchange(true, std::memory_order_acquire)) {
            for (int i = 0; i < backoff; ++i) \{
                __builtin_ia32_pause();  // CPU暂停指令，降低功耗
            \}
            backoff = std::min(backoff * 2, max_backoff);
        }
    }
    
    void unlock() {
        locked_.store(false, std::memory_order_release);
    }
};

// 指针原子操作：无锁链表基础
struct Node \{
    int data;
    Node* next;
\};

std::atomic head\{nullptr\};

void push_front(int value) {
    Node* new_node = new Node\{value, head.load(std::memory_order_relaxed)\};
    // CAS循环：直到成功插入
    while (!head.compare_exchange_weak(
        new_node->next, new_node,
        std::memory_order_release,
        std::memory_order_relaxed)) \{
        // compare_exchange_weak可能虚假失败，需要重试
    \}
}

compare_exchange_weak与compare_exchange_strong的区别在于：weak允许虚假失败（spurious failure），通常在循环中使用以获得更好的性能（在LL/SC架构如ARM上）；strong保证只有在期望值确实不匹配时才失败。选择哪个取决于使用场景：循环中使用weak，单次尝试使用strong。

lock-free 队列与无锁哈希表

无锁数据结构是并发编程的巅峰挑战。它们的核心思路是使用CAS操作替代互斥锁，使得线程永远不会因为等待锁而阻塞。

// Michael-Scott无锁队列（SPSC简化版）
template
class LockFreeQueue \{
    struct Node \{
        std::shared_ptr data;
        std::atomic next;
        Node() : next(nullptr) \{\}
        explicit Node(T val) 
            : data(std::make_shared(std::move(val)))
            , next(nullptr) \{\}
    \};
    
    std::atomic head_;
    std::atomic tail_;
    // 注意：此实现省略了完整的内存回收机制
    // 生产环境需配合Hazard Pointer或Epoch-based回收
    
public:
    LockFreeQueue() \{
        Node* sentinel = new Node();
        head_.store(sentinel, std::memory_order_relaxed);
        tail_.store(sentinel, std::memory_order_relaxed);
    \}
    
    ~LockFreeQueue() \{
        while (dequeue());  // 清空队列
        delete head_.load(std::memory_order_relaxed);
    \}
    
    void enqueue(T value) \{
        Node* new_node = new Node(std::move(value));
        Node* old_tail = tail_.load(std::memory_order_relaxed);
        
        while (true) \{
            Node* next = old_tail->next.load(std::memory_order_relaxed);
            if (old_tail == tail_.load(std::memory_order_acquire)) \{
                if (next == nullptr) \{
                    if (old_tail->next.compare_exchange_weak(
                        next, new_node,
                        std::memory_order_release,
                        std::memory_order_relaxed)) \{
                        break;  // 成功链接
                    \}
                \} else \{
                    // 帮助其他线程推进tail
                    tail_.compare_exchange_weak(
                        old_tail, next,
                        std::memory_order_release,
                        std::memory_order_relaxed);
                \}
            \}
            old_tail = tail_.load(std::memory_order_relaxed);
        \}
        // 尝试推进tail（可以失败，其他线程会帮助）
        tail_.compare_exchange_weak(
            old_tail, new_node,
            std::memory_order_release,
            std::memory_order_relaxed);
    \}
    
    bool dequeue(T& result) \{
        Node* old_head = head_.load(std::memory_order_relaxed);
        while (true) \{
            Node* next = old_head->next.load(std::memory_order_acquire);
            if (old_head == head_.load(std::memory_order_acquire)) \{
                if (next == nullptr) return false;  // 队列空
                result = std::move(*next->data);
                if (head_.compare_exchange_weak(
                    old_head, next,
                    std::memory_order_release,
                    std::memory_order_relaxed)) \{
                    // 注意：不能立即delete old_head（ABA问题）
                    // 需要延迟回收机制
                    break;
                \}
            \}
            old_head = head_.load(std::memory_order_relaxed);
        \}
        return true;
    \}
\};

ABA 问题与 Hazard Pointer

ABA问题是所有CAS-based无锁算法的天敌。它发生在以下场景：线程A读取值V，线程A被挂起，线程B将V改为U再改回V，线程A恢复后CAS成功但数据已被破坏。

// ABA问题的经典场景
// 线程A读取head = node_A
// 线程B: pop node_A, node_A被回收
// 线程B: push node_C, node_C被分配到node_A相同的地址
// 线程A: CAS(head, node_A, node_A->next) 成功！
// 但node_A->next已经是node_C的next，数据被破坏

// 解决方案1：带标签的指针
template
struct TaggedPtr \{
    T* ptr;
    uint64_t tag;
\};

// 使用64位原子操作（在64位平台上，ptr+tag压缩到一个128位CAS）
// 解决方案2：Hazard Pointer

template
class HazardPointer \{
    static constexpr size_t HP_COUNT = 
        std::thread::hardware_concurrency();
    static constexpr size_t HP_PER_THREAD = 2;
    
    struct HPRecord \{
        std::atomic pointer\{nullptr\};
        std::atomic next\{nullptr\};
        std::thread::id owner;
    \};
    
    std::atomic head_\{nullptr\};
    
    // 每线程Hazard Pointer槽位
    static thread_local HPRecord* my_hp[HP_PER_THREAD];
    
public:
    HPRecord* acquire_hp(size_t index) \{
        if (!my_hp[index]) \{
            my_hp[index] = new HPRecord();
            my_hp[index]->owner = std::this_thread::get_id();
            HPRecord* old_head = head_.load(std::memory_order_relaxed);
            do \{
                my_hp[index]->next.store(old_head, std::memory_order_relaxed);
            \} while (!head_.compare_exchange_weak(
                old_head, my_hp[index],
                std::memory_order_release,
                std::memory_order_relaxed));
        \}
        return my_hp[index];
    \}
    
    void protect(size_t index, void* ptr) \{
        my_hp[index]->pointer.store(ptr, std::memory_order_release);
    \}
    
    void release(size_t index) \{
        my_hp[index]->pointer.store(nullptr, std::memory_order_release);
    \}
    
    // 检查指针是否被其他线程保护
    bool is_protected(void* ptr) \{
        HPRecord* curr = head_.load(std::memory_order_acquire);
        while (curr) \{
            if (curr->pointer.load(std::memory_order_acquire) == ptr)
                return true;
            curr = curr->next.load(std::memory_order_acquire);
        \}
        return false;
    \}
    
    // 延迟回收：检查retire list中的节点
    void retire(void* ptr, void (*deleter)(void*)) \{
        // 简化：如果未被保护则立即删除
        // 完整实现需要维护retire list并定期扫描
        if (!is_protected(ptr)) \{
            deleter(ptr);
        \}
        // 否则加入retire list，等待后续回收
    \}
\};

除了Hazard Pointer，还有Epoch-based Reclamation（基于纪元的回收）和RCU（Read-Copy-Update）等方案。选择哪种取决于场景：Hazard Pointer适合低竞争环境，Epoch-based适合读多写少，RCU适合极低延迟的读路径。

实战：实现 MPMC 无锁队列

MPMC（Multi-Producer Multi-Consumer）无锁队列是并发编程的终极挑战之一。以下实现基于Vyukov队列的变体，使用固定大小的环形缓冲区来消除内存回收的复杂性。

#include 
#include 
#include 
#include 

template
class MPMCBoundedQueue \{
    // 每个槽位的状态
    enum class CellState : uint8_t \{
        Empty = 0,
        Writing,
        Ready,
        Reading
    \};
    
    struct Cell \{
        std::atomic state\{CellState::Empty\};
        alignas(alignof(T)) unsigned char storage[sizeof(T)];
        
        T* ptr() \{ return reinterpret_cast(storage); \}
        const T* ptr() const \{ 
            return reinterpret_cast(storage); 
        \}
    \};
    
    // 缓存行对齐，避免false sharing
    struct alignas(64) PaddedIndex \{
        std::atomic value\{0\};
    \};
    
    std::vector buffer_;
    PaddedIndex enqueue_pos_;
    PaddedIndex dequeue_pos_;
    
    static constexpr size_t MASK = Capacity - 1;
    static_assert((Capacity & (Capacity - 1)) == 0,
        "Capacity must be power of 2");
    
public:
    MPMCBoundedQueue() : buffer_(Capacity) \{\}
    ~MPMCBoundedQueue() \{
        // 析构残留元素
        T dummy;
        while (try_dequeue(dummy));
    \}
    
    bool try_enqueue(T item) \{
        CellState expected = CellState::Empty;
        size_t pos = enqueue_pos_.value.load(
            std::memory_order_relaxed);
        
        while (true) \{
            Cell& cell = buffer_[pos & MASK];
            CellState state = cell.state.load(
                std::memory_order_acquire);
            
            switch (state) \{
                case CellState::Empty:
                    expected = CellState::Empty;
                    if (cell.state.compare_exchange_weak(
                        expected, CellState::Writing,
                        std::memory_order_acquire,
                        std::memory_order_relaxed)) \{
                        goto write;
                    \}
                    break;
                case CellState::Ready:
                case CellState::Writing:
                    // 槽位被占用，推进位置
                    size_t next = enqueue_pos_.value.load(
                        std::memory_order_acquire);
                    if (pos == next) \{
                        // 尝试推进enqueue_pos
                        enqueue_pos_.value.compare_exchange_strong(
                            pos, pos + 1,
                            std::memory_order_relaxed);
                        pos = next + 1;
                    \} else \{
                        pos = next;
                    \}
                    break;
                case CellState::Reading:
                    // 消费者正在读取，跳过
                    if (pos == enqueue_pos_.value.load(
                        std::memory_order_relaxed)) \{
                        return false;  // 队列满
                    \}
                    ++pos;
                    break;
            \}
        \}
        
    write:
        // 在指定槽位写入数据（placement new）
        new (cell.ptr()) T(std::move(item));
        cell.state.store(CellState::Ready, 
            std::memory_order_release);
        
        size_t old_pos = pos;
        enqueue_pos_.value.compare_exchange_strong(
            old_pos, pos + 1,
            std::memory_order_release);
        return true;
    \}
    
    bool try_dequeue(T& result) \{
        size_t pos = dequeue_pos_.value.load(
            std::memory_order_relaxed);
        
        while (true) \{
            Cell& cell = buffer_[pos & MASK];
            CellState state = cell.state.load(
                std::memory_order_acquire);
            
            switch (state) \{
                case CellState::Ready:
                    CellState expected = CellState::Ready;
                    if (cell.state.compare_exchange_weak(
                        expected, CellState::Reading,
                        std::memory_order_acquire,
                        std::memory_order_relaxed)) \{
                        goto read;
                    \}
                    break;
                case CellState::Empty:
                    // 尝试推进dequeue_pos
                    \{
                        size_t old_pos = pos;
                        dequeue_pos_.value.compare_exchange_strong(
                            old_pos, pos + 1,
                            std::memory_order_relaxed);
                    \}
                    pos = dequeue_pos_.value.load(
                        std::memory_order_acquire);
                    if (pos > enqueue_pos_.value.load(
                        std::memory_order_acquire)) \{
                        return false;  // 队列空
                    \}
                    break;
                default:
                    ++pos;
                    break;
            \}
        \}
        
    read:
        result = std::move(*cell.ptr());
        cell.ptr()->~T();
        cell.state.store(CellState::Empty, 
            std::memory_order_release);
        
        size_t old_pos = pos;
        dequeue_pos_.value.compare_exchange_strong(
            old_pos, pos + 1,
            std::memory_order_release);
        return true;
    \}
\};

这个MPMC队列的关键设计决策包括：(1) 固定容量消除内存回收问题；(2) 四状态机（Empty/Writing/Ready/Reading）替代简单的布尔标志；(3) 缓存行对齐避免false sharing；(4) CAS循环中的帮助机制确保并发安全。在实际生产中，还需要考虑队列满时的背压策略、大对象的存储优化（使用索引而非直接存储对象）等。

结语

C++内存模型为并发编程提供了坚实的理论基础。从relaxed到seq_cst的内存序梯度，为架构师提供了性能与正确性之间的精细控制能力。无锁数据结构虽然实现复杂，但在特定场景下（高吞吐、低延迟、锁竞争剧烈）能提供显著的性能优势。然而，工程实践中应当遵循"先正确后优化"的原则：优先使用互斥锁和标准库并发原语，只有在性能剖析（profiling）证明锁是瓶颈之后，才考虑无锁方案。成熟的并发库（如Intel TBB、Folly、concurrentqueue）已经封装了复杂的技术细节，架构师应当在理解原理的基础上善用这些工具。