揭秘Rust异步的“心跳”：Poll机制与状态机转换

2301_81549453

372人浏览 · 2025-10-30 18:37:48

2301_81549453 · 2025-10-30 18:37:48 发布

揭秘Rust异步的“心跳”：Poll机制与状态机转换 🤖

当我们写下 async fn 时，我们并不仅仅是在定义一个函数。我们实际上是在定义一个状态机（State Machine）。而 Poll 机制，就是驱动这个状态机运转的唯一“心跳”或“引擎”。

async/await 语法糖极大地简化了异步代码的编写，但它隐藏了底层的核心抽象：Future trait。t。

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

这，就是 Rust 异步的全部。

Future：一个代表“未来某个时刻会完成的工作”的类型。
poll 方法：执行器（Executor）用来“询问” Future 状态的****方法。
Poll<T> 枚举：poll 的返回值，只有两种可能：
** Poll::Pending: “我现在还没准备好，请（通过 Waker）稍后_再来 poll 我。”
- `Poll:Ready(T): “我完成了！这是你的结果 T`。”

💡 核心解读：`async` 如何变成状态机？

Rust 编译器是真正的魔法师。当你写下这样的 async 函数时：

async fn check_database_and_network(db_query: String) -> u32 {
    // 状态 A
    let db_result = database::query(db_query).await; // <- 暂停点 1
    
    // 状态 B
    let network_result = network::fetch(db_result).await; // <- 暂停点 2
    
    // 状态 C
    network_result.status_code
}

编译器会将其“解糖”（Desugar）成一个类似这样的状态机结构体：

// 伪代码，展示编译器的思路
enum CheckDatabaseState {
    // 初始状态，持有初始参数
    A { db_query: String },
    
    // 正在等待数据库
    B { db_future: impl Future<Output = DbResult> },
    
    // 正在等待网络
    C { network_future: impl Future<Output = NetworkResult> },
    
    // 最终状态
    Terminated,
}

struct CheckDatabaseFuture {
    state: CheckDatabaseState,
}

这个 CheckDatabaseFuture 会自动为你实现 Future trait。

**而 poll 方法，就是这个状态 match 语句！**

// 伪代码，展示 poll 的核心逻辑
impl Future for CheckDatabaseFuture {
    type Output = u32;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // 'loop' 确保状态机尽可能地向前推进
        loop {
            match self.state {
                // 状态 A: 初始状态
                State::A { db_query } => {
                    // 创建第一个 future
                    let db_future = database::query(db_query);
                    // 转换到下一个状态
                    self.state = State::B { db_future };
                    // 继续 loop，立即 poll State::B
                }
                
                // 状态 B: 等待数据库
                State::B { ref mut db_future } => {
                    // 轮询子 future
                    match db_future.poll(cx) {
                        Poll::Ready(db_result) => {
                            // 数据库准备好了！
                            // 创建第二个 future
                            let network_future = network::fetch(db_result);
                            // 转换到下一个状态
                            self.state = State::C { network_future };
                            // 继续 loop，立即 poll State::C
                        }
                        Poll::Pending => {
                            // 数据库没准备好，整个状态机也必须暂停
                            return Poll::Pending;
                        }
                    }
                }

                // 状态 C: 等待网络
                State::C { ref mut network_future } => {
                    match network_future.poll(cx) {
                        Poll::Ready(network_result) => {
                            // 网络准备好了！
                            let status_code = network_result.status_code;
                            // 转换到最终状态
                            self.state = State::Terminated;
                            // 返回最终结果！
                            return Poll::Ready(status_code);
                        }
                        Poll::Pending => {
                            // 网络没准备好，暂停
                            return Poll::Pending;
                        }
                    }
                }
                
                State::Terminated => {
                    panic!("Future polled after completion");
                }
            }
        }
    }
}

🚀 深度实践：手动实现一个带状态的 `Future`

让我们通过实践来内化这个知识。我们将手动实现一个 Future，它会“连接”两个 Future：fut1 和 fut2。fut2 的输入依赖于 fut1 的输出。

这正是 `async { fut2(fut1.wait).await }` 的手动实现。

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

// 我们将手动实现 'async { fut2(fut1.await).await }'
// F1 必须是 Future, F2 必须是返回 Future 的函数
struct Then<F1, F2, Fut2>
where
    F1: Future,
    F2: FnOnce(F1::Output) -> Fut2, // FnOnce 因为它只被调用一次
    Fut2: Future,
{
    // 我们的状态机
    state: State<F1, F2, Fut2>,
}

// 状态机的枚举
enum State<F1, F2, Fut2>
where
    F1: Future,
    F2: FnOnce(F1::Output) -> Fut2,
    Fut2: Future,
{
    // 状态1：等待第一个 Future (F1)
    WaitingOnFirst { fut1: F1, func: F2 },
    
    // 状态2：等待第二个 Future (Fut2)
    WaitingOnSecond { fut2: Fut2 },
    
    // 状态3：完成
    Done,
}

// 辅助函数，用于创建我们的状态机
fn my_then<F1, F2, Fut2>(fut1: F1, func: F2) -> Then<F1, F2, Fut2>
where
    F1: Future,
    F2: FnOnce(F1::Output) -> Fut2,
    Fut2: Future,
{
    Then {
        state: State::WaitingOnFirst { fut1, func },
    }
}


impl<F1, F2, Fut2> Future for Then<F1, F2, Fut2>
where
    F1: Future + Unpin, // 为简单起见，我们用 Unpin。否则需要 `Pin::new_unchecked`
    F2: FnOnce(F1::Output) -> Fut2 + Unpin,
    Fut2: Future + Unpin,
{
    type Output = Fut2::Output;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // loop 确保我们完成 fut1 后立即 poll fut2
        loop {
            // 我们需要对 state 进行可变借用
            let state = &mut self.state;

            match state {
                State::WaitingOnFirst { fut1, func } => {
                    // Pin<&mut F1>
                    let pin_fut1 = Pin::new(fut1); 
                    
                    match pin_fut1.poll(cx) {
                        Poll::Ready(output1) => {
                            // F1 完成！
                            // 我们需要取出 func，但这会使当前状态失效
                            // 所以我们先用 State::Done 替换
                            let old_state = std::mem::replace(state, State::Done);
                            
                            // 从旧状态中提取 func
                            let func = match old_state {
                                State::WaitingOnFirst { func, .. } => func,
                                _ => unreachable!(),
                            };

                            // 调用 func 来创建 fut2
                            let fut2 = func(output1);
                            
                            // 转换到新状态：WaitingOnSecond
                            *state = State::WaitingOnSecond { fut2 };
                            
                            // ⭐️ 关键：继续 loop，立即 poll fut2
                        }
                        Poll::Pending => {
                            // F1 没准备好，我们必须等待
                            return Poll::Pending;
                        }
                    }
                }

                State::WaitingOnSecond { fut2 } => {
                    let pin_fut2 = Pin::new(fut2);
                    
                    match pin_fut2.poll(cx) {
                        Poll::Ready(output2) => {
                            // F2 完成！
                            *state = State::Done; // 转换到最终状态
                            return Poll::Ready(output2); // 返回最终结果
                        }
                        Poll::Pending => {
                            // F2 没准备好，等待
                            return Poll::Pending;
                        }
                    }
                }
                
                State::Done => {
                    panic!("Poll after completion");
                }
            }
        }
    }
}

🧠 专业思考：`Pin` 与自引用

你可能注意到了 poll 中的 `Pin<&mutlf>。为什么需要 Pin`？

答案是：状态机是“自引用”的。

在编译器生成的真实状态机中，变量的生命周期需要跨越 .await。例如：

async fn self_ref() {
    let data = vec![1, 2, 3];
    let data_slice = &data[..]; // 引用
    
    // .await
    some_other_future(data_slice).await; 
}

编译器生成的 Future 结构体必须同时存储 data (Vec) 和 data_slice (&u8u8)。data_slice 指向了同一结构体中的 data 字段。

如果这个 Future 结构体在内存中被移动（Move），data 的地址会改变，但 `data_slice仍然指向旧地址，导致悬垂指针！

Pin 是 Rust 对这个问题的回答。它是一种类型级的“固定”保证。poll 接收 Pin<&mut Self>，向编译器承诺：“这个 Future 实例的内存地址不会改变了”，因此在它内部的自引用（如 data_slice）是安全的。