★ FEATURED ARTICLE 无锁数据结构里最反直觉的一件事你从无锁栈pop()拿到的节点不能随手delete掉。因为你没法保证没有别的线程此刻正握着这个节点的指针、马上就要解引用它。这一篇讲清楚无锁环境下的内存回收为什么难、有哪三种主流思路每种都给一个能跑的简化实现最后落到一句话生产级直接上现成库。1. 引子pop 出来的节点凭什么不能 delete先看一个「看起来正确、实则危险」的无锁栈骨架。它的 push/pop 用 CAS 维护头指针正确性没问题但pop()里那句delete是颗雷// 反例不要这么写pop 后直接 delete 节点别的线程可能还在用它templatetypenameTstructUnsafeStack{structNode{T value;Node*next;// 反例裸指针只为演示内存回收问题};std::atomicNode*head_{nullptr};voidpush(T v){Node*nnewNode{v,head_.load()};while(!head_.compare_exchange_weak(n-next,n)){}}boolpop(Tout){Node*oldhead_.load();while(old!head_.compare_exchange_weak(old,old-next)){}if(!old)returnfalse;outold-value;deleteold;// ← 危险old 可能正被另一个线程读取returntrue;}};危险来自下面这种交错线程 A 和线程 B 同时操作无锁栈A pop 出的节点被 B 读取 时间 ──────────────────────────────────────────────────────────► 线程 A load head → old(N1) ── CAS 成功head 指向 N2 ── delete N1 ──► 线程 B load head → 拿到 N1 的指针 ── 解引用 N1-value ──► ✗ use-after-free 问题B 在「load 到 N1」之后、「解引用」之前A 已经 delete 了 N1。 所以「节点不在结构里了」≠「节点没人在用了」。这就是无锁内存回收的核心矛盾结构上的移除逻辑删除和物理上的释放delete之间隔着一段「可能还有人引用」的窗口。有锁的世界里锁天然保证了「拿到锁 别人进不来 没人再用」所以delete是安全的无锁的世界里没有这层保护必须自己造一套机制。2. 三种回收思路总览解决这个矛盾工业界有三条主流路线本质都是「延迟释放 判定无人引用」方案核心思想读侧开销适用场景引用计数每个节点带引用计数没人引用才释放一次原子自增/自减通用实现简单开销略高hazard pointer读者声明「我正在用这几个指针」回收者只释放没人声明的一次原子写声明读多写少节点数有限epoch-based reclamation读者只登记自己「处于第几个 epoch」回收者等所有读者都走过静止点再释放极低一次 store读侧极频繁、对延迟最敏感RCU 思路下面逐个给简化实现。先声明这三段的代码都是「讲原理」的最小实现管理结构里有意用了锁来简化正确性真正的无锁内存回收器folly::hazptr、libcds、Linux 内核的 RCU把「声明」「退休」「回收」全做成了无锁并处理了 ABA、内存序等细节工程里不要自己重造。3. 方案一引用计数shared_ptr 原子操作最简单也最通用的办法让节点自带引用计数谁拿到指针谁 1用完 -1归零自动释放。C 里现成就有std::shared_ptr而std::atomic_load/std::atomic_store/std::atomic_compare_exchange_weak对std::shared_ptr的重载memory提供C11 起就有能让我们把它塞进无锁结构里还不产生 data race// rc_stack.cpp — 编译: g -stdc17 -Wall -O2 -pthread rc_stack.cpp -o rc_stack#includeatomic#includecstdio#includememory#includethread#includevectortemplatetypenameTclassRcStack{structNode{T value;std::shared_ptrNodenext;};std::shared_ptrNodehead_;// 头节点shared_ptr 自动管引用计数public:voidpush(T v){autonstd::make_sharedNode();n-valuev;n-nextstd::atomic_load(head_);while(!std::atomic_compare_exchange_weak(head_,n-next,n)){}}boolpop(Tout){autooldstd::atomic_load(head_);while(old!std::atomic_compare_exchange_weak(head_,old,old-next)){}if(!old)returnfalse;outold-value;// old 是 shared_ptr此刻引用计数 ≥1节点不会被释放returntrue;}};intmain(){RcStackints;std::vectorstd::threadws;for(intt0;t4;t){ws.emplace_back([s,t]{for(intk0;k2500;k)s.push(t*2500k);});}for(autow:ws)w.join();intgot0;intv;while(s.pop(v))got;std::printf(pushed10000 popped%d\n,got);}pushed10000 popped10000关键在pop()里的局部变量old是shared_ptrCAS 失败也好、成功也罢只要old这个局部shared_ptr还活着它指向的节点引用计数就 ≥1别的线程无论怎么折腾都delete不掉它。等pop返回、old析构引用计数归零节点才被自动释放。这就是「用引用计数把 use-after-free 堵死」。引用计数就是最朴素的「延迟释放 无人引用判定」。用 ThreadSanitizerclang 19复验一遍确认这套共享指针的原子操作没有 data race$ clang -stdc17 -O1 -fsanitizethread -g -pthread rc_stack.cpp -o rc_tsan ./rc_tsan pushed10000 popped10000 stderr 无任何 ThreadSanitizer 报告 —— 说明没有 data race代价也摆在那里每次 push/pop 都伴随一次原子引用计数的加减shared_ptr的控制块操作在高竞争下比纯裸指针慢。所以它适合「正确性优先、竞争不极端」的场景。官方文档std::atomic_load / atomic_store / atomic_compare_exchangeshared_ptr 重载 · std::shared_ptr4. 方案二hazard pointer声明「我在用」引用计数每次访问都要动引用计数读侧不便宜。hazard pointer 换个思路读者在读之前把「我正握着的指针」写进一个全局可见的 hazard 槽回收者要释放某个节点前先扫一遍所有 hazard 槽只有没人声明的节点才delete。读侧只需一次 store回收的代价被摊到回收者身上。下面是简化实现hazard 槽用原子变量退休列表用锁保护// hazptr_demo.cpp — 编译: g -stdc17 -Wall -O2 -pthread hazptr_demo.cpp -o hazptr#includeatomic#includecstdio#includemutex#includevector// 简化 hazard pointerhazard 槽是原子的退休列表用锁保护// 生产级如 folly::hazptr 两者都无锁这里用锁只为把机制讲清楚classMiniHazptr{public:staticconstexprintkReaders2;std::atomicint*slots_[kReaders]{};// 每个读者一个 hazard 槽std::atomicintreclaimed_{0};private:std::mutex mu_;std::vectorint*retired_;public:voidacquire(intreader,int*p){slots_[reader].store(p);}voidrelease(intreader){slots_[reader].store(nullptr);}voidretire(int*p){std::lock_guardstd::mutexlock(mu_);retired_.push_back(p);}boolis_protected(int*p)const{for(intr0;rkReaders;r){if(slots_[r].load()p)returntrue;}returnfalse;}voidreclaim(){std::lock_guardstd::mutexlock(mu_);for(autoitretired_.begin();it!retired_.end();){if(is_protected(*it)){it;// 有人正在用跳过}else{delete*it;// 演示内存回收机制此处故意裸 deletereclaimed_.fetch_add(1);itretired_.erase(it);}}}};intmain(){MiniHazptr hp;int*anewint(1);// 演示内存回收机制故意裸 new真实代码交给现成库int*bnewint(2);hp.acquire(0,a);// 读者 0 声明「正在用 a」hp.retire(a);// 回收者退休 ahp.retire(b);// 退休 bhp.reclaim();// a 被保护不删只删 bstd::printf(第一次 reclaim回收 %d 个ba 仍在用%d\n,hp.reclaimed_.load(),*a);hp.release(0);// 读者 0 用完 ahp.reclaim();// 现在 a 才被删std::printf(第二次 reclaim累计回收 %d 个\n,hp.reclaimed_.load());}第一次 reclaim回收 1 个ba 仍在用1 第二次 reclaim累计回收 2 个两次reclaim()的结果正好说明机制第一次a在 hazard 槽里、被跳过等读者release之后a才被回收。时序是这样的hazard pointer 的「声明 → 退休 → 回收」时序 读者 0 acquire(a) ──────────── release(a) ─────────────────► 回收者 retire(a) ── reclaim() reclaim() ──► │扫 hazard │a 已不在 │ 槽里有 a跳过 │ 任何槽里 ▼ ▼ a 保留不删 delete a b 从未被声明 → 第一次 reclaim 就直接 delete bhazard pointer 的局限也很明确读者必须显式、成对地acquire/release漏写一次就是 use-after-free而且一个读者同时最多能用多少个节点就得准备多少个 hazard 槽。这些正是生产库要替你兜住的地方。官方文档std::atomic::storehazard 槽的原子写 · C Core Guidelines CP并发章节5. 方案三epoch-based reclamation读侧免锁hazard pointer 要求读者「精确声明每个指针」epoch-based 更粗放读者只登记「我正处于第几个 epoch」用完清零回收者想释放某批节点时先把全局 epoch 推进一位然后确认所有读者都已经走过旧的 epoch静止点quiescent state这时退休于旧 epoch 的节点就能安全释放。读侧只是两次store几乎没有竞争 —— 这正是 Linux 内核 RCUread-copy-update读侧免锁的简化版思路。// epoch_demo.cpp — 编译: g -stdc17 -Wall -O2 -pthread epoch_demo.cpp -o epoch#includeatomic#includecstdio#includemutex#includevector// 简化 epoch-based reclamationQSBR 风格读侧只登记 epoch免锁classMiniEpoch{public:staticconstexprintkReaders2;std::atomicunsignedlongglobal_epoch_{1};std::atomicunsignedlongreader_epoch_[kReaders]{};// 0 空闲std::atomicintreclaimed_{0};private:std::mutex mu_;std::vectorint*retired_;public:voidenter(intreader){reader_epoch_[reader].store(global_epoch_.load());}voidexit(intreader){reader_epoch_[reader].store(0);}boolall_idle()const{for(intr0;rkReaders;r){if(reader_epoch_[r].load()!0)returnfalse;}returntrue;}voidretire(int*p){std::lock_guardstd::mutexlock(mu_);retired_.push_back(p);}// 回收者等所有读者穿过静止点推进 epoch再释放旧 epoch 退休的节点voidadvance_and_reclaim(){while(!all_idle()){}// 简化忙等静止点constunsignedlongoldglobal_epoch_.load();global_epoch_.store(old1);// 推进 epochstd::lock_guardstd::mutexlock(mu_);for(autoitretired_.begin();it!retired_.end();){delete*it;// 演示内存回收机制此处故意裸 deletereclaimed_.fetch_add(1);itretired_.erase(it);}}};intmain(){MiniEpoch e;int*anewint(1);// 演示内存回收机制故意裸 newe.enter(0);// 读者 0 进入读临界区e.retire(a);// 退休 astd::printf(读者在读时能回收吗 %d\n,static_castint(e.all_idle()));e.exit(0);// 读者 0 退出到达静止点e.advance_and_reclaim();// 推进 epoch 并回收std::printf(读者退出后累计回收 %d新 epoch %lu\n,e.reclaimed_.load(),e.global_epoch_.load());}读者在读时能回收吗 0 读者退出后累计回收 1新 epoch 2注意enter/exit只是两次对原子变量的store读者之间互不干扰 —— 这就是「读侧免锁」的含义。代价转移到回收者它必须等所有读者都走到静止点任何一个读者卡在临界区里回收就整体被拖住。所以 epoch-based 适合「读极多、写极少、读者绝不在临界区里阻塞」的场景比如路由表、配置的热更新。6. 完整示例引用计数的多生产者多消费者无锁栈把最实用的引用计数方案做成完整版4 个生产者各 push 一批4 个消费者并发 pop用「总和守恒」验证每个值恰好被取走一次、没有丢也没有重复// rc_stack_full.cpp — 编译: g -stdc17 -Wall -O2 -pthread rc_stack_full.cpp -o rcfull#includeatomic#includecstdio#includememory#includethread#includevectortemplatetypenameTclassRcStack{structNode{T value;std::shared_ptrNodenext;};std::shared_ptrNodehead_;public:voidpush(T v){autonstd::make_sharedNode();n-valuev;n-nextstd::atomic_load(head_);while(!std::atomic_compare_exchange_weak(head_,n-next,n)){}}boolpop(Tout){autooldstd::atomic_load(head_);while(old!std::atomic_compare_exchange_weak(head_,old,old-next)){}if(!old)returnfalse;outold-value;returntrue;}};intmain(){constexprintkProducers4;constexprintkConsumers4;constexprintkPerProducer2500;constexprintkTotalkProducers*kPerProducer;RcStackints;std::vectorstd::threadproducers;for(intp0;pkProducers;p){producers.emplace_back([s]{for(intk1;kkPerProducer;k)s.push(k);// 每个生产者 push 1..2500});}for(autot:producers)t.join();std::atomiclonglongsum{0};std::atomicintcount{0};std::vectorstd::threadconsumers;for(intc0;ckConsumers;c){consumers.emplace_back([s,sum,count]{intv;while(s.pop(v)){sum.fetch_add(v,std::memory_order_relaxed);count.fetch_add(1,std::memory_order_relaxed);}});}for(autot:consumers)t.join();constlonglongexpectedstatic_castlonglong(kProducers)*kPerProducer*(kPerProducer1)/2;std::printf(pop 总数 %d期望 %d\n,count.load(),kTotal);std::printf(pop 总和 %lld期望 %lld\n,sum.load(),expected);}pop 总数 10000期望 10000 pop 总和 12505000期望 12505000两个数字都精确对得上说明在引用计数保护下多生产者多消费者各自并发地 CAS 头指针没有丢节点、没有重复取、也没有一次 use-after-free。这正是引用计数作为「最省心的内存回收」的底气。工程上std::atomicstd::shared_ptrT在 C20 里把上面那三个自由函数收编成了成员函数写起来更顺手但在 C17 环境下自由函数版本就是标准写法。7. 延伸阅读std::atomic_load / atomic_store / atomic_compare_exchange_weakshared_ptr 特化— cppreference —— 无锁结构里安全操作shared_ptr的官方入口这篇示例 1 和示例 6 就靠它std::shared_ptr — cppreference —— 引用计数与make_shared控制块分配理解「引用计数回收」的基础std::memory_order — cppreference —— 无锁代码里每个原子操作都要显式定序这是内存序的权威参考C Core Guidelines · 并发章节 — isocpp —— 无锁/内存回收属于高风险代码指南明确要求「先证明必要再写」Compiler Explorer — godbolt.org —— 把上面的无锁栈贴进去开-O2看atomic_load/compare_exchange到底生成了什么指令本知识库内的相关篇目《无锁栈Treiber 栈的 CAS、ABA 问题与 tagged pointer》 —— 无锁栈Treiber stack是理解无锁编程的入门课《无锁队列MPSC/MPMCVyukov 环形队列与 slot 未就绪问题》 —— 队列比栈难因为它有两个端点——头尾都要协调。《锁进阶atomic_flag 自旋锁、shared_mutex 读写锁与无锁的代价》 —— std::mutex 一竞争就把失败线程挂进内核futex8. 一句话总结无锁结构里「从结构移除节点」和「物理delete节点」之间隔着一段无人知道的引用窗口所以不能直接delete三种回收思路本质都是「延迟释放 判定无人引用」。引用计数靠shared_ptr最省心、hazard pointer 靠读者声明「我在用」省读侧开销、epoch-based 靠读者登记 epoch 做到读侧免锁这三段的代码都是讲原理的最小实现生产级请直接用folly::hazptr、libcds或内核 RCU 这类经过验证的现成库别自己重造。
阅读完成 · 觉得有帮助？
