从‘++i’崩溃说起：深入理解C++ atomic的compare_exchange_weak与强内存屏障

从‘++i’崩溃说起：深入理解C++ atomic的compare_exchange_weak与强内存屏障

在某个深夜的调试中，你盯着屏幕上那个看似简单的计数器——++shared_counter——它本应在多线程环境下稳定递增，却总是莫名其妙地丢失更新。这个场景或许唤起了许多C++开发者的痛苦记忆：当并发编程遇上硬件优化，简单的自增操作背后隐藏着令人抓狂的复杂性。

1. 为什么++i不是线程安全的？

假设我们有一个全局变量int counter = 0，四个线程同时执行以下代码：

for (int i = 0; i < 10000; ++i) { ++counter; }

理论上最终结果应该是40000，但实际运行可能得到32768、28491等随机值。这种诡异现象源于三个层面的交互：

1. 编译器优化：编译器可能将++counter优化为寄存器操作，延迟写回内存
2. CPU流水线：现代CPU的乱序执行会打乱指令顺序
3. 缓存一致性：多核CPU的缓存更新存在延迟

考虑这段代码的汇编展开：

mov eax, [counter] ; 读取counter到寄存器 inc eax ; 寄存器加1 mov [counter], eax ; 写回内存

当两个线程同时执行时，可能出现以下交错执行：

最终counter=1，而两个线程各执行了一次自增，这就是典型的丢失更新问题。

2. atomic的基础救赎

C++11引入的std::atomic模板类提供了真正的原子操作。将变量声明为：

std::atomic<int> counter(0);

此时++counter会生成原子性的fetch_add指令。x86架构下对应的汇编是：

lock xadd dword ptr [counter], eax

lock前缀会触发CPU的缓存锁定机制，确保该指令执行期间其他核心无法访问同一缓存行。但原子操作只是解决了可见性问题，真正的挑战在于内存顺序。

3. 内存序：性能与正确性的权衡

C++提供了六种内存序，从强到弱排列：

考虑一个典型的生产者-消费者场景：

std::atomic<bool> ready{false}; int data = 0; // 生产者线程 data = 42; // (1) ready.store(true, std::memory_order_release); // (2) // 消费者线程 while (!ready.load(std::memory_order_acquire)); // (3) assert(data == 42); // (4)

这里release与acquire形成同步关系，确保(1)的写入对(4)可见。如果使用memory_order_relaxed，断言可能失败。

4. compare_exchange：无锁算法的基石

compare_exchange_weak和compare_exchange_strong是构建无锁数据结构的核心操作，其伪代码逻辑如下：

bool compare_exchange(T& expected, T desired) { if (*this == expected) { *this = desired; return true; } else { expected = *this; return false; } }

两者的区别在于：

• weak版本允许虚假失败（即使值匹配也可能返回false）
• strong版本保证成功当且仅当值匹配

在x86架构下，典型实现是lock cmpxchg指令。下面是一个无锁栈的push操作示例：

template<typename T> class lock_free_stack { struct node { T data; node* next; }; std::atomic<node*> head; public: void push(const T& data) { node* new_node = new node{data, nullptr}; new_node->next = head.load(std::memory_order_relaxed); while (!head.compare_exchange_weak( new_node->next, new_node, std::memory_order_release, std::memory_order_relaxed)) { // 循环直到成功 } } };

这里的内存序参数表示：

• 成功时采用release语义：确保新节点的构造对pop线程可见
• 失败时采用relaxed语义：因为只是重试，不需要同步

5. 实战：实现无锁队列

结合上述技术，我们实现一个多生产者多消费者的无锁队列：

template<typename T> class lock_free_queue { struct node { std::atomic<node*> next; T data; }; std::atomic<node*> head; std::atomic<node*> tail; public: lock_free_queue() : head(new node), tail(head.load()) {} void enqueue(T value) { node* new_node = new node{nullptr, 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 (!next) { if (old_tail->next.compare_exchange_weak( next, new_node, std::memory_order_release, std::memory_order_relaxed)) { break; } } else { tail.compare_exchange_weak( old_tail, next, std::memory_order_relaxed, std::memory_order_relaxed); } old_tail = tail.load(std::memory_order_relaxed); } 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 (!next) return false; if (head.compare_exchange_weak( old_head, next, std::memory_order_release, std::memory_order_relaxed)) { result = std::move(next->data); delete old_head; return true; } } } };

关键设计点：

1. 使用dummy节点简化边界条件处理
2. enqueue时的两步CAS：先链接新节点，再移动tail
3. 内存序的选择：
   • enqueue成功时用release确保数据可见
   • dequeue时用acquire获取最新数据

6. 性能优化与陷阱

在实际项目中应用无锁结构时，需要注意：

ABA问题解决方案

// 使用带标签的指针 template<typename T> struct tagged_ptr { T* ptr; uintptr_t tag; }; std::atomic<tagged_ptr<node>> head;

缓存行优化

struct alignas(64) cache_line_padded { std::atomic<int> counter; char padding[64 - sizeof(std::atomic<int>)]; };

平台特定优化x86架构下，某些内存序可以降级：

// 在x86上等同于memory_order_seq_cst bool compare_exchange_strong( T& expected, T desired, std::memory_order_acq_rel, std::memory_order_acquire);

经过大量测试，不同内存序在x86上的性能对比：

在实现无锁结构时，一个常见错误是过度使用memory_order_seq_cst。实际上，大多数场景只需要acq_rel即可保证正确性。