Rust in Everything
基于 crossbeam-epoch 0.9.21

内部的无锁链表和队列

主线位置：回收主线里两处「全局」的数据结构。try_advance 遍历的「所有参与者」是一条无锁链表；push_bag 推进去、collect 弹出来的「全局队列」是一个 Michael-Scott 队列。两者都是 crate 私有的。

flowchart LR
  subgraph global["Global"]
    locals["locals: List<Local>"]
    queue["queue: Queue<SealedBag>"]
    epoch["epoch: CachePadded<AtomicEpoch>"]
  end
  reg["Local::register"] -->|"insert 到表头"| locals
  fin["Local::finalize"] -->|"entry.delete：next 打标记"| locals
  adv["try_advance"] -->|"iter：顺手摘除已标记节点"| locals
  pb["push_bag"] -->|"push"| queue
  col["collect"] -->|"try_pop_if(过期)"| queue

两个结构自己也用 Atomic / Shared / Guard 写成，摘下的节点同样交给 defer_destroy：回收器用自己回收自己的节点。

参与者链表：sync/list.rs

文件开头注明出处：Michael，《High Performance Dynamic Lock-Free Hash Tables and List-Based Sets》，SPAA 2002（sync/list.rs）。

侵入式

链表节点不单独分配，而是嵌在元素里：

pub(crate) struct Entry {
    /// The next entry in the linked list.
    /// If the tag is 1, this entry is marked as deleted.
    next: Atomic<Entry>,
}

pub(crate) trait IsElement<T> {
    fn entry_of(_: &T) -> &Entry;
    unsafe fn element_of(_: &Entry) -> &T;
    unsafe fn finalize(_: &Entry, _: &Guard);
}

（第 16 行、第 67 行）

Local 是 #[repr(C)]，entry 是第一个字段（internal.rs 第 288 行），所以 Entry 和 Local 的地址相同，entry_of / element_of 就是两次指针转换（第 555–573 行）。finalize 是「节点已从链表上摘下」时的回调，对 Local 来说就是 guard.defer_destroy(...) 整个 Local。internal.rs 原来用 memoffset 的 offset_of! 算字段偏移，2023 年 4aeb87eb 去掉这个依赖，改成靠 repr(C) + 第一个字段。

插入、删除

插入只发生在表头（insert 第 172 行）：

let mut next = to.load(Relaxed, guard);
loop {
    entry.next.store(next, Relaxed);
    match to.compare_exchange_weak(next, entry_ptr, Release, Relaxed, guard) {
        Ok(_) => break,
        Err(err) => next = err.current,
    }
}

CAS 用 Release：遍历者用 Acquire 读到新表头时，entry.next 和整个 Local 的初始化都已可见。

删除分两步，delete 只做第一步（第 148 行）：

pub(crate) unsafe fn delete(&self, guard: &Guard) {
    self.next.fetch_or(1, Release, guard);
}

在自己的 next 指针上打标记位，逻辑删除。标记打在「自己的 next」上而不是「前驱指向自己的指针」上，是 Michael 链表的关键：此后任何想在本节点后面插入或摘除的 CAS（期望值是不带标记的 next）都会失败，不会出现「节点刚被摘掉、别人又往它后面挂了东西」的丢失。

遍历时摘除，以及 Stalled

真正把节点从链表上摘下来的是迭代器（第 238 行）：

while let Some(c) = unsafe { self.curr.as_ref() } {
    let succ = c.next.load(Acquire, self.guard);

    if succ.tag() == 1 {
        // This entry was removed. Try unlinking it from the list.
        let succ = succ.with_tag(0);

        let succ = match self
            .pred
            .compare_exchange(self.curr, succ, Acquire, Acquire, self.guard)
        {
            Ok(_) => {
                unsafe {
                    C::finalize(self.curr.deref(), self.guard);
                }
                succ
            }
            Err(e) => e.current,
        };

        // If the predecessor node is already marked as deleted, we need to restart from
        // `head`.
        if succ.tag() != 0 {
            self.pred = self.head;
            self.curr = self.head.load(Acquire, self.guard);

            return Some(Err(IterError::Stalled));
        }

        self.curr = succ;
        continue;
    }

    self.pred = &c.next;
    self.curr = succ;

    return Some(Ok(unsafe { C::element_of(c) }));
}

遇到标了删除的节点，就 CAS 前驱的 next，把它跳过去：

  • CAS 成功：本线程负责回收，调 finalize（defer_destroy 这个 Local）。
  • CAS 失败：别人改了前驱的 next。拿到的当前值如果带标记，说明前驱自己也被删了，从它往后走已经不可靠，只能从表头重来，并把这件事报告给调用者：Err(IterError::Stalled)。

调用者只有一个，try_advance。它对 Stalled 的处理是直接放弃这次推进（defer 与回收）。这样做总是安全的：不推进只是推迟回收，不会提前释放。

由此有一个推论：退出线程的 Local 只有在某个线程跑 try_advance 时才会被真正摘除；摘除之后还要再等两个纪元才会被释放。一个进程里线程来来去去、又很少有人 pin，链表上会留着一串已标记的节点。

List 的 Drop 要求所有元素都已经删除（assert_eq!(succ.tag(), 1)，第 226 行），用 unprotected() 逐个 finalize。Global 只有在最后一个 Collector 克隆和最后一个 Local 都没了之后才会 drop，那时所有 Local 都已经 delete 过。

垃圾袋队列：sync/queue.rs

文件头注明 Michael 和 Scott 的 PODC 1996 论文，以及 2004 年对它的形式化验证（sync/queue.rs）。

pub(crate) struct Queue<T> {
    head: CachePadded<Atomic<Node<T>>>,
    tail: CachePadded<Atomic<Node<T>>>,
}

struct Node<T> {
    data: MaybeUninit<T>,
    next: Atomic<Node<T>>,
}

经典结构：单链表，表头是一个不存数据的哨兵节点；tail 可能落后于真正的尾巴，谁看见了谁帮忙往前推。（结构体上方的注释还提到「Blocked 节点」「唤醒阻塞在 pop 上的线程」，现在的代码里没有这些，是旧版本留下的描述。）

push：发现落后就帮忙

push_internal 第 66 行：

let o = unsafe { onto.deref() };
let next = o.next.load(Acquire, guard);
if unsafe { next.as_ref().is_some() } {
    // if not, try to "help" by moving the tail pointer forward
    let _ = self
        .tail
        .compare_exchange(onto, next, Release, Relaxed, guard);
    false
} else {
    // looks like the actual tail; attempt to link in `n`
    let result = o
        .next
        .compare_exchange(Shared::null(), new, Release, Relaxed, guard)
        .is_ok();
    if result {
        // try to move the tail pointer forward
        let _ = self
            .tail
            .compare_exchange(onto, new, Release, Relaxed, guard);
    }
    result
}

真正的「入队」是把新节点 CAS 进尾节点的 next；之后推 tail 只是优化，失败了也没关系，别人会帮忙。

try_pop_if：只看队头

collect 用的是带条件的弹出（pop_if_internal 第 146 行）：

let head = self.head.load(Acquire, guard);
let h = unsafe { head.deref() };
let next = h.next.load(Acquire, guard);
match unsafe { next.as_ref() } {
    Some(n) if condition(unsafe { &*n.data.as_ptr() }) => unsafe {
        self.head
            .compare_exchange(head, next, Release, Relaxed, guard)
            .map(|_| {
                let tail = self.tail.load(Relaxed, guard);
                // Advance the tail so that we don't retire a pointer to a reachable node.
                if head == tail {
                    let _ = self
                        .tail
                        .compare_exchange(tail, next, Release, Relaxed, guard);
                }
                guard.defer_destroy(head);
                Some(n.data.assume_init_read())
            })
            .map_err(|_| ())
    },
    None | Some(_) => Ok(None),
}
  • 条件在 CAS 之前对队头数据求值，而这时别的线程可能也在读同一个节点，所以 try_pop_if 要求 T: Sync。SealedBag 里的 Bag 装着 !Sync 的闭包，于是 SealedBag 手写了 unsafe impl Sync，注释说明安全的理由是「is_expired 只看纪元」（internal.rs 第 143–144 行）。
  • 弹出成功后，原来的哨兵被 defer_destroy，数据被 assume_init_read 搬出来，next 那个节点成为新哨兵（它的 data 已经搬空，MaybeUninit 保证不会再 drop 一次）。
  • 如果 tail 还指着要退休的旧哨兵，先把 tail 推过去再退休它，注释说这是为了「不退休一个仍然可达的节点」。
  • 条件不满足就返回 None，不往后找。

try_pop_if 外面是个循环，只在 CAS 失败（Err(())）时重试。

回收产生回收

collect 每弹出一个袋子就 defer_destroy 一个队列节点，这个节点进了本线程的袋子，袋子满了又会 push_bag 进同一个队列。所以回收本身会产生新垃圾，不过每 64 次弹出才产生一个新袋子，是收敛的。defer 与回收的 E3 里，钉住者离开后本线程袋子里残留的最后 64 个对象也被回收了，就是被这些新垃圾挤出去的。

反方向的递归要避免：Local::finalize 的注释特别说明「push_bag 不能再 defer 新的垃圾」，否则「注销时把袋子推出去」会无限地产生要推的袋子。

两个结构对照

参与者链表 List<Local>垃圾队列 Queue<SealedBag>
论文Michael 2002（有序链表 / 集合）Michael、Scott 1996
节点侵入式，嵌在 Local 里独立分配，一个节点装一整个 SealedBag（2064 字节）
写入线程注册时插到表头每满 64 个对象、或每次 flush 时入队
删除打标记（逻辑），遍历时摘除（物理）带条件的出队，只看队头
读者只有 try_advance只有 collect
摘下的节点defer_destroy 整个 Localdefer_destroy 旧哨兵
竞争失败迭代返回 Stalled，推进放弃重试

try_advance 的源码注释里留着一句 TODO（原注释带名字，此处略去）：用链表存参与者是因为链表容易做成无锁的，但遍历时缓存未命中和数据依赖会让它变慢，「应该试试别的数据结构」。到 0.9.21 仍是链表。

小结

  • 参与者链表是 Michael 2002 的无锁链表：插入在表头，删除先在自己的 next 上打标记，遍历时再 CAS 前驱把它摘掉；前驱也被删了就返回 Stalled，try_advance 遇到就放弃本次推进。
  • 已退出线程的 Local 只在有人 try_advance 时被摘除，摘除后再等两个纪元才释放。
  • 垃圾队列是 Michael-Scott 队列：入队 CAS 尾节点的 next，tail 落后时谁看见谁帮忙；try_pop_if 只看队头，条件在 CAS 前求值，所以要求 T: Sync，SealedBag 为此手写了 Sync。
  • 两个结构都用 crate 自己的 Atomic 和 defer_destroy 回收节点；回收产生的新垃圾每 64 次弹出才多一个袋子，是收敛的。

💬 关联讨论

发起讨论
加载中...