取舍与对比

主线位置：主线之外，回头看。宏展开一段的选择是「改写语法」，运行时一段的选择是「线程局部指针」，本章分别看它们换来了什么、付出了什么。

flowchart TB
  d1{"值怎么从 async 块递给消费者"}
  d1 -->|"通道"| ch["要分配、要同步，值经过缓冲区"]
  d1 -->|"线程局部指针（0.3.6）"| tl["值直接写进 poll_next 栈上的 Option"]
  tl --> rule["前提：send 只在本 stream 的 poll 里被 poll"]
  rule --> who["谁来保证：宏的改写规则"]
  who --> h106["#106：别的宏把 yield 挪进 async 块"]
  who --> h107["#107：Sender 协变"]
  d2{"yield 怎么认出来"}
  d2 -->|"等语言的 gen / async gen"| lang["stable 上还不能用"]
  d2 -->|"proc-macro 改写（0.3.6）"| pm["只认字面位置；报错是编译器原话"]

线程局部指针，而不是通道

「在 async 块里产出一串值」最直接的实现是通道：yield v 就是 tx.send(v).await，poll_next 就是 rx.poll_recv(cx)。async-stream 的名字 Sender / Receiver 还留着这个影子，但 yielder 里它们都是零大小类型，中间只有一个线程局部的裸指针。

这样做能省掉什么，可以直接量。同一个「依次产出 0 到 999 999」的流，分别用 stream!、futures::stream::unfold、mpsc + spawn + ReceiverStream 写，消费者用 fold 求和，全局分配器数堆分配次数：

fn with_macro() -> impl Stream<Item = u64> {
  stream! {
    for i in 0..N {
      yield i;
    }
  }
}

fn with_unfold() -> impl Stream<Item = u64> {
  futures::stream::unfold(0u64, |i| async move { (i < N).then_some((i, i + 1)) })
}

fn with_channel(cap: usize) -> impl Stream<Item = u64> {
  let (tx, rx) = mpsc::channel(cap);
  tokio::spawn(async move {
    for i in 0..N {
      if tx.send(i).await.is_err() {
        break;
      }
    }
  });
  ReceiverStream::new(rx)
}

async fn measure(name: &str, s: impl Stream<Item = u64>) {
  let before = ALLOCS.load(Relaxed);
  let started = Instant::now();
  let sum = s.fold(0u64, |acc, v| async move { acc + v }).await;
  let ns = started.elapsed().as_nanos() as f64 / N as f64;
  let allocs = ALLOCS.load(Relaxed) - before;
  println!("{name:<16} sum={sum} {ns:>6.2} ns/item  allocs={allocs}");
}

ALLOCS 是一个包住 System 的 #[global_allocator]，每次 alloc 加一。--release、#[tokio::main(flavor = "current_thread")]（单线程，分配计数里没有别的工作线程），M2 Max、rustc 1.99.0，三次中的一次：

stream!          sum=499999500000   3.96 ns/item  allocs=0
unfold           sum=499999500000   3.77 ns/item  allocs=0
mpsc(1)+spawn    sum=499999500000 12546.92 ns/item  allocs=1
mpsc(1024)+spawn sum=499999500000 137.35 ns/item  allocs=7815

三次的范围：stream! 3.94–3.97 ns，unfold 3.70–3.79 ns，两者都是 0 次分配；mpsc(1) 12.5 µs 左右，mpsc(1024) 134–138 ns、7815 次分配。换成多线程运行时，stream! 仍是 3.9 ns 左右，mpsc(1) 是 7.5–7.7 µs，mpsc(1024) 是 104–125 ns；那时分配计数会混进运行时工作线程自己的分配，stream! 一栏显示 2 到 5 次，不能当成流本身的分配。

几点读法：

• stream! 和 unfold 一个量级，都是 0 次分配：值从 async 块直接写进 poll_next 栈上的 Option，没有经过任何缓冲区，这和 poll_next 里读到的一致。
• 通道慢的主要原因不是分配：mpsc(1) 在计时期间只分配了 1 次（通道和任务在计时开始前就建好了），却慢了三个数量级。容量为 1 时，每一项都要让生产者任务挂起、消费者被唤醒，来回切换一次；这 12 微秒具体花在调度器的哪一步，本章没有再往下拆。容量加大后切换变少，每项开销降到一百多纳秒，但要为缓冲块分配内存。
• 这是「每项什么都不做」时的纯开销。每一项如果要等一次网络或数据库，几纳秒和一百纳秒的差别就看不出来了。选通道的理由应当是为什么需要它里说的「生产和消费要并发」，而不是性能。

换来的东西不只是快：值不跨任务，stream! 的块可以借用调用方的局部变量（通道 + spawn 要求 'static）；不需要运行时，async-stream 只依赖 futures-core 和 pin-project-lite。

付出的东西在另一边：通道的两端由类型系统连在一起，送错地方是不可能的；线程局部指针把「这个值该写进哪个槽」交给了「此刻是谁在 poll」。STORE 是 *mut ()，Send::poll 转回 *mut Option<T> 时不做任何检查，旁边那句 TODO: Ensure wakers match? 到 0.3.6 也没做。正确的前提是「send 只会在本 stream 的 poll_next 里被 poll」，这个前提由宏的改写规则、标签检查、Sender 的不变性和 unsafe fn pair 一起保证（yielder 的「类型安全靠什么」有一张表）。规则漏掉一种情况，结果不是逻辑错误，而是内存错误。0.3.6 修的两个洞都是这样来的。

0.3.6 修掉的两个洞

两个洞在同一个提交里修复（#109），0.3.6 的 CHANGELOG 只写了一行「Fix soundness bugs (#109)」。修复的 diff 很小：yielder.rs 改了两行，async-stream-impl/src/lib.rs 加了一个标签块和一句 break，运行时的逻辑一行没动。

#106：yield 被别的宏挪进了 async 块

tests/ui/unsoundness_issue_106.rs 的关键几行：

macro_rules! asynk {
    ($e:expr) => {
        async { $e }
    };
}

// ...
    pin!(stream! {
        let yield_42 = asynk!(yield 42_usize);
        let s = stream! {
            yield Box::new(12345);
            yield_42.await; // yield 42 -- wait that's not a Box!?
        };
        // ...

按用例注释里记录的输出，0.3.5 里这段代码能编译、能运行，直到段错误。问题出在三条规则叠在一起：

1. Scrub 会进宏调用的 token 找 yield（为了支持 select!），所以 asynk!(yield 42_usize) 里的 yield 被改写成了外层的 __yield_tx.send(42_usize).await。
2. asynk! 展开后把它包进了 async { }。Scrub 不进 async 块，指的是用户字面写下的 async 块；宏展开之后才出现的 async 块，它看不见。
3. 这个 async 块被交给内层 stream，在内层的 poll_next 里被 poll。此时 STORE 指向的是内层的槽。

sequenceDiagram
  participant O as 外层 poll_next
  participant I as 内层 poll_next
  participant T as STORE
  participant Y as yield_42：外层的 Send<usize>
  O->>T: 写入外层槽 Option<usize>
  O->>I: for await 拿下一项
  I->>T: 写入内层槽 Option<Box<i32>>
  I->>Y: 内层的块 await yield_42
  Y->>T: 读出指针，当成 Option<usize>
  Y->>Y: 把 Some(42) 写进内层槽
  I-->>O: 把槽里的东西当 Box<i32> 交出

Send<usize> 把指针当成 *mut Option<usize> 写入，可它指向的是一个 Option<Box<i32>>。本机实测两者大小分别是 16 和 8 字节：写入越过了槽的边界，内层再把这块内存当成 Box<i32> 交给消费者，用例注释里记录的结果是段错误。0.3.5 不在本机的 cargo 缓存里，这个段错误没有在本机复现；06 动手用自己写的最小 stream! 去掉检查后复现了同样的「送错槽」，两边都用 u32，只送错、不越界。

修法是给 async 块加一个标签，在每个 yield 改写结果里放一句永不执行的 break 到这个标签（第 128–136 行、第 237 行）。Rust 的标签穿不过函数、闭包和 async 块，所以只要 yield 最终落在别的 async 块或闭包里，不管是谁把它挪进去的，编译就失败。用例在 rustc 1.99.0 上的报错见 Scrub。

为什么是标签，而不是别的办法？源码和提交说明里没有讨论，下面是对照几种可能做法得出的分析，不是作者的原话：

标签检查不区分「真的会出错」和「其实没事」：tokio::join!(async { yield "a"; }, ..) 这种两个分支都在外层 stream 自己的 poll 里执行、槽也是同一个类型的写法，同样会被挡住。0.3.6 上的报错和改法见 04 常见坑；0.3.5 是否接受这种写法，本机没有 0.3.5，未实测。

#107：Sender 对 T 协变

这个洞和修法在 yielder 里已经完整走过一遍：Sender<T> 原来是 PhantomData<T>，对 T 协变，编译器允许把 Sender<&'long i32> 当成 Sender<&'short i32> 用，一个活得不够长的引用就能从发送端进去、从接收端以长生命周期出来。修法是把字段改成 PhantomData<fn(T) -> T>，对 T 不变。

共同点

两个洞的共同点是：yielder.rs 和 async_stream.rs 的运行时逻辑都没有错，错在「send 和 STORE 一定配对」这个前提的某一条保证漏了。#106 漏在改写规则（宏能把 yield 挪到改写者看不见的地方），#107 漏在类型（两端的 T 可以不完全相同）。把正确性交给宏的设计，审计时要看的就不是那 94 行 yielder.rs，而是宏所有可能的输入。

改写语法，而不是等语言

yield 在 stable 上不是合法语法，async-stream 让它「能用」的办法是在编译器看到它之前就把它改掉。这让 0.3.6 在 MSRV 1.65 的 stable 上就能用，代价有三个：

• 只认字面位置。 Scrub 只改写直接写在块里的 yield；隔着闭包、async 块、嵌套 fn 就不改写（Scrub 的「改写只看字面位置」有完整的表）。同一个 yield 搬进 for_each 的闭包，代码一个字没改，就从能用变成编译不过。
• 报错是编译器的原话。 没被改写的 yield 原样留给编译器，编译器报的是「yield syntax is experimental」「yield can only be used in #[coroutine] closures, or gen blocks」，还会建议你加 #[coroutine]，没有一条提到 stream!。标签检查的报错提到了 '__async_stream_private_check_scope，但也要知道它的来历才看得懂。04 常见坑把这几条报错逐条对回到宏的哪一步。
• 类型推断看到的是展开后的代码。 ? 被改写成 send(Err(e.into()))，into() 的目标类型只能从别处推出来，推不出时报的是 AsyncStream<Result<i32, _>, ..> 这种用户从没写过的类型。

另外，改写依赖 syn 的 full 和 visit-mut 两个 feature，第一次编译要把 syn 完整编一遍。本卷没有单独量这部分编译时间。

和其他写法比

「emitter 参数」这种做法在机制上值得一提：yield 变成了普通的方法调用加 .await，不需要改写语法，也就没有「只认字面位置」的问题；但递值时是否也走线程局部指针、闭包里能不能调用它，取决于具体实现，这里没法验证，不下结论。

原生 gen 与 async gen：本机实测

stable 1.99.0（edition 2024）上：

fn main() { let g = gen { yield 1; }; for x in g { println!("{x}"); } }

error[E0658]: gen blocks are experimental
 --> stable.rs:1:21
  |
1 | fn main() { let g = gen { yield 1; }; for x in g { println!("{x}"); } }
  |                     ^^^
  |
  = note: see issue #117078 <https://github.com/rust-lang/rust/issues/117078> for more information

error[E0658]: yield syntax is experimental
 --> stable.rs:1:27

async gen { yield 1; } 在 stable 上报的也是这两种。nightly 1.101.0（2026-10-05）打开 gen_blocks 后，两种块都能跑：

#![feature(gen_blocks, async_iterator)]
use std::async_iter::AsyncIterator;
use std::pin::pin;
use std::task::{Context, Poll, Waker};

fn main() {
  let g = gen {
    for i in 0..3 {
      yield i * 10;
    }
  };
  println!("gen: {:?}", g.collect::<Vec<_>>());

  let mut a = pin!(async gen {
    for i in 0..2 {
      let v = std::future::ready(i).await;
      yield v;
    }
  });
  let mut cx = Context::from_waker(Waker::noop());
  loop {
    match a.as_mut().poll_next(&mut cx) {
      Poll::Ready(Some(v)) => println!("async gen: Ready(Some({v}))"),
      Poll::Ready(None) => {
        println!("async gen: Ready(None)");
        break;
      }
      Poll::Pending => println!("async gen: Pending"),
    }
  }
}

gen: [0, 10, 20]
async gen: Ready(Some(0))
async gen: Ready(Some(1))
async gen: Ready(None)

gen 块是一个普通的 Iterator；async gen 块实现的是标准库里还没稳定的 AsyncIterator，这里手动 poll 它。它和 futures 的 Stream 是两个 trait，要交给 StreamExt 或 axum 的 Sse 还得有一层适配，本章没有写这层适配，未实测。

两种块的 yield 由编译器生成状态机，闭包、async 块里的 yield 有编译器自己的规则，也不需要线程局部指针递值。这些块何时稳定，本卷不作判断：上面只是 2026-10 本机 stable 和 nightly 的实测结果。

小结

• 线程局部指针换来了零分配、零同步：实测一百万项，stream! 每项约 4 ns、0 次分配，和 unfold 相当；通道慢在任务切换，容量 1 时每项约 12.5 µs。这只是空载开销，有真实 IO 时差别会被淹没。
• 代价是正确性压在宏的改写规则上：STORE 是 *mut ()，Send::poll 不做检查。0.3.6 修的 #106（别的宏把 yield 挪进 async 块）和 #107（Sender 协变）都是「保证漏了一条」，运行时逻辑没变。
• #106 的修法是一个标签加一句永不执行的 break：零运行时开销，借编译器「标签穿不过 async 块和闭包」的规则；副作用是一些本来不会出错的写法（比如 join! 里的 yield）也被挡住。
• proc-macro 改写让它在 stable 上可用，代价是只认字面位置、报错是编译器原话、推断看到的是展开后的类型。
• 本机 stable 1.99.0 上 gen / async gen 都报「gen blocks are experimental」；nightly 1.101.0 打开 gen_blocks 后都能跑，async gen 实现的是 AsyncIterator 而不是 Stream。async-fn-stream、futures-async-stream 本机没有，未实测。