最小 EBR

主线位置：主线 3「一块内存的回收」的缩小版。pin 对照 pin 与 Guard，退休和推进对照 defer 与回收。代码在 exp-crossbeam/hands-on/src/mini_ebr.rs。

flowchart TB
  pin["pin(slot)：读全局纪元 e，本地 = (e << 1) | 1，SeqCst fence"] --> read["读共享指针，用它"]
  read --> unpin["Guard drop：本地 = 0（Release）"]
  retire["retire：拿锁，读全局纪元 e，放进 bags[e % 3]"] --> adv
  adv["try_advance：拿锁，读 e，SeqCst fence，所有已 pin 的本地纪元都是 e？"] -->|是| step["全局 = e + 1；取出 bags[(e + 2) % 3]，即纪元 e - 1 的那一袋，放锁后释放"]
  adv -->|否| no["返回 false"]

数据结构

pub struct Ebr {
  global: AtomicUsize,
  /// 0 = not pinned; (epoch << 1) | PINNED = pinned in `epoch`.
  locals: [AtomicUsize; SLOTS],
  bags: Mutex<[Vec<Garbage>; 3]>,
  fences: Fences,
}

和 crossbeam 的对应关系：

Garbage 是一个裸指针加一个释放函数（free_box::<T> 把它还原成 Box<T> 再 drop），相当于 crossbeam 的 Deferred 只支持 defer_destroy 一种。

pin

pub fn pin(&self, slot: usize) -> Guard<'_> {
  let e = self.global.load(Ordering::Relaxed);
  self.locals[slot].store((e << 1) | PINNED, Ordering::Relaxed);
  // store(local) -> load(shared pointer) must not be reordered: pairs with
  // the fence in `try_advance` (store-buffering otherwise).
  if self.fences.pin {
    fence(Ordering::SeqCst);
  }
  Guard { ebr: self, slot }
}

和 crossbeam 的 Local::pin 一样：Relaxed 读全局、Relaxed 写本地、一道 SeqCst fence（pin 与 Guard）。Guard drop 时把本地纪元 Release 写成 0。

retire 与 try_advance

pub unsafe fn retire<T>(&self, _guard: &Guard<'_>, ptr: *mut T) {
  // The epoch is read under the same lock `try_advance` holds, so a bag is
  // stamped either before an advance (and that advance happens after the
  // unlink) or after it (and gets the newer epoch). No fence needed here.
  let mut bags = self.bags.lock().unwrap();
  let e = self.global.load(Ordering::Relaxed);
  bags[e % 3].push(Garbage { ptr: ptr as *mut (), free: free_box::<T> });
}

pub fn try_advance(&self) -> bool {
  let mut bags = self.bags.lock().unwrap();
  let e = self.global.load(Ordering::Relaxed);
  if self.fences.advance {
    fence(Ordering::SeqCst);
  }
  for local in &self.locals {
    let l = local.load(Ordering::Relaxed);
    if l & PINNED != 0 && (l >> 1) != e {
      return false;
    }
  }
  fence(Ordering::Acquire);
  let new = e + 1;
  // bags[(new + 1) % 3] == bags[(e - 1) % 3]: retired at epoch new - 2
  let expired = std::mem::take(&mut bags[(new + 1) % 3]);
  self.global.store(new, Ordering::Release);
  drop(bags);
  for g in expired {
    unsafe { (g.free)(g.ptr) };
  }
  true
}

规则和 crossbeam 相同：所有已 pin 的参与者都在当前纪元 e 才推进；在纪元 s 退休的垃圾，等全局纪元到 s + 2 才释放（crossbeam 的 global - bag.epoch >= 2，defer 与回收）。三个袋子轮换正好实现这一点：推进到 new 时清空的 bags[(new + 1) % 3] 就是 bags[(new - 2) % 3]。

retire 要求调用方已 pin（参数里的 &Guard）：这保证在它放进 bags[e % 3] 之前，全局纪元最多再走一步，不会走到把这个袋子清掉的 e + 2。

为什么 retire 里没有 fence

crossbeam 的 push_bag 在读全局纪元给袋子盖戳之前有一道 SeqCst fence。这里没有，因为读纪元和推进纪元都在同一把锁里：要么这次 retire 的临界区在某次推进之前，那么摘链（swap）先于这次推进对全局纪元的 Release 写，之后读到新纪元的读者经 pin 里的 fence 与它同步，一定看得到摘链；要么在推进之后，袋子盖的就是推进后的纪元，晚一个纪元释放。crossbeam 的封袋走无锁的 MS 队列，没有这把锁可借，才需要那道 fence。

下面的 loom 实验验证了这一点，也顺带说明了 fence 放在哪里的问题。

loom 检查

tests/ebr_loom.rs，--features loom 时 mini_ebr 的原子类型、fence、Mutex 都换成 loom 的。模型（照 Collector 与 LocalHandle 里 E5 第二部分的写法）：

• 读者：pin(1)，Acquire 读共享指针；如果读到的是旧节点，断言旧节点的「已释放」标志还是假。读者不解引用，坏的情况表现为断言失败，而不是真的读已释放内存。
• 写者：pin(0)，swap 进新节点，retire 旧节点，unpin；然后 try_advance 三次（够从退休纪元走两步）。
• 两线程模型里写者也负责推进；三线程模型里写者只 swap + retire，推进由主线程做两次。

fn model(fences: Fences, pin_reader: bool) {
  loom::model(move || {
    let ebr = Arc::new(Ebr::with_fences(fences));
    // ... old / new 两个节点，shared = AtomicPtr::new(old)
    let reader = thread::spawn(move || {
      let guard = if pin_reader { Some(ebr.pin(1)) } else { None };
      let p = shared.load(Ordering::Acquire);
      if p as usize == old_addr {
        assert!(!old_freed.load(Ordering::Acquire), "old node freed while a reader still held it (execution {iter})");
      }
      drop(guard);
    });
    {
      let guard = ebr.pin(0);
      let prev = shared.swap(new, Ordering::AcqRel);
      unsafe { ebr.retire(&guard, prev) };
    }
    for _ in 0..3 {
      ebr.try_advance();
    }
    reader.join().unwrap();
  });
}

（片段省略了节点构造、Arc 克隆和收尾释放，完整的在测试文件里。）LOOM_MAX_PREEMPTIONS=2，release：

LOOM_MAX_PREEMPTIONS=3 时两个无改动的模型分别是 457 和 29 237 个交错，全部通过；反向的五种改动同样全部失败（三线程的两种在第 35 842、35 094 个交错）。失败时的交错数可以比通过时的总数大：去掉 fence 以后读操作能读到的值更多，状态空间本身变大了。

两道 fence 缺一不可，是标准的 store-buffering 形状：读者「写本地纪元，再读共享指针」，推进者「（经锁）看到摘链，再读本地纪元」。两边各一道 SeqCst fence，才能排除「读者读到旧指针、推进者又读到读者未 pin」同时发生。

一个没有被抓到的版本

这一页的代码最初在 retire 里也放了一道 SeqCst fence（照 crossbeam 的 push_bag）。那一版在两线程模型下去掉 try_advance 的 fence，loom 跑完 584 个交错没有报错。原因是两线程模型里写者自己也是推进者：retire 里那道 fence 恰好位于「摘链」和「读本地纪元」之间、在同一个线程上，起了和推进者 fence 一样的作用。去掉 retire 的 fence 之后，同样的改动在第 2 个交错就失败了；三线程模型（推进者是另一个线程）则不管 retire 有没有 fence 都能抓到。

这说明两件事：反向验证的结论依赖模型的形状，一个改动「没被抓到」不等于它安全；以及 loom 自己也写明了它模拟不了某些跨变量的重排（loom 0.7.2 文档「Relaxed Memory Ordering」一节），所以「loom 通过」只是在它能模拟的范围内通过。

多线程测试

tests/ebr_std.rs 不用 loom，用真线程：3 个读者循环「pin、读指针、解引用并检查节点的 alive 标志、unpin」，1 个写者 20 万次「pin、换新节点、retire、unpin、try_advance」。读者共读了约 570 万次，没有一次读到已释放的节点；结束时全局纪元约 18.7 万，20 万个节点里 199 999 个已经释放（剩下的一个在最后两个纪元的袋子里），Ebr drop 时其余全部释放，计数正好 200 001（含最后仍挂在共享指针上的那个）。这个测试只能说明「没有撞上」，正确性的论证靠上面的 loom。

省掉了什么

这里的 retire 是 unsafe fn、Guard 也不限制读到的裸指针活多久：调用方要自己保证「只在 guard 活着时用读到的指针」。crossbeam 用 Shared<'g> 把这条规则交给了编译器。

小结

• 一个全局纪元、每个参与者一个本地纪元、三个轮换的垃圾袋，就是 EBR 的全部骨架：已 pin 的都在当前纪元才推进，退休满两个纪元才释放。
• 读侧只有「写本地纪元 + SeqCst fence」；推进侧在读所有本地纪元前也要一道 SeqCst fence，两边配成 store-buffering 的一对。
• 退休和推进共用一把锁时，退休不需要 fence；crossbeam 的封袋是无锁的，所以需要。
• loom：两线程 188、三线程 3 727 个交错通过；读者不 pin、去掉任一道 fence 都会被抓到。一个早期版本「没被抓到」的原因是模型里写者兼推进者，fence 恰好在同一线程上。