一条消息

主线位置：主线 1 的全程。本页只画时序和关键数值，每一步的细节在 06 部分：array、Context 与 Waker、API 与 flavor。

场景：容量为 1 的有界 channel，里面已经有一条消息；一个线程再 send 一条，阻塞；另一个线程 recv，把它叫醒。

let (tx, rx) = crossbeam_channel::bounded::<u32>(1);
tx.send(1).unwrap();            // 第一次：直接写进槽
let t = std::thread::spawn(move || tx.send(2).unwrap()); // 第二次：满了，阻塞
assert_eq!(rx.recv(), Ok(1));   // 取走一条，叫醒发送方
t.join().unwrap();
assert_eq!(rx.recv(), Ok(2));

sequenceDiagram
  participant S as 发送线程
  participant C as array Channel
  participant W as senders SyncWaker
  participant R as 接收线程
  S->>C: start_send：tail=4，槽 stamp=1，判定满
  S->>S: snooze 11 次，仍满
  S->>W: Context::with，register(oper, cx)
  S->>C: 再查 is_full()，仍满
  S->>S: wait_until(None)，thread::park
  R->>C: start_recv：stamp=1 等于 head+1，CAS head 0 到 4
  R->>C: read：取出消息，stamp 写 4（Release）
  R->>W: notify：try_select CAS 成 Operation，unpark
  W-->>S: 醒来，select 不再是 Waiting
  S->>C: 重新 start_send：stamp=4 等于 tail，CAS tail 4 到 8
  S->>C: write：写消息，stamp 写 5（Release），receivers.notify

第 0 步：建通道

bounded(cap) 在 cap == 0 时建 zero flavor，否则建 array flavor（with_capacity）。head 和 tail 都是一个打包的「stamp」：低位是槽下标，往上一位是 mark 位（只在 tail 里用，表示断开），再往上是圈数（lap）。

cap = 1 时：

槽的 stamp 是这个算法的核心：stamp == tail 表示「这一圈的这个槽空着，可以写」，stamp == head + 1 表示「写好了，可以读」。

第 1 步：第一次 send，直接写

start_send：tail = 0，下标 0，槽 0 的 stamp 也是 0，相等，可以写。下一个 tail：下标 + 1 不小于容量，于是进一圈，lap + one_lap = 4。compare_exchange_weak(0, 4, SeqCst, Relaxed) 成功，在 token 里记下槽地址和「写完以后 stamp 应该是 tail + 1 = 1」。

write 把消息写进槽，stamp.store(1, Release)，然后 receivers.notify()。此时没有接收方在等，SyncWaker 的 is_empty 是 true，notify 读一个原子布尔就返回，不碰锁。

第 2 步：第二次 send，满了

tail = 4：下标 4 & (mark_bit - 1) = 0，lap 是 4。槽 0 的 stamp 是 1，不等于 tail。再看 stamp + one_lap == tail + 1：1 + 4 = 5，4 + 1 = 5，成立，意思是「这个槽装着上一圈的消息，还没被读」。于是 fence(SeqCst) 之后读 head：head + one_lap = 0 + 4 = tail，head 也落后一整圈，通道满，start_send 返回 false。

send 不会马上睡：

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();
    }
}

snooze 前 7 次自旋 1 到 64 次，后 4 次 yield_now，第 11 次之后 is_completed() 为真（Backoff，本机实测这 11 次总共一两微秒）。这期间接收方只要取走一条，发送方就不用睡。

第 3 步：注册，再查一次，park

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(_) => {}
    }
});

• Context::with：从线程本地变量里取出缓存的 Context（一个 Arc<Inner>，里面是 select: AtomicUsize、packet、Thread 句柄、ThreadId），reset() 把 select 置回 0（Waiting）。线程本地变量已析构时临时新建一个。
• Operation::hook(token)：拿 token 这个栈变量的地址当操作 ID，断言它大于 2，因为 0、1、2 被 Waiting、Aborted、Disconnected 占了。
• SyncWaker::register：拿锁，把 Entry{oper, packet, cx} 推进 selectors，is_empty.store(false, SeqCst)。
• 再查一次：注册之后、睡之前，用 SeqCst 读 head / tail 看是否还满。这是防丢唤醒的那一半：如果接收方恰好在「第 2 步判满」和「注册」之间取走了消息，它的 notify 会看到 is_empty == true 而什么也不做；这里的再查一次会发现不满，自己把 select 设成 Aborted，不睡。两边都用 SeqCst，保证至少一方看到对方（本库里的内存序 第 ② 类）。
• wait_until(None)：select 还是 Waiting 就 thread::park()，醒来再看，循环到不是 Waiting 为止。

第 4 步：接收方取走消息，叫醒发送方

start_recv：head = 0，槽 0 的 stamp = 1 = head + 1，可以读；新 head 进一圈为 4，CAS 成功；读完以后槽的 stamp 应该是 head + one_lap = 4，也就是「下一圈的 tail 到这里时可以写」。

read 取出消息，stamp.store(4, Release)，然后 senders.notify()。这次 is_empty 是 false，notify 拿锁，调 Waker::try_select：

self.selectors
    .iter()
    .position(|selector| {
        // Does the entry belong to a different thread?
        selector.cx.thread_id() != thread_id
            && selector // Try selecting this operation.
                .cx
                .try_select(Selected::Operation(selector.oper))
                .is_ok()
            && {
                // Provide the packet.
                selector.cx.store_packet(selector.packet);
                // Wake the thread up.
                selector.cx.unpark();
                true
            }
    })
    // Remove the entry from the queue to keep it clean and improve
    // performance.
    .map(|pos| self.selectors.remove(pos))

跳过本线程注册的条目；对第一个别的线程的条目，Context::try_select 用 compare_exchange(0, oper, AcqRel, Acquire) 把它的 select 从 Waiting 改成这个操作的地址。CAS 成功才 unpark，并把条目从队列里删掉。CAS 失败说明那个线程已经被别人选中或超时了，看下一个。

第 5 步：发送方醒来，重试

wait_until 看到 select 不再是 Waiting，返回 Selected::Operation。match 里这个分支什么也不做（条目已经被唤醒方删了），回到外层 loop，新建一个 Backoff 重新 start_send：tail = 4，槽 0 的 stamp 现在是 4，等于 tail，可以写；新 tail 为 8，CAS 成功；write 写入消息、stamp.store(5, Release)、receivers.notify()。

注意：被叫醒不等于拿到了槽。notify 只是说「现在可能有空位了」，发送方醒来后和其他发送方公平竞争 tail，输了就再走一遍第 2、3 步。

每一步在哪一章

小结

• array channel 的 head、tail 和每个槽的 stamp 都是 {lap, mark, index} 打包的整数；stamp == tail 可写，stamp == head + 1 可读，stamp + one_lap == tail + 1 且 head 落后一圈时为满。
• 满了的 send 先 snooze 到 is_completed（11 次），再 Context::with 取线程本地的 Context、注册到 senders、再查一次、park。
• 接收方 read 后改写 stamp（Release），notify 用 CAS 把一个等待者的 select 改成操作地址再 unpark；没人等时 is_empty 让 notify 免锁。
• 被叫醒只表示「可能有空位」，发送方回到循环开头重新竞争 tail。