Channel

主线位置：跨任务主线的全程。一个任务 CH.send(v).await，另一个任务 CH.receive().await：RawMutex 护住内部状态，空 / 满时登记 waker，对端操作后叫醒，执行器把被叫醒的任务放回运行队列再 poll。最后对照中断处理函数里的 try_send。

sequenceDiagram
  participant R as 接收任务
  participant Ch as Channel
  participant Ex as 执行器
  participant S as 发送任务
  R->>Ch: receive poll
  Ch-->>R: 空 登记 receiver_waker
  Ex->>Ex: 无事可做 WFE
  S->>Ch: send poll
  Ch->>Ch: push_back
  Ch->>Ex: receiver_waker.wake
  Ex->>Ex: 入运行队列 SEV
  Ch-->>S: Ready
  Ex->>R: 再次 poll
  R->>Ch: pop_front
  Ch-->>R: Ready v

三个文件：channel.rs（1206 行，末尾 148 行是测试）是普通的多生产者多消费者队列，priority_channel.rs（847 行）把 Deque 换成二叉堆，zerocopy_channel.rs（360 行）是单生产者单消费者、直接借出缓冲区槽位的变体。本页 std 实验在 exp-embassy/sync-std（platform-std 执行器，release，M2 Max、macOS 27、rustc 1.99.0；同机有别的程序在跑，uptime 负载 20 到 44），QEMU 实验在 exp-embassy/sync-qemu（QEMU 11.1.2 b-l475e-iot01a，release，embassy-stm32 用注释掉 rcc/bd.rs 里 while !cr.read().dbp() {} 那一行的补丁版本，时间驱动是模板里的 1 kHz SysTick 驱动）。

结构：一个 Deque，两个 waker 槽

#[derive(Debug)]
struct ChannelState<T, const N: usize> {
    queue: Deque<T, N>,
    receiver_waker: WakerRegistration,
    senders_waker: WakerRegistration,
}

ChannelState 就这三样：heapless 0.9 的定长双端队列，加接收方、发送方各一个 WakerRegistration。字段名 senders_waker 是复数，类型却只能存一个 waker。Channel 把它包进 Mutex<M, RefCell<..>>，所有操作都经过 lock：

    fn lock<R>(&self, f: impl FnOnce(&mut ChannelState<T, N>) -> R) -> R {
        self.inner.lock(|rc| f(&mut *unwrap!(rc.try_borrow_mut())))
    }

try_borrow_mut 失败就 panic，原始互斥 一页在 QEMU 上撞到过这一行。new() 是 const fn，整个 Channel 可以直接是 static，在 .bss 里占的空间编译期就定了：thumbv7em 上 Channel<CriticalSectionRawMutex, u32, 4> 是 48 字节，N = 64 是 288 字节，每多一个槽多 4 字节，固定开销 32 字节（两个 Option<Waker> 各 8 字节，加 Deque 的下标和 RefCell 的借用计数）。

Sender 和 Receiver 只是 &'ch Channel 的包装，都是 Copy，可以随便复制给多个任务；Channel 自己也有 send / receive，直接用 &'static Channel 也行。

send：消息放在 future 里

SendFuture 持有 channel: &Channel 和 message: Option<T>。poll 把消息拿出来试一次，满了再放回去：

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match self.message.take() {
            Some(m) => match self.channel.try_send_with_context(m, Some(cx)) {
                Ok(..) => Poll::Ready(()),
                Err(TrySendError::Full(m)) => {
                    self.message = Some(m);
                    Poll::Pending
                }
            },
            None => panic!("Message cannot be None"),
        }
    }

真正干活的是 ChannelState::try_send_with_context：

    fn try_send_with_context(&mut self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), TrySendError<T>> {
        match self.queue.push_back(message) {
            Ok(()) => {
                self.receiver_waker.wake();
                Ok(())
            }
            Err(message) => {
                if let Some(cx) = cx {
                    self.senders_waker.register(cx.waker());
                }
                Err(TrySendError::Full(message))
            }
        }
    }

• 推进去了就叫醒接收方。receiver_waker 是空的（没人在等）时 wake() 什么也不做。
• 推不进去，push_back 把消息原样还回来；有 Context 才登记 waker。try_send 传的是 None，所以 try_send 从不登记任何东西，只返回 TrySendError::Full(v)。
• 消息一直在 SendFuture 里，没进队列。SendFuture 被 drop（例如在 select 里输了），消息随之 drop，channel 里什么也没留下，也没有半条消息。
• SendFuture 实现了 Unpin（不管 T 是什么），所以可以不 pin! 直接 poll。

receive：先叫醒发送方，再取

try_receive_with_context：

    fn try_receive_with_context(&mut self, cx: Option<&mut Context<'_>>) -> Result<T, TryReceiveError> {
        if self.queue.is_full() {
            self.senders_waker.wake();
        }

        if let Some(message) = self.queue.pop_front() {
            Ok(message)
        } else {
            if let Some(cx) = cx {
                self.receiver_waker.register(cx.waker());
            }
            Err(TryReceiveError::Empty)
        }
    }

叫醒发送方的条件是「取之前是满的」，也就是这一次取走会腾出第一个空位。叫醒在锁里发生，但不会重入 Channel：embassy 执行器的 waker 只做 wake_task，把任务压进运行队列，被叫醒的发送方要等这次 poll 返回、执行器轮到它时才会再进 lock。poll_receive、try_peek_with_context 开头也是同样的「满了先叫醒发送方」，所以对一个满的 channel 调 try_peek，发送方也会被叫醒一次，再 poll 时发现还是满的，重新登记。

ReceiveFuture 没有自己的状态，只有 &Channel；它被 drop 时什么也不会丢，消息还在队列里。这就是 receive() 可以放心放进 select 的原因（实验见 embassy-futures）。

clear 也按同样的规则：满着的时候清空，先叫醒发送方。

走一遍跨任务主线

接收任务先跑、channel 是空的：

1. 执行器 poll 接收任务，ReceiveFuture::poll 调 try_receive_with_context(Some(cx))：Channel::lock 进 RawMutex（CriticalSectionRawMutex 就是存 PRIMASK、cpsid i），借出 RefCell，pop_front 得到 None，把接收任务的 waker 存进 receiver_waker，返回 Pending。
2. 执行器发现运行队列空了。Cortex-M 线程模式执行器执行 WFE 睡下去（见 各平台的执行器）。
3. 发送任务被别的事件（定时器、中断）叫醒、被 poll，SendFuture::poll → try_send_with_context：push_back 成功，receiver_waker.wake()。waker 的 vtable 调 wake_task，任务状态置为已排队，压进运行队列；运行队列从空变非空，SyncExecutor::enqueue 调 pender，线程模式执行器的 __pender 执行 SEV。发送方返回 Ready。
4. 执行器下一轮从运行队列取出接收任务 poll，pop_front 拿到消息，返回 Ready(v)。

channel 满、发送方先等的情况对称：发送方登记在 senders_waker，接收方取走一条之前看到「满」，叫醒它。

满了以后：多个发送方

一个槽只能存一个 waker。第二个发送方来登记时，WakerRegistration::register 把第一个换出来并叫醒它；第一个被 poll，发现还是满的，再登记，又把第二个换出来叫醒……只要 channel 一直满着，几个发送方就一直互相叫醒，运行队列永远不空。waker_registration.rs 的注释预告了这一点：「This wastes CPU but things will still work」。

sync-std/src/bin/chan_wake.rs：Channel<CriticalSectionRawMutex, u32, 1> 先 try_send(0) 塞满，按 s1、s2、s3 的顺序 spawn 三个发送任务，每个 send(id * 10) 一次；Counted 包装数每个 SendFuture 被 poll 了多少次。主任务每隔 100 ms 取一条：

#[embassy_executor::task(pool_size = 3)]
async fn sender(id: usize) {
  Counted { inner: CH.send(id as u32 * 10), id }.await;
  println!("  s{id} send done at {} ms, polls so far {:?}", Instant::now().as_millis(), polls());
}

round 0: before receive, polls [850094, 1700187, 850094]
round 0: received 0
  s1 send done at 105 ms, polls so far [850095, 1700187, 850094]
round 1: before receive, polls [850095, 2813311, 1963217]
round 1: received 10
  s3 send done at 209 ms, polls so far [850095, 2813311, 1963218]
round 2: before receive, polls [850095, 2813312, 1963218]
round 2: received 30
  s2 send done at 312 ms, polls so far [850095, 2813313, 1963218]
round 3: before receive, polls [850095, 2813313, 1963218]
round 3: received 20
end: polls [850095, 2813313, 1963218]

• 前 100 ms 里三个 SendFuture 一共被 poll 了 340 万次，s2 是另外两个的两倍：每一轮交换里 s2 都在场。
• s1 进去以后剩 s2、s3 两个，又互相叫醒了 100 ms，各一百多万次。
• 最后只剩 s2 一个等着时，没人跟它抢槽，100 ms 里只被 poll 了 1 次。
• 进入顺序是 s1、s3、s2，不是 spawn 的顺序。再跑三次，第一个进去的分别是 s3、s3、s3，第二个是 s2、s1、s1。腾出空位的那一刻，谁恰好排在运行队列前面谁就进去。
• /usr/bin/time 看同一个程序：墙上时间 0.43 s，用户态 CPU 0.21 s。至少两个发送方在等的那 200 ms，CPU 一直在跑。

QEMU 上同样的结构（sync-qemu/src/bin/chan_wake.rs，加 -icount shift=0，SysTick 计时）：

one sender waiting, after 100 ms: polls [1, 0]
two senders waiting, after 100 ms: polls [275101, 275101]
received 0, one left waiting, after 100 ms: polls [275102, 275102] at 300 ms

一个发送方等的时候它只被 poll 1 次，执行器在 WFE 里睡着；两个发送方一起等，100 ms 里各被 poll 27.5 万次，执行器一刻也没睡。在电池供电的板子上，这就是功耗。

这是 Channel 的设计选择，不是 bug：多数嵌入式程序里一个 channel 只有一个发送任务，或者满的时间很短。需要多个发送方排队等的场合，可以每个发送方一个 channel，或者在发送方之间先过一个 FairSemaphore。tokio 的有界 mpsc 正相反：发送方在信号量的等待链表里 FIFO 排队，一个许可只叫醒一个（见 tokio 卷 mpsc 的「满了以后按顺序恢复」实验）。

在中断里：try_send

中断处理函数不是 async 的，只能用不等待的 try_send，而且 channel 必须是 CriticalSectionRawMutex：NoopRawMutex 的 channel 进不了 static，用 unsafe impl Sync 硬塞进去会在 try_borrow_mut 处 panic（两者都在 原始互斥 里实测过）。

sync-qemu/src/bin/irq_send.rs 借一个没用到的中断 UART4 当软件中断：主任务通过 NVIC 把它挂起（pend），处理函数里 try_send 一个序号，另一个任务 receive().await：

static CH: Channel<CriticalSectionRawMutex, u32, 2> = Channel::new();

#[interrupt]
fn UART4() {
  let n = SEQ.fetch_add(1, Ordering::Relaxed) + 1;
  if let Err(TrySendError::Full(_)) = CH.try_send(n) {
    FULL.fetch_add(1, Ordering::Relaxed);
  }
}

// in main, after `unsafe { interrupt::UART4.enable() }`:
for _ in 0..3 {
  hprintln!("pend UART4 at {} ms", Instant::now().as_millis());
  interrupt::UART4.pend();
  Timer::after_millis(10).await;
}

hprintln!("burst: pend UART4 5 times without awaiting");
for _ in 0..5 {
  interrupt::UART4.pend();
}
hprintln!("after burst: sent {}, full {}, len {}", SEQ.load(Ordering::Relaxed), FULL.load(Ordering::Relaxed), CH.len());

前三次的输出：

pend UART4 at 5 ms
  receiver got 1 at 5 ms
pend UART4 at 15 ms
  receiver got 2 at 15 ms
pend UART4 at 25 ms
  receiver got 3 at 25 ms

pend() 写 NVIC 的挂起寄存器，线程模式下中断开着，处理函数立刻抢占主任务；try_send 成功后 receiver_waker.wake() 把接收任务压进运行队列并 SEV。处理函数返回后主任务继续跑到 Timer::after_millis(10).await 让出，执行器接着 poll 接收任务，所以「got」和「pend」在同一毫秒。中断处理函数从头到尾没有碰执行器的任何 API，它只调了一个 waker；从中断到任务这一跳，和 一次中断唤醒 里 HAL 驱动用的 AtomicWaker 是同一条路。

动态版本、Stream 与 Sink

Sender<'ch, M, T, N> 的类型里带着 M 和 N，写进函数签名或结构体字段很啰嗦。DynamicSender / DynamicReceiver 存的是 &dyn DynamicChannel<T>（DynamicChannel 是 crate 内部的 trait，try_send_with_context 等五六个方法），只剩 T 一个参数，代价是每次操作一次间接调用。它们不是 Send（dyn DynamicChannel<T> 没有 Sync 约束）；0.7.0 加的 SendDynamicSender / SendDynamicReceiver 用 unsafe impl Send + Sync 补上，只能从 M: RawMutex + Sync + Send 的 Sender 转换过来。

Receiver 和 Channel 实现了 futures_core::Stream（0.7.0，poll_next 就是 poll_receive(cx).map(Some)，永远不会返回 None，因为 channel 没有「关闭」的概念）；0.8.0 又给 Sender 和 Channel 实现了 futures_sink::Sink，Sink for Channel 的 poll_ready 是 poll_ready_to_send，start_send 是 try_send，poll_flush / poll_close 直接 Ready(Ok(()))。

和 tokio 的 mpsc 比，这里没有「所有发送方都 drop 了」的检测：Sender 是 Copy，没有引用计数，也没有 Drop。接收方会一直等下去。

代码体积

sync-qemu 里四个只差一处的程序：生产者任务每 1 ms 产生一个 u32，消费者任务收 10 个求和后退出。「无 channel」版本用一个 AtomicU32 加 Timer 代替。llvm-size（thumbv7em-none-eabihf，opt-level = "s"，含 embassy-stm32 的 init 和执行器）：

用上一个 channel，.text 多 628 字节。llvm-nm --size-sort 看去处：send / receive 全部内联进了两个任务的 TaskStorage::poll（生产者那个从 0x76 涨到 0x146 字节，消费者从 0xd0 涨到 0x132），另外有两个没内联的函数 WakerRegistration::register（64 字节）和 WakerRegistration::wake（26 字节）。容量从 4 到 64，.text 只多 8 字节，.bss 多 240 字节，正好是 60 个 u32。换成 ThreadModeRawMutex 反而多 232 字节，是 lock 里的断言和 panic 信息。

PriorityChannel

PriorityChannel<M, T, K, N> 和 Channel 结构相同，ChannelState 里的 Deque<T, N> 换成 heapless::BinaryHeap<T, K, N>，K 是 Max 或 Min，T: Ord 决定谁先出队。等待逻辑一行没变，还是两个 WakerRegistration。多了一个 remove_if(predicate)（0.7.0），实现是遍历堆、把不满足条件的元素 clone 进一个新堆，再整个换掉；它腾出了空位，却不叫醒 senders_waker，和上面冷知识里 clear() 修好之前是同一类问题。sync-std/src/bin/prio_remove.rs：容量 1 的 PriorityChannel 先塞满，一个任务 send(2).await 等着，主任务 remove_if(|_| true) 清掉唯一的元素：

remove_if at 14 ms, len 0
after 100 ms: len 0

channel 是空的，发送方却一直没醒（它的「sender done」始终没有打印）。之后的 receive 也叫不醒它，因为 channel 不再是满的。main 上这段代码到写作时仍未改动。它复用了 channel.rs 的 TrySendError、DynamicSender 等类型。

0.8.0 的 ChannelState::poll_ready_to_send 有一个优先级写错的 bug：

    fn poll_ready_to_send(&mut self, cx: &mut Context<'_>) -> Poll<()> {
        self.senders_waker.register(cx.waker());

        if !self.queue.len() == self.queue.capacity() {
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    }

! 先作用在 usize 上，是按位取反，!0 是 usize::MAX，几乎永远不等于容量。sync-std/src/bin/prio_ready.rs 对一个空的 PriorityChannel 和一个空的 Channel 各 poll_once 一次：

PriorityChannel empty: Pending
Channel empty:         Ready(())
PriorityChannel try_send: Ok(())

空 channel 明明能发（try_send 成功），poll_ready_to_send 却说 Pending。用 poll_fn 或 Sink 风格等「可以发了」的代码会一直等下去。main 上的 #5996（提交「embassy-sync: fix priority_channel poll_ready_to_send」）改成了 !=，尚未发版。

zerocopy_channel

zerocopy_channel::Channel<'a, M, T> 不自己存数据。缓冲区由调用方给（new(buf: &'a mut [T])，长度就是容量），split 要 &mut self，得到唯一的一对 Sender / Receiver，所以是单生产者单消费者。发送方 send() 返回的不是 ()，而是下一个空槽的 &mut T，调用方原地写好，再调 send_done() 把它交出去；接收方 receive() 拿到 &mut T 原地读，receive_done() 归还。大块数据（网络帧、DMA 缓冲区）就不用在任务之间拷来拷去，embassy-net-driver-channel 用的正是它。

它的 State 是 front、back、full 三个下标加两个 WakerRegistration。字段名容易看反：等空槽的发送方登记在 receive_waker 里，由 receive_done 叫醒；等数据的接收方登记在 send_waker 里，由 send_done 叫醒。名字说的是「被谁叫醒」。

0.8.0 的 Sender::clear 能在安全代码里造出两个指向同一个槽的 &mut T（sync-std/src/bin/zc_clear.rs）：

let r: &mut u32 = rx.try_receive().unwrap(); // receiver is reading slot 0
tx.clear();
let w: &mut u32 = tx.try_send().unwrap(); // sender gets slot 0 again
println!("receiver slot {:p}, sender slot {:p}", r as *const u32, w as *const u32);
*w = 99;
println!("receiver now reads {}", *r);

receiver slot 0x16bbe9bb8, sender slot 0x16bbe9bb8
receiver now reads 99

接收方还拿着槽 0 的 &mut，clear 把下标归零，发送方又借到了槽 0。这是安全代码里的别名可变引用，属于未定义行为。main 上 #6004 直接删掉了 Sender::clear（提交「embassy-sync: remove unsound Sender::clear」），尚未发版。Receiver::clear 和 Channel::clear 不受影响：前者要 &mut Receiver，此时不可能还有借出的接收槽；后者要 &mut Channel，split 出来的两半都得先结束。

动脑筋答案

sync-qemu/src/bin/irq_send.rs 的后半段输出：

burst: pend UART4 5 times without awaiting
after burst: sent 8, full 3, len 2
  receiver got 4 at 35 ms
  receiver got 5 at 35 ms

• 每次 pend() 之后中断立刻执行，五次 pend() 就是五次处理函数，序号 4 到 8，sent 是 8（含前面的 3 次）。
• 接收任务这时没机会运行（主任务循环里没有 .await），channel 容量 2：4、5 进去，6、7、8 三次 try_send 返回 Full，full 是 3，len 是 2。
• try_send 失败时消息原样还给调用方（TrySendError::Full(v)），处理函数把它丢了。主任务 await 之后接收任务收到 4 和 5，6 到 8 永远丢了。中断里用 try_send，要么容量留足，要么自己决定满了怎么办（计数、丢最旧的、换 PubSubChannel::publish_immediate 那种覆盖语义）。

小结

• Channel<M, T, N> = Mutex<M, RefCell<ChannelState>>，ChannelState 是 heapless::Deque<T, N> 加接收方、发送方各一个 WakerRegistration。thumbv7em 上 u32 × 4 占 48 字节，用一个 channel 多 628 字节代码。
• send：消息在 SendFuture 里，push_back 成功就叫醒接收方，满了登记 waker；future 被 drop，消息跟着 drop。receive：取之前是满的就先叫醒发送方，空了登记 waker；future 被 drop 不丢消息。
• 跨任务主线：登记 waker → 执行器 WFE → 对端 wake() → wake_task 入运行队列、SEV → 再 poll。
• 多个任务同时等同一方向时互相叫醒：std 上 100 ms 各一百多万次 poll，QEMU 上 27.5 万次，执行器不睡；进入顺序不固定。
• 中断里只能 try_send，channel 必须是 CriticalSectionRawMutex；满了 try_send 把消息还回来。
• 0.8.0 里 PriorityChannel::poll_ready_to_send 永远 Pending，zerocopy_channel::Sender::clear 不健全，这两处 main 上已修、尚未发版；PriorityChannel::remove_if 腾出空位不叫醒发送方，main 上也没改。