2. 手动实现 Future trait 的简单状态机

为什么需要手动实现 Future

async/await 是语法糖，编译器会把 async 块展开成一个实现了 Future 的状态机。要理解异步运行时如何驱动任务，最直接的方式就是自己写一个 Future：在 poll 中先返回 Poll::Pending，把 Waker 保存下来，等条件满足后再调用 waker.wake()，让执行器重新 poll 一次并返回 Poll::Ready。

本例实现一个模拟延时任务 Delay：它持有一个截止时间，第一次 poll 时若未到期就返回 Pending 并保存 waker；到期后返回 Ready。为了不依赖真实定时器，用一个后台线程在到期时唤醒 waker，模拟运行时的事件通知。

实现 Delay：保存 waker 并返回 Pending

核心结构体持有 deadline 和 Option<Waker>。第一次 poll 时若未到期，把 cx.waker().clone() 存入 waker，并启动一个后台线程在剩余时间后调用 wake()；到期后返回 Poll::Ready(())。

use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};

struct Delay {
    deadline: Instant,
    waker: Arc<Mutex<Option<Waker>>>,
}

impl Delay {
    fn new(after: Duration) -> Self {
        Delay {
            deadline: Instant::now() + after,
            waker: Arc::new(Mutex::new(None)),
        }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if Instant::now() >= self.deadline {
            return Poll::Ready(());
        }

        // 保存 waker，供后台线程唤醒
        let mut slot = self.waker.lock().unwrap();
        *slot = Some(cx.waker().clone());
        drop(slot);

        // 启动一次后台唤醒线程（简化：每次 poll 都启动，实际应做去重）
        let waker = Arc::clone(&self.waker);
        let remaining = self.deadline.saturating_duration_since(Instant::now());
        thread::spawn(move || {
            thread::sleep(remaining);
            if let Some(w) = waker.lock().unwrap().take() {
                w.wake();
            }
        });

        Poll::Pending
    }
}

用极简执行器驱动它

标准库没有提供公开的执行器，所以写一个最小 block_on：循环 poll，遇到 Pending 就阻塞等待 waker 被调用，被唤醒后继续 poll，直到 Ready。

use std::sync::{Arc, Condvar, Mutex};
use std::task::{Wake, Waker};

struct ThreadWaker {
    pair: Arc<(Mutex<bool>, Condvar)>,
}

impl Wake for ThreadWaker {
    fn wake(self: Arc<Self>) {
        let (lock, cvar) = &*self.pair;
        *lock.lock().unwrap() = true;
        cvar.notify_one();
    }
}

fn block_on<F: Future>(fut: F) -> F::Output {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let waker = Waker::from(Arc::new(ThreadWaker { pair: Arc::clone(&pair) }));
    let mut cx = Context::from_waker(&waker);

    let mut fut = Box::pin(fut);
    loop {
        match fut.as_mut().poll(&mut cx) {
            Poll::Ready(v) => return v,
            Poll::Pending => {
                let (lock, cvar) = &*pair;
                let mut woken = lock.lock().unwrap();
                while !*woken {
                    woken = cvar.wait(woken).unwrap();
                }
                *woken = false;
            }
        }
    }
}

fn main() {
    let start = Instant::now();
    block_on(Delay::new(Duration::from_millis(300)));
    println!("elapsed ≈ {:?}", start.elapsed());
}

运行与验证

• 运行后应看到 elapsed ≈ 300ms 左右，说明 Pending 期间线程确实被阻塞，而不是忙等。
• 把 Delay::new 的时长改为 Duration::ZERO，第一次 poll 就应直接返回 Ready，不进入等待分支。
• 若把 w.wake() 注释掉，程序会永久阻塞在 cvar.wait，这正好验证了「没有 waker 唤醒，执行器不会重新 poll」这一底层机制。