无锁并发数据结构

从一次成功 CAS 的生效点开始

上一章你给数据结构配了锁：谁要动它就先拿钥匙，别人在门口等。锁很好用，但有个隐患：万一拿着钥匙的厨师突然被叫走（线程被操作系统挂起），手里攥着钥匙不放，其他人就全卡在门口——干等。

这一章换个思路：根本不发钥匙。每个厨师改账前先看一眼「账上还是不是我以为的那个数」——是，就动手改；不是（被别人改过了），就重新看一遍、重新算、再试。没人持钥匙，自然没人能把全场卡死；总有人能往前走。

代价是两个新麻烦。一是你离开一小会儿，账上的数被人换走又换回成原样，你以为没人动过——其实暗地里全变了。二是你把一张单子从台上撤下来，刚要扔掉，却发现还有同事正盯着它看——这单子还不能扔。这一章就教你怎么不上锁地干活，又怎么躲开这两个坑。

什么叫「无锁」：关键不在没锁，在进展保证

无锁（lock-free） 的重点是系统级进展。在参与线程持续执行的前提下，竞争失败意味着别的操作取得了进展；某个暂停线程不会因持有 mutex 阻止所有其他操作。但如果关键 std::atomic<TaggedPtr> 在目标平台内部用锁实现，整个算法不能据此宣称 lock-free。

更强的 无等待（wait-free） 要求每个操作都在有界自身步骤内完成。CAS 重试循环通常只能证明 lock-free，不能证明某个线程不会一直失败。

无锁栈的 push：CAS 重试循环

无锁的活怎么干？核心套路就是上一章 CAS 的「重试循环」形态。以无锁栈的 push 为例，它要把新节点接到栈顶，靠的是 CAS 重试循环：读一眼 head 当作 expected、把新节点的 next 接到它上面，再 CAS(head, expected, 新节点)；如果中途别的线程改了 head，CAS 失败、把真实的 head 回填进来，循环重新接好、再试。下面这张动画把这套自旋一拍一拍演给你看。先猜一猜，再单步：

猜一猜：当 CAS 因为 head 被别的线程抢先改走而失败时，它会做什么——直接报错让你重写，还是顺手把真实的 head 塞给你、让你接着重试？

盯住「失败」那两拍：CAS 不报错，而是把真实的 head 回填进 expected，于是循环体几乎什么都不用写——下一轮自动拿新值重算。这就是无锁「看一眼是不是原样、是才改、不是重来」的全部精髓。代码见第六节。

内存回收难题：弹出的节点为什么不能马上 delete

push 简单，pop 才是无锁的真正硬骨头。pop 用 CAS 把栈顶节点从链上摘下来之后，能不能马上 delete 它？不能。问题在于：别的线程此刻可能正拿着指向这个节点的裸指针——它在自己的 pop 里读了同一个 head、还没来得及解引用。你一 delete，它接下来访问的就是已释放的内存，读到垃圾或直接崩溃。这就是无锁结构的 内存回收（reclamation） 难题：摘下容易，安全释放难。

经典解法之一是 风险指针（hazard pointer）：线程访问某节点前，先在一块大家都能看到的「正在使用」公告板上登记一个指向它的指针；回收线程想删某节点前，先扫这块公告板——只要还有人挂着牌，就把它挂进「待回收」推迟，等没人挂牌了再删。下面这张动画把「挂牌—扫板—推迟—摘牌—重扫—安全删」演一遍：

看清楚：回收不是「摘下就删」，而是「摘下 → 看还有没有人在用 → 没人用了才删」。这正是上面学习目标里那道自测题的答案。引用计数是另一条解法（思路不同，第六节对照），它给节点自带一个计数：有人开始用就 +1、用完 -1，归零才释放。

ABA 问题：被改走又改回，CAS 看不出来

无锁的第二个坑藏在 CAS 自己身上：CAS 只比较「值相不相等」，看不出「被改走又改回」。设想线程1 读到 head 是地址 0xA、准备 CAS，却被打断；线程2 趁机把 0xA 这个节点弹出、释放，又 push 一个复用了 0xA 地址的新节点。等线程1 恢复执行 CAS，发现 head 还是 0xA → 误判「没人动过」→ 成功，可栈结构早已面目全非。这就是 ABA 问题，像你离开一会儿、账上的数被换走又换回成原样、你没察觉。下面这张动画把这条阴险的时间线拆开演：

ABA 的解法之一是 标签指针（tagged pointer，又叫带版本号的指针）：把指针和一个每次修改都自增的版本号捆在一起 CAS——地址变回 0xA、版本号却变了，CAS 一比「指针 + 版本号」整体就发现「动过」，不再误判。另一条思路是让被弹出的地址短期内不被复用（风险指针/引用计数的延迟回收顺带就做到了），从源头掐断 ABA。代码见第六节。

上手玩一玩这三张图

这三张图都是可单步的教具，配合第六节代码一起玩比读十遍管用：

• CAS 重试循环 ``（主 Demo）：单步看「读 head → 备新节点 → CAS 失败（head 被抢成 M）→ 回填真实 head → 重做 → 再 CAS 成功」。重点体会失败不报错、只是带着真实值重来——这就是无锁的自旋。
• 风险指针 ``：单步看线程A 挂牌 hazard→N、线程B 扫板见牌→推迟、线程A 摘牌、线程B 重扫无牌→安全删。盯住「回收前一定先扫板」这一步，理解为什么节点不能摘下就删。
• ABA 问题 ``：单步看线程1 读 A 被打断，线程2 pop A、pop B、又 push 复用 0xA 的 A′，线程1 的 CAS 见地址仍是 0xA 而误判成功。体会「CAS 只看值相等、看不出改走又改回」。

玩熟这三张图，下一节每一句 compare_exchange_weak、每一处「不能 delete」、每一个版本号自增，你都能对上图里某个动作。

代码：无锁栈与两种回收方案

无锁栈 push：CAS 重试循环

push 的全部秘密就在那个 while 里。读一眼 head 接好新节点，再 CAS——失败就靠回填的真实值重试：

#include <atomic>
#include <memory>
 
template <typename T>
class LockFreeStack {
    struct Node {
        std::shared_ptr<T> data;
        Node* next;
        explicit Node(T const& v) : data(std::make_shared<T>(v)) {}
    };
    std::atomic<Node*> head{nullptr};
public:
    void push(T const& value) {
        Node* const new_node = new Node(value);
        new_node->next = head.load(std::memory_order_relaxed);
        while (!head.compare_exchange_weak(         // CAS：head 还是我以为的吗？
                   new_node->next, new_node,        // 是→换成 new_node；不是→真实 head 回填进 new_node->next
                   std::memory_order_release,       // 成功：发布新节点的数据
                   std::memory_order_relaxed)) {    // 失败：只重试，无数据要发布
            // 循环体空着：失败时 new_node->next 已被刷成真实 head，下一轮自动重 CAS
        }
    }
};

循环体之所以能空着，正是因为 CAS 失败时已替你把真实的 head 回填进了 new_node->next——下一轮 compare_exchange_weak 自动拿新值重比（对照那张 CAS 重试动画）。

无锁栈 pop：摘下容易，删除是个雷

pop 同样用 CAS 把栈顶摘下来。但摘下之后那一行，藏着无锁最大的雷：

#include <atomic>
#include <memory>
 
template <typename T>
class LockFreeStack {
    struct Node {
        std::shared_ptr<T> data;
        Node* next;
    };
    std::atomic<Node*> head{nullptr};
public:
    std::shared_ptr<T> pop() {
        Node* old_head = head.load(std::memory_order_acquire);
        // 把栈顶摘下：head 还是 old_head 吗？是→换成 old_head->next
        while (old_head &&
               !head.compare_exchange_weak(
                   old_head, old_head->next,
                   std::memory_order_acquire,
                   std::memory_order_relaxed)) {
            // 失败时 old_head 已被回填成真实 head，重试
        }
        std::shared_ptr<T> res;
        if (old_head) res.swap(old_head->data);   // 取出数据
        // 危险：此刻别的线程的 pop 可能正持有指向 old_head 的指针！
        // 所以这里【绝不能】写 delete old_head; —— 见下文延迟回收
        return res;
    }
};

这段故意不 delete，因此会泄漏节点，只用于展示 CAS 摘链；它不是完整容器。成功 CAS 的 acquire 读取 push 的 release 发布，使节点数据可见。真正实现必须在解引用 old_head->next 之前先完成风险保护或引用计数，否则一旦其他实现开始回收，CAS 前就可能 use-after-free。

解法一：风险指针（公告板登记）

风险指针的骨架是一块全局公告板，加上「访问前登记、回收前扫板」两个动作。先看公告板和扫板：

#include <atomic>
 
// 固定槽表：atomic_flag 负责认领，pointer 公布正在访问的节点
struct HazardPointer {
    std::atomic_flag claimed = ATOMIC_FLAG_INIT;
    std::atomic<void*> pointer{nullptr};
};
constexpr unsigned MAX_HAZARD_POINTERS = 100;
HazardPointer hazard_pointers[MAX_HAZARD_POINTERS];
 
// 扫公告板：是否还有任何线程的风险指针指向 p（指向→不能删）
bool outstanding_hazard_pointers_for(void* p) {
    for (unsigned i = 0; i < MAX_HAZARD_POINTERS; ++i) {
        if (hazard_pointers[i].pointer.load(std::memory_order_acquire) == p)
            return true;
    }
    return false;                                                  // 无牌→可删
}

挂牌后不能立刻解引用：在“第一次读 head”和“风险指针发布”之间，节点可能已经被摘下。必须发布后重新读取 head，确认入口仍指向同一节点；若变了就更新风险指针并重试：

template <typename Node>
Node* protect_head(std::atomic<Node*>& head,
                   std::atomic<void*>& hazard) {
    Node* candidate = nullptr;
    do {
        candidate = head.load(std::memory_order_acquire);
        hazard.store(candidate); // 先公开“我将解引用它”
    } while (candidate != head.load(std::memory_order_acquire));
    return candidate;            // 此后才能读取 candidate->next
}

摘下后，回收方扫描所有 hazard 槽；有牌就放入 retired list，无牌才 delete。实际方案还要定义槽位数量、retired list 批量扫描阈值、线程退出清理和内存序，并用成熟实现验证，不应只复制这段骨架。

解法二：引用计数（split reference count）

另一条思路完全不同：不搞外部公告板，而是给「握住 head」这件事配一个计数。无锁实现常用 split reference count（拆分引用计数）——把计数拆成「外部」「内部」两半：

#include <atomic>
 
template <typename T>
class RefCountedStack {
    struct Node;
    // 外部计数 + 指针：捆在一起 CAS，记「有几个线程正握着这个 head」
    struct CountedNodePtr {
        int external_count;
        Node* ptr;
    };
    struct Node {
        T data;
        std::atomic<int> internal_count;   // 内部计数
        CountedNodePtr next;
    };
    std::atomic<CountedNodePtr> head{CountedNodePtr{0, nullptr}};
    // 读 head 时把外部计数 +1：宣告「我握住它了，别删」
    void increase_head_count(CountedNodePtr& old_counter) {
        CountedNodePtr new_counter;
        do {
            new_counter = old_counter;
            ++new_counter.external_count;
        } while (!head.compare_exchange_strong(old_counter, new_counter));
        old_counter.external_count = new_counter.external_count;
    }
    // 外部 + 内部计数合并归零 → 没人引用 → 真正释放
};

要点：读 head 时先把外部计数 +1（宣告「我握住它了，别删」），用完再把对应的量并进内部计数；只有外部、内部合并归零，才说明真的没人引用、可以释放。拆成两个计数是为了绕开「单个原子计数在并发读取与递增之间的窗口」难题。比起风险指针，它把「谁在用」的信息自带在节点里，而不是放在外部公告板。

解法（防 ABA）：标签指针带版本号

回收方案顺带能缓解 ABA（地址短期不复用），但若要直接在 CAS 上防 ABA，可给指针配版本号：

#include <atomic>
 
template <typename T>
class TaggedStack {
    struct Node;
    // 指针 + 版本号捆成一个可原子 CAS 的整体：每次改都让 version+1
    struct TaggedPtr {
        Node* ptr;
        unsigned version;            // 地址变回原样、version 也变了 → CAS 能识别
    };
    struct Node { T data; TaggedPtr next; };
    std::atomic<TaggedPtr> head{TaggedPtr{nullptr, 0}};
public:
    void push(Node* n) {
        TaggedPtr old_head = head.load();
        TaggedPtr new_head;
        do {
            n->next = old_head;
            new_head.ptr = n;
            new_head.version = old_head.version + 1;   // 版本号 +1
        } while (!head.compare_exchange_weak(old_head, new_head));
    }
};

关键在 version：地址复用时版本不同，CAS 能识别变化。但有限宽度标签最终会回绕，必须证明在任何旧观察仍存活期间不可能绕回同一组合，或改用不会过早复用地址的回收方案。TaggedPtr 还应具有稳定、可比较的对象表示，避免未控制的填充位参与 C++17 CAS。std::atomic<TaggedPtr> 是否无锁取决于双宽 CAS 支持，必须查询；内部用锁时算法不再具有 lock-free 进展保证。

无锁队列：哨兵节点、帮助推进与两个线性化点

无锁队列不能只把栈的 head 改名。Michael-Scott 队列保留一个哨兵节点，head 指向待移除的哨兵，tail 是可能暂时落后的提示。enqueue 先 CAS 链接 tail->next，再尝试推进 tail；若发现 tail 落后，当前线程会帮助把它向前推，而不是等待原线程恢复。

void enqueue(T value) {
    Node* node = new Node(std::move(value));
    for (;;) {
        Node* last = protect_tail(); // 由 hazard/epoch 保护后才能解引用
        Node* next = last->next.load(std::memory_order_acquire);
        if (last != tail.load(std::memory_order_acquire)) continue;
        if (next == nullptr) {
            if (last->next.compare_exchange_weak(
                    next, node, std::memory_order_release,
                    std::memory_order_relaxed)) {
                tail.compare_exchange_strong(last, node);
                return;
            }
        } else {
            tail.compare_exchange_weak(last, next); // 帮助落后的 tail
        }
    }
}

成功链接 last->next 是 enqueue 的 线性化点（linearization point）；推进 tail 只是优化，即使当前线程暂停，别的线程也能帮助完成。dequeue 的线性化点通常是把 head CAS 到下一哨兵。两端在解引用节点前都必须进入风险指针、epoch 等回收协议；上述片段只展示入队核心，不能脱离 protect_tail() 和完整析构协议单独使用。

无锁编程准则：先证明，再优化

原书的无锁编程准则可以整理成五项验收：为每个返回路径指出线性化点；证明竞争失败意味着其他操作前进；先写 seq_cst 正确版，再逐个降级并记录同步证据；在第一次解引用前完成回收保护；最后检查目标 atomic 的 is_lock_free()、标签回绕和高争用退避。

测试只能发现反例，不能证明无锁正确性。ThreadSanitizer 可找普通数据竞争，却不理解所有自定义回收协议；压力测试也无法穷尽内存模型。生产项目应优先使用经过长期验证的容器或回收库，并把平台、编译器与最大线程数写进支持矩阵。

容易踩的坑

小结

• 无锁（lock-free） 关键不在「没有锁」，而在进展保证：任一线程被挂起都不卡死全场；更强的无等待（wait-free） 要求每个线程都在有限步内完成（最难，本章栈做不到）
• CAS 结构：栈 push 在线性化成功替换 head；无锁队列 enqueue 在线性化链接 tail->next，其他线程可帮助推进落后的 tail
• 内存回收难题：pop 摘下的节点不能马上 delete——别的线程可能正持有它的裸指针（use-after-free）；必须延迟回收
• 风险指针（访问前挂牌、回收前扫板、见牌不删）与引用计数（计数归零才释放）是两条延迟回收解法
• ABA 问题：CAS 只比值相等、看不出「改走又改回」；用标签指针（带版本号）或延迟回收（地址短期不复用）解决

#include <atomic>
#include <stdexcept>
 
struct HazardPointer {
    std::atomic_flag claimed = ATOMIC_FLAG_INIT;
    std::atomic<void*> pointer{nullptr};
};
constexpr unsigned MAX_HAZARD_POINTERS = 100;
HazardPointer hazard_pointers[MAX_HAZARD_POINTERS];
 
// 每个线程拿到自己专属的一个风险指针槽（首次访问时认领，析构时归还）
class HpOwner {
    HazardPointer* hp;
public:
    HpOwner() : hp(nullptr) {
        for (unsigned i = 0; i < MAX_HAZARD_POINTERS; ++i) {
            if (!hazard_pointers[i].claimed.test_and_set(
                    std::memory_order_acq_rel)) {
                hp = &hazard_pointers[i];
                break;
            }
        }
        if (!hp) throw std::runtime_error("no hazard pointer slots");
    }
    std::atomic<void*>& get_pointer() { return hp->pointer; }
    HpOwner(const HpOwner&) = delete;
    HpOwner& operator=(const HpOwner&) = delete;
    ~HpOwner() {
        hp->pointer.store(nullptr, std::memory_order_release);
        hp->claimed.clear(std::memory_order_release);
    }
};