并发跳表

主线位置：主线 3「一块内存的回收」的一个用户。跳表摘下一个节点后，自己先用引用计数确认没有 Entry 还指着它，计数归零时调 guard.defer_unchecked(finalize)，后面的事交给 defer 与回收。

flowchart TB
  ins["insert(k, v)"] --> sp["search_position：从最高层往下，记录每层 left / right"]
  sp -->|找到同键| mark1["mark_tower 旧节点（insert 默认替换）"]
  mark1 --> sp
  sp -->|没找到| alloc["Node::alloc(random_height, refs = 2)"]
  alloc --> l0["CAS 第 0 层：left[0] 从 right[0] 改成新节点"]
  l0 -->|失败| sp
  l0 --> up["逐层往上接：每接一层 refs + 1"]
  rm["remove(k)"] --> sp2["search_position 找到节点"]
  sp2 --> acq["try_acquire：refs + 1，作为返回的 Entry"]
  acq --> mt["mark_tower：从顶层到第 0 层逐个 fetch_or(1)"]
  mt --> unl["逐层 CAS 摘下，每摘一层 decrement"]
  unl --> zero{"refs 归零？"}
  zero -->|是| def["guard.defer_unchecked(finalize)"]

版本说明

本章以 tag crossbeam-skiplist-0.1.3（3aa2598e，2024-01-08）为准。这个 tag 在 main 上，不是本卷其他 crate 共用的发布提交 2920c984 的祖先；2920c984 里的 skiplist 目录反而更旧（Cargo.toml 写 0.0.0，没有 0.1.2 加的 compare_insert）。registry 里下载的 crossbeam-skiplist-0.1.3 四个源文件和这个 tag 逐字节相同（本章写作时 cmp 核对过）。

src/ 四个文件：base.rs 2123 行（SkipList 本体）、map.rs 783 行、set.rs 629 行、lib.rs 257 行，共 3792 行，含文档注释。SkipMap / SkipSet 是 base::SkipList 的薄包装，本章不逐个讲它们的方法。

Node：一次分配，塔是变长的

const HEIGHT_BITS: usize = 5;
const MAX_HEIGHT: usize = 1 << HEIGHT_BITS;          // 32
const HEIGHT_MASK: usize = (1 << HEIGHT_BITS) - 1;

#[repr(C)]
struct Tower<K, V> {
    pointers: [Atomic<Node<K, V>>; 0],
}

#[repr(C)]
struct Node<K, V> {
    value: V,
    key: K,
    refs_and_height: AtomicUsize,
    tower: Tower<K, V>,
}

（base.rs:17、Node。）

• Tower 是一个长度为 0 的数组，真正的长度在分配时决定：Node::alloc 按高度算 Layout（结构体大小加 height 个指针），一次 alloc，把塔清零，键和值留给调用方写。所以一个节点是一次分配，高度 1 的节点只多带一个指针。
• refs_and_height 把两样东西塞进一个 usize：低 5 位是 高度 - 1（0 到 31），其余位是引用计数。高度在节点一生中不变，和计数放在一起不会冲突，还省一个字。
• 塔里的指针是 crossbeam-epoch 的 Atomic<Node>，最低位当「已删除」标记（epoch 的标签指针，见 Atomic、Owned、Shared）。

SkipList 本身是一个头节点（一座 32 层的塔，Head）、一个 epoch Collector、一个缓存行对齐的 HotData{seed, len, max_height}（base.rs:300）。len 用 Relaxed 增减，可能短暂下溢成很大的数，len() 把大于 isize::MAX 的值当 0；文档说并发修改时只是近似值。

随机高度

random_height：

let mut num = self.hot_data.seed.load(Ordering::Relaxed);
num ^= num << 13;
num ^= num >> 17;
num ^= num << 5;
self.hot_data.seed.store(num, Ordering::Relaxed);

let mut height = cmp::min(MAX_HEIGHT, num.trailing_zeros() as usize + 1);
unsafe {
    // Keep decreasing the height while it's much larger than all towers currently in the
    // skip list.
    while height >= 4
        && self.head[height - 2]
            .load(Ordering::Relaxed, epoch::unprotected())
            .is_null()
    {
        height -= 1;
    }
}

• 随机数是 xorshift，种子放在 HotData 里，所有线程共用，Relaxed 读写：两个线程同时读到同一个种子会得到同样的高度，这不影响正确性，只影响分布，作者接受了。注释说用的是「32 位整数」的那组移位常数（13、17、5），变量是 usize。种子初始为 1。
• 高度 = 末尾 0 的个数 + 1：高度为 k 的概率是 2^-k，即每升一层概率减半，期望高度 2。
• 再往下压：高度 ≥ 4 时，如果头节点的第 height - 2 层还是空的（现有节点都没那么高），就降一层。新节点最多只比现有最高的塔高一点点，避免一个偶然的 30 层节点让每次查找都从 30 层开始。
• max_height 只增不减，查找从这一层开始。

查找：顺手帮忙摘除

search_position 从 max_height 开始（跳过头节点上为空的层），每层往右走到第一个键 ≥ 目标的节点，记下这一层的前驱 left[level] 和后继 right[level]，再下一层。返回 Position{found, left, right}，插入和删除都要用这两组数组。

每读一个 succ，如果它带删除标记（tag() == 1），说明当前节点 c 已被删除，就调 help_unlink 在这一层把 c 摘掉：CAS 前驱的指针从 c 改成 succ，成功就给 c 的引用计数减一。CAS 失败（前驱也变了）就从头重新搜。所以删除操作不用自己负责每一层都摘干净，后来的查找会帮忙。

指针用 load_consume 读：在 ARM / AArch64 上是 Relaxed 加编译器屏障，其他平台退回 Acquire（见本库里的内存序）。读到一个节点指针后紧接着解引用它，数据依赖本身保证了顺序。

插入

insert_internal 有一个 replace: Fn(&V) -> bool 参数，三个公开入口只差这个参数：

步骤：

1. 搜索。找到同键且要替换，就 mark_tower 旧节点，回到 1；找到同键且不替换，try_acquire 返回已有的。
2. Node::alloc(height, 2)：初始引用计数 2，一个给即将返回的 RefEntry，一个给第 0 层的链接。len 乐观地加一。
3. 第 0 层：新节点的 tower[0] 指向 right[0]，CAS left[0] 从 right[0] 改成新节点。失败就重新搜索（期间若发现别人插了同键，按 replace 规则处理，必要时把自己刚分配的节点 finalize 掉）。第 0 层接上，插入就生效了。
4. 往上逐层接：每层先 CAS 自己的 tower[level] 指向 succ，引用计数加一，再 CAS 前驱。前驱 CAS 失败就把计数减回去、重新搜索。发现自己已被标记删除（别人在我搭塔时删了我）就停止。注释原话：「Only the lowest level really matters, and all the higher levels are there just to make searching faster.」
5. 搭完如果发现塔顶被标记了，再搜一次，让查找把刚接上的高层摘掉。

引用计数 = 持有它的 Entry 数 + 它被链接在跳表里的层数。

删除：先标记整座塔，再逐层摘

remove：

1. 搜到节点，try_acquire（计数加一，作为返回给调用方的 RefEntry）。计数已经是 0 说明节点正在被回收，重新搜。
2. mark_tower：从最高层到第 0 层，对每个塔指针 fetch_or(1, SeqCst)。第 0 层的标记是「谁删除了这个节点」的唯一裁决：fetch_or 返回的旧值里第 0 层已经有标记，说明别人先删了，返回 false。标记以后这些指针再也 CAS 不进去，任何人都不能在这个节点后面接新节点。
3. 是我删的：len 减一，从顶层往下，用搜索时记下的 left[level] CAS 把节点摘掉，每摘一层 decrement 一次；某一层 CAS 失败（前驱变了），就调一次搜索，让 help_unlink 把剩下的层摘干净，然后结束。
4. 返回 RefEntry。调用方还能读键和值，直到放手。

先标记、后摘除，是 Harris 无锁链表的做法，推广到每一层：标记保证没人能在被删节点后面插入，否则插入会随着被删节点一起丢掉。

两层回收：引用计数，然后 epoch

decrement：

unsafe fn decrement(&self, guard: &Guard) {
    if self
        .refs_and_height
        .fetch_sub(1 << HEIGHT_BITS, Ordering::Release)
        >> HEIGHT_BITS
        == 1
    {
        fence(Ordering::Acquire);
        unsafe { guard.defer_unchecked(move || Self::finalize(self)) }
    }
}

计数归零（fetch_sub 返回 1）时，不直接释放，而是 defer_unchecked：别的线程可能正在遍历时读到了这个节点的指针，还没来得及 try_increment。try_increment 在计数为 0 时拒绝增加（base.rs:190），保证不会「复活」一个已进入回收的节点；而这些读者手里裸指针的安全，由它们所在的 epoch Guard 保证：只要它们还 pin 着，finalize 就不会运行。Release 减加 Acquire fence 是和 Arc 一样的引用计数写法（见 bytes 卷克隆与提升的对照）。

finalize 对键和值 drop_in_place，再按高度算 Layout 释放。

一个小实验（exp-crossbeam/skiplist-reclaim/）：SkipMap 里插一个带 Drop 计数的值，get 拿一个 Entry，remove 再拿一个，然后依次放手：

after remove, entries still held:       drops = 0
after dropping both entries:            drops = 0
after 2 pin().flush() calls:            drops = 1
insert over existing key, immediately:  extra drops = 0
after 2 pin().flush() calls:            extra drops = 1
map.len() = 1

两个 Entry 都放手后计数已归零，但值还没析构，它在 epoch 的延迟队列里；本线程又 pin().flush() 两次（纪元推进两格，见 defer 与回收）后才析构。insert 覆盖已有键也一样：旧值不是立即析构的。如果某个线程长期 pin 着，这些旧值会一直留着，这是 EBR 的代价（为什么需要内存回收）。

为什么 SkipMap 要求 'static

源码里有一段 TODO（base.rs:313，原文署名已略去）解释了：SkipMap::new 用的是全局默认收集器（epoch::default_collector().clone()，map.rs:32），全局收集器可能在跳表被 drop 很久以后才执行延迟的 finalize，所以 insert / remove 等方法要求 K: 'static、V: 'static，不能存借用数据。TODO 设想给每个跳表内嵌一个自己的 Collector，跳表 drop 时把垃圾一起清掉，就能去掉 'static 约束；到 0.1.3 还没做。base::SkipList::new(collector) 允许传入自定义收集器，每个方法开头 check_guard 校验传进来的 Guard 属于同一个收集器（见 Collector 与 LocalHandle）。

SkipMap / SkipSet

SkipMap 只有一个字段 inner: base::SkipList<K, V>。每个方法自己 epoch::pin()，调 inner 的同名方法，把返回的 RefEntry 包成 map::Entry。map::Entry drop 时调 release_with_pin，再 pin 一次去减计数。SkipSet 是 SkipMap<T, ()> 的包装。

所以 SkipMap 的每次操作至少有一次 pin（进入临界区写一次本地纪元，加一次 SeqCst fence；见 pin 与 Guard），持有 Entry 则在放手时再 pin 一次。迭代器 iter() 用的是 ref_iter，每前进一步都 pin 一次、对下一个节点 try_increment，不会在整个迭代期间一直 pin 着。

谁在用

pingora 的 TinyUFO 缓存有两种存储：new 用 flurry 的并发 HashMap，new_compact 用分片的 crossbeam SkipMap，文档说后者省内存、get 慢一个常数（TinyUFO）。跳表在这里不是为了有序，而是为了没有大块的预分配表：每个条目一个节点，节点大小随高度变化，期望高度 2。

发布之后

main 上修了 RefRange 的 Send / Sync 约束，0.1.3 没有带上；本章不展开。

小结

• 0.1.3 以 main 上的 tag 为准，registry 包与之逐字节相同；2920c984 里的 skiplist 更旧。
• Node 一次分配：值、键、refs_and_height（低 5 位高度、其余位引用计数）、变长的塔；最高 32 层。
• 随机高度：共享 xorshift 种子，末尾 0 个数加一，每层概率减半；比现有最高塔高出太多时往下压。
• 查找顺手 help_unlink 已标记的节点；插入先接第 0 层（这一步生效），再尽力搭高层；删除先 mark_tower 整座塔（第 0 层的标记裁定谁删的），再逐层摘。
• 引用计数 = Entry 数 + 链接层数；归零后 defer_unchecked(finalize) 交给 epoch。实测放手后要再推进两次纪元，值才析构；insert 覆盖旧键时旧值同样延迟析构。
• SkipMap 用全局收集器，所以要求 'static；每个方法自己 pin。