C++ CAS 原子操作深度解析与实战

这里需要特别注意 expected 是 T& 引用类型。当 CAS 失败时，函数不仅返回 false，还会自动将内存中最新的实际值写入 expected 变量中，这正是我们需要的行为——让下一次循环使用最新的状态。

4. ABA 问题

ABA 问题是指在使用 CAS 操作时，一个变量的值经历了 A -> B -> A 的变化过程，但 CAS 仅仅检查值是否相等，因此误认为该变量从未被修改过，从而导致逻辑错误。

核心本质在于 CAS 只能检测'状态'是否一致，无法检测'时序'是否发生过变化。

我们可以举一个例子来形象地形容：桌子上有一杯水（状态 A）。你看到桌上有杯水，转身去接电话。室友过来把水喝完了（变成状态 B），又接了一杯水放在原处（变回状态 A）。你回来看到水还在，端起来就喝。虽然看起来还是水，但杯子可能被污染了，或者水的温度变了。你以为没变，实际上已经'偷梁换柱'了。

ABA 问题对于一些纯数值修改中通常无害，但在涉及指针和内存管理的动态数据结构中是致命的。

假设我们有一个无锁栈，栈顶指针为 Top。栈内元素：Top -> Node1 -> Node2。

故障出现过程：

1. 线程 1 准备出栈（Pop）：读取 Top 为 Node1。读取 Node1->next 为 Node2。（此时线程 1 被操作系统调度挂起）
2. 线程 2 疯狂操作：
   • Pop：将 Node1 弹出，此时 Top 变为 Node2。
   • Pop：将 Node2 弹出，此时 Top 变为 nullptr。
   • Push：线程 2 创建了一个新节点，恰好操作系统的内存分配器将刚刚释放的 Node1 的内存地址重用给了这个新节点。
   • Push：将这个新节点（地址仍然是 Node1）压入栈。
   • 现状：Top 指向了新 Node1，但 Node1->next 不再指向 Node2（因为 Node2 已经被释放了）。
3. 线程 1 恢复执行：
   • 执行 CAS 操作：compare_exchange(Top, expected=Node1, new=Node2)。
   • 判断：检查 Top 当前的值，发现地址确实是 Node1（尽管内容变了，但地址没变）。
   • 交换：CAS 成功！线程 1 将 Top 修改为 Node2。

后果：Top 现在指向了 Node2。但是 Node2 早在步骤 2 中被线程 2 释放回内存池了！任何后续访问 Top 的操作都会访问野指针，导致程序崩溃或数据错乱。

从底层原因看，首先是操作系统或内存分配器倾向于复用刚刚释放的内存块（热缓存），而且 CAS 是浅层比较，它只管二进制数据是否相同。

对于以上 ABA 问题，我们可以采取版本号、std::atomic<std::shared_ptr>、Hazard Pointers 来解决。在此，我们介绍下利用版本号进行控制。

首先定义一个结构体，它将指针和版本号捆绑在一起：

// 假设 Node 是栈中的节点结构
struct Node {
    int data;
    Node* next;
};

// 复合状态结构体
struct PointerWithVersion {
    Node* pointer; // 栈顶指针 (Top)
    unsigned int version; // 版本计数器
};

// 栈顶指针现在是这个复合结构的原子变量
std::atomic<PointerWithVersion> top;

然后实现正确的 Pop 操作：

// 模拟无锁栈的 Pop 操作
Node* pop() {
    // 1. 原子地读取当前的 (指针 A, 版本 v1) 状态
    PointerWithVersion expected = top.load();
    
    // 2. 构造线程 1 想要的新状态：指针变为 A->next，版本号递增
    PointerWithVersion desired;
    if (expected.pointer == nullptr) {
        return nullptr;
    }
    
    // 线程 1 进入 CAS 循环
    do {
        // 确保在循环内读取最新状态，因为 expected 可能会在 CAS 失败时被更新
        Node* current_node = expected.pointer;
        
        // 构造新状态：指向下一个节点，版本号递增
        desired.pointer = current_node->next;
        desired.version = expected.version + 1;
        
        // 关键的原子操作
        // 如果 top 仍等于 expected，则 top = desired，返回 true
        // 如果 top 不等于 expected，则 top 被更新为 top 的最新值，返回 false，继续循环
    } while (!top.compare_exchange_weak(expected, desired));
    
    // CAS 成功，current_node 已被弹出
    return expected.pointer;
}

现在我们来分析一下 ABA 问题的解决过程，利用表格进行对照：

版本号将 CAS 的比较粒度从一个地址提升到了地址与历史状态的组合。即使指针 A 变回了 A，但版本号 v1 已经变成了 v3。线程 1 的预期值 A_{v1} 永远不会等于内存中的最新值 A_{v3}，因此 CAS 会失败，线程被迫读取最新的 (Node1, 3) 状态并重试，从而阻止了它将栈顶链接到早已被释放的旧节点上。

5. 自旋锁

5.1 原子'交换'

使用 std::atomic<bool> 实现锁，并不是简单的 if 判断，而是利用了 原子的'读 - 改 - 写' 能力。

加锁逻辑：

线程尝试将布尔值设置为 true（表示'我占用了'）。exchange(true) 函数会把变量设为 true，并返回旧值。如果返回 false：说明之前是解锁状态，我现在抢到了锁。成功！如果返回 true：说明之前已经是 true（被别人占用了）。失败，继续循环重试（自旋）。

解锁逻辑：

将布尔值设回 false（store(false)）。

5.2 代码实现

#include <atomic>
#include <thread>

class SpinLock {
private:
    std::atomic<bool> flag = {false}; // false 代表未加锁，true 代表已加锁

public:
    void lock() {
        // exchange(true) 会把 flag 设为 true，并返回原来的值。
        // 只要返回 true，说明原来就被锁住了，我就在 while 里一直空转（Spin）。
        // 只有返回 false，说明原来没锁，我成功锁住了，跳出循环。
        // std::memory_order_acquire 保证之后的代码不会被重排到这一行之前
        while (flag.exchange(true, std::memory_order_acquire)) {
        }
    }

    void unlock() {
        // 解锁很简单，直接设为 false。
        // std::memory_order_release 保证之前的代码不会被重排到这一行之后
        flag.store(false, std::memory_order_release);
    }
};

可以利用该自旋锁代替 std::mutex，进行性能对比：

在 C++ 中，最标准的自旋锁其实推荐使用 std::atomic_flag，而 std::atomic<bool> 上述实现逻辑是为了更好的理解自旋锁的形成。它的接口非常简单，专门为自旋锁设计：

test_and_set(): 相当于 exchange(true)。 clear(): 相当于 store(false)。