无锁栈（lock-free stack，通常指 Treiber stack）是理解无锁编程的第一课：它只用一条compare_exchange循环就把push和pop写完了，全程没有一把锁。但「没有锁」不代表「没有坑」。最著名的ABA 问题就藏在这里：一个节点的地址被别的线程弹出又塞回来后，CAS 会因为「值没变」而误判成功，悄无声息地丢掉节点。这篇把 Treiber 栈完整写出来真跑，讲透 ABA 的成因，再用 tagged pointer（带版本号的指针）破解它，最后用 TSan 验证并发下没有 data race。

1. 引子：一句head->next里藏着的并发语义

先看无锁栈的核心。它的数据结构极简。一个原子头指针 + 单向链表：

无锁栈（Treiber stack）的内存结构 head（std::atomic<Node*>） │ ▼ ┌─────┐ ┌─────┐ ┌─────┐ │ B │───►│ A │───►│null │ └─────┘ └─────┘ └─────┘ 栈顶 B，下面是 A，栈底 null push(n)：n->next = head; CAS(head, n) ← 把头指针换成新节点 pop() ：读 head； CAS(head, head->next) ← 把头指针换成下一个

push和pop各自只有一个「读头 → 改头」的动作，compare_exchange保证这个动作是原子的：只有当「头的当前值」还等于你读到的那份时，才写入新值；否则失败重来。这一条原语就足以让整个栈在无锁下保持结构一致。

官方文档：std::atomic::compare_exchange · std::atomic

2. 最小实现：能真跑、TSan 干净的 Treiber 栈

下面是一个完整实现。为聚焦并发本身，节点用std::vector预分配、不做回收。这避开了「节点被释放后再被访问」的内存回收问题（那是 hazard pointer 的事，见 B9），也让 TSan 能干净地验证结构正确性：

// treiber_stack.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread treiber_stack.cpp -o ts#include<atomic>#include<cstdio>#include<thread>#include<vector>structNode{intvalue;Node*next{nullptr};};classTreiberStack{public:voidpush(Node*n){Node*head=head_.load(std::memory_order_relaxed);do{n->next=head;// 新节点指向旧栈顶}while(!head_.compare_exchange_weak(head,n,std::memory_order_release,std::memory_order_relaxed));}Node*pop(){Node*head=head_.load(std::memory_order_relaxed);while(head!=nullptr&&!head_.compare_exchange_weak(head,head->next,std::memory_order_acquire,std::memory_order_relaxed)){}returnhead;// nullptr 表示栈空}private:std::atomic<Node*>head_{nullptr};};intmain(){constexprintkNodes=8;constexprintkThreads=4;std::vector<Node>pool(static_cast<std::size_t>(kNodes));TreiberStack stack;for(inti=0;i<kNodes;++i){// 预先把全部节点压栈（单线程）pool[static_cast<std::size_t>(i)].value=i;stack.push(&pool[static_cast<std::size_t>(i)]);}std::atomic<int>count{0};std::atomic<longlong>sum{0};std::vector<std::thread>workers;for(intt=0;t<kThreads;++t){// 4 线程并发 pop，直到栈空workers.emplace_back([&]{for(;;){Node*n=stack.pop();if(n==nullptr)break;count.fetch_add(1,std::memory_order_relaxed);sum.fetch_add(n->value,std::memory_order_relaxed);}});}for(auto&t:workers)t.join();std::printf("弹出总数 = %d（期望 %d）\n",count.load(),kNodes);std::printf("值之和 = %lld（期望 %d）\n",sum.load(),kNodes*(kNodes-1)/2);}

弹出总数 = 8（期望 8） 值之和 = 28（期望 28）

28 = 0+1+…+7，8 个节点一个没丢、一个没重。这段用 clang 19 配-fsanitize=thread真跑过，stderr 为空、没有 data race。因为并发阶段只有「读」和原子的 CAS，没有对同一内存的普通读写竞争。

两个容易漏的细节：

• push里n->next = head写在 CAS失败循环内，因为每次失败head都被compare_exchange_weak刷成了最新值，必须重新挂链。
• 内存序是release（push 发布）/acquire（pop 获取）配对，保证「拿到节点就能看到它的完整内容」。

官方文档：std::memory_order

3. ABA 问题：地址没变，世界已经变了

上面的例子是「先全部压栈、再全部弹出」，没有节点复用，所以一切正常。真正的坑出现在节点被弹出后又被塞回来（node reuse）的场景。看这个交错：

ABA 问题（无锁栈丢节点）—— 时间从上往下 初始栈：A -> B -> C （A 是栈顶） 线程 T1 线程 T2 ──────────── ──────────── [1] 读 head = A [2] 记录 A->next = B（准备 CAS 后让 head = B） [3] pop：CAS(head A->B) 成功，拿走 A [4] pop：CAS(head B->C) 成功，拿走 B [5] push(A)：A->next = C，CAS(head C->A) 成功 现在栈：A -> C [6] T1 执行 CAS(head, 期望 A -> 实际 B) 但此刻 head 恰好又是 A！CAS 成功，head 变成 B 结果：栈变成了 B -> C，而 A 丢了（它还指着 C，成了游离节点） ──────────────────────────────────────────────── 根因：T1 的 CAS 只比较「头指针的值」，而「头指针的值」从 A 变成了 B 又变回了 A —— 值相同，但中间已经发生了两次 pop + 一次 push。

这就是ABA 问题：compare_exchange只能判断「值是否还是我期望的那个」，无法判断「中间有没有变过又变回来」。当节点 A 的地址被复用后，头指针的值又回到了 A，T1 的 CAS 误以为「一切照旧」，把 head 设成了一个早已不在栈里的 B。于是 A 丢失，栈被破坏。

注意 ABA 和内存回收是两个不同的问题：ABA 是「逻辑上的误判」，导致丢节点；而「节点被释放后指针悬垂」是 use-after-free，需要 hazard pointer / epoch-based reclamation（B9）来兜。两者常被混为一谈，务必分开。

4. tagged pointer：给指针加上版本号

破解 ABA 的思路很直接：让 CAS 比较的不只是地址，还要带上一个每次修改都自增的版本号。这样「A 弹出又塞回」虽然地址还是 A，版本号却从 5 变成了 7，T1 拿旧版本 5 去 CAS 就会失败。

在 x86-64 上，用户态指针实际只用低 48 位，高 16 位可以白嫖来放版本号：这就是 tagged pointer（带标签的指针），把「指针 + 版本号」打包进一个std::uintptr_t里做原子 CAS：

// tagged_stack.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread tagged_stack.cpp -o tgs#include<atomic>#include<cstdint>#include<cstdio>structNode{intvalue;Node*next{nullptr};};// x86-64 用户态指针只用低 48 位，高 16 位塞版本号constexprstd::uintptr_t kPtrMask=(std::uintptr_t{1}<<48)-1;constexprintkTagShift=48;voidpush(std::atomic<std::uintptr_t>&head,Node*n){std::uintptr_t h=head.load(std::memory_order_relaxed);for(;;){n->next=reinterpret_cast<Node*>(h&kPtrMask);conststd::uintptr_t d=reinterpret_cast<std::uintptr_t>(n)|(((h>>kTagShift)+1)<<kTagShift);if(head.compare_exchange_weak(h,d,std::memory_order_release,std::memory_order_relaxed))return;}}Node*pop(std::atomic<std::uintptr_t>&head){std::uintptr_t h=head.load(std::memory_order_relaxed);for(;;){Node*n=reinterpret_cast<Node*>(h&kPtrMask);if(n==nullptr)returnnullptr;conststd::uintptr_t d=reinterpret_cast<std::uintptr_t>(n->next)|(((h>>kTagShift)+1)<<kTagShift);if(head.compare_exchange_weak(h,d,std::memory_order_acquire,std::memory_order_relaxed))returnn;}}intmain(){Node a{1},b{2};std::atomic<std::uintptr_t>head{0};push(head,&a);std::printf("push(a) 后 版本 = %llu\n",static_cast<unsignedlonglong>(head.load()>>kTagShift));Node*x=pop(head);std::printf("pop() 后 版本 = %llu（拿到 %d）\n",static_cast<unsignedlonglong>(head.load()>>kTagShift),x->value);push(head,x);// 同一个节点 a 的地址又回来了，但版本号已从 1 变到 3std::printf("push(a) 回 版本 = %llu\n",static_cast<unsignedlonglong>(head.load()>>kTagShift));push(head,&b);std::printf("push(b) 后 版本 = %llu\n",static_cast<unsignedlonglong>(head.load()>>kTagShift));}

push(a) 后 版本 = 1 pop() 后 版本 = 2（拿到 1） push(a) 回 版本 = 3 push(b) 后 版本 = 4

每次push/pop都把版本号 +1，所以同一个地址出现两次，版本号一定不同。回到第 3 节的交错：T1 在第 [1] 步读到的「A + 版本 5」，等它第 [6] 步去 CAS 时，栈顶已经是「A + 版本 7」，compare_exchange发现版本号对不上，失败重读。A 不会再丢了。

需要说明的是：这个实现把版本号塞进指针高位，依赖 x86-64 的 48 位地址约定（AArch64 有 top-byte-ignore 也类似，但不通用）。跨平台要可靠，得用 128 位的双字 CAS（cmpxchg16b/lock cmpxchg16b，对应 GCC/Clang 的__int128原子），或者干脆用现成库。

官方文档：std::uintptr_t · C++ Core Guidelines · CP.8

5.compare_exchange_weakvsstrong，以及「无锁 vs 有锁」的取舍

compare_exchange有两个变体，区别只有一个，是否允许「伪失败（spurious failure）」：

无锁栈的push/pop都在while/for循环里，所以一律用weak。

至于「无锁一定比有锁快吗」，结论务实：不一定。无锁栈在低到中等竞争下往往比std::mutex+std::stack快（避免内核态 futex 与上下文切换），但在高竞争下，CAS 的失败重试（缓存行乒乓）也会剧烈退化。真实项目里，无锁栈的价值是免死锁、免优先级反转、在中断/信号上下文里也能用，而不单纯是快。

6. 完整示例：多生产者 + 多消费者的 tagged 无锁栈

把 tagged pointer 和多生产者/多消费者串起来：4 个生产者各压 4 个互不重复的节点，4 个消费者再并发弹出全部 16 个，验证总和确定、无丢失：

// treiber_final.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread treiber_final.cpp -o tf#include<atomic>#include<cstdint>#include<cstdio>#include<thread>#include<vector>structNode{intvalue;Node*next{nullptr};};constexprstd::uintptr_t kPtrMask=(std::uintptr_t{1}<<48)-1;constexprintkTagShift=48;classTaggedStack{public:voidpush(Node*n){std::uintptr_t h=head_.load(std::memory_order_relaxed);for(;;){n->next=reinterpret_cast<Node*>(h&kPtrMask);conststd::uintptr_t d=reinterpret_cast<std::uintptr_t>(n)|(((h>>kTagShift)+1)<<kTagShift);if(head_.compare_exchange_weak(h,d,std::memory_order_release,std::memory_order_relaxed))return;}}Node*pop(){std::uintptr_t h=head_.load(std::memory_order_relaxed);for(;;){Node*n=reinterpret_cast<Node*>(h&kPtrMask);if(n==nullptr)returnnullptr;conststd::uintptr_t d=reinterpret_cast<std::uintptr_t>(n->next)|(((h>>kTagShift)+1)<<kTagShift);if(head_.compare_exchange_weak(h,d,std::memory_order_acquire,std::memory_order_relaxed))returnn;}}private:std::atomic<std::uintptr_t>head_{0};};intmain(){constexprintkProducers=4;constexprintkPerProducer=4;// 共 16 个节点，值 0..15constexprintkConsumers=4;std::vector<Node>pool(static_cast<std::size_t>(kProducers*kPerProducer));TaggedStack stack;std::vector<std::thread>producers;for(intt=0;t<kProducers;++t){// 多生产者并发压栈producers.emplace_back([&,t]{for(inti=0;i<kPerProducer;++i){std::size_t idx=static_cast<std::size_t>(t*kPerProducer+i);pool[idx].value=t*kPerProducer+i;stack.push(&pool[idx]);}});}for(auto&t:producers)t.join();std::atomic<int>count{0};std::atomic<longlong>sum{0};std::vector<std::thread>consumers;for(intt=0;t<kConsumers;++t){// 多消费者并发弹出consumers.emplace_back([&]{for(;;){Node*n=stack.pop();if(n==nullptr)break;count.fetch_add(1,std::memory_order_relaxed);sum.fetch_add(n->value,std::memory_order_relaxed);}});}for(auto&t:consumers)t.join();constexprintkTotal=kProducers*kPerProducer;std::printf("弹出总数 = %d（期望 %d）\n",count.load(),kTotal);std::printf("值之和 = %lld（期望 %d）\n",sum.load(),kTotal*(kTotal-1)/2);}

弹出总数 = 16（期望 16） 值之和 = 120（期望 120）

120 = 0+1+…+15。这个版本在 clang 19 的 TSan 下同样没有 data race（节点只被压入一次、弹出一次，next字段在并发阶段只读不写）。

最后必须强调：这里的节点都是预分配、永不回收的。真正的生产级无锁栈，节点要在pop后被释放，这时「读head->next的瞬间节点可能正被别的线程释放」的 use-after-free 问题就冒出来了，需要 hazard pointer 或 epoch-based reclamation（B9）配合。所以务实的建议是：要无锁栈，用现成库（Boost.Lockfree 的boost::lockfree::stack、liblfds、concurrentqueue），它们把这些坑都踩平了；自己手写只适合学习和验证理解。

7. 延伸阅读

• std::atomic::compare_exchange — cppreference —— CAS 语义、weak/strong与内存序参数
• std::memory_order — cppreference —— release/acquire 配对的正式定义
• std::atomic — cppreference —— 原子类型总览与 lock-free 相关说明
• std::uintptr_t — cppreference —— tagged pointer 打包时用到的整数类型
• C++ Core Guidelines · 并发章节 CP.1/CP.8 —— 无锁编程的规范性建议与陷阱提醒

本知识库内的相关篇目：

• 《无锁内存回收：hazard pointer 与 epoch-based reclamation 原理》 —— 无锁栈 pop 出来的节点为什么不能立刻 delete？因为另一个线程可能还攥着它的指针。
• 《无锁队列（MPSC/MPMC）：Vyukov 环形队列与 slot 未就绪问题》 —— 队列比栈难，因为它有两个端点——头尾都要协调。
• 《锁进阶：atomic_flag 自旋锁、shared_mutex 读写锁与无锁的代价》 —— std::mutex 一竞争就把失败线程挂进内核（futex）

8. 一句话总结

无锁栈靠一条compare_exchange_weak循环就实现无锁的push/pop，但compare_exchange只比较「值」、不比较「中间有没有变过又变回来」——节点地址被复用后会产生 ABA 问题，CAS 误判成功而丢节点；破解办法是 tagged pointer，把版本号塞进指针高位、每次修改自增，让「同地址重现」也能被识别出来；再往深走还有节点回收的 use-after-free（要 hazard pointer，见 B9），所以生产级别请直接用现成库。