yielder：线程局部指针

主线位置：运行时的发送端。宏把 yield v 改写成了 __yield_tx.send(v).await（见 Scrub），本章讲这个 send 返回的 future 被 poll 时做了什么：从线程局部变量里取出一个指针，把 v 写过去，返回 Pending。接收端那个指针从哪来，见下一章 poll_next。

flowchart TB
  pair["unsafe pair()"] --> tx["Sender：零大小"]
  pair --> rx["Receiver：零大小"]
  rx -->|"enter(&mut dst)"| store["thread_local STORE：一个擦掉类型的裸指针"]
  tx -->|"send(v)"| fut["Send { value: Some(v) }"]
  fut -->|poll| read["读 STORE，转回 Option 的指针"]
  store --> read
  read --> w{"槽是空的"}
  w -->|是| put["写入 v"] --> pend["Pending"]
  w -->|否| pend
  fut -->|"再 poll，value 已取走"| ready["Ready(())"]

两个零大小的端点

yielder.rs 第 8–32 行：

#[derive(Debug)]
pub struct Sender<T> {
    _p: PhantomData<fn(T) -> T>,
}

#[derive(Debug)]
pub struct Receiver<T> {
    _p: PhantomData<T>,
}

// ...

// Note: It is considered unsound for anyone other than our macros to call
// this function. This is a private API intended only for calls from our
// macros, and users should never call it, but some people tend to
// misinterpret it as fine to call unless it is marked unsafe.
#[doc(hidden)]
pub unsafe fn pair<T>() -> (Sender<T>, Receiver<T>) {
    let tx = Sender { _p: PhantomData };
    let rx = Receiver { _p: PhantomData };
    (tx, rx)
}

名字叫 Sender / Receiver，看起来像通道，其实两边都只有一个 PhantomData，什么数据都不存，彼此之间也没有任何连接。它们的作用只在类型上：把「这个 stream 的 Item 是 T」这件事同时钉在发送端和接收端。实测：

size_of Sender<String> = 0
size_of Receiver<String> = 0

pair 没有任何不安全的操作，却标成了 unsafe fn。注释写得很直白：除了宏，谁调用它都是 unsound 的，不标 unsafe 就总有人以为能随便调。原因在后面：值能安全地送到，靠的是宏保证 send 只会在对应的那个 stream 的 poll 里被 poll。

Sender 为什么是 PhantomData<fn(T) -> T>

0.3.5 里 Sender 的字段是 PhantomData<T>，0.3.6 改成了 PhantomData<fn(T) -> T>（v0.3.5..v0.3.6 的 diff 里 yielder.rs 只改了这一行和 STORE 的初始化）。区别在型变：PhantomData<T> 对 T 协变，fn(T) -> T 对 T 不变。

协变意味着编译器可以把 Sender<&'long i32> 当成 Sender<&'short i32> 用。于是一对 pair::<&'long i32>() 里，接收端承诺产出 &'long i32，发送端却能塞进一个只活到 'short 的引用。仓库里的 tests/ui/unsoundness_issue_107.rs（#107）就是这样把一个已经释放的 Vec 的引用带出了作用域：

    let mut outer = vec![];
    {
        let v = vec![0; 10];
        let v_ref = &v;
        let mut s = pin!(stream! {
            for x in v_ref {
                yield x
            }
        });
        while let Some(x) = s.next().await {
            outer.push(x);
        }
    };
    // use-after-free
    println!("{outer:?}"); // […garbage allocator internals…, 0, 0, 0]

改成不变之后，两端的 T 必须完全相同，T 只能是 &'short i32，outer 存不下它。抄进实验工程，rustc 1.99.0 的报错（节选）：

error[E0597]: `v` does not live long enough
  --> src/bin/issue107.rs:11:21
   |
10 |         let v = vec![0; 10];
   |             - binding `v` declared here
11 |         let v_ref = &v;
   |                     ^^ borrowed value does not live long enough
...
20 |     };
   |     - `v` dropped here while still borrowed
21 |     // use-after-free
22 |     println!("{outer:?}"); // […garbage allocator internals…, 0, 0, 0]
   |                ----- borrow later used here

这就是全部修复：一个 PhantomData 的写法。0.3.6 的 CHANGELOG 只写了「Fix soundness bugs (#109)」，#107 和 #106 两个洞是一起修的，另一个见 Scrub 的标签检查。两者为什么是同一类问题，放到 03 设计取舍。

STORE：一个线程局部的裸指针

第 34–37 行：

// Tracks the pointer to `Option<T>`.
//
// TODO: Ensure wakers match?
thread_local!(static STORE: Cell<*mut ()> = const { Cell::new(ptr::null_mut()) });

• 每个线程一个，存一个指向 Option<T> 的指针。T 每个 stream 都不一样，线程局部变量不能是泛型的，所以存成擦掉类型的 *mut ()，用的时候再转回来。
• 初始值是空指针。const { .. } 初始化（0.3.6 加的）让它不需要运行时的惰性初始化。tokio 的 CONTEXT 也是这样，见 tokio 卷线程本地上下文。
• 这个指针只在一次 poll_next 内部有效。在两次 poll 之间，即使 async 块所在的任务被调度到另一个线程，也没有关系：每次 poll_next 都在当前线程重新写一次。
• 注释里的 TODO: Ensure wakers match? 到 0.3.6 都还在。它问的是：Send::poll 要不要确认自己拿到的 cx 和设置 STORE 的那次 poll 是同一个？现在不检查，下面会看到为什么平时不需要。

Send：写进槽，然后 Pending

Sender::send 和它返回的 future：

impl<T> Sender<T> {
    pub fn send(&mut self, value: T) -> impl Future<Output = ()> {
        Send { value: Some(value) }
    }
}

struct Send<T> {
    value: Option<T>,
}

impl<T> Unpin for Send<T> {}

impl<T> Future for Send<T> {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {
        if self.value.is_none() {
            return Poll::Ready(());
        }

        STORE.with(|cell| {
            let ptr = cell.get() as *mut Option<T>;
            let option_ref = unsafe { ptr.as_mut() }.expect("invalid usage");

            if option_ref.is_none() {
                *option_ref = self.value.take();
            }

            Poll::Pending
        })
    }
}

send 本身什么都不做，只把值装进 Send。一个 Send 被 poll 时有三种情况：

正常的 yield 只会走第一行和第三行：第一次 poll 写入、挂起；poll_next 把值交给消费者；消费者下次要值时再 poll，Send 发现值已经交出去了，Ready，async 块继续往下跑，直到下一个 yield 或结束。第二行是防御性的：宏改写出来的永远是 send(v).await，构造完立刻 await，同一次 poll 里不会有两个 send 都被 poll 到；万一出现，后一个值不会覆盖前一个，而是留在 Send 里等下次 poll。

几个细节：

• _cx 没用。 Send 返回 Pending 却不登记 waker，违反了 Future 的常规约定（返回 Pending 时要安排好唤醒，见 tokio 卷 poll 与唤醒）。这里不会挂死，因为这个 Pending 永远到不了执行器：poll_next 看到槽里有值，返回的是 Ready(Some(v))。下一章用一个数唤醒次数的 waker 实测这一点。

• impl Unpin for Send<T> 是手写的：Send 从不把 T pin 住，只是 take() 出来移走，所以不管 T 是不是 Unpin，Send 都可以是 Unpin，poll 里才能用 self.value 拿到 &mut。0.3.1 的 #50（「Allow yielding !Unpin values」）正是把原来的 impl<T: Unpin> Future for Send<T> 改成了这一行。

• expect("invalid usage")：STORE 是空指针时 panic。只有在 stream 的 poll 之外 poll 了 Send 才会这样，宏生成的代码做不到这一点。手动调用 unsafe 的 pair 就能触发：

  Copy

  let (mut tx, _rx) = unsafe { async_stream::__private::yielder::pair::<u32>() };
  tx.send(1).await;
  

  Copy

  thread 'main' (2070795) panicked at .../async-stream-0.3.6/src/yielder.rs:63:54:
  invalid usage
  

  （路径前缀已省略。）yielder.rs:63 就是那句 expect("invalid usage")。

Receiver::enter 与 Enter：写入，再恢复

接收端只有一个方法，第 76–94 行：

impl<T> Receiver<T> {
    pub(crate) fn enter<'a>(&'a mut self, dst: &'a mut Option<T>) -> Enter<'a, T> {
        let prev = STORE.with(|cell| {
            let prev = cell.get();
            cell.set(dst as *mut _ as *mut ());
            prev
        });

        Enter { _rx: self, prev }
    }
}

// ===== impl Enter =====

impl<'a, T> Drop for Enter<'a, T> {
    fn drop(&mut self) {
        STORE.with(|cell| cell.set(self.prev));
    }
}

enter 把 dst 的地址写进 STORE，同时记下原来的值；返回的守卫 Enter 在 drop 时把原值写回去。Enter 借着 &'a mut Receiver<T> 和 &'a mut Option<T>，所以守卫活着的时候，dst 不能被别人碰，也不能被移走。

「记下原值、drop 时恢复」是为了嵌套。一个 stream 的 async 块里 poll 另一个 stream（比如 for await），调用栈是这样的：

flowchart TB
  o1["外层 poll_next：STORE = 外层槽"] --> o2["poll 外层 async 块"]
  o2 --> i1["内层 poll_next：记下外层槽，STORE = 内层槽"]
  i1 --> i2["poll 内层 async 块：yield 写进内层槽"]
  i2 --> i3["Enter drop：STORE = 外层槽"]
  i3 --> o3["外层 async 块拿到内层的值，yield 写进外层槽"]
  o3 --> o4["Enter drop：STORE = 原来的值"]

内层的 yield 发生在内层 enter 和它的 Enter drop 之间，写的是内层的槽；内层 poll 一返回，STORE 已经指回外层的槽，外层接下来的 yield 写的是外层的槽。这就是一个按调用栈后进先出的「当前槽」。实测内层 yield "inner-a"; yield "inner-b";，外层 for await 再 yield format!("outer saw {v}")：

["outer saw inner-a", "outer saw inner-b"]

类型安全靠什么

Send::poll 把 STORE 里的 *mut () 直接转成 *mut Option<T>，没有任何检查。这一步正确的前提是：被 poll 的这个 Send<T> 和设置 STORE 的那个 Receiver<T> 来自同一次 pair()，否则就是把一个类型的值写进另一个类型的槽。yielder.rs 自己不保证这一点，它是由几处一起保证的：

在这些前提下，「当前 STORE 指向谁」和「正在执行哪个 async 块」总是一致的，所以也就不需要再核对 waker，那个 TODO 一直没做。代价是：正确性不在 yielder.rs 这 94 行里，而散在宏的改写规则上，宏的任何一条规则出现漏洞，都会直接变成内存安全问题。0.3.6 修的两个洞就是这样来的。

小结

• Sender、Receiver 都只有一个 PhantomData，实测大小为 0；它们只在类型上把两端的 T 绑在一起。pair 标成 unsafe fn，只许宏调用。
• Sender 用 PhantomData<fn(T) -> T> 让它对 T 不变，这是 #107 的修复，unsoundness_issue_107.rs 在 rustc 1.99.0 上报 E0597。
• STORE 是 thread_local! 的 Cell<*mut ()>，指向当前正在 poll 的那个 stream 的槽，只在一次 poll_next 内有效。
• Send::poll：槽空就写入、返回 Pending；值已交出就 Ready(())。它不登记 waker，因为这个 Pending 不会传到执行器。STORE 为空时 panic invalid usage。
• Receiver::enter 写入地址并记下旧值，Enter drop 时恢复，嵌套的 stream 各写各的槽。
• *mut () 转回 *mut Option<T> 不做检查，类型安全靠宏的改写规则、标签检查、Sender 的不变性、卫生性和 unsafe fn pair 一起保证。