无锁栈lock-free stack通常指 Treiber stack是理解无锁编程的第一课它只用一条compare_exchange循环就把push和pop写完了全程没有一把锁。但「没有锁」不代表「没有坑」。最著名的ABA 问题就藏在这里一个节点的地址被别的线程弹出又塞回来后CAS 会因为「值没变」而误判成功悄无声息地丢掉节点。这篇把 Treiber 栈完整写出来真跑讲透 ABA 的成因再用 tagged pointer带版本号的指针破解它最后用 TSan 验证并发下没有 data race。1. 引子一句head-next里藏着的并发语义先看无锁栈的核心。它的数据结构极简。一个原子头指针 单向链表无锁栈Treiber stack的内存结构 headstd::atomicNode* │ ▼ ┌─────┐ ┌─────┐ ┌─────┐ │ 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::atomic2. 最小实现能真跑、TSan 干净的 Treiber 栈下面是一个完整实现。为聚焦并发本身节点用std::vector预分配、不做回收。这避开了「节点被释放后再被访问」的内存回收问题那是 hazard pointer 的事见 B9也让 TSan 能干净地验证结构正确性// treiber_stack.cpp — 编译: g -stdc17 -Wall -O2 -pthread treiber_stack.cpp -o ts#includeatomic#includecstdio#includethread#includevectorstructNode{intvalue;Node*next{nullptr};};classTreiberStack{public:voidpush(Node*n){Node*headhead_.load(std::memory_order_relaxed);do{n-nexthead;// 新节点指向旧栈顶}while(!head_.compare_exchange_weak(head,n,std::memory_order_release,std::memory_order_relaxed));}Node*pop(){Node*headhead_.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::atomicNode*head_{nullptr};};intmain(){constexprintkNodes8;constexprintkThreads4;std::vectorNodepool(static_caststd::size_t(kNodes));TreiberStack stack;for(inti0;ikNodes;i){// 预先把全部节点压栈单线程pool[static_caststd::size_t(i)].valuei;stack.push(pool[static_caststd::size_t(i)]);}std::atomicintcount{0};std::atomiclonglongsum{0};std::vectorstd::threadworkers;for(intt0;tkThreads;t){// 4 线程并发 pop直到栈空workers.emplace_back([]{for(;;){Node*nstack.pop();if(nnullptr)break;count.fetch_add(1,std::memory_order_relaxed);sum.fetch_add(n-value,std::memory_order_relaxed);}});}for(autot: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期望 2828 01…78 个节点一个没丢、一个没重。这段用 clang 19 配-fsanitizethread真跑过stderr 为空、没有 data race。因为并发阶段只有「读」和原子的 CAS没有对同一内存的普通读写竞争。两个容易漏的细节push里n-next head写在 CAS失败循环内因为每次失败head都被compare_exchange_weak刷成了最新值必须重新挂链。内存序是releasepush 发布/acquirepop 获取配对保证「拿到节点就能看到它的完整内容」。官方文档std::memory_order3. ABA 问题地址没变世界已经变了上面的例子是「先全部压栈、再全部弹出」没有节点复用所以一切正常。真正的坑出现在节点被弹出后又被塞回来node reuse的场景。看这个交错ABA 问题无锁栈丢节点—— 时间从上往下 初始栈A - B - C A 是栈顶 线程 T1 线程 T2 ──────────── ──────────── [1] 读 head A [2] 记录 A-next B准备 CAS 后让 head B [3] popCAS(head A-B) 成功拿走 A [4] popCAS(head B-C) 成功拿走 B [5] push(A)A-next CCAS(head C-A) 成功 现在栈A - C [6] T1 执行 CAS(head, 期望 A - 实际 B) 但此刻 head 恰好又是 ACAS 成功head 变成 B 结果栈变成了 B - C而 A 丢了它还指着 C成了游离节点 ──────────────────────────────────────────────── 根因T1 的 CAS 只比较「头指针的值」而「头指针的值」从 A 变成了 B 又变回了 A —— 值相同但中间已经发生了两次 pop 一次 push。这就是ABA 问题compare_exchange只能判断「值是否还是我期望的那个」无法判断「中间有没有变过又变回来」。当节点 A 的地址被复用后头指针的值又回到了 AT1 的 CAS 误以为「一切照旧」把 head 设成了一个早已不在栈里的 B。于是 A 丢失栈被破坏。注意 ABA 和内存回收是两个不同的问题ABA 是「逻辑上的误判」导致丢节点而「节点被释放后指针悬垂」是 use-after-free需要 hazard pointer / epoch-based reclamationB9来兜。两者常被混为一谈务必分开。4. tagged pointer给指针加上版本号破解 ABA 的思路很直接让 CAS 比较的不只是地址还要带上一个每次修改都自增的版本号。这样「A 弹出又塞回」虽然地址还是 A版本号却从 5 变成了 7T1 拿旧版本 5 去 CAS 就会失败。在 x86-64 上用户态指针实际只用低 48 位高 16 位可以白嫖来放版本号这就是 tagged pointer带标签的指针把「指针 版本号」打包进一个std::uintptr_t里做原子 CAS// tagged_stack.cpp — 编译: g -stdc17 -Wall -O2 -pthread tagged_stack.cpp -o tgs#includeatomic#includecstdint#includecstdiostructNode{intvalue;Node*next{nullptr};};// x86-64 用户态指针只用低 48 位高 16 位塞版本号constexprstd::uintptr_t kPtrMask(std::uintptr_t{1}48)-1;constexprintkTagShift48;voidpush(std::atomicstd::uintptr_thead,Node*n){std::uintptr_t hhead.load(std::memory_order_relaxed);for(;;){n-nextreinterpret_castNode*(hkPtrMask);conststd::uintptr_t dreinterpret_caststd::uintptr_t(n)|(((hkTagShift)1)kTagShift);if(head.compare_exchange_weak(h,d,std::memory_order_release,std::memory_order_relaxed))return;}}Node*pop(std::atomicstd::uintptr_thead){std::uintptr_t hhead.load(std::memory_order_relaxed);for(;;){Node*nreinterpret_castNode*(hkPtrMask);if(nnullptr)returnnullptr;conststd::uintptr_t dreinterpret_caststd::uintptr_t(n-next)|(((hkTagShift)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::atomicstd::uintptr_thead{0};push(head,a);std::printf(push(a) 后 版本 %llu\n,static_castunsignedlonglong(head.load()kTagShift));Node*xpop(head);std::printf(pop() 后 版本 %llu拿到 %d\n,static_castunsignedlonglong(head.load()kTagShift),x-value);push(head,x);// 同一个节点 a 的地址又回来了但版本号已从 1 变到 3std::printf(push(a) 回 版本 %llu\n,static_castunsignedlonglong(head.load()kTagShift));push(head,b);std::printf(push(b) 后 版本 %llu\n,static_castunsignedlonglong(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 位的双字 CAScmpxchg16b/lock cmpxchg16b对应 GCC/Clang 的__int128原子或者干脆用现成库。官方文档std::uintptr_t · C Core Guidelines · CP.85.compare_exchange_weakvsstrong以及「无锁 vs 有锁」的取舍compare_exchange有两个变体区别只有一个是否允许「伪失败spurious failure」变体伪失败适用场合底层差异compare_exchange_weak可能即使值相等也返回 false写在循环里失败重试即可某些平台少一条指令更快compare_exchange_strong不会只试一次、或循环体很贵不宜重试内部会处理伪失败无锁栈的push/pop都在while/for循环里所以一律用weak。至于「无锁一定比有锁快吗」结论务实不一定。无锁栈在低到中等竞争下往往比std::mutexstd::stack快避免内核态 futex 与上下文切换但在高竞争下CAS 的失败重试缓存行乒乓也会剧烈退化。真实项目里无锁栈的价值是免死锁、免优先级反转、在中断/信号上下文里也能用而不单纯是快。维度无锁栈Treiber有锁栈mutex stack竞争低时快纯用户态 CAS加锁解锁有固定开销竞争高时CAS 失败重试缓存行乒乓失败者进内核挂起唤醒有上下文切换死锁 / 优先级反转无无锁可能持锁线程被抢占实现复杂度高ABA、内存回收都是坑低std::stack直接套正确性验证要 TSan 模型检查锁本身提供保证好推理6. 完整示例多生产者 多消费者的 tagged 无锁栈把 tagged pointer 和多生产者/多消费者串起来4 个生产者各压 4 个互不重复的节点4 个消费者再并发弹出全部 16 个验证总和确定、无丢失// treiber_final.cpp — 编译: g -stdc17 -Wall -O2 -pthread treiber_final.cpp -o tf#includeatomic#includecstdint#includecstdio#includethread#includevectorstructNode{intvalue;Node*next{nullptr};};constexprstd::uintptr_t kPtrMask(std::uintptr_t{1}48)-1;constexprintkTagShift48;classTaggedStack{public:voidpush(Node*n){std::uintptr_t hhead_.load(std::memory_order_relaxed);for(;;){n-nextreinterpret_castNode*(hkPtrMask);conststd::uintptr_t dreinterpret_caststd::uintptr_t(n)|(((hkTagShift)1)kTagShift);if(head_.compare_exchange_weak(h,d,std::memory_order_release,std::memory_order_relaxed))return;}}Node*pop(){std::uintptr_t hhead_.load(std::memory_order_relaxed);for(;;){Node*nreinterpret_castNode*(hkPtrMask);if(nnullptr)returnnullptr;conststd::uintptr_t dreinterpret_caststd::uintptr_t(n-next)|(((hkTagShift)1)kTagShift);if(head_.compare_exchange_weak(h,d,std::memory_order_acquire,std::memory_order_relaxed))returnn;}}private:std::atomicstd::uintptr_thead_{0};};intmain(){constexprintkProducers4;constexprintkPerProducer4;// 共 16 个节点值 0..15constexprintkConsumers4;std::vectorNodepool(static_caststd::size_t(kProducers*kPerProducer));TaggedStack stack;std::vectorstd::threadproducers;for(intt0;tkProducers;t){// 多生产者并发压栈producers.emplace_back([,t]{for(inti0;ikPerProducer;i){std::size_t idxstatic_caststd::size_t(t*kPerProduceri);pool[idx].valuet*kPerProduceri;stack.push(pool[idx]);}});}for(autot:producers)t.join();std::atomicintcount{0};std::atomiclonglongsum{0};std::vectorstd::threadconsumers;for(intt0;tkConsumers;t){// 多消费者并发弹出consumers.emplace_back([]{for(;;){Node*nstack.pop();if(nnullptr)break;count.fetch_add(1,std::memory_order_relaxed);sum.fetch_add(n-value,std::memory_order_relaxed);}});}for(autot:consumers)t.join();constexprintkTotalkProducers*kPerProducer;std::printf(弹出总数 %d期望 %d\n,count.load(),kTotal);std::printf(值之和 %lld期望 %d\n,sum.load(),kTotal*(kTotal-1)/2);}弹出总数 16期望 16 值之和 120期望 120120 01…15。这个版本在 clang 19 的 TSan 下同样没有 data race节点只被压入一次、弹出一次next字段在并发阶段只读不写。最后必须强调这里的节点都是预分配、永不回收的。真正的生产级无锁栈节点要在pop后被释放这时「读head-next的瞬间节点可能正被别的线程释放」的 use-after-free 问题就冒出来了需要 hazard pointer 或 epoch-based reclamationB9配合。所以务实的建议是要无锁栈用现成库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/MPMCVyukov 环形队列与 slot 未就绪问题》 —— 队列比栈难因为它有两个端点——头尾都要协调。《锁进阶atomic_flag 自旋锁、shared_mutex 读写锁与无锁的代价》 —— std::mutex 一竞争就把失败线程挂进内核futex8. 一句话总结无锁栈靠一条compare_exchange_weak循环就实现无锁的push/pop但compare_exchange只比较「值」、不比较「中间有没有变过又变回来」——节点地址被复用后会产生 ABA 问题CAS 误判成功而丢节点破解办法是 tagged pointer把版本号塞进指针高位、每次修改自增让「同地址重现」也能被识别出来再往深走还有节点回收的 use-after-free要 hazard pointer见 B9所以生产级别请直接用现成库。