defer 与回收

主线位置：回收主线的后半段。guard.defer_destroy(old) 把析构放进本线程的袋子；袋子满了封上纪元交给全局队列；某次 pin 或 flush 触发 collect，先试着推进全局纪元，再把「老了两个纪元」的袋子拿出来执行。

flowchart LR
  d["defer_destroy(old)"] --> def["Deferred：fn 指针 + 3 个字"]
  def --> bag["本地 Bag：最多 64 个"]
  bag -->|"满了，或 flush"| seal["push_bag：SeqCst fence，读全局纪元，封成 SealedBag"]
  seal --> q["全局 MS 队列"]
  pin["每 128 次 pin，或 flush"] --> col["collect"]
  col --> adv["try_advance：所有已钉住的参与者都在当前纪元才 +1"]
  adv --> pop["弹出 global - bag.epoch >= 2 的袋子，最多 8 个"]
  q --> pop
  pop --> run["Bag::drop：逐个执行析构"]

Deferred：放进袋子的闭包

deferred.rs 把任意 FnOnce() 装进一个固定大小的结构体：

const DATA_WORDS: usize = 3;
type Data = [usize; DATA_WORDS];

pub(crate) struct Deferred {
    call: unsafe fn(*mut u8),
    data: MaybeUninit<Data>,
    _marker: PhantomData<*mut ()>, // !Send + !Sync
}

闭包不超过 3 个字、对齐不超过 usize 时直接存在 data 里，否则装箱，data 里只存 Box 指针。call 是针对闭包类型单态化出来的函数，负责从 data 里把闭包读出来调用。在 64 位平台上一个 Deferred 占 32 字节。

注释说「3 个字够大多数情况用」，例子是一个函数指针加一个胖指针。常见调用方都放得下：

Guard::defer_unchecked 在外面再包一层 move || drop(f())（guard.rs 第 188 行），所以闭包可以有返回值，返回值在执行后立即 drop。安全版本 defer 要求 F: Send + 'static：析构可能在别的线程上执行。defer_destroy 不要求 T: Send，文档解释说 Shared 本身不是 Send，类型系统证明不了，但到了宽限期之后对象已经不再共享，交给别的线程析构是安全的。

Bag：64 个一袋

#[cfg(not(any(crossbeam_sanitize, miri)))]
const MAX_OBJECTS: usize = 64;
// Makes it more likely to trigger any potential data races.
#[cfg(any(crossbeam_sanitize, miri))]
const MAX_OBJECTS: usize = 4;

pub(crate) struct Bag {
    deferreds: [Deferred; MAX_OBJECTS],
    len: usize,
}

（internal.rs 第 55–67 行）

64 × 32 + 8 = 2056 字节。它直接内嵌在每个线程的 Local 里；反汇编实验程序，Local::pin 访问 guard_count 用的偏移是 0x818（2072 = 链表节点 8 + Collector 指针 8 + Bag 2056），和这个算法对得上。Bag 的 Drop 逐个执行里面的闭包，「执行一袋垃圾」就是 drop 一个 Bag。

MAX_OBJECTS 在 sanitizer 和 Miri 下缩成 4，让袋子更频繁地进全局队列、更容易暴露竞争。64 这个值有过反复：2020 年 9 月 b5ca3b27 为了让一个 Local 不超过 2048 字节（#551），把它改成 62 并加了大小断言；2022 年 7 月 #879（f2ca66ba）又改回 64，说明是回滚 #552，「降低下游有缺陷的实现发生段错误的风险」（见 #869），大小断言被注释掉，测试 local_size 现在是空的。

Local::defer 很短：

pub(crate) unsafe fn defer(&self, mut deferred: Deferred, guard: &Guard) {
    let bag = self.bag.with_mut(|b| &mut *b);

    while let Err(d) = bag.try_push(deferred) {
        self.global().push_bag(bag, guard);
        deferred = d;
    }
}

放不下就把整袋交出去，换一个空袋再放。袋子是线程本地的，try_push 没有任何同步，这是把「每个对象一次全局同步」摊成「每 64 个对象一次」。

push_bag：盖上纪元戳

第 182 行：

pub(crate) fn push_bag(&self, bag: &mut Bag, guard: &Guard) {
    let bag = mem::replace(bag, Bag::new());

    atomic::fence(Ordering::SeqCst);

    let epoch = self.epoch.load(Ordering::Relaxed);
    self.queue.push(bag.seal(epoch), guard);
}

先一道 SeqCst fence，再读全局纪元，封成 SealedBag { epoch, bag } 推进全局队列。fence 保证袋子里那些对象的「摘除」先于「读全局纪元」对所有线程可见，和 pin 里的 fence 配对（pin 与 Guard）。全局队列是一个 Michael-Scott 无锁队列，下一章讲。

try_advance：全局纪元什么时候能走

第 228 行，去掉 TSan 分支：

pub(crate) fn try_advance(&self, guard: &Guard) -> Epoch {
    let global_epoch = self.epoch.load(Ordering::Relaxed);
    atomic::fence(Ordering::SeqCst);

    for local in self.locals.iter(guard) {
        match local {
            Err(IterError::Stalled) => {
                // A concurrent thread stalled this iteration. That thread might also try to
                // advance the epoch, in which case we leave the job to it. Otherwise, the
                // epoch will not be advanced.
                return global_epoch;
            }
            Ok(local) => {
                let local_epoch = local.epoch.load(Ordering::Relaxed);

                // If the participant was pinned in a different epoch, we cannot advance the
                // global epoch just yet.
                if local_epoch.is_pinned() && local_epoch.unpinned() != global_epoch {
                    return global_epoch;
                }
            }
        }
    }
    atomic::fence(Ordering::Acquire);

    let new_epoch = global_epoch.successor();
    self.epoch.store(new_epoch, Ordering::Release);
    new_epoch
}

遍历所有注册过的参与者：没钉住的不管；钉住的必须都钉在当前全局纪元上，有一个落在旧纪元就放弃。链表遍历被别的线程打断（Stalled）也放弃，注释说那个线程多半也在推进，交给它。全部通过才把全局纪元加一。

几个细节：

• 读各参与者的本地纪元用的是 Relaxed，循环结束后补一道 fence(Acquire)，和那些参与者 unpin 时的 Release 写配对：它们在旧纪元里的访问，都在本线程宣布推进之前完成。
• 推进用的是普通 store，不是 CAS。注释的论证：调用者自己钉在 global_epoch 上，全局纪元最多被别人推到 global_epoch + 1，不可能更远；别人抢先推进了，这里写入的也是同一个值。
• TSan 不理解 fence，crossbeam_sanitize_thread 下把「Relaxed 读 + fence」改写成「先把参与者收集进一个 Vec，再逐个 Acquire 读一遍」（第 240–270 行）。注释承认要分配内存「很遗憾」，但不这样做会在不该同步的地方制造同步、导致误报。见 怎么验证并发代码。

差两个纪元：is_expired

fn is_expired(&self, global_epoch: Epoch) -> bool {
    // A pinned participant can witness at most one epoch advancement. Therefore, any bag that
    // is within one epoch of the current one cannot be destroyed yet.
    global_epoch.wrapping_sub(self.epoch) >= 2
}

（第 148 行）

为什么是 2：设袋子盖戳时读到的全局纪元是 E。

1. 袋子里的对象在盖戳之前已经摘除（push_bag 的 fence 之前）。在这之后才 pin 的读者读不到它们。
2. 能读到它们的读者，pin 时读到的全局纪元 r 不会超过 E。全局纪元在任何时刻最多比任何一个仍钉住的参与者领先 1（try_advance 只在所有钉住者都在当前纪元时推进）。
3. 所以只要这个读者还钉着，全局纪元最多到 r + 1 ≤ E + 1。全局纪元到了 E + 2，说明它已经离开了。

注释那句「一个已钉住的参与者最多看到一次推进」就是第 2 条。

collect：每次最多 8 袋

#[cold]
pub(crate) fn collect(&self, guard: &Guard) {
    let global_epoch = self.try_advance(guard);

    let steps = if cfg!(crossbeam_sanitize) {
        usize::MAX
    } else {
        Self::COLLECT_STEPS
    };

    for _ in 0..steps {
        match self.queue.try_pop_if(
            &|sealed_bag: &SealedBag| sealed_bag.is_expired(global_epoch),
            guard,
        ) {
            None => break,
            Some(sealed_bag) => drop(sealed_bag),
        }
    }
}

（第 199 行）

COLLECT_STEPS = 8：一次 collect 最多析构 8 袋。try_pop_if 只看队头，队头没过期就停，不往后找。袋子大致按纪元顺序入队，所以队头没过期，后面的通常也没过期。

谁调 collect：

两条加起来能推出一个上限：**不调 flush 时，一个线程每 128 次 pin 最多析构 8 × 64 = 512 个对象，平均每次 pin 约 4 个。**每次 pin 都 defer 超过 4 个对象、又没有人 flush，垃圾会越积越多。下面的实验 E3 量到了这一点。

还有一个容易漏掉的地方：从全局队列弹出一个袋子时，装它的队列节点本身也要交给 epoch 回收（guard.defer_destroy(head)），所以回收会产生新的垃圾，collect 的文档注释第一句就是「This may itself produce garbage and in turn allocate new bags」。

flush

Local::flush：本地袋子不空就交出去，然后 collect 一次。Guard::flush 的文档说它让延迟的函数「尽快」执行，因为进了全局队列以后任何线程都有机会执行它。

它不能绕过纪元条件：flush 之后袋子仍要等全局纪元前进两步。deque 扩容时，如果新 buffer 不小于 1 KiB 就 flush 一次（扩容与回收），目的是让大块的旧 buffer 早点进全局队列，不至于在本线程袋子里等另外 63 个对象。

实验 E3：长期 pin 住、flush 与回收速度

断言：一个线程长期钉住时，所有线程 defer 的对象一个都不会析构，内存线性增长；flush 改变不了这一点。

方法（exp-crossbeam/epoch-rss/）：辅助线程 epoch::pin() 后睡眠（hold），或 pin 一下就放开（free）。主线程循环 20 万次：pin，swap 进一个新的 1 KiB 对象，defer_destroy 旧的（flush 模式下再调一次 guard.flush()）。每 2 万次打印已析构个数和进程 RSS（ps -o rss=）。之后放开辅助线程，主线程反复 pin，数要多少次 pin 才回收完。release，各模式重复 3 次，数字完全一致（单线程推进，结果是确定的）。M2 Max（8P + 4E）。

结果：

（起始 RSS 1 920 KiB。）

• hold：析构 0 个，RSS 每退休一个对象涨约 1.05 KiB（对象本身 1 KiB，加上每 64 个对象一个约 2 KiB 的袋子）。
• hold + flush：仍是 0 个，RSS 反而涨到 3.5 倍，每个对象约 3.68 KiB。因为每次 flush 都把只装了 1 个对象的袋子推进全局队列，每个队列节点里是一整个 2056 字节的 Bag（加上纪元戳和 next 指针共 2072 字节；macOS 分配器按 512 字节档位取整到 2560，这一步是推测）。长期 pin 住时 flush 是负优化。
• free：始终只积压约 223 个（3 袋多），说明正常情况下回收跟得上。

放开辅助线程后，回收完要多少次 pin：

两个数都和「每 128 次 pin 回收 8 袋」吻合。hold 模式最后那 64 个还留在本线程袋子里的对象也被回收了，是因为弹出队列节点产生的新垃圾把本地袋子挤满、推了出去，正是上面说的「回收会产生垃圾」。

再看回收速度上限：辅助线程不钉住，主线程每次 pin 下 defer k 个对象，一共 20 万个，不 flush，循环结束时的析构数：

k ≤ 4 时基本跟得上，k = 8 时只回收了一半，k = 16 只有四分之一，和「平均每次 pin 最多约 4 个」一致。同一个 k = 8 改成每次 pin 后 flush，积压只剩 8 个。所以批量删除（一次 pin 摘掉大量节点）之后，调一次 flush 是有意义的；长期 pin 住的线程存在时，flush 就没有意义。

反向验证：free 模式本身就是对照，同样的退休节奏，没有长期钉住者时每个检查点都只差约 200 个。另外 hold 模式放开钉住者后很快全部回收（上表），说明前面的 0 确实是钉住造成的，不是程序本身不回收。

历史：三个纪元，5 天

2020 年 5 月 20 日合并的 #416（「Require three epoch advancements for deallocation」，提交 52a4e31c，作者日期 2018 年 11 月）把规则改成：

• push_bag 不再做 SeqCst fence、不再读全局纪元，袋子盖的是本线程的本地纪元（self.epoch.load(Relaxed).unpinned()）。
• 本地纪元可能比全局纪元落后 1，于是过期条件从 >= 2 改成 >= 3，注释改成「一个已钉住的参与者最多看到两次推进」。
• pin 里加了一个校验循环：写完本地纪元、做完 fence 后再读一次全局纪元，变了就用新值重来，保证钉住时抄下的纪元不是过时的。
• skiplist 的 drops 测试相应多加了一次 handle.pin().flush()：要多推进一次纪元才能回收。

两种论证的差别在「袋子的戳从哪来」：现行版本在 push_bag 里付一道 SeqCst fence，换来一个准确的全局纪元，于是两步就够；#416 省掉这道 fence，用一个可能落后一步的本地纪元盖戳，于是要多等一步，并把正确性的一部分挪到 pin 的校验循环里。

5 天后，5 月 25 日合并的 #517 把它整个回滚（389a60b0）。#517 的说明是：怀疑 #416 引起了 #514 里报告的 SIGSEGV，先回滚看 SIGSEGV 会不会消失。git 历史里之后再没有改回三个纪元，现行代码就是回滚后的样子。#514 的根因到底是不是 #416，从仓库历史里看不出结论。

小结

• 一个 Deferred 是函数指针加 3 个字的内联存储，32 字节；64 个装一袋，袋子内嵌在每线程的 Local 里，填满前不做任何同步。
• 袋子满了或 flush 时，push_bag 做一道 SeqCst fence、读全局纪元盖戳，推进全局 MS 队列。
• try_advance 遍历所有参与者，钉住的都在当前纪元才把全局纪元加一；袋子要等全局纪元比戳大 2 才析构，因为钉住的读者最多看到一次推进。
• 一次 collect 最多析构 8 袋；不 flush 时每线程平均每次 pin 最多回收约 4 个对象，实测每次 pin 退休 8 个时只回收一半。
• 长期 pin 住时垃圾零回收（实测 20 万个 1 KiB 对象，RSS 涨到约 206 MiB）；此时 flush 让每个对象多占一整个袋子，RSS 涨到约 721 MiB。
• 2020 年「三个纪元」的改动上线 5 天就因疑似引起 SIGSEGV 被回滚。