Scrub：改写 yield 与 ?

主线位置：宏展开的核心一步。stream_inner 把解析出来的每条语句交给 Scrub，Scrub 在语法树里找到 yield v，改写成 __yield_tx.send(v).await。主线上「yield 怎么递出值」的编译期一半就在这一步完成，运行时那一半从 yielder 开始。

flowchart TB
  e["visit_expr_mut(表达式)"] --> k{哪种表达式}
  k -->|Yield| y["先改写里面的表达式，再换成 { 标签检查; send(v).await }"]
  k -->|Try| t["先改写里面的表达式，再换成 match：Err 就 send(Err(e.into())) 后 return"]
  k -->|"Closure / Async"| skip["原样跳过"]
  k -->|ForLoop| f["带 #[await_] 就展开成 next 循环"]
  k -->|其他| rec["默认递归"]
  m["visit_macro_mut(宏调用)"] --> mk{"是 stream! / try_stream!"}
  mk -->|是| skip2["跳过"]
  mk -->|否| tok["在 token 里找 yield"]
  it["visit_item_mut(条目)"] --> im{"是宏调用"}
  im -->|是| m
  im -->|否| skip3["不进 fn、impl、mod"]

Scrub 是什么

Scrub 只有四个字段：

struct Scrub<'a> {
    /// Whether the stream is a try stream.
    is_try: bool,
    /// The unit expression, `()`.
    unit: Box<syn::Expr>,
    has_yielded: bool,
    crate_path: &'a TokenStream2,
}

is_try 区分两个宏；unit 是预先解析好的 ()，给不带值的 yield; 用；has_yielded 记录见没见过 yield，决定要不要插 dummy_yield；crate_path 是外壳传进来的 $crate，for await 展开时要用。

它实现了 syn 的 VisitMut。VisitMut 为语法树的每种节点提供一个 visit_*_mut 方法，默认实现只是递归访问子节点；Scrub 覆盖了其中三个：visit_expr_mut、visit_macro_mut、visit_item_mut。没覆盖的节点走默认递归，所以 Scrub 能走到 if、match、loop、let 里面的任何表达式。

yield：换成 send().await

visit_expr_mut 的 Yield 分支：

            syn::Expr::Yield(yield_expr) => {
                self.has_yielded = true;

                syn::visit_mut::visit_expr_yield_mut(self, yield_expr);

                let value_expr = yield_expr.expr.as_ref().unwrap_or(&self.unit);

                // ...
                let yield_expr = if self.is_try {
                    quote! { __yield_tx.send(::core::result::Result::Ok(#value_expr)).await }
                } else {
                    quote! { __yield_tx.send(#value_expr).await }
                };
                *i = syn::parse_quote! {
                    {
                        #[allow(unreachable_code)]
                        if false {
                            break '__async_stream_private_check_scope (loop {});
                        }
                        #yield_expr
                    }
                };
            }

按顺序：

记下见过 yield。
先递归改写 yield 后面的表达式。yield parse(s)? 里的 ? 在这一步被改写，所以 yield 和 ? 能写在同一行（0.3.3 修的 #66 就是这种情况）。
取出要发的值，yield; 发 ()。
try_stream! 里包一层 Ok(..)。这是 is_try 唯一起作用的地方。
整个 yield 表达式换成一个块：一句永不执行的 break，加上 __yield_tx.send(v).await。块的值是 send(v).await 的值 ()，和 yield 原来的位置兼容。

send 返回的 future 在运行时做什么，见 yielder：它把 v 写进消费者的槽，然后让 async 块在这里停一次。

那句永不执行的 break

#[allow(unreachable_code)]
if false {
    break '__async_stream_private_check_scope (loop {});
}

if false 保证它永远不执行，它存在只是为了让编译器检查一件事：这个 yield 是不是直接位于本 stream 的 async 块里，中间没有隔着别的 async 块或闭包。Rust 的标签不能穿过函数、闭包、async 块，如果 yield 被挪到了别的 async 块里，break 找不到标签，编译失败。

几个细节：

break 'label value 要求 value 的类型和标签块的类型一致。loop {} 的类型是 !，能转成任何类型，所以不管标签块是什么类型都能通过。
loop {} 之后的代码不可达，#[allow(unreachable_code)] 把这个警告压掉。
正常写的 yield 不可能跑到别的 async 块里：Scrub 根本不进 async 块（下面会讲）。能把它挪进去的只有别的宏：Scrub 在宏调用的 token 里也会改写 yield，但那个宏展开后可能把这段代码放进它自己的 async { }。0.3.6 之前这正是一个 soundness 洞（#106），仓库里的 tests/ui/unsoundness_issue_106.rs 用一个 macro_rules! asynk { ($e:expr) => { async { $e } } } 把 yield 42_usize 藏进 async 块，让它在另一个 stream! 的 poll 里执行，把一个 usize 写进了一个 Option<Box<i32>> 槽。

把这个用例抄进实验工程，rustc 1.99.0 的报错（节选）：

error[E0767]: use of unreachable label `'__async_stream_private_check_scope`
  --> src/bin/issue106.rs:14:10
   |
14 |       pin!(stream! {
   |  __________^
15 | |         let yield_42 = asynk!(yield 42_usize);
...
   = note: labels are unreachable through functions, closures, async blocks and modules

报错正文是编译器的原话：标签穿不过函数、闭包、async 块和模块。这个洞为什么能写坏内存、为什么修复选了标签而不是别的办法，放到 03 设计取舍；看到这条报错该怎么改，放到 04 常见坑。

?：Err 就发出去，然后 return

Try 分支：

            syn::Expr::Try(try_expr) => {
                syn::visit_mut::visit_expr_try_mut(self, try_expr);
                // ...
                let e = &try_expr.expr;

                *i = syn::parse_quote! {
                    match #e {
                        ::core::result::Result::Ok(v) => v,
                        ::core::result::Result::Err(e) => {
                            __yield_tx.send(::core::result::Result::Err(e.into())).await;
                            return;
                        }
                    }
                };
            }

普通 async 块里的 ? 会把错误作为 async 块的返回值交出去。但 stream! 的 async 块返回 ()，错误没处去，所以 Scrub 把 ? 改写成：Ok 就取出值；Err 就把 Err(e.into()) 当成流的一项发出去，然后 return 结束 async 块，流也就结束了。cargo expand 一个 try_stream! 的结果：

for t in input {
    let v: i32 = match t.parse() {
        ::core::result::Result::Ok(v) => v,
        ::core::result::Result::Err(e) => {
            __yield_tx
                .send(::core::result::Result::Err(e.into()))
                .await;
            return;
        }
    };
    {
        #[allow(unreachable_code)]
        if false {
            break '__async_stream_private_check_scope (loop {});
        }
        __yield_tx.send(::core::result::Result::Ok(v)).await
    };
}

注意这个分支没有看 is_try。stream! 里的 ? 也照样改写，只要流的 Item 是 Result，就能用：

const INPUT: [&str; 3] = ["1", "x", "3"];

// stream! 里也能写 `?`：Item 必须是 Result，Err 作为最后一项发出
fn plain() -> impl Stream<Item = Result<i32, ParseIntError>> {
  stream! {
    for t in INPUT {
      let v: i32 = t.parse()?;
      yield Ok(v);
    }
  }
}

// try_stream!：yield 的值自动包成 Ok
fn tried() -> impl Stream<Item = Result<i32, ParseIntError>> {
  try_stream! {
    for t in INPUT {
      let v: i32 = t.parse()?;
      yield v;
    }
  }
}

stream!:     [Ok(1), Err(ParseIntError { kind: InvalidDigit })]
try_stream!: [Ok(1), Err(ParseIntError { kind: InvalidDigit })]

两个宏的结果完全一样："x" 解析失败，Err 作为第二项发出，然后流结束，"3" 根本没被处理。两个宏的区别只在 yield 要不要你自己写 Ok(..)。

由此推出几条：

? 之后流就结束了。 改写里是 return，不是 continue。想「记下错误、接着处理下一项」，就别用 ?，自己 match 然后 yield Err(e)。

? 只认 Result。 改写写死了 Result::Ok / Result::Err，对 Option 用 ? 会类型不匹配。rustc 1.99.0：

error[E0308]: mismatched types
 --> src/main.rs:4:12
  |
4 |     let _s = stream! {
  |  ____________^
5 | |     let v: Option<u32> = Some(1);
6 | |     let x = v?;
  | |             - this expression has type `Option<u32>`
7 | |     yield x;
8 | |   };
  | |___^ expected `Option<u32>`, found `Result<_, _>`

e.into() 让错误类型能转换，就像普通的 ? 一样，但也带来推断问题：错误类型如果只出现在 ? 里，编译器推不出 into() 的目标。这条报错放到 04 常见坑。
用户自己写的 return; 也一样能提前结束流。

闭包和 async 块：不进去

            syn::Expr::Closure(_) | syn::Expr::Async(_) => {
                // Don't transform inner closures or async blocks.
            }

这个分支（第 153–155 行）什么都不做，也不递归。理由是语义上的：

闭包里的 ? 属于闭包自己，应当从闭包返回，改写成「发出去再 return」就错了。
闭包和 async 块里的 yield 即使改写成 send(v).await，也不在本 stream 的 async 块里执行：闭包可能被存起来以后再调，async 块可能被交给别处 poll。

所以这些位置的 yield 原样留给编译器。stable 上编译器不认 yield，报的是语言本身的错误，比如 tests/ui/yield_in_async.rs 在本机 rustc 1.99.0 上的第一条：

error[E0658]: yield syntax is experimental
 --> src/bin/yield_in_async.rs:6:13
  |
6 |             yield 123;
  |             ^^^^^^^^^

报错里看不出和 stream! 有关。三个 yield_in_* 用例的完整报错和改法放到 04 常见坑。

宏调用：在 token 里找 yield

syn 不解析宏调用的参数，tokio::select! { .. } 在语法树里只是「一个路径 + 一串 token」。如果 Scrub 只走语法树，select! 分支里的 yield 就漏掉了。0.3.2（#57）起，visit_macro_mut 会进到宏调用的 token 里：

    fn visit_macro_mut(&mut self, mac: &mut syn::Macro) {
        let mac_ident = mac.path.segments.last().map(|p| &p.ident);
        if mac_ident.map_or(false, |i| i == "stream" || i == "try_stream") {
            return;
        }

        self.visit_token_stream(&mut mac.tokens);
    }

嵌套的 stream! / try_stream! 跳过：它们里面的 yield 属于内层的 stream，等内层宏自己展开时再改写。其他宏的 token 交给 visit_token_stream_impl 逐个扫描：

    while let Some(tt) = tokens.next() {
        match tt {
            TokenTree::Ident(i) if i == "yield" => {
                let stream = std::iter::once(TokenTree::Ident(i)).chain(tokens).collect();
                match syn::parse2(stream) {
                    Ok(Partial(yield_expr, rest)) => {
                        let mut expr = syn::Expr::Yield(yield_expr);
                        visitor.visit_expr_mut(&mut expr);
                        expr.to_tokens(out);
                        *modified = true;
                        tokens = rest.into_iter().peekable();
                    }
                    Err(e) => {
                        out.append_all(e.to_compile_error().into_iter());
                        *modified = true;
                        return;
                    }
                }
            }
            TokenTree::Ident(i) if i == "stream" || i == "try_stream" => {
                // ... 原样输出 `stream ! { .. }`，不进去
            }
            TokenTree::Group(group) => {
                // ... 递归进括号组
            }
            other => out.append(other),
        }
    }

遇到标识符 yield，就从这里开始解析一个 ExprYield。Partial<T> 是一个小技巧：先解析出一个 T，剩下的 token 原样留着。解析出来的 yield 表达式交给 visit_expr_mut，走的就是上面的 Yield 分支，然后接着扫剩下的 token。
yield 后面不是合法表达式时，把 syn 的错误变成 compile_error! 放进输出（tests/ui/yield_bad_expr_in_macro.rs 测的就是这条路）。
遇到 stream / try_stream 后面紧跟 ! 和一个组，原样输出，不进去。
括号组递归进去。

用 tokio::select! 合并两个通道，yield 写在分支的处理部分：

fn merged(mut a: mpsc::Receiver<u32>, mut b: mpsc::Receiver<u32>) -> impl Stream<Item = String> {
  stream! {
    loop {
      tokio::select! {
        Some(v) = a.recv() => yield format!("a{v}"),
        Some(v) = b.recv() => yield format!("b{v}"),
        else => break,
      }
    }
  }
}

两个通道各放一个值后关闭，收集结果并排序：

["a1", "b2"]

cargo expand 里，select! 展开后的 match 分支中，yield 已经是带标签检查的 send().await 了：

match output {
    __tokio_select_util::Out::_0(Some(v)) => {
        #[allow(unreachable_code)]
        if false {
            break '__async_stream_private_check_scope (loop {});
        }
        __yield_tx.send(v).await
    }
    // ...
}

select! 的处理部分在 select! 自己的 poll_fn 之外、直接在外层 async 块里执行，所以标签可达，编译通过。select! 本身的机制见 tokio 卷 select! 与 join!。

token 里只找 yield，不找 ?

visit_token_stream_impl 只认 yield 这个标识符，不认 ?。所以宏调用参数里的 ? 不会被改写，留下来的是普通的 ?：

fn parsed() -> impl Stream<Item = Result<i32, ParseIntError>> {
  try_stream! {
    // `?` 写在宏调用的参数里
    println!("parsed {}", "x".parse::<i32>()?);
    yield 1;
  }
}

rustc 1.99.0：

error[E0277]: the `?` operator can only be used in an async block that returns `Result` or `Option` (or another type that implements `FromResidual`)
  --> src/bin/q_in_macro.rs:8:45
   |
 6 | /   try_stream! {
 7 | |     // `?` 写在宏调用的参数里
 8 | |     println!("parsed {}", "x".parse::<i32>()?);
   | |                                             ^ cannot use the `?` operator in an async block that returns `()`

例外是 yield 后面那个表达式里的 ?：它会随 yield 一起交给 visit_expr_mut，被正常改写。把 select! 分支写成 v = async { t.parse::<i32>() } => yield v?,，对输入 ["1", "x"] 的结果是：

[Ok(1), Err(ParseIntError { kind: InvalidDigit })]

想在宏参数里用 ?，先在宏外面 let x = expr?;，再把 x 传进去。

嵌套的 fn：不进去

visit_item_mut 覆盖了条目（fn、struct、impl、mod ……）的访问：

    fn visit_item_mut(&mut self, i: &mut syn::Item) {
        // Recurse into macros but otherwise don't transform inner items.
        if let syn::Item::Macro(i) = i {
            self.visit_macro_mut(&mut i.mac);
        }
    }

块里定义的函数和 stream 没有关系，它的 yield 和 ? 都不该改写，所以只处理条目位置上的宏调用，其余一概不进。tests/ui/yield_in_nested_fn.rs 测的就是这条：嵌套 fn 里的 yield 原样留下，编译器报 yield syntax is experimental。

改写只看字面位置

把三种「不进去」放在一起，规则是：Scrub 只改写字面上直接写在 stream! 块里的 yield 和 ?，中间可以隔着 if、match、loop 和其他宏调用，不能隔着闭包、async 块和条目。这不是运行时的判断，是语法上的：同一个 yield 搬进一个闭包，代码一个字没改，就从「能用」变成「编译不过」。

位置	`yield`	`?`
块里、`if` / `match` / `loop` 里	改写	改写
别的宏调用的参数里	改写（之后再由标签检查兜底）	不改写，除非在 `yield` 的表达式里
嵌套的 `stream!` / `try_stream!` 里	不改写，留给内层宏	不改写，留给内层宏
闭包、async 块里	不改写，编译器报错	不改写，属于闭包或 async 块自己
嵌套的 `fn` 里	不改写，编译器报错	不改写，属于那个函数

小结

Scrub 是 syn 的 VisitMut，覆盖了 visit_expr_mut、visit_macro_mut、visit_item_mut 三个方法。
yield v 改写成 { if false { break '__async_stream_private_check_scope (loop {}); } __yield_tx.send(v).await }；try_stream! 里发 Ok(v)。那句 break 让编译器检查 yield 没有被别的宏挪进 async 块或闭包，这是 0.3.6 修 #106 的办法。
? 改写成 match：Err 时 send(Err(e.into())).await 然后 return。这一步不看 is_try，stream! 里的 ? 也一样，实测两个宏都产出 [Ok(1), Err(..)] 后结束。
闭包、async 块、嵌套 fn 不进去，里面的 yield 留给编译器报 yield syntax is experimental。
宏调用的 token 里只找 yield，所以 select! 分支里能 yield；但宏参数里的 ? 不改写，实测报 E0277。