無鎖棧lock-free stack通常指 Treiber stack是理解無鎖編程的第一課它只用一條compare_exchange循環(huán)就把push和pop寫完了全程沒有一把鎖。但「沒有鎖」不代表「沒有坑」。最著名的ABA 問題就藏在這里一個節(jié)點的地址被別的線程彈出又塞回來后CAS 會因為「值沒變」而誤判成功悄無聲息地丟掉節(jié)點。這篇把 Treiber 棧完整寫出來真跑講透 ABA 的成因再用 tagged pointer帶版本號的指針破解它最后用 TSan 驗證并發(fā)下沒有 data race。1. 引子一句head-next里藏著的并發(fā)語義先看無鎖棧的核心。它的數(shù)據(jù)結構極簡。一個原子頭指針 單向鏈表無鎖棧Treiber stack的內(nèi)存結構 headstd::atomicNode* │ ▼ ┌─────┐ ┌─────┐ ┌─────┐ │ B │───?│ A │───?│null │ └─────┘ └─────┘ └─────┘ 棧頂 B下面是 A棧底 null push(n)n-next head; CAS(head, n) ← 把頭指針換成新節(jié)點 pop() 讀 head CAS(head, head-next) ← 把頭指針換成下一個push和pop各自只有一個「讀頭 → 改頭」的動作compare_exchange保證這個動作是原子的只有當「頭的當前值」還等于你讀到的那份時才寫入新值否則失敗重來。這一條原語就足以讓整個棧在無鎖下保持結構一致。官方文檔std::atomic::compare_exchange · std::atomic2. 最小實現(xiàn)能真跑、TSan 干凈的 Treiber 棧下面是一個完整實現(xiàn)。為聚焦并發(fā)本身節(jié)點用std::vector預分配、不做回收。這避開了「節(jié)點被釋放后再被訪問」的內(nèi)存回收問題那是 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;// 新節(jié)點指向舊棧頂}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){// 預先把全部節(jié)點壓棧單線程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 線程并發(fā) 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(彈出總數(shù) %d期望 %d\n,count.load(),kNodes);std::printf(值之和 %lld期望 %d\n,sum.load(),kNodes*(kNodes-1)/2);}彈出總數(shù) 8期望 8 值之和 28期望 2828 01…78 個節(jié)點一個沒丟、一個沒重。這段用 clang 19 配-fsanitizethread真跑過stderr 為空、沒有 data race。因為并發(fā)階段只有「讀」和原子的 CAS沒有對同一內(nèi)存的普通讀寫競爭。兩個容易漏的細節(jié)push里n-next head寫在 CAS失敗循環(huán)內(nèi)因為每次失敗head都被compare_exchange_weak刷成了最新值必須重新掛鏈。內(nèi)存序是releasepush 發(fā)布/acquirepop 獲取配對保證「拿到節(jié)點就能看到它的完整內(nèi)容」。官方文檔std::memory_order3. ABA 問題地址沒變世界已經(jīng)變了上面的例子是「先全部壓棧、再全部彈出」沒有節(jié)點復用所以一切正常。真正的坑出現(xiàn)在節(jié)點被彈出后又被塞回來node reuse的場景?？催@個交錯ABA 問題無鎖棧丟節(jié)點—— 時間從上往下 初始棧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) 成功 現(xiàn)在棧A - C [6] T1 執(zhí)行 CAS(head, 期望 A - 實際 B) 但此刻 head 恰好又是 ACAS 成功head 變成 B 結果棧變成了 B - C而 A 丟了它還指著 C成了游離節(jié)點 ──────────────────────────────────────────────── 根因T1 的 CAS 只比較「頭指針的值」而「頭指針的值」從 A 變成了 B 又變回了 A —— 值相同但中間已經(jīng)發(fā)生了兩次 pop 一次 push。這就是ABA 問題compare_exchange只能判斷「值是否還是我期望的那個」無法判斷「中間有沒有變過又變回來」。當節(jié)點 A 的地址被復用后頭指針的值又回到了 AT1 的 CAS 誤以為「一切照舊」把 head 設成了一個早已不在棧里的 B。于是 A 丟失棧被破壞。注意 ABA 和內(nèi)存回收是兩個不同的問題ABA 是「邏輯上的誤判」導致丟節(jié)點而「節(jié)點被釋放后指針懸垂」是 use-after-free需要 hazard pointer / epoch-based reclamationB9來兜。兩者常被混為一談務必分開。4. tagged pointer給指針加上版本號破解 ABA 的思路很直接讓 CAS 比較的不只是地址還要帶上一個每次修改都自增的版本號。這樣「A 彈出又塞回」雖然地址還是 A版本號卻從 5 變成了 7T1 拿舊版本 5 去 CAS 就會失敗。在 x86-64 上用戶態(tài)指針實際只用低 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 用戶態(tài)指針只用低 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);// 同一個節(jié)點 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所以同一個地址出現(xiàn)兩次版本號一定不同?；氐降?3 節(jié)的交錯T1 在第 [1] 步讀到的「A 版本 5」等它第 [6] 步去 CAS 時棧頂已經(jīng)是「A 版本 7」compare_exchange發(fā)現(xiàn)版本號對不上失敗重讀。A 不會再丟了。需要說明的是這個實現(xiàn)把版本號塞進指針高位依賴 x86-64 的 48 位地址約定AArch64 有 top-byte-ignore 也類似但不通用?？缙脚_要可靠得用 128 位的雙字 CAScmpxchg16b/lock cmpxchg16b對應 GCC/Clang 的__int128原子或者干脆用現(xiàn)成庫。官方文檔std::uintptr_t · C Core Guidelines · CP.85.compare_exchange_weakvsstrong以及「無鎖 vs 有鎖」的取舍compare_exchange有兩個變體區(qū)別只有一個是否允許「偽失敗spurious failure」變體偽失敗適用場合底層差異compare_exchange_weak可能即使值相等也返回 false寫在循環(huán)里失敗重試即可某些平臺少一條指令更快compare_exchange_strong不會只試一次、或循環(huán)體很貴不宜重試內(nèi)部會處理偽失敗無鎖棧的push/pop都在while/for循環(huán)里所以一律用weak。至于「無鎖一定比有鎖快嗎」結論務實不一定。無鎖棧在低到中等競爭下往往比std::mutexstd::stack快避免內(nèi)核態(tài) futex 與上下文切換但在高競爭下CAS 的失敗重試緩存行乒乓也會劇烈退化。真實項目里無鎖棧的價值是免死鎖、免優(yōu)先級反轉、在中斷/信號上下文里也能用而不單純是快。維度無鎖棧Treiber有鎖棧mutex stack競爭低時快純用戶態(tài) CAS加鎖解鎖有固定開銷競爭高時CAS 失敗重試緩存行乒乓失敗者進內(nèi)核掛起喚醒有上下文切換死鎖 / 優(yōu)先級反轉無無鎖可能持鎖線程被搶占實現(xiàn)復雜度高ABA、內(nèi)存回收都是坑低std::stack直接套正確性驗證要 TSan 模型檢查鎖本身提供保證好推理6. 完整示例多生產(chǎn)者 多消費者的 tagged 無鎖棧把 tagged pointer 和多生產(chǎn)者/多消費者串起來4 個生產(chǎn)者各壓 4 個互不重復的節(jié)點4 個消費者再并發(fā)彈出全部 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 個節(jié)點值 0..15constexprintkConsumers4;std::vectorNodepool(static_caststd::size_t(kProducers*kPerProducer));TaggedStack stack;std::vectorstd::threadproducers;for(intt0;tkProducers;t){// 多生產(chǎn)者并發(fā)壓棧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){// 多消費者并發(fā)彈出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(彈出總數(shù) %d期望 %d\n,count.load(),kTotal);std::printf(值之和 %lld期望 %d\n,sum.load(),kTotal*(kTotal-1)/2);}彈出總數(shù) 16期望 16 值之和 120期望 120120 01…15。這個版本在 clang 19 的 TSan 下同樣沒有 data race節(jié)點只被壓入一次、彈出一次next字段在并發(fā)階段只讀不寫。最后必須強調(diào)這里的節(jié)點都是預分配、永不回收的。真正的生產(chǎn)級無鎖棧節(jié)點要在pop后被釋放這時「讀head-next的瞬間節(jié)點可能正被別的線程釋放」的 use-after-free 問題就冒出來了需要 hazard pointer 或 epoch-based reclamationB9配合。所以務實的建議是要無鎖棧用現(xiàn)成庫Boost.Lockfree 的boost::lockfree::stack、liblfds、concurrentqueue它們把這些坑都踩平了自己手寫只適合學習和驗證理解。7. 延伸閱讀std::atomic::compare_exchange — cppreference —— CAS 語義、weak/strong與內(nèi)存序參數(shù)std::memory_order — cppreference —— release/acquire 配對的正式定義std::atomic — cppreference —— 原子類型總覽與 lock-free 相關說明std::uintptr_t — cppreference —— tagged pointer 打包時用到的整數(shù)類型C Core Guidelines · 并發(fā)章節(jié) CP.1/CP.8 —— 無鎖編程的規(guī)范性建議與陷阱提醒本知識庫內(nèi)的相關篇目《無鎖內(nèi)存回收hazard pointer 與 epoch-based reclamation 原理》 —— 無鎖棧 pop 出來的節(jié)點為什么不能立刻 delete因為另一個線程可能還攥著它的指針?！稛o鎖隊列MPSC/MPMCVyukov 環(huán)形隊列與 slot 未就緒問題》 —— 隊列比棧難因為它有兩個端點——頭尾都要協(xié)調(diào)?！舵i進階atomic_flag 自旋鎖、shared_mutex 讀寫鎖與無鎖的代價》 —— std::mutex 一競爭就把失敗線程掛進內(nèi)核futex8. 一句話總結無鎖?？恳粭lcompare_exchange_weak循環(huán)就實現(xiàn)無鎖的push/pop但compare_exchange只比較「值」、不比較「中間有沒有變過又變回來」——節(jié)點地址被復用后會產(chǎn)生 ABA 問題CAS 誤判成功而丟節(jié)點破解辦法是 tagged pointer把版本號塞進指針高位、每次修改自增讓「同地址重現(xiàn)」也能被識別出來再往深走還有節(jié)點回收的 use-after-free要 hazard pointer見 B9所以生產(chǎn)級別請直接用現(xiàn)成庫。 無鎖棧：Treiber 棧的 CAS、ABA 問題與 tagged pointer
無鎖棧lock-free stack通常指 Treiber stack是理解無鎖編程的第一課它只用一條compare_exchange循環(huán)就把push和pop寫完了全程沒有一把鎖。但「沒有鎖」不代表「沒有坑」。最著名的ABA 問題就藏在這里一個節(jié)點的地址被別的線程彈出又塞回來后CAS 會因為「值沒變」而誤判成功悄無聲息地丟掉節(jié)點。這篇把 Treiber 棧完整寫出來真跑講透 ABA 的成因再用 tagged pointer帶版本號的指針破解它最后用 TSan 驗證并發(fā)下沒有 data race。1. 引子一句head-next里藏著的并發(fā)語義先看無鎖棧的核心。它的數(shù)據(jù)結構極簡。一個原子頭指針 單向鏈表無鎖棧Treiber stack的內(nèi)存結構 headstd::atomicNode* │ ▼ ┌─────┐ ┌─────┐ ┌─────┐ │ B │───?│ A │───?│null │ └─────┘ └─────┘ └─────┘ 棧頂 B下面是 A棧底 null push(n)n-next head; CAS(head, n) ← 把頭指針換成新節(jié)點 pop() 讀 head CAS(head, head-next) ← 把頭指針換成下一個push和pop各自只有一個「讀頭 → 改頭」的動作compare_exchange保證這個動作是原子的只有當「頭的當前值」還等于你讀到的那份時才寫入新值否則失敗重來。這一條原語就足以讓整個棧在無鎖下保持結構一致。官方文檔std::atomic::compare_exchange · std::atomic2. 最小實現(xiàn)能真跑、TSan 干凈的 Treiber 棧下面是一個完整實現(xiàn)。為聚焦并發(fā)本身節(jié)點用std::vector預分配、不做回收。這避開了「節(jié)點被釋放后再被訪問」的內(nèi)存回收問題那是 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;// 新節(jié)點指向舊棧頂}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){// 預先把全部節(jié)點壓棧單線程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 線程并發(fā) 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(彈出總數(shù) %d期望 %d\n,count.load(),kNodes);std::printf(值之和 %lld期望 %d\n,sum.load(),kNodes*(kNodes-1)/2);}彈出總數(shù) 8期望 8 值之和 28期望 2828 01…78 個節(jié)點一個沒丟、一個沒重。這段用 clang 19 配-fsanitizethread真跑過stderr 為空、沒有 data race。因為并發(fā)階段只有「讀」和原子的 CAS沒有對同一內(nèi)存的普通讀寫競爭。兩個容易漏的細節(jié)push里n-next head寫在 CAS失敗循環(huán)內(nèi)因為每次失敗head都被compare_exchange_weak刷成了最新值必須重新掛鏈。內(nèi)存序是releasepush 發(fā)布/acquirepop 獲取配對保證「拿到節(jié)點就能看到它的完整內(nèi)容」。官方文檔std::memory_order3. ABA 問題地址沒變世界已經(jīng)變了上面的例子是「先全部壓棧、再全部彈出」沒有節(jié)點復用所以一切正常。真正的坑出現(xiàn)在節(jié)點被彈出后又被塞回來node reuse的場景?？催@個交錯ABA 問題無鎖棧丟節(jié)點—— 時間從上往下 初始棧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) 成功 現(xiàn)在棧A - C [6] T1 執(zhí)行 CAS(head, 期望 A - 實際 B) 但此刻 head 恰好又是 ACAS 成功head 變成 B 結果棧變成了 B - C而 A 丟了它還指著 C成了游離節(jié)點 ──────────────────────────────────────────────── 根因T1 的 CAS 只比較「頭指針的值」而「頭指針的值」從 A 變成了 B 又變回了 A —— 值相同但中間已經(jīng)發(fā)生了兩次 pop 一次 push。這就是ABA 問題compare_exchange只能判斷「值是否還是我期望的那個」無法判斷「中間有沒有變過又變回來」。當節(jié)點 A 的地址被復用后頭指針的值又回到了 AT1 的 CAS 誤以為「一切照舊」把 head 設成了一個早已不在棧里的 B。于是 A 丟失棧被破壞。注意 ABA 和內(nèi)存回收是兩個不同的問題ABA 是「邏輯上的誤判」導致丟節(jié)點而「節(jié)點被釋放后指針懸垂」是 use-after-free需要 hazard pointer / epoch-based reclamationB9來兜。兩者常被混為一談務必分開。4. tagged pointer給指針加上版本號破解 ABA 的思路很直接讓 CAS 比較的不只是地址還要帶上一個每次修改都自增的版本號。這樣「A 彈出又塞回」雖然地址還是 A版本號卻從 5 變成了 7T1 拿舊版本 5 去 CAS 就會失敗。在 x86-64 上用戶態(tài)指針實際只用低 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 用戶態(tài)指針只用低 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);// 同一個節(jié)點 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所以同一個地址出現(xiàn)兩次版本號一定不同?；氐降?3 節(jié)的交錯T1 在第 [1] 步讀到的「A 版本 5」等它第 [6] 步去 CAS 時棧頂已經(jīng)是「A 版本 7」compare_exchange發(fā)現(xiàn)版本號對不上失敗重讀。A 不會再丟了。需要說明的是這個實現(xiàn)把版本號塞進指針高位依賴 x86-64 的 48 位地址約定AArch64 有 top-byte-ignore 也類似但不通用?？缙脚_要可靠得用 128 位的雙字 CAScmpxchg16b/lock cmpxchg16b對應 GCC/Clang 的__int128原子或者干脆用現(xiàn)成庫。官方文檔std::uintptr_t · C Core Guidelines · CP.85.compare_exchange_weakvsstrong以及「無鎖 vs 有鎖」的取舍compare_exchange有兩個變體區(qū)別只有一個是否允許「偽失敗spurious failure」變體偽失敗適用場合底層差異compare_exchange_weak可能即使值相等也返回 false寫在循環(huán)里失敗重試即可某些平臺少一條指令更快compare_exchange_strong不會只試一次、或循環(huán)體很貴不宜重試內(nèi)部會處理偽失敗無鎖棧的push/pop都在while/for循環(huán)里所以一律用weak。至于「無鎖一定比有鎖快嗎」結論務實不一定。無鎖棧在低到中等競爭下往往比std::mutexstd::stack快避免內(nèi)核態(tài) futex 與上下文切換但在高競爭下CAS 的失敗重試緩存行乒乓也會劇烈退化。真實項目里無鎖棧的價值是免死鎖、免優(yōu)先級反轉、在中斷/信號上下文里也能用而不單純是快。維度無鎖棧Treiber有鎖棧mutex stack競爭低時快純用戶態(tài) CAS加鎖解鎖有固定開銷競爭高時CAS 失敗重試緩存行乒乓失敗者進內(nèi)核掛起喚醒有上下文切換死鎖 / 優(yōu)先級反轉無無鎖可能持鎖線程被搶占實現(xiàn)復雜度高ABA、內(nèi)存回收都是坑低std::stack直接套正確性驗證要 TSan 模型檢查鎖本身提供保證好推理6. 完整示例多生產(chǎn)者 多消費者的 tagged 無鎖棧把 tagged pointer 和多生產(chǎn)者/多消費者串起來4 個生產(chǎn)者各壓 4 個互不重復的節(jié)點4 個消費者再并發(fā)彈出全部 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 個節(jié)點值 0..15constexprintkConsumers4;std::vectorNodepool(static_caststd::size_t(kProducers*kPerProducer));TaggedStack stack;std::vectorstd::threadproducers;for(intt0;tkProducers;t){// 多生產(chǎn)者并發(fā)壓棧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){// 多消費者并發(fā)彈出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(彈出總數(shù) %d期望 %d\n,count.load(),kTotal);std::printf(值之和 %lld期望 %d\n,sum.load(),kTotal*(kTotal-1)/2);}彈出總數(shù) 16期望 16 值之和 120期望 120120 01…15。這個版本在 clang 19 的 TSan 下同樣沒有 data race節(jié)點只被壓入一次、彈出一次next字段在并發(fā)階段只讀不寫。最后必須強調(diào)這里的節(jié)點都是預分配、永不回收的。真正的生產(chǎn)級無鎖棧節(jié)點要在pop后被釋放這時「讀head-next的瞬間節(jié)點可能正被別的線程釋放」的 use-after-free 問題就冒出來了需要 hazard pointer 或 epoch-based reclamationB9配合。所以務實的建議是要無鎖棧用現(xiàn)成庫Boost.Lockfree 的boost::lockfree::stack、liblfds、concurrentqueue它們把這些坑都踩平了自己手寫只適合學習和驗證理解。7. 延伸閱讀std::atomic::compare_exchange — cppreference —— CAS 語義、weak/strong與內(nèi)存序參數(shù)std::memory_order — cppreference —— release/acquire 配對的正式定義std::atomic — cppreference —— 原子類型總覽與 lock-free 相關說明std::uintptr_t — cppreference —— tagged pointer 打包時用到的整數(shù)類型C Core Guidelines · 并發(fā)章節(jié) CP.1/CP.8 —— 無鎖編程的規(guī)范性建議與陷阱提醒本知識庫內(nèi)的相關篇目《無鎖內(nèi)存回收hazard pointer 與 epoch-based reclamation 原理》 —— 無鎖棧 pop 出來的節(jié)點為什么不能立刻 delete因為另一個線程可能還攥著它的指針?！稛o鎖隊列MPSC/MPMCVyukov 環(huán)形隊列與 slot 未就緒問題》 —— 隊列比棧難因為它有兩個端點——頭尾都要協(xié)調(diào)?！舵i進階atomic_flag 自旋鎖、shared_mutex 讀寫鎖與無鎖的代價》 —— std::mutex 一競爭就把失敗線程掛進內(nèi)核futex8. 一句話總結無鎖?？恳粭lcompare_exchange_weak循環(huán)就實現(xiàn)無鎖的push/pop但compare_exchange只比較「值」、不比較「中間有沒有變過又變回來」——節(jié)點地址被復用后會產(chǎn)生 ABA 問題CAS 誤判成功而丟節(jié)點破解辦法是 tagged pointer把版本號塞進指針高位、每次修改自增讓「同地址重現(xiàn)」也能被識別出來再往深走還有節(jié)點回收的 use-after-free要 hazard pointer見 B9所以生產(chǎn)級別請直接用現(xiàn)成庫。