Rust in Everything
基于 embassy-sync 0.8.0

Signal、Watch 与 PubSub

主线位置：跨任务主线的三个变体。Channel 是「每条消息恰好一个人收到」；这里的三个类型改的正是这一点：Signal 只留最新的一个值，Watch 让多个接收者都看最新值，PubSubChannel 让每个订阅者都收到每一条。锁和 waker 槽和 Channel 一样，见 原始互斥 和 等待队列。

flowchart TB
  q{"每条消息给谁"} -->|恰好一个接收者| ch["Channel"]
  q -->|所有订阅者| ps["PubSubChannel"]
  q -->|只要最新值| last{"几个接收者"}
  last -->|一个| sig["Signal"]
  last -->|多个| w["Watch"]

文件：signal.rs（144 行）、watch.rs（1127 行，第 398 行以后是 Sender / Receiver / AnonReceiver 及其 Dyn 版本的包装，786 行以后是测试）、pubsub/（mod.rs 789 行，publisher.rs 320 行，subscriber.rs 195 行）。实验都在 exp-embassy/sync-std（platform-std 执行器，release，M2 Max、macOS 27、rustc 1.99.0）。

Signal：一个 Cell，三种状态

Signal<M, T> 是 Mutex<M, Cell<State<T>>>，State 三选一：

enum State<T> {
    None,
    Waiting(Waker),
    Signaled(T),
}

值和 waker 不会同时存在：有人在等的时候里面没有值，有值的时候没人在等。signal 无条件把状态换成 Signaled(val)，换出来的是 Waiting 就叫醒它，换出来的是旧的 Signaled 就让它随之 drop。poll_wait：

    fn poll_wait(&self, cx: &mut Context<'_>) -> Poll<T> {
        self.state.lock(|cell| {
            let state = cell.replace(State::None);
            match state {
                State::None => {
                    cell.set(State::Waiting(cx.waker().clone()));
                    Poll::Pending
                }
                State::Waiting(w) if w.will_wake(cx.waker()) => {
                    cell.set(State::Waiting(w));
                    Poll::Pending
                }
                State::Waiting(w) => {
                    cell.set(State::Waiting(cx.waker().clone()));
                    w.wake();
                    Poll::Pending
                }
                State::Signaled(res) => Poll::Ready(res),
            }
        })
    }

Cell 里的值要先 replace 出来才能看，看完再放回去，所以每个分支都有一次 set。第三个分支和 WakerRegistration 的规则一样：别的任务来等，就把原来的等待者换出来叫醒。Signal 的文档说它是「for a single consumer」，两个任务同时 wait() 的后果是互相叫醒，直到值来了被其中一个拿走：

static SIG2: Signal<CriticalSectionRawMutex, u32> = Signal::new();

#[embassy_executor::task(pool_size = 2)]
async fn waiter(id: usize) {
  let v = Counted { inner: Box::pin(SIG2.wait()), id }.await;
  println!("  waiter {id} got {v} at {} ms, polls {}", Instant::now().as_millis(), POLLS[id].load(Ordering::Relaxed));
}

sync-std/src/bin/signal_watch.rs 先演示覆盖，再 spawn 两个 waiter，100 ms 后 signal(42)：

signaled 1,2,3; signaled() = true
wait() -> 3
try_take() after wait -> None
after 100 ms: polls 1202027 1202027
  waiter 2 got 42 at 100 ms, polls 1202028
after another 100 ms: polls 1202028 1202028, signaled() = false
  • 连着 signal(1)、signal(2)、signal(3)，wait() 只拿到 3，1 和 2 没有任何痕迹。这是 Signal 的本意：传「状态」而不是「事件」，接收方只关心最新的。
  • 两个等待者 100 ms 里各被 poll 120 万次。值来了以后，正好排在前面的那个（这次是 waiter 2，再跑一次是 waiter 1）拿走它；另一个最后一次 poll 看到 None，登记自己，然后安静地睡下去，再也收不到这个 42。
  • wait() 是 cancel-safe 的：值只在返回 Ready 的那一次 poll 里被取走，future 中途被 drop 不会丢值。0.8.0 前夕的提交 98012435e「document signal cancel safety」把这句写进了文档。

try_take 只在 Signaled 时取走值，其他状态原样放回；reset 就是调它。signaled() 只看不取。

📝NOTE
冷知识：reset() 以前会把等待者的 waker 扔掉

2025 年 8 月的 #4598（首次进入 embassy-sync 0.8.0，CHANGELOG 记作「Fix wakers getting dropped by Signal::reset」）之前，reset 是一行 self.state.lock(|cell| cell.set(State::None));。如果此时有任务在 wait()，状态是 Waiting(waker)，这一行把它连同 waker 一起丢掉；之后 signal(v) 换出来的是 None，没有人可叫，等待的任务就一直睡着，直到它碰巧因为别的原因被 poll。提交「embassy-sync: Don't drop wakers in Signal::reset」把它改成 self.try_take();：try_take 遇到 Waiting 会原样放回去。try_take 自己在 2023 年 12 月加入时就有「Put waiting state back if any」这一步（提交 13c107e81），reset 当时没有跟着改。

Watch：一个值，一个版本号

Watch<M, T, N> 的状态 WatchState：

struct WatchState<T: Clone, const N: usize> {
    data: Option<T>,
    current_id: u64,
    wakers: MultiWakerRegistration<N>,
    receiver_count: usize,
}

send 写入 data、current_id += 1、叫醒 wakers 里所有人。每个接收者记着自己看过的版本号 at_id，poll_changed 只在 current_id > *id 时返回值（clone 一份）并把 *id 更新成 current_id；poll_get 不管版本号，有值就返回。

  • 接收者有上限：receiver 在 receiver_count == N 时返回 None，Receiver drop 时减回去。N 同时是 MultiWakerRegistration<N> 的槽数，所以正常情况下槽够用。
  • anon_receiver 不计数，也只有 try_get / try_changed 这些不等待的方法，从不登记 waker，可以有任意多个。中断处理函数里想看一眼最新值就用它。
  • 新接收者的 at_id 是 0。Watch::new() 的 current_id 也从 0 开始，第一次 send 之后才是 1，所以第一次 changed() 能等到它。Watch::new_with(v) 带着初始值但版本号仍是 0，changed() 认为「没变过」（sync-std/src/bin/watch_init.rs）：
get():     Ready(7)
changed(): Pending
third receiver: false
after send(8), changed(): Ready(8)

signal_watch.rs 的第三段：两个接收者各自循环 changed() 两次，主任务连发 1、2、3，10 ms 后再发 4：

sent 1,2,3 back to back
  watcher 1 changed -> 3 at 220 ms
  watcher 2 changed -> 3 at 220 ms
sent 4
  watcher 2 changed -> 4 at 233 ms
  watcher 1 changed -> 4 at 233 ms

三次 send 发生在接收者被 poll 之前，current_id 从 0 跳到 3，接收者醒来只看到 3。中间值被合并，和 tokio 的 watch 一样（tokio 卷 oneshot、watch、broadcast、Notify 里「中间值被合并」的实验）；区别在于 tokio 用读写锁护值、等待者数量不限，这里是一把 RawMutex、最多 N 个会等待的接收者。

clear() 把 data 置为 None，既不加版本号也不叫醒任何人。send_if_modified(f) 只在闭包返回 true 时加版本号、叫醒。

PubSubChannel：每条消息记着还有几个人没读

PubSubChannel<M, T, CAP, SUBS, PUBS> 有五个参数：容量、订阅者上限、发布者上限。PubSubState：

struct PubSubState<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> {
    /// The queue contains the last messages that have been published and a countdown of how many subscribers are yet to read it
    queue: Deque<(T, usize), CAP>,
    /// Every message has an id.
    /// Don't worry, we won't run out.
    /// If a million messages were published every second, then the ID's would run out in about 584942 years.
    next_message_id: u64,
    /// Collection of wakers for Subscribers that are waiting.  
    subscriber_wakers: MultiWakerRegistration<SUBS>,
    /// Collection of wakers for Publishers that are waiting.  
    publisher_wakers: MultiWakerRegistration<PUBS>,
    /// The amount of subscribers that are active
    subscriber_count: usize,
    /// The amount of publishers that are active
    publisher_count: usize,
}

每条消息在队列里带一个倒计数：发布时等于当时的订阅者数，每个订阅者读一次减一。每个 Subscriber 自己记 next_message_id；队首消息的编号是 next_message_id - queue.len()，算出来就知道自己要读的那条在队列的第几个。

try_publish：没有订阅者时直接返回 Ok(())，消息被丢掉；队列满了返回 Err(message)；否则 push_back((message, subscriber_count))、编号加一、叫醒所有订阅者。Publisher::publish().await 在 Err 时登记到 publisher_wakers 等着。

get_message：

    fn get_message(&mut self, message_id: u64) -> Option<WaitResult<T>> {
        let start_id = self.next_message_id - self.queue.len() as u64;

        if message_id < start_id {
            return Some(WaitResult::Lagged(start_id - message_id));
        }

        let current_message_index = (message_id - start_id) as usize;

        if current_message_index >= self.queue.len() {
            return None;
        }

        // We've checked that the index is valid
        let queue_item = self.queue.iter_mut().nth(current_message_index).unwrap();

        // We're reading this item, so decrement the counter
        queue_item.1 -= 1;

        let message = if current_message_index == 0 && queue_item.1 == 0 {
            let (message, _) = self.queue.pop_front().unwrap();
            self.publisher_wakers.wake();
            // Return pop'd message without clone
            message
        } else {
            queue_item.0.clone()
        };

        Some(WaitResult::Message(message))
    }
  • 订阅者要读的编号比队首还小，说明它要的消息已经不在队列里了，返回 Lagged(差几条)，调用方把自己的编号加上这个数，下次从队首读。
  • 最后一个读者读的若是队首，消息被 pop_front 移出来直接交给它，不 clone，并叫醒等着的发布者。
  • 正常的 publish 永远不会让人 Lagged：队首那条只要还有一个订阅者没读，计数就不为 0，不会出队，队列满了发布者就等。能挤掉没读完的消息的只有 publish_immediate：满了先 pop_front，再发。ImmediatePublisher 不占 PUBS 的名额，也从不等待，所以中断处理函数里发布就用它。

订阅者 drop 时，unregister_subscriber 把它还没读的那些消息的计数都减一，再把队首计数为 0 的消息弹出，必要时叫醒发布者。否则一个被丢掉的订阅者会让队列永远卡满。

sync-std/src/bin/pubsub_lag.rs 第二段：PubSubChannel<_, u32, 2, 2, 1>，一个订阅者立刻读，另一个每条之前睡 100 ms，发布者用会等待的 publish：

published 0 at +0 ms
published 1 at +0 ms
  fast sub got Message(0) at 1 ms
  fast sub got Message(1) at 1 ms
  slow sub got Message(0) at 104 ms
published 2 at +102 ms
  fast sub got Message(2) at 104 ms
  slow sub got Message(1) at 208 ms
published 3 at +206 ms
  fast sub got Message(3) at 208 ms
  slow sub got Message(2) at 313 ms
published 4 at +311 ms
  fast sub got Message(4) at 313 ms
  slow sub got Message(3) at 418 ms
  slow sub got Message(4) at 523 ms

容量 2，前两条立刻发出去；第三条要等慢订阅者读完 0，0 才出队。之后发布者和快订阅者都被慢订阅者拖成每 100 ms 一条。这是 publish 的语义：所有人都要收到，最慢的人决定速度。

❗IMPORTANT
动脑筋：Lagged 几条

pubsub_lag.rs 第一段：

static PS: PubSubChannel<CriticalSectionRawMutex, u32, 4, 2, 1> = PubSubChannel::new();

let mut a = PS.subscriber().unwrap();
let mut b = PS.subscriber().unwrap();
let p = PS.immediate_publisher();
for v in 0..6 {
  p.publish_immediate(v);
}
println!("len after 6 publish_immediate: {}", PS.len());
println!("a: {:?}", a.try_next_message());
println!("a: {:?}", a.try_next_message());
两个订阅者都还没读，容量 4，立即发布 0 到 5。三行各打印什么？答案在本页末尾。

订阅者的内部类型 Sub（Subscriber 解引用得到它）实现了 Stream，它的 poll_next 遇到 Lagged 不返回，而是 wake_by_ref 再返回 Pending，下一次 poll 从队首继续，相当于自动跳过丢失的部分；想自己处理就用 next_message()，想忽略就用 next_message_pure()。

怎么选

需求用满 / 慢的时候
每条消息恰好一个接收者处理Channel发送方等（或 try_send 失败）
一个接收者，只要最新状态Signal新值覆盖旧值，不等
多个接收者，只要最新状态Watch新值覆盖旧值，不等；中间值被合并
多个订阅者，每条都要收到PubSubChannel + publish发布者等最慢的订阅者
多个订阅者，发布方不能等（中断）PubSubChannel + publish_immediate挤掉最旧的，慢订阅者收到 Lagged

动脑筋答案

len after 6 publish_immediate: 4
a: Some(Lagged(2))
a: Some(Message(2))
  • 容量 4，发到第 5 条（值 4）时队列满了，publish_immediate 弹出 0；发第 6 条时弹出 1。队列里剩 2、3、4、5，长度 4。
  • a 的编号还是 0，队首编号是 6 - 4 = 2，0 < 2，返回 Lagged(2)，a 的编号跳到 2。
  • 第二次读到 2，计数从 2 减到 1（b 还没读），所以 2 留在队列里，是 clone 出来的。b 接着读会得到 [Lagged(2), Message(2), Message(3), Message(4), Message(5)]（同一程序的下一行输出）。

小结

  • Signal：Cell<State>，None / Waiting(Waker) / Signaled(T) 三选一；signal 覆盖旧值；第二个等待者会把第一个换出来叫醒，两者互相 poll 直到有值，没拿到的那个睡下去。0.8.0 修了 reset 丢 waker 的问题。
  • Watch：一个 Option<T> 加 u64 版本号，send 叫醒全部 N 个槽；changed 比版本号，中间值合并；new_with 的初始值不算一次「变化」；AnonReceiver 不登记 waker、不占名额。
  • PubSubChannel：Deque<(T, 剩余读者数), CAP> 加全局 u64 编号；publish 等最慢的订阅者；publish_immediate 挤掉队首，慢订阅者得到 Lagged(n)；最后一个读者把消息移走不 clone；没有订阅者时发布的消息直接丢弃。

💬 关联讨论

发起讨论
加载中...