Rust 异步运行时：从 Future 到 Waker

Rust 的异步编程模型以其零成本抽象著称，但背后的 Future 状态机、Waker 唤醒机制和 Reactor 事件循环却鲜有人深入理解。本文将自底向上实现一个迷你 Tokio，彻底揭开 Rust 异步运行时的神秘面纱。

一、Future 状态机

Rust 编译器将每个 async 函数或块转换为一个匿名的 Future 状态机。每个 .await 点对应状态机的一个状态转换。Future 通过 poll 方法驱动执行，返回 Poll::Ready 或 Poll::Pending。

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

// 手动实现一个 Future：延迟执行
struct DelayFuture {
    delay_ms: u64,
    start: Option,
}

impl Future for DelayFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let now = std::time::Instant::now();
        if self.start.is_none() {
            self.start = Some(now);
        }
        if now.duration_since(self.start.unwrap()).as_millis() >= self.delay_ms as u128 {
            Poll::Ready(())
        } else {
            // 注册 waker，当时间到达时唤醒
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

二、Waker 与 Reactor 事件循环

Waker 是 Future 与 Executor 之间的桥梁。当 Future 返回 Pending 时，它需要将 Waker 注册到某个事件源（如 epoll、定时器）。当事件就绪时，Waker 被调用，通知 Executor 重新 poll 该 Future。

use std::sync::{Arc, Mutex};
use std::collections::VecDeque;

struct MiniTokio {
    ready_queue: Arc>>,
}

impl MiniTokio {
    fn new() -> Self {
        MiniTokio { ready_queue: Arc::new(Mutex::new(VecDeque::new())) }
    }

    fn spawn(&self, future: F)
    where F: Future
 + Send + 'static
    {
        let waker = self.create_waker();
        let mut future = Box::pin(future);
        let queue = self.ready_queue.clone();

        std::thread::spawn(move || {
            let mut cx = Context::from_waker(&waker);
            loop {
                match future.as_mut().poll(&mut cx) {
                    Poll::Ready(()) => break,
                    Poll::Pending => {
                        // 等待被唤醒
                        std::thread::park();
                    }
                }
            }
        });
    }

    fn run(&self) {
        loop {
            let waker = {
                let mut queue = self.ready_queue.lock().unwrap();
                queue.pop_front()
            };
            match waker {
                Some(w) => w.wake(),
                None => std::thread::sleep(std::time::Duration::from_millis(1)),
            }
        }
    }
}

三、Reactor：I/O 事件驱动

真正的异步运行时需要与操作系统 I/O 多路复用机制集成。Reactor 负责监听 epoll/kqueue/IOCP 事件，当 I/O 就绪时唤醒对应的 Future。

use std::os::unix::io::RawFd;

struct Reactor {
    epoll_fd: RawFd,
    wakers: Arc>>,
}

impl Reactor {
    fn register_read(&self, fd: RawFd, waker: Waker, token: u64) {
        let mut event = libc::epoll_event {
            events: libc::EPOLLIN as u32 | libc::EPOLLONESHOT as u32,
            u64: token,
        };
        unsafe {
            libc::epoll_ctl(self.epoll_fd, libc::EPOLL_CTL_ADD, fd, &mut event);
        }
        self.wakers.lock().unwrap().insert(token, waker);
    }

    fn event_loop(&self) {
        let mut events = vec![libc::epoll_event { events: 0, u64: 0 }; 1024];
        loop {
            let n = unsafe {
                libc::epoll_wait(self.epoll_fd, events.as_mut_ptr(), 1024, -1)
            };
            for i in 0..n {
                let token = events[i as usize].u64;
                if let Some(waker) = self.wakers.lock().unwrap().remove(&token) {
                    waker.wake();
                }
            }
        }
    }
}

四、Pin 与自引用结构

Rust 的 async 函数生成的 Future 可能包含自引用字段（如引用同一结构体中的另一个字段）。Pin 保证 Future 在内存中不会被移动，从而保证自引用指针的有效性。这是 Rust 异步模型中最精妙的设计之一。

五、与 Tokio 的对比

Tokio 在此基础上增加了工作窃取调度器、协作式调度、I/O 驱动抽象层（mio）、定时器轮（hashed wheel timer）等高级特性。但核心原理与我们的迷你实现完全一致。

理解 Rust 异步运行时的内部机制，不仅能写出更高效的异步代码，更能深入理解零成本抽象这一 Rust 核心哲学。