Context 与 Waker

主线位置：一条消息的阻塞部分。array 的 send 自旋完仍然满，就 Context::with 取出线程本地的 Context，senders.register(oper, cx) 挂进 SyncWaker，复查一次，cx.wait_until(None) 里 park；接收方 read 完调 senders.notify()，在 Context 上 CAS 出 Selected::Operation 并 unpark。

sequenceDiagram
  participant S as 发送方线程
  participant CX as S 的 Context
  participant W as senders: SyncWaker
  participant R as 接收方线程
  S->>CX: Context::with：取线程本地缓存，reset 为 Waiting
  S->>W: register(oper, cx)：加锁，push Entry，is_empty = false
  S->>S: 复查 is_full()，仍满
  S->>CX: wait_until：select 仍是 Waiting，park
  R->>R: read 一条，stamp 更新
  R->>W: notify：is_empty 为 false，加锁
  W->>CX: try_select：CAS Waiting → Operation(oper)
  W->>S: unpark，Entry 移出队列
  S->>CX: 醒来，读到 Operation，返回
  S->>S: 回到外层循环，重新 start_send

两个文件，一共三百来行：context.rs 178 行，waker.rs 287 行。普通的阻塞收发和 select! 用的是同一套：一个线程阻塞时，不管等的是一个操作还是十个，都只有一个 Context，被叫醒时 Context 里记着是哪个操作赢了。

Selected 与 Operation：一个整数编码四种状态

Selected 是阻塞操作的当前状态：

Operation::hook 把一个 &mut T 的地址当作操作 id：

pub fn hook<T>(r: &mut T) -> Operation {
    let val = r as *mut T as usize;
    // Make sure that the pointer address doesn't equal the numerical representation of
    // `Selected::{Waiting, Aborted, Disconnected}`.
    assert!(val > 2);
    Operation(val)
}

普通 send 用 Operation::hook(token)：token 是 send 栈帧上的局部变量，整个阻塞期间都活着，地址在本线程内唯一。select 用 handles 切片里每个元素的地址。地址不可能是 0、1、2，所以四种状态塞进一个 usize，用一个 AtomicUsize 就能原子地切换。

Context

pub struct Context {
    inner: Arc<Inner>,
}

struct Inner {
    select: AtomicUsize,
    packet: AtomicPtr<()>,
    thread: Thread,
    thread_id: ThreadId,
}

select 就是上面那个 Selected 的编码；packet 是给 zero 传 Packet 指针用的；thread 是 unpark 的目标。Arc 是因为别的线程的 Waker 里要持有它的克隆。

线程本地缓存

Context::with：

std::thread_local! {
    /// Cached thread-local context.
    static CONTEXT: Cell<Option<Context>> = Cell::new(Some(Context::new()));
}

CONTEXT
    .try_with(|cell| match cell.take() {
        None => f(&Context::new()),
        Some(cx) => {
            cx.reset();
            let res = f(&cx);
            cell.set(Some(cx));
            res
        }
    })
    .unwrap_or_else(|_| f(&Context::new()))

每个线程第一次阻塞时建一个 Context（一次 Arc 分配，加一次 thread::current()），之后反复用，reset 把 select 置回 Waiting、packet 置空。用 Cell<Option<_>> 加 take，是为了处理重入：闭包 f 里如果又调了 Context::with（比如在阻塞路径上触发了另一个 channel 的阻塞操作），缓存已经被拿走了，take 得到 None，就临时新建一个。线程析构期间 TLS 不可用，try_with 返回错误，也临时新建。

try_select：一次 CAS 决定谁赢

try_select：

pub fn try_select(&self, select: Selected) -> Result<(), Selected> {
    self.inner
        .select
        .compare_exchange(
            Selected::Waiting.into(),
            select.into(),
            Ordering::AcqRel,
            Ordering::Acquire,
        )
        .map(|_| ())
        .map_err(|e| e.into())
}

只有从 Waiting 出发的 CAS 能成功。一个线程在 select! 里同时挂在三个 channel 上，三个 channel 的对端可能同时来叫它，只有第一个 CAS 成功的那个算数；后面的 CAS 失败，就知道「这个线程已经被别人选走了」，去试队列里的下一个等待者。超时（Aborted）、断开（Disconnected）也走同一个 CAS，所以「超时」和「被选中」不会同时成立。

AcqRel 的 Release 一半和等待方 wait_until 里的 Acquire 读配对：选中方在 CAS 之前做的写（比如 zero 写进 Packet 的指针）对醒来的线程可见。

wait_until：直接 park

wait_until：

pub fn wait_until(&self, deadline: Option<Instant>) -> Selected {
    loop {
        // Check whether an operation has been selected.
        let sel = Selected::from(self.inner.select.load(Ordering::Acquire));
        if sel != Selected::Waiting {
            return sel;
        }

        // If there's a deadline, park the current thread until the deadline is reached.
        if let Some(end) = deadline {
            let now = Instant::now();

            if now < end {
                thread::park_timeout(end - now);
            } else {
                // The deadline has been reached. Try aborting select.
                return match self.try_select(Selected::Aborted) {
                    Ok(()) => Selected::Aborted,
                    Err(s) => s,
                };
            }
        } else {
            thread::park();
        }
    }
}

park 可能虚假唤醒，所以外面是 loop，每次醒来先看 select。到了截止时间也不是直接返回 Aborted，而是 CAS 一次：如果恰好在最后一刻被选中，CAS 失败，返回的是那个 Operation，操作照样完成。

0.5.14 之前，这个函数开头还有一段「乐观自旋」：先用 Backoff 自旋、让出约 11 次，看 select 会不会变，再 park。8144fbb4（0.5.14，CHANGELOG 记作 #1154「Remove inefficient spinning」）删掉了这一段。调用 wait_until 之前，array / list 的 send / recv 已经自旋过一轮，这里再转一轮是重复的。std 的移植版也没有这一段。

thread::park 依赖「unpark 先于 park 发生时，park 立即返回」的语义（std 给每个线程一个令牌），所以下面「先 CAS 再 unpark」和「先看 select 再 park」之间不会丢唤醒：选中方 unpark 在等待方 park 之前也没关系。

Waker：等待者列表

pub(crate) struct Entry {
    pub(crate) oper: Operation,
    pub(crate) packet: *mut (),
    pub(crate) cx: Context,
}

pub(crate) struct Waker {
    selectors: Vec<Entry>,
    observers: Vec<Entry>,
}

两个列表：

• selectors：要完成操作的等待者（阻塞的 send / recv、select! 的分支）。叫醒一个就够，用 try_select。
• observers：只想知道就绪的观察者（Select::ready() 一类，见 select），用 watch / unwatch 增删。就绪时全部叫醒，用 notify。

Waker::try_select 找第一个「别的线程的、CAS 能成功的」条目：

let thread_id = current_thread_id();

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))

跳过本线程的条目，是因为一个线程可以在 select! 里对同一个 zero channel 同时挂一个 send 分支和一个 recv 分支；它不能和自己配对。CAS 失败的条目（那个线程已被别人选走）也跳过，但不移除：它醒来后自己 unregister。选中的条目从 Vec 里 remove，这是 O(n) 的，但队列通常很短。

disconnect 把所有 selectors CAS 成 Disconnected 并 unpark，条目留在原地，由各线程醒来后自己 unregister（注释说它们可能要从 packet 里取回消息）。

current_thread_id 把 thread::current().id() 缓存在线程本地里；TLS 不可用时退回每次调 thread::current()。std 的移植版换成了「一个线程本地 u8 的地址」，析构期间也可用，见 vs-std。

SyncWaker：没人等时不加锁

array 和 list 的 write / read 每条消息都要调一次 notify，如果每次都加锁，无锁的快路径就白做了。SyncWaker 在 Mutex<Waker> 外面加了一个 is_empty: AtomicBool：

pub(crate) fn notify(&self) {
    if !self.is_empty.load(Ordering::SeqCst) {
        let mut inner = self.inner.lock();
        if !self.is_empty.load(Ordering::SeqCst) {
            inner.try_select();
            inner.notify();
            self.is_empty.store(
                inner.selectors.is_empty() && inner.observers.is_empty(),
                Ordering::SeqCst,
            );
        }
    }
}

（waker.rs:221）没人等时，notify 只是一次 SeqCst 读。register / unregister 在锁内更新 is_empty。

这里用 SeqCst 而不是 Acquire / Release，防的是经典的丢唤醒：

需要的保证是：② 读到旧状态（仍满）和 ④ 读到旧的 is_empty == true 不能同时发生，否则发送方去睡、接收方也不叫它。这是两个线程各「先写一个、再读另一个」的 store-load 模式，只有 SeqCst（或 SeqCst fence）能排除，Acquire / Release 不够。分类见本库里的内存序。

锁是 crate::utils::Mutex（utils.rs:61），一个包着 std::sync::Mutex 的新类型，lock() 用 unwrap_or_else(PoisonError::into_inner) 忽略 poison。SyncWaker 持锁期间执行的是等待队列的增删、CAS 和 unpark，不调用户提供的闭包。这个新类型是 c0c466e5「channel: Use non-poison Mutex」加的，随 0.5.16 发布；在那之前是 std Mutex 加 .lock().unwrap()，std 的移植版至今如此。

一次阻塞要花多少

阻塞路径上每次都有：一次 TLS 访问取 Context，register / unregister 各一次加锁（Vec push / remove），park / unpark 各一次系统调用。实验 E6 的 zero flavor 每条消息都要走一遍完整的阻塞路径，本机约 1.4 微秒（1P1C 0.72 Mmsg/s，见 zero）。array / list 在快路径上只有原子操作和一次 is_empty 读，1P1C 能到 30–100 Mmsg/s。两者之间的差距，就是 send / recv 在阻塞前先自旋的理由；自旋多少才合适，本机的二分实验给出了反直觉的结果，见 vs-std。

小结

• 阻塞操作的状态是一个 AtomicUsize：0 等待、1 放弃、2 断开、其他值是被选中操作的 id（某个栈上变量的地址）。
• Context 线程本地缓存，每次阻塞前 reset；重入或 TLS 不可用时临时新建。
• 谁叫醒一个线程，取决于谁先把它的 select 从 Waiting CAS 掉；超时、断开、被选中三者互斥。
• wait_until 循环 park / park_timeout，0.5.14 起不再在这里乐观自旋。
• Waker 分 selectors（叫醒一个、完成操作）和 observers（全部叫醒、只报就绪）；try_select 跳过本线程的条目。
• SyncWaker 的 is_empty 让没人等时 notify 只是一次 SeqCst 读；SeqCst 是为了和「注册后复查」配对，排除丢唤醒。