AsyncStream::poll_next
主线位置：运行时的接收端，也是整条主线的终点。消费者每次要下一项，都调到这里；它准备好槽、poll 宏生成的 async 块，再看槽里有没有
Send::poll写进来的值。
flowchart TB
start["poll_next(cx)"] --> d0{"done"}
d0 -->|是| none0["Ready(None)，不碰 async 块"]
d0 -->|否| slot["栈上 dst = None；enter(&mut dst)"]
slot --> poll["generator.poll(cx)"]
poll --> leave["Enter drop，STORE 恢复"]
leave --> setd["done = 这次 poll 是否 Ready"]
setd --> has{"dst 里有值"}
has -->|有| some["Ready(Some(v))"]
has -->|没有| d1{"done"}
d1 -->|是| none1["Ready(None)"]
d1 -->|否| pend["Pending：内层在等 IO 或定时器"]
结构
pin_project! {
#[doc(hidden)]
#[derive(Debug)]
pub struct AsyncStream<T, U> {
rx: Receiver<T>,
done: bool,
#[pin]
generator: U,
}
}
impl<T, U> AsyncStream<T, U> {
#[doc(hidden)]
pub fn new(rx: Receiver<T>, generator: U) -> AsyncStream<T, U> {
AsyncStream {
rx,
done: false,
generator,
}
}
}
三个字段：零大小的接收端、一个结束标志、宏生成的 async 块（字段名还叫 generator，是 async-stream 早年的叫法）。只有 generator 标了 #[pin]：pin-project-lite 的 project() 给它一个 Pin<&mut U>，给另外两个普通的 &mut。
整个 AsyncStream 就是这些，没有缓冲区，没有通道，也没有装箱。stream! { yield 1u8; } 实测 size_of_val 是 4 字节，async 块的状态机直接存在里面。
poll_next 逐行
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let me = self.project();
if *me.done {
return Poll::Ready(None);
}
let mut dst = None;
let res = {
let _enter = me.rx.enter(&mut dst);
me.generator.poll(cx)
};
*me.done = res.is_ready();
if dst.is_some() {
return Poll::Ready(dst.take());
}
if *me.done {
Poll::Ready(None)
} else {
Poll::Pending
}
}
- 已经结束就直接返回
None，不再 poll async 块。async 块完成之后再 poll 它会 panic（「async fnresumed after completion」），done挡住了这一点。 - 槽
dst是poll_next的局部变量，在栈上，每次调用都是新的。它的地址只通过enter写进STORE。 _enter守卫只活在内层的块里：enter写入地址，generator.poll(cx)执行 async 块，块结束时_enterdrop，STORE恢复成进来之前的值。指向dst的指针只在这一次poll期间可见，poll_next返回之前就已经撤掉了，不会留下悬空指针。cx原样交给 async 块。async 块里如果在等真正的 IO 或定时器，用的就是这个cx登记 waker。done记录 async 块是否完成。- 先看槽：有值就返回
Ready(Some(v))，不管 async 块这次是Pending还是Ready。 - 槽是空的：async 块完成了就
Ready(None)；没完成就Pending，这时 async 块一定是在等别的东西，而那个东西已经用cx登记了 waker。
对照 Send::poll：yield 让 async 块返回 Pending，但同时槽里有了值，于是走第 6 步，调用方看到的是 Ready(Some(v))。yield 造成的 Pending 从不传出 poll_next，这就是 Send 可以不登记 waker 的原因。
实验：数一数唤醒
用一个会计数的 waker 手动 poll 一个 stream：先 yield 1，再睡 20ms，再 yield 2。每次 Pending 之后不 poll，只等 50ms，看 waker 被叫了几次。
struct CountWake(AtomicUsize);
impl ArcWake for CountWake {
fn wake_by_ref(arc_self: &Arc<Self>) {
arc_self.0.fetch_add(1, Ordering::SeqCst);
}
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
let counter = Arc::new(CountWake(AtomicUsize::new(0)));
let waker = waker(counter.clone());
let mut cx = Context::from_waker(&waker);
let mut s = pin!(stream! {
yield 1;
tokio::time::sleep(Duration::from_millis(20)).await;
yield 2;
});
for n in 1..=6 {
let r = s.as_mut().poll_next(&mut cx);
println!(
"poll #{n}: {:<16} wakes={} size_hint={:?}",
format!("{r:?}"),
counter.0.load(Ordering::SeqCst),
s.size_hint()
);
if r.is_pending() {
// 不再 poll，只让时间过去；看是谁叫醒了我们
tokio::time::sleep(Duration::from_millis(50)).await;
println!(" after 50ms: wakes={}", counter.0.load(Ordering::SeqCst));
}
if let Poll::Ready(None) = r {
// 再 poll 一次：done 已经置位，不会再碰 async 块
let again = s.as_mut().poll_next(&mut cx);
println!("poll #{}: {again:?} (fused)", n + 1);
break;
}
}
}
输出（跑 3 次相同）：
poll #1: Ready(Some(1)) wakes=0 size_hint=(0, None)
poll #2: Pending wakes=0 size_hint=(0, None)
after 50ms: wakes=1
poll #3: Ready(Some(2)) wakes=1 size_hint=(0, None)
poll #4: Ready(None) wakes=1 size_hint=(0, Some(0))
poll #5: Ready(None) (fused)
sequenceDiagram participant C as 手动 poll participant A as poll_next participant G as async 块 participant T as tokio 定时器 C->>A: poll #1 A->>G: poll G-->>A: Pending（send 写入 1） A-->>C: Ready(Some(1))，没有登记 waker C->>A: poll #2 A->>G: poll G->>T: Sleep 登记 cx 里的 waker G-->>A: Pending，槽空 A-->>C: Pending T-->>C: 20ms 后 wake，wakes=1 C->>A: poll #3 A->>G: poll G-->>A: Pending（send 写入 2） A-->>C: Ready(Some(2)) C->>A: poll #4 A->>G: poll G-->>A: Ready(()) A-->>C: Ready(None)，done = true
逐行对应：
- #1：async 块跑到
yield 1，Send写入、返回Pending，没碰 waker。poll_next返回Ready(Some(1))，唤醒次数是 0。 - #2：async 块从
yield 1之后继续，Send这次Ready；接着sleep返回Pending，并用传进来的cx向 tokio 的定时器登记了 waker。槽是空的，poll_next返回Pending。50ms 后唤醒次数变成 1：叫醒我们的是定时器，不是 async-stream。 - #3：睡醒了，跑到
yield 2，同 #1。 - #4：
yield 2之后块结束，Ready(())，槽空，done = true，返回Ready(None)。size_hint从这一刻起变成(0, Some(0))。 - #5：
done已经置位，第 1 步直接返回None，没有再 poll 完成了的 async 块（否则会 panic）。
整个过程 waker 只被叫过一次，就是 sleep 那一次。yield 不产生任何唤醒，也不需要：消费者拿到 Ready(Some(v)) 之后，想要下一项自然会再 poll。
值和结束在同一次 poll
第 6 步先看槽、再看 done，所以如果同一次 poll 里 async 块既写了值又结束了，这次返回值，下次才返回 None。宏生成的代码里 send 写完值后 async 块一定停在那个 .await 上，造不出这种情况；用 __private 手写一个「展开结果」就可以，只为观察这个分支：
// 宏不会生成这种代码，这里直接用 __private 手写一个「展开结果」
let (mut tx, rx) = unsafe { async_stream::__private::yielder::pair::<u32>() };
let mut s = pin!(async_stream::__private::AsyncStream::new(rx, async move {
let send = Box::pin(tx.send(7));
let _ = select(send, ready(())).await; // send 先被 poll：写入 7、返回 Pending；ready 随即完成
println!(" [block] finished in the same poll");
}));
let mut cx = Context::from_waker(noop_waker_ref());
println!("poll #1: {:?} is_terminated={}", s.as_mut().poll_next(&mut cx), s.is_terminated());
println!("poll #2: {:?}", s.as_mut().poll_next(&mut cx));
[block] finished in the same poll
poll #1: Ready(Some(7)) is_terminated=true
poll #2: Ready(None)
第一次 poll：futures::future::select 先 poll send，7 写进槽；再 poll ready(())，立即完成，async 块随之结束。poll_next 里 done 已经是 true，但槽里有 7，先返回 7。值不会丢。注意此时 is_terminated() 已经是 true，而这一项刚刚才交出去，下一节说这意味着什么。
FusedStream 与 size_hint
impl<T, U> FusedStream for AsyncStream<T, U>
where
U: Future<Output = ()>,
{
fn is_terminated(&self) -> bool {
self.done
}
}
// ...
fn size_hint(&self) -> (usize, Option<usize>) {
if self.done {
(0, Some(0))
} else {
(0, None)
}
}
FusedStream：is_terminated就是done。futures::select!之类的组合子靠它跳过已经结束的流。正常的宏代码里，done变成true的那次 poll 一定返回的是None（槽一定是空的），所以is_terminated()为真时后面确实不会再有值。上一节手写的特殊情况是唯一的例外。size_hint：async 块里有多少个yield在运行前不可知，所以没结束时只能说「至少 0 个、上限未知」，结束后说「正好 0 个」。这个实现是 0.3.1（#40）加的，在那之前用的是Stream的默认实现，始终是(0, None)。实验里 #4 之后size_hint变成(0, Some(0))，和这段代码一致。
!Unpin、Send、Sync 与 Debug
AsyncStream 自己没有手写任何 auto trait 的实现，全部跟着字段走，而字段里唯一有意思的是 async 块 U：
Unpin：pin_project!生成的Unpin实现要求#[pin]字段U: Unpin。async 块不是Unpin（它的状态里可能有指向自己局部变量的引用），所以stream!返回的流是!Unpin，调用StreamExt::next之前要先pin!/pin_mut!/Box::pin。消费方式本身会 pin 的（collect、for_each、for await）则不用。忘了 pin 时的报错放到 04 常见坑。Send：async 块跨.await持有的变量都是Send，流就是Send。Sender/Receiver是PhantomData，PhantomData<fn(T) -> T>总是Send + Sync，PhantomData<T>跟着T。Sync：同样跟着 async 块走。块里跨yield持有一个Cell，流就不是Sync：rustc 1.99.0（节选）：let s = stream! { let c = Cell::new(1); yield c.get(); }; need_send(&s); // Cell 是 Send：通过 need_sync(&s); // Cell 不是 Sync：0.3.6 里 AsyncStream 跟着内层 future 走error: future cannot be shared between threads safely --> src/main.rs:14:3 | 14 | need_sync(&s); // Cell 不是 Sync：0.3.6 里 AsyncStream 跟着内层 future 走 | ^^^^^^^^^^^^^ future created by async block is not `Sync` | = help: within `AsyncStream<i32, {async block@lib.rs:193:9}>`, the trait `Sync` is not implemented for `Cell<i32>`need_send那一行没有报错。其实&AsyncStream上能做的事只有size_hint、is_terminated和Debug，都不碰 async 块，按理可以无条件Sync。master 上的 #115 就是这么做的：把 async 块包进一个只给&mut访问的SyncWrapper，AsyncStream就不再因为 async 块而失去Sync（同一个提交还给SyncWrapper手写了Debug）。这个改动不在 0.3.6 里。多数场景只要求Send（比如tokio::spawn、axum 的响应体），不受影响；要把流放进要求Sync的地方，0.3.6 下得让块里跨yield的变量都是Sync。Debug：#[derive(Debug)]要求U: Debug，而 async 块没有实现Debug，所以stream!返回的流实际上不能{:?}打印。rustc 1.99.0 的报错开头是：error[E0277]: `{async block@lib.rs:193:9}` doesn't implement `Debug`
小结
AsyncStream只有三个字段：零大小的Receiver、done、#[pin]的 async 块；没有缓冲区，stream! { yield 1u8; }实测 4 字节。poll_next：done就返回None；否则在栈上放dst，enter登记地址，poll async 块，Enterdrop 撤掉地址；有值返回Ready(Some(v))，没值看 async 块是否完成。yield造成的Pending从不传出poll_next，所以Send不登记 waker；实测整个流程只有sleep唤醒过一次。- 有值优先：同一次 poll 里写了值又结束，先返回值，下次返回
None。 FusedStream::is_terminated就是done；size_hint只有(0, None)和(0, Some(0))。- 流是
!Unpin，Send/Sync跟着 async 块走；0.3.6 里块跨yield持有Cell就不是Sync（master 的 #115 改了，未发布）；Debug实际不可用。