async-stream结合anyhow返回Stream报非Send错误的原因是什么

核心结论 #

这个问题和anyhow完全无关，anyhow::Error不会捕获栈上的局部变量，问题根源是Rust异步代码的Send安全检查逻辑，以及async_stream生成状态机的特性。 #

具体原因拆解 #

• 前提知识：parking_lot::MutexGuard本身就不实现Send trait，这是锁的设计约束：锁守卫只能在加锁的线程内释放，跨线程传递会导致未定义行为。
• async_stream::try_stream!宏会把你写的代码转换为一个基于状态机的Stream实现：代码中的?（返回错误）、yield（返回成功值）都会作为状态机的挂起点，每次poll_next被调用时会从上次的挂起点继续执行。
• Rust对异步代码的Send安全检查有保守性：编译器会检查所有可能存活在挂起点之间的变量是否满足Send约束，当前版本的编译器不会做非常精细的控制流层面的drop位置判断，只会保守地认为变量只要定义在挂起点之前，就可能存活到挂起点之后。
• 你最初的代码中，_guard定义在Err(..)?这个挂起点之前，即使从同步代码的逻辑看?返回错误时已经退出了_guard的作用域，_guard应该被销毁，但编译器依旧会保守判定_guard可能被保存到状态机中跨挂起点存在，因此判定返回的Stream不满足Send约束，调用boxed()的时候就会报错（boxed()要求Stream是Send的）。
• 调整代码顺序后，_guard定义在Err(..)?之后，即使?触发错误返回，_guard还没有被创建，自然不会被存入状态机，也就不会触发Send检查错误。

可选修复方案 #

如果你需要保留先拿锁再返回错误的逻辑，可以手动显式销毁守卫，避免编译器误判：

{
    let mut _guard = a.try_lock().expect("aa");
    // 提前把错误保存下来，手动drop守卫之后再返回错误
    let res = Err(anyhow::anyhow!("asdf"));
    drop(_guard);
    res?;
}

或者把锁操作和错误处理封装到同步函数中，让守卫在函数返回前自动销毁：

fn operate_lock(a: &Arc<Mutex<()>>) -> anyhow::Result<()> {
    let mut _guard = a.try_lock().expect("aa");
    Err(anyhow::anyhow!("asdf"))
}

// 然后在try_stream里调用
operate_lock(&a)?;

内容的提问来源于stack exchange，提问作者mq7