Collector 与 LocalHandle

主线位置：回收主线的「参与者」从哪来、到哪去。epoch::pin() 背后是默认收集器和线程本地 HANDLE，这一章把这两层拆开：Collector 是一个独立的「全局纪元 + 参与者链表 + 垃圾队列」，LocalHandle 是一个参与者。

flowchart TB
  c["Collector：global 是 Arc<Global>"] -->|"register()"| h["LocalHandle：local 是 *const Local"]
  h -->|"pin()"| g["Guard"]
  h -->|"drop：handle_count -= 1"| rel{"guard_count == 0 且这是最后一个 handle?"}
  g -->|"drop：unpin"| un{"guard_count 归零且 handle_count == 0?"}
  rel -->|"是"| fin["Local::finalize"]
  un -->|"是"| fin
  fin --> push["pin 一下，把本地 Bag 推进全局队列"]
  push --> del["链表节点打删除标记"]
  del --> drop["drop 自己持有的 Collector 克隆"]
  drop -->|"若是最后一个"| gd["Global drop：队列里所有袋子立即执行"]

两个类型

collector.rs，去掉测试只有一百来行：

pub struct Collector {
    pub(crate) global: Arc<Global>,
}

unsafe impl Send for Collector {}
unsafe impl Sync for Collector {}

pub struct LocalHandle {
    pub(crate) local: *const Local,
}

• Collector 是 Arc<Global> 的包装，Clone 共享同一个 Global，PartialEq 比的是 Arc::ptr_eq。可以跨线程传。
• LocalHandle 是一个参与者，裸指针让它 !Send + !Sync：参与者的本地袋子、guard_count 都是不加同步的 Cell，只能在一个线程里用。
• LocalHandle::pin() 就是 (*self.local).pin()，和默认收集器走的是同一个 Local::pin。

默认收集器只是「一个全局 Collector + 每线程一个 LocalHandle」（pin 与 Guard）。

为什么要自定义收集器

每个 Collector 有自己的全局纪元、自己的参与者链表、自己的垃圾队列。两个收集器之间完全不相干：

• 隔离：A 结构的读者长时间钉在 A 的收集器上，不会挡住 B 结构的回收。用默认收集器时，进程里任何一个地方长期 pin 住，所有用默认收集器的结构（包括 deque、skiplist）都停止回收（defer 与回收 的 E3）。
• 及时释放：Collector 的最后一个克隆和最后一个参与者都没了以后，Global 被 drop，队列里剩下的袋子不看纪元全部执行（Queue::drop 用 unprotected() 弹空，sync/queue.rs 第 203 行）。默认收集器在 static 里，永远不会被 drop。
• 测试：crate 自己的单元测试和 loom 测试都用 Collector::new()，测试之间互不影响。

代价是每次操作都要拿到对应的 LocalHandle 再 pin；线程本地缓存要调用方自己做。

crossbeam-skiplist 的底层 SkipList::new(collector: Collector) 让用户传入收集器，公开的方法都接受 &Guard，并在入口处检查 guard 属于同一个收集器（crossbeam-skiplist-0.1.3 的 base.rs 第 376 行）：

fn check_guard(&self, guard: &Guard) {
    if let Some(c) = guard.collector() {
        assert!(c == &self.collector);
    }
}

Guard::collector() 对 unprotected() 返回 None，所以无保护的 guard 能通过这个检查，后果由调用方负责。上层的 SkipMap / SkipSet 固定用 epoch::default_collector().clone()，方法内部自己 epoch::pin()。

注册

Local::register：

pub(crate) fn register(collector: &Collector) -> LocalHandle {
    unsafe {
        // Since we dereference no pointers in this block, it is safe to use `unprotected`.

        let local = Owned::new(Local {
            entry: Entry::default(),
            collector: UnsafeCell::new(ManuallyDrop::new(collector.clone())),
            bag: UnsafeCell::new(Bag::new()),
            guard_count: Cell::new(0),
            handle_count: Cell::new(1),
            pin_count: Cell::new(Wrapping(0)),
            epoch: CachePadded::new(AtomicEpoch::new(Epoch::starting())),
        })
        .into_shared(unprotected());
        collector.global.locals.insert(local, unprotected());
        LocalHandle {
            local: local.as_raw(),
        }
    }
}

一次堆分配，Local 里自己持有一个 Collector 克隆（ManuallyDrop，注销时手动 drop），所以 Collector 可以先于 handle 被 drop：collector.rs 开头的文档例子就是 drop(collector) 之后 handle.pin().flush() 照常工作。

两个计数，谁最后走谁收尾

Local 有两个计数：

两者都归零时注销。最后归零的那一方负责调用 finalize：

• release_handle：guard_count == 0 && handle_count == 1（减之前）。
• unpin：最后一个 guard 离开，且 handle_count == 0。

第二种情况在正常使用里很少见，但默认收集器在线程退出时会碰到：HANDLE 已经析构，with_handle 临时注册一个 handle，这个 handle 在表达式结束就 drop 了，guard 还在，注销只能等 guard drop。

注销：finalize

第 521 行：

fn finalize(&self) {
    // Temporarily increment handle count. This is required so that the following call to `pin`
    // doesn't call `finalize` again.
    self.handle_count.set(1);
    unsafe {
        // Pin and move the local bag into the global queue. It's important that `push_bag`
        // doesn't defer destruction on any new garbage.
        let guard = &self.pin();
        self.global()
            .push_bag(self.bag.with_mut(|b| &mut *b), guard);
    }
    // Revert the handle count back to zero.
    self.handle_count.set(0);

    unsafe {
        // Take the reference to the `Global` out of this `Local`. Since we're not protected
        // by a guard at this time, it's crucial that the reference is read before marking the
        // `Local` as deleted.
        let collector: Collector = ptr::read(self.collector.with(|c| &*(*c)));

        // Mark this node in the linked list as deleted.
        self.entry.delete(unprotected());

        // Finally, drop the reference to the global. Note that this might be the last reference
        // to the `Global`. If so, the global data will be destroyed and all deferred functions
        // in its queue will be executed.
        drop(collector);
    }
}

四步：

1. 把 handle_count 临时设为 1 再 pin：push_bag 需要一个 guard；guard 在块尾 drop 时会 unpin，如果 handle_count 是 0，unpin 会再次调用 finalize。
2. 本地袋子整袋推进全局队列，哪怕没满。线程退出时残留的垃圾就是这样交给别人的。它们仍然要等纪元前进两步，由之后某个线程的 collect 执行。
3. 先把 Collector 读出来，再打删除标记。注释强调顺序：打完标记以后，别的线程的 try_advance 随时可能把这个 Local 摘下并 defer_destroy，而本线程此刻没有被 guard 保护，之后不能再碰 Local 的任何字段。
4. drop 这个 Collector 克隆。如果它是最后一个引用，Global 立即 drop，队列里所有袋子执行，链表里所有 Local 释放。

Local 的内存本身是由别的线程释放的：被 try_advance 遍历时摘下、defer_destroy、再等两个纪元（内部的无锁链表和队列）。

main 上有一个和这里相关、没有进入 0.9.21 的修复：c8153dbe 修了「collector 和 handle 都 drop 后再使用 guard」时 Guard::drop 持有的 &Local 跨过 finalize 的 Stacked Borrows 问题，见 pin 与 Guard 末尾。

实验 E5（第二部分）：在 loom 下跑 crossbeam-epoch

crossbeam-epoch 是 crossbeam 里少数能直接用 loom 检查的 crate。lib.rs 里有一层 crate::primitive（第 71–133 行）：--cfg crossbeam_loom 时原子类型、UnsafeCell、Arc、thread_local! 全部换成 loom 的；默认收集器的 OnceLock 换成 loom::lazy_static!；compiler_fence 因为 loom 不支持，用 fence 代替（注释说这可能漏掉一些竞争）。CI 的 ci/crossbeam-epoch-loom.sh 用 --cfg crossbeam_loom --cfg crossbeam_sanitize、LOOM_MAX_PREEMPTIONS=2 跑 tests/loom.rs，怎么验证并发代码讲 CI 的部分。

断言：读者钉住时，它读到的旧对象不会被析构；读者不钉住（用 unprotected()），loom 能找到旧对象在读者手里时被析构的交错。

方法（exp-crossbeam/epoch-loom/）：依赖 registry 里的 crossbeam-epoch =0.9.21（开 loom feature），用 RUSTFLAGS="--cfg crossbeam_loom" 和单独的 target 目录编译，所以跑的是发布版的 epoch 代码本身。模型里一个 Collector、一个 Atomic<Obj>，Obj 析构时把一个 loom AtomicBool 置位：

• 读者线程：注册 handle，pin（或取 unprotected()），load 指针；如果读到的是旧对象，断言旧对象的析构标志还没置位。读者不解引用，所以反向的情况表现为断言失败，而不是真的读到已释放内存。
• 写者（主线程）：pin，swap 进新对象，defer_destroy 旧对象，flush；再 pin().flush() 两次。没人挡着时这足够推进两次纪元、析构旧对象。

结果：

反向验证就是第三行：同一个模型只把读者的 guard 换成 unprotected()，loom 很快找出了写者推进两次纪元、析构旧对象，而读者刚好读到旧指针的交错。这说明前两行的「通过」不是因为模型太弱、根本碰不到回收。

这个实验的覆盖面是有限的：两个线程、一个对象、抢占上限 2 或 3；crossbeam_loom 下 compiler_fence 被换成了更强的 fence，x86 的 CAS 分支在 AArch64 上编译不到。它验证的是 pin / defer / try_advance 这一套在这个场景下的内存序，不是整个 crate。

小结

• Collector 是 Arc<Global>：一套独立的全局纪元、参与者链表、垃圾队列。LocalHandle 是一个参与者，!Send。
• 自定义收集器带来隔离（别处长期 pin 不影响它）和及时释放（最后一个引用没了，剩下的垃圾立即执行）；skiplist 用 check_guard 校验 guard 属于同一个收集器，unprotected() 不受检查。
• Local 用 handle_count 和 guard_count 决定何时注销，最后归零的一方调用 finalize：把本地袋子推进全局队列、打删除标记、drop 自己持有的 Collector 克隆。Local 的内存由之后遍历链表的线程摘除、再经两个纪元释放。
• 在 --cfg crossbeam_loom 下编译 crossbeam-epoch 0.9.21，loom 检查了 10 374 个交错（抢占上限 2）没有发现钉住的读者手里的对象被析构；把读者换成 unprotected()，第 21 个交错就报错。