Rust in Everything
基于 async-stream 0.3.6

运行时

01 宏展开的终点是一句 __yield_tx.send(v).await 和一个 AsyncStream::new(rx, async move { .. })。这一部分讲它们在运行时做什么。代码只有两个文件:yielder.rs 94 行、async_stream.rs 79 行。没有通道,没有锁,没有堆分配,值从 async 块到消费者只经过一个裸指针。

两章分别站在两端:

  1. yielder:线程局部指针:发送的一端。Sender、Receiver 都是零大小类型;thread_local! 的 STORE 存着当前槽的地址;Receiver::enter 写入、Enter drop 时恢复旧值,所以嵌套的 stream 互不干扰;Send::poll 把值写进槽后返回 Pending,不碰 waker;类型安全不靠 STORE,靠宏和 Sender 的不变性(#107 的修复)。
  2. AsyncStream::poll_next:接收的一端。槽 dst 放在 poll_next 的栈上,只在一次 generator.poll 期间登记;有值就返回值,哪怕 async 块同时结束了;done 与 FusedStream;yield 造成的 Pending 从不暴露给调用方,真正的等待由内层自己登记 waker;size_hint、!Unpin,以及 0.3.6 里 Sync 跟着内层 future 走。
sequenceDiagram
  participant C as 消费者
  participant A as AsyncStream::poll_next
  participant R as Receiver::enter / Enter
  participant T as STORE
  participant G as async 块
  participant S as Send::poll
  C->>A: poll_next(cx)
  A->>R: enter(&mut dst)
  R->>T: 存旧值,写 &mut dst
  A->>G: poll(cx)
  G->>S: send(v).await
  S->>T: 读出指针
  S->>S: 槽空就写入 v
  S-->>G: Pending
  G-->>A: Pending
  A->>R: Enter drop
  R->>T: 恢复旧值
  A-->>C: Ready(Some(v))

💬 关联讨论

发起讨论
加载中...