手撕 Rust 异步执行器：从 Future trait 到 Waker 机制的完整实现

Rust 的异步编程模型与其他语言有本质区别 —— 它采用零成本抽象，将调度权完全交给开发者。理解这一机制的最佳方式不是阅读 Tokio 源码，而是亲手实现一个最小化的异步执行器。本文将从 Future trait 的定义出发，逐步构建包含 Spawner、Executor、Task 和 Waker 的完整运行时，揭示异步任务如何在单线程环境下协作调度。

Future Trait：异步计算的基石

Rust 的异步核心是一个极其简洁的 trait 定义：

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

pub enum Poll<T> {
    Ready(T),
    Pending,
}

poll 方法是整个异步系统的入口点。它接收两个参数：Pin<&mut Self> 确保 Future 在内存中的位置不会被移动（这对自引用结构至关重要），Context 则携带了与执行器通信的 Waker。返回值只有两种状态：Ready(T) 表示任务完成，Pending 表示需要等待外部事件。

这里的 Pin 不是可选的语法糖，而是内存安全的保障。当 Future 内部持有指向自身的指针（如异步函数中的局部变量引用）时，移动 Future 会导致悬垂指针。Pin 通过类型系统禁止这种移动，确保自引用结构在多次 poll 调用间保持有效。

Waker：连接异步事件与执行器的桥梁

Waker 是 Rust 异步模型中最精妙的设计。它是一个线程安全的句柄，允许异步事件源（定时器、I/O 完成通知）在准备就绪时通知执行器重新调度任务。

pub struct Waker {
    // 内部使用 RawWakerVTable 实现类型擦除
}

impl Waker {
    pub fn wake(self) { /* 通知执行器 */ }
    pub fn wake_by_ref(&self) { /* 不消耗所有权版本 */ }
}

Waker 的核心语义是：当 Future 返回 Pending 时，它必须将 cx.waker() 克隆并存储；当外部事件就绪时，调用该 waker 的 wake() 方法，将任务重新推入执行队列。这一机制解耦了事件源与执行器 —— 定时器不需要知道 Executor 的实现细节，只需持有 Waker 即可。

最小化执行器的组件设计

一个可运行的执行器需要四个核心组件协同工作：

Task：Future 的包装器，包含可轮询的状态和重新入队的能力

struct Task {
    future: Mutex<Option<BoxFuture<'static, ()>>>,
    task_sender: SyncSender<Arc<Task>>,
}

使用 Mutex 包裹 Future 是为了满足编译器的线程安全检查。在生产环境中，单线程执行器可以使用 UnsafeCell 避免锁开销。BoxFuture 是 Pin<Box<dyn Future<Output = T> + Send>> 的别名，将具体 Future 类型擦除为动态分发对象。

Spawner：任务创建器，将 Future 装箱并推入执行队列

impl Spawner {
    fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) {
        let boxed = future.boxed();
        let task = Arc::new(Task {
            future: Mutex::new(Some(boxed)),
            task_sender: self.task_sender.clone(),
        });
        self.task_sender.send(task).unwrap();
    }
}

Executor：事件循环主体，从队列取出任务并轮询

impl Executor {
    fn run(&self) {
        while let Ok(task) = self.ready_queue.recv() {
            let mut future_slot = task.future.lock().unwrap();
            if let Some(mut future) = future_slot.take() {
                let waker = waker_ref(&task);
                let context = &mut Context::from_waker(&waker);
                
                if future.as_mut().poll(context).is_pending() {
                    *future_slot = Some(future); // 未完成，放回槽位
                }
            }
        }
    }
}

ArcWake：将 Task 转换为 Waker 的桥梁

impl ArcWake for Task {
    fn wake_by_ref(arc_self: &Arc<Self>) {
        let cloned = arc_self.clone();
        arc_self.task_sender.send(cloned).unwrap();
    }
}

当 Future 调用 waker.wake() 时，实际触发的是 wake_by_ref，它将 Task 的克隆重新推入通道，等待 Executor 的下一次轮询。

完整执行流程：以定时器为例

让我们追踪一个 TimerFuture 的完整生命周期：

impl Future for TimerFuture {
    type Output = ();
    
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.shared_state.lock().unwrap();
        if state.completed {
            return Poll::Ready(());
        }
        state.waker = Some(cx.waker().clone()); // 存储 waker
        Poll::Pending
    }
}

1. 创建阶段：Spawner 将 TimerFuture 装箱，包装为 Arc<Task>，推入 sync_channel
2. 首次轮询：Executor 从通道接收 Task，创建 Waker，调用 poll。TimerFuture 发现时间未到，存储 Waker 并返回 Pending，Future 被放回 Mutex 槽位
3. 等待阶段：独立线程休眠指定时长，期间 Executor 可以处理其他任务
4. 唤醒阶段：定时器线程完成，调用存储的 waker.wake()，触发 ArcWake::wake_by_ref，Task 被重新推入通道
5. 完成阶段：Executor 再次轮询，TimerFuture 返回 Ready(())，Future 从槽位取出后不再放回，Task 完成

生产环境的演进方向

手撕执行器揭示了核心机制，但生产级运行时（Tokio、async-std）在以下维度做了大量优化：

工作窃取（Work Stealing）：多线程执行器使用本地任务队列 + 全局队列的混合架构。线程优先消费本地队列，空闲时从其他线程 "窃取" 任务，减少锁竞争。

I/O 驱动集成：真实运行时整合 epoll/kqueue/IOCP，将文件描述符就绪事件与 Waker 绑定。AsyncFd 类型封装了这种注册 / 唤醒机制。

内存布局优化：使用 slab 分配器管理 Task，避免 Arc 的引用计数开销。Waker 的 vtable 设计允许无分配唤醒。

协作式调度：通过 yield_now() Future 让出执行权，防止单个任务长时间占用线程。

调试与诊断要点

实现自定义执行器时，常见问题包括：

• Waker 丢失：Future 返回 Pending 但未存储 Waker，导致永久挂起。应在 poll 入口处断言 cx.waker() 已被保存
• Pin 违规：尝试移动被 Pin 的 Future。使用 Pin::as_mut() 获取可变引用，而非解引用
• 线程安全：Waker 需实现 Send + Sync，确保可从任意线程唤醒。使用 sync_channel 而非 mpsc::channel 保证线程安全

理解这些底层机制后，阅读 Tokio 源码将不再是黑箱操作。当你看到 runtime::scheduler 模块时，会认出这与手撕版本相同的 Spawner/Executor/Task 三元组，只是经过了工业级优化。

参考来源

• Asynchronous Programming in Rust - Build an Executor
• Async Runtimes - redixhumayun.github.io

systems