array：有界环形缓冲

主线位置：一条消息的中段。Sender::send 进到 array::Channel::send，start_send 抢到一个槽、write 写进去、叫醒接收方；满了就自旋，再不行就交给 Context 与 Waker 去 park。

flowchart TB
  send["send(msg)"] --> ss["start_send：读 tail 和槽的 stamp"]
  ss -->|"stamp == tail"| cas["CAS tail 前进一格"]
  cas -->|成功| w["write：写消息，stamp = tail + 1"]
  w --> nr["receivers.notify()"]
  cas -->|失败| ss
  ss -->|"stamp + one_lap == tail + 1"| full{"SeqCst fence，再读 head：真满？"}
  full -->|是| spin["Backoff 自旋 / 让出，最多约 11 次"]
  spin -->|仍满| reg["senders.register，再查一次，park"]
  full -->|否| ss
  ss -->|其他| snooze["snooze：别人正在写这个槽"]
  snooze --> ss

array flavor 是 Dmitry Vyukov 的有界 MPMC 队列（文件开头的注释给了 1024cores 上的原文链接），外面套上阻塞和断开。ArrayQueue 是同一个环，去掉阻塞和断开。

数据结构

struct Slot<T> {
    stamp: AtomicIndex,
    msg: UnsafeCell<MaybeUninit<T>>,
}

pub(crate) struct Channel<T> {
    head: CachePadded<AtomicIndex>,
    tail: CachePadded<AtomicIndex>,
    buffer: Box<[Slot<T>]>,
    cap: usize,
    one_lap: Index,
    mark_bit: Index,
    senders: SyncWaker,
    receivers: SyncWaker,
}

（Slot、Channel。）

• AtomicIndex 在有 64 位原子的平台上是 AtomicU64，否则 AtomicUsize（array.rs:29-36）。这是 0.5.17 的 #1321：32 位平台上 lap 只有二十几位，绕一圈的时间短到可能出 ABA，换成 64 位就安全得多。64 位平台上两者一样。
• head、tail 各包一层 CachePadded（aarch64 / x86_64 上对齐 128 字节，见 CachePadded）：生产者只写 tail，消费者只写 head，分开两条缓存行，互不打扰。
• 两个 SyncWaker：满了等着的发送方、空了等着的接收方。

stamp：lap、mark、index 打包

head、tail 和每个槽的 stamp 都是同一种编码：低位是槽下标 index，中间一位是 mark，高位是圈数 lap。with_capacity 算两个常数：

let mark_bit = (cap as Index)
    .checked_add(1)
    .and_then(Index::checked_next_power_of_two)
    .expect("bounded channel capacity is too large");
let one_lap = mark_bit
    .checked_mul(2)
    .expect("bounded channel capacity is too large");

checked_* 是 0.5.17 的 #1301：以前 bounded(usize::MAX) 这类参数会在 next_power_of_two 里溢出，现在明确 panic。

以 bounded(3) 为例：mark_bit = 4（二进制 100），one_lap = 8（1000）。一个值 0b1_0_10 读作 lap 1、mark 0、index 2。

槽 i 的 stamp 初始化为 i（lap 0、index i）；head、tail 都从 0 开始。mark 位只出现在 tail 上，表示 channel 已断开。

三个判定

start_send 读 tail，取出它指向的槽，看槽的 stamp：

写完把 stamp 置为 tail + 1，意思是「这一圈、这个槽，已有消息」。start_recv 是镜像：

拿 bounded(3) 走一遍（one_lap = 8，只看低 4 位）：

stamp 的含义就是「这个槽下一次轮到谁」：等于某个 tail 值时等写者，等于某个 head + 1 时等读者。

当 index 走到 cap - 1 时，下一个值不是 tail + 1，而是 lap + one_lap（下一圈、index 0），所以 index 永远小于 cap，mark_bit 那一位只给断开用。

两个 SeqCst fence

「看起来满」和「看起来空」两个分支各有一个 atomic::fence(Ordering::SeqCst)（array.rs:212、295），再 Relaxed 读对面的索引。这是本库里的内存序分类里的第二类：fence 做 store-load 屏障。发送方先读到槽的 stamp（上一圈的消息还在），再去读 head；如果没有 fence，读 head 可能被提到读 stamp 之前，读到一个旧的 head，把「刚有人读走」误判成「满」。误判满不会出错，只会多走一趟阻塞路径，但阻塞路径之后还有一次 is_full() 复查，见下。

数据本身的发布靠 stamp 的 Release / Acquire：write 先写消息、再 stamp.store(.., Release)；start_recv 用 Acquire 读 stamp，看到 head + 1 就一定能看到消息。tail / head 的 CAS 用 SeqCst，失败分支用 Relaxed，CAS 失败时直接用返回的当前值继续循环（Err(t) => { tail = t; backoff.spin(); }）。std 的移植版这里不一样，失败后先退避再重新 load，这个差别在 vs-std 的二分实验里是 4 倍吞吐的来源之一（那次是 list flavor）。

send 与 recv：先自旋，再阻塞

send：

let token = &mut Token::default();
loop {
    // Try sending a message several times.
    let backoff = Backoff::new();
    loop {
        if self.start_send(token) {
            let res = unsafe { self.write(token, msg) };
            return res.map_err(SendTimeoutError::Disconnected);
        }
        if backoff.is_completed() {
            break;
        } else {
            backoff.snooze();
        }
    }
    // 截止时间检查
    Context::with(|cx| {
        // Prepare for blocking until a receiver wakes us up.
        let oper = Operation::hook(token);
        self.senders.register(oper, cx);

        // Has the channel become ready just now?
        if !self.is_full() || self.is_disconnected() {
            let _ = cx.try_select(Selected::Aborted);
        }

        // Block the current thread.
        let sel = cx.wait_until(deadline);

        match sel {
            Selected::Waiting => unreachable!(),
            Selected::Aborted | Selected::Disconnected => {
                self.senders.unregister(oper).unwrap();
            }
            Selected::Operation(_) => {}
        }
    });
}

三段：

1. 自旋：start_send 失败后 snooze，前 7 次是 1、2、4……64 次 spin_loop，之后 4 次 yield_now，第 11 次 snooze 后 is_completed() 为真。一共最多 12 次尝试。
2. 注册后复查：把自己挂进 senders，然后再看一次是否已经不满。不复查就有丢唤醒：接收方可能恰好在「我判断满」和「我注册」之间读走了一条，它 notify 的时候我还不在队列里，我注册以后就没人叫我了。复查发现不满，就自己把 select 状态 CAS 成 Aborted，wait_until 立即返回。
3. 阻塞：wait_until 里 thread::park，直到接收方 read 完调 senders.notify() 把我选中、unpark。醒来后不直接写，而是回到外层循环重新 start_send：被叫醒只说明「刚才有空位」，不保证现在还有。

recv 是镜像，register 到 receivers、复查 !is_empty() || is_disconnected()。

第 1 段的自旋，std 的移植版没有：std 的 send 一次 start_send 失败就去注册、park。vs-std 里把它加回 std 的复制版，1P1C 有界吞吐从 81 降到 55 Mmsg/s（三次交替测量的中位数）。

write 和 read 各自在最后叫醒对面：write 调 receivers.notify()，read 调 senders.notify()。SyncWaker::notify 在没人等的时候只读一个 AtomicBool 就返回（见 Context 与 Waker），所以不阻塞的快路径上没有锁。

len：等 tail 稳定

len 读 tail、读 head、再读一次 tail，两次 tail 相同才算数，否则重来。然后按 index 算：hix < tix 是 tix - hix，hix > tix 是绕了一圈，cap - hix + tix；相等时看 lap 区分空和满。发送方很忙时这个循环可能转好几圈，sqlx 卷里那条注释说的就是这个（ArrayQueue::len 同一段代码，见 ArrayQueue 与 SegQueue）。返回的数在返回那一刻就可能过时。

disconnect 与 Drop

disconnect 用 fetch_or 给 tail 置上 mark_bit，第一次置上的人叫醒两边所有等待者：

pub(crate) fn disconnect(&self) -> bool {
    let tail = self.tail.fetch_or(self.mark_bit, Ordering::SeqCst);

    if tail & self.mark_bit == 0 {
        self.senders.disconnect();
        self.receivers.disconnect();
        true
    } else {
        false
    }
}

置了 mark 以后，start_send 第一行就看到它，返回「断开」；start_recv 只在看到空的时候才检查 mark，所以断开后剩下的消息还能一条条收完。

消息的析构在 Drop for Channel：按 head、tail 算出还有几条，逐个 assume_init_drop。T 不需要 drop 时整段跳过（mem::needs_drop，0.5.10 的 #1057「Optimize Drop」）。

最后一个接收端走了，消息还在

Channel 的 Drop 只在 Counter 被释放时运行，也就是发送端和接收端都走光以后（见 API 与 flavor）。array 没有 list 那样的 disconnect_receivers，所以「接收端全走了、发送端还活着」这段时间，缓冲里的消息一条也不析构。

实验 E8（exp-crossbeam/rx-drop-discard/、exp-crossbeam/rx-drop-0515/）：发 10 条带 Drop 计数的消息，drop 接收端，读计数，再 drop 发送端：

crossbeam 0.5.17 bounded(16)   after rx drop:  0, after tx drop: 10
crossbeam 0.5.17 unbounded     after rx drop: 10, after tx drop: 10
std sync_channel(16)           after rx drop: 10, after tx drop: 10
std channel                    after rx drop: 10, after tx drop: 10
crossbeam 0.5.15 bounded(16)   after rx drop:  0, after tx drop: 10
crossbeam 0.5.15 unbounded     after rx drop: 10, after tx drop: 10

0.5.15（本站 Cargo.lock 锁的版本，实验 crate 里配的是 crossbeam-utils 0.8.23）和 0.5.17 行为相同。四个组合里只有 crossbeam 的有界 channel 不丢。如果消息里攥着文件句柄、连接或大块内存，而发送端是一个长期存活的对象（比如一个池子或一个后台线程持有的 Sender），这些资源会一直留到发送端也 drop。

std 的移植版有 disconnect_receivers + discard_all_messages（std array.rs:470、495，见 vs-std），所以 std 的 sync_channel 立即丢。

发布之后

main 上的 #1121（2026-06-08 合入）给 array 补上了 disconnect_receivers 和丢弃逻辑，0.5.17 没有带上。升级到包含它的版本后，上面第一行会变成 10 → 10：接收端一走，消息就析构。依赖「接收端走了消息还能留着」的代码（例如 drop 接收端后再从别处 Sender 端取回东西）不存在公开 API 能做到，所以这个变化只影响析构的时机。

小结

• array 是 Vyukov 的有界 MPMC 环：每个槽一个 stamp，head / tail 各占一条缓存行，三个值都把 lap、mark、index 打包进一个整数，one_lap 是大于 cap 的最小 2 的幂的两倍。
• stamp == tail 可写、stamp == head + 1 可读；stamp + one_lap == tail + 1 加上 head + one_lap == tail 才判满；看起来满 / 空的分支用 SeqCst fence 再读对面。
• 写完 stamp 置 tail + 1，读完置 head + one_lap：stamp 记的是「这个槽下一次等谁」。
• send / recv 先 Backoff 自旋加让出约 11 次，再注册、复查、park；醒来后重新抢槽。std 删掉了第一段自旋。
• 断开是给 tail 置 mark 位；断开后剩余消息仍可收完。
• 0.5.17（和本站的 0.5.15）的有界 channel 在最后一个接收端 drop 时不丢弃缓冲消息，要等发送端也走光（E8：0 → 10）；std 和 crossbeam 的无界都立即丢，main 的 #1121 已修、未发布。