for await
主线位置：宏展开里的一条支线。
for await x in s { .. }让一个 stream 消费另一个 stream，它本身不产出值，但循环体里的yield仍由 Scrub 改写。本章讲它的两步改写：先在 token 层换掉，再在语法树上展开。
flowchart TB
src["for await v in s { body }"] --> tok["replace_for_await：token 层"]
tok --> attr["#[await_] for v in s { body }"]
attr --> parse["syn 解析：普通的 ExprForLoop，带一个属性"]
parse --> visit["Scrub 的 ForLoop 分支：先改写 body"]
visit --> check{"恰好一个属性且是 await_"}
check -->|否| keep["当普通 for 循环，保持不动"]
check -->|是| exp["展开：遮蔽 + Pin::new_unchecked + loop/next().await"]
第一步：token 层换成 #[await_] for
for await v in s 不是合法的 Rust 语法，syn 不可能把它解析成表达式。宏的办法是先在 token 层把它改成一个合法的形状：给 for 循环加一个属性。原本想用的是 #[await]，但 await 是关键字，syn 2 不肯把它当属性名解析，于是用 await_（第 282–284 行）：
// syn 2.0 wont parse `#[await] for x in xs {}`
// because `await` is a keyword, use `await_` instead
const AWAIT_ATTR_NAME: &str = "await_";
replace_for_await 在 parse_input 里、语法解析之前运行：
fn replace_for_await(input: impl IntoIterator<Item = TokenTree>) -> TokenStream2 {
let mut input = input.into_iter().peekable();
let mut tokens = Vec::new();
while let Some(token) = input.next() {
match token {
TokenTree::Ident(ident) => {
match input.peek() {
Some(TokenTree::Ident(next)) if ident == "for" && next == "await" => {
let next_span = next.span();
let next = syn::Ident::new(AWAIT_ATTR_NAME, next_span);
tokens.extend(quote!(#[#next]));
let _ = input.next();
}
_ => {}
}
tokens.push(ident.into());
}
TokenTree::Group(group) => {
// ... 递归进括号组，保留原 span
}
_ => tokens.push(token),
}
}
tokens.into_iter().collect()
}
看到标识符 for 后面紧跟标识符 await，就先输出 #[await_]，吞掉 await，再输出 for。属性名 await_ 借用了原来 await 的 span，报错时能指回你写的 await。
这是纯 token 操作，不知道语法结构，后果有两个，后面都会实测：
- 它会进到所有括号组里，包括 async 块和闭包。但
Scrub不进 async 块和闭包，那里的#[await_] for不会被展开。 - 它把属性插在
for紧前面。如果for前面有标签，结果是'label: #[await_] for，属性跑到了标签和循环中间。
第二步：ForLoop 分支展开
语法树上，#[await_] for v in s { .. } 是一个带一个属性的普通 ExprForLoop。Scrub 的 ForLoop 分支：
syn::Expr::ForLoop(expr) => {
syn::visit_mut::visit_expr_for_loop_mut(self, expr);
// TODO: Should we allow other attributes?
if expr.attrs.len() != 1 || !expr.attrs[0].meta.path().is_ident(AWAIT_ATTR_NAME) {
return;
}
let syn::ExprForLoop {
attrs,
label,
pat,
expr,
body,
..
} = expr;
attrs.pop().unwrap();
let crate_path = self.crate_path;
*i = syn::parse_quote! {{
let mut __pinned = #expr;
let mut __pinned = unsafe {
::core::pin::Pin::new_unchecked(&mut __pinned)
};
#label
loop {
let #pat = match #crate_path::__private::next(&mut __pinned).await {
::core::option::Option::Some(e) => e,
::core::option::Option::None => break,
};
#body
}
}}
}
- 先递归改写循环的各部分，循环体里的
yield、?在这一步改写。普通的for循环也走这一步，所以普通for里的yield照样能用。 - 不带属性、或者属性不止一个、或者那个属性不是
await_，就当普通for循环，到此为止。源码里留着一句TODO: Should we allow other attributes?：for await前面再加别的属性（比如#[allow(..)]），它就不认了。 - 是
for await就拆开，换成一个块：把被迭代的 stream 存进__pinned，用同名变量遮蔽成一个Pin<&mut _>，再用loop+next().await逐个取值，取到None就break。原来的标签#label放在loop前面，所以循环体里的break 'label/continue 'label还指向这个循环。
一个 stream! 里套 for await 消费另一个 stream!，cargo expand 的输出（节选，内层 stream! 的展开省略）：
'__async_stream_private_check_scope: {
let inner = { /* 内层 stream! 的展开 */ };
{
let mut __pinned = inner;
let mut __pinned = unsafe {
::core::pin::Pin::new_unchecked(&mut __pinned)
};
loop {
let v = match ::async_stream::__private::next(&mut __pinned)
.await
{
::core::option::Option::Some(e) => e,
::core::option::Option::None => break,
};
{
{
#[allow(unreachable_code)]
if false {
break '__async_stream_private_check_scope (loop {});
}
__yield_tx.send(v + 10).await
};
}
}
}
}
Pin::new_unchecked 为什么成立
Pin::new_unchecked 是 unsafe 的：调用方要保证被指向的值从此不再移动，直到它被 drop。这里的保证来自两点：
- 同名遮蔽。 第二个
let mut __pinned遮蔽了第一个，此后没有任何代码能再以值的方式拿到原来的 stream，也就没法把它移走。这和futures::pin_mut!、std::pin::pin!用的是同一个思路。另外，__pinned这个名字来自quote!，用户代码因为卫生性看不见它（见外壳）。 - 它住在 async 块的状态里。
next(..).await会让 async 块挂起，__pinned跨越了.await，所以它是 async 块状态机的一个字段。async 块被 poll 时本身已经被 pin 住了，它的字段也就不会再移动。
所以 for await 能直接消费 !Unpin 的 stream，比如另一个 stream!，用户不用自己先 pin!。上面这个例子（内层 yield "inner-a"; yield "inner-b";，外层 yield format!("outer saw {v}")）的输出：
["outer saw inner-a", "outer saw inner-b"]
自己写的 next
__private::next 就是 StreamExt::next：
// This is equivalent to the `futures::StreamExt::next` method.
// But we want to make this crate dependency as small as possible, so we define our `next` function.
#[doc(hidden)]
pub fn next<S>(stream: &mut S) -> impl Future<Output = Option<S::Item>> + '_
where
S: Stream + Unpin,
{
Next { stream }
}
Next::poll 只有一行 Pin::new(&mut self.stream).poll_next(cx)。它要求 S: Unpin，而传进去的是 &mut Pin<&mut T>，Pin<&mut T> 总是 Unpin，所以对任何 stream 都成立。自己写这 30 行，是为了让 async-stream 只依赖 futures-core（只有 trait 定义），不必拉进整个 futures-util。
实测：两个边界
带标签的 for await 编译不过
源码里的 #label 看起来支持标签，但标签写在 for await 前面时，第一步的 token 替换已经把语法弄坏了：
stream! {
'outer: for await a in rows() {
for await b in rows() {
if b > a {
continue 'outer;
}
yield (a, b);
}
}
}
rustc 1.99.0：
error: expected loop or block expression
--> src/bin/label_fail.rs:6:3
|
6 | / stream! {
7 | | 'outer: for await a in rows() {
8 | | for await b in rows() {
9 | | if b > a {
... |
15 | | }
| |___^
|
= note: this error originates in the macro `$crate::__private::stream_inner` which comes from the expansion of the macro `stream` (in Nightly builds, run with -Z macro-backtrace for more info)
replace_for_await 产出的是 'outer: #[await_] for a in rows()，标签后面必须紧跟循环或块，属性夹在中间，syn 解析失败，报错就是 syn 的原话。反过来，如果手写成 #[await_] 'outer: for a in rows()，syn 能解析，ForLoop 分支也会把 'outer 放到 loop 前面，实测输出：
[(0, 0), (1, 0), (1, 1), (2, 0), (2, 1), (2, 2)]
这说明展开逻辑本身支持标签，坏在 token 替换那一步。但 await_ 是内部名字，master 上 syn 已经升到 3（#119），不应该依赖它。需要标签时，自己 pin 再用 while let：
stream! {
'outer: for a in 0..3u32 {
// 不用 for await：自己 pin、自己 next().await，标签照常用
let mut inner = pin!(rows());
while let Some(b) = inner.next().await {
if b > a {
continue 'outer;
}
yield (a, b);
}
}
}
输出相同。这里带标签的外层循环是普通 for，没有问题；出错的只是写在 for await 前面的标签。
async 块里的 for await
第一步会进到 async 块里替换，第二步不会进去展开，于是 async 块里留下一个没人处理的 #[await_] for：
fn sum<S: Stream<Item = u32> + Unpin>(input: S) -> impl Stream<Item = u32> {
stream! {
let fut = async move {
let mut total = 0;
for await v in input {
total += v;
}
total
};
yield fut.await;
}
}
rustc 1.99.0 报三条错（节选）：
error[E0658]: attributes on expressions are experimental
...
error: cannot find attribute `await_` in this scope
--> src/main.rs:8:11
|
8 | for await v in input {
| ^^^^^
...
error[E0277]: `S` is not an iterator
--> src/main.rs:8:22
|
8 | for await v in input {
| ^^^^^ `S` is not an iterator
第二条的位置指向 await，就是因为 await_ 借了 await 的 span。三条报错都在说那个残留的属性和普通 for 循环，没有一条提到「async 块里不支持 for await」。for await 和 yield 一样只在 stream! 块的字面位置上有效；async 块里要消费 stream，用 StreamExt::next 写 while let。
小结
for await分两步改写：replace_for_await在 token 层把for await换成#[await_] for（await是关键字，syn 2 不认#[await]）；Scrub的ForLoop分支再把恰好带一个await_属性的for展开。- 展开结果：
let mut __pinned = s; let mut __pinned = unsafe { Pin::new_unchecked(&mut __pinned) }; loop { match next(&mut __pinned).await { Some(e) => e, None => break } .. }。同名遮蔽保证原值拿不到，__pinned又住在已经 pin 住的 async 块里，所以new_unchecked成立，for await能直接消费!Unpin的 stream。 __private::next是自己写的StreamExt::next，为了只依赖futures-core。- 0.3.6 里
'label: for await编译不过（expected loop or block expression），因为属性被插在标签后面；需要标签就自己 pin 再while let。 - async 块里的
for await只被替换、不被展开，报的是「attributes on expressions are experimental」「cannot find attribute」这类看不出和for await有关的错误。