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