Rust异步运行时核心原理：手搓极简Async Runtime实现物理射线检测

1. 项目概述：为什么要在 Rust 中手搓一个极简 Async Runtime？

如果你写过 Rust，尤其是涉足网络服务或并发编程，那么async/await肯定不陌生。它让异步代码写起来像同步一样直观，但背后那个负责调度和执行这些异步任务的runtime，却常常像个黑盒。我们用的tokio或async-std功能强大，但也复杂。有没有可能，只用 200 多行代码，就窥见其核心奥秘，甚至实现一个能跑起来的、具备特定功能的极简 runtime 呢？这个项目就是答案：用 Rust 实现一个专为“物理射线检测”场景优化的极简 async runtime。

物理射线检测（比如游戏中的子弹命中判定、光线追踪的初级采样）是个典型的计算密集型且高并发的场景。它不涉及复杂的 I/O 等待，但需要快速生成海量射线，并高效地调度这些检测任务。主流的通用 runtime 在这里可能“杀鸡用牛刀”，带来不必要的开销。我们的目标，就是剥离所有无关特性，构建一个纯粹为“任务生成、调度、执行”而生的核心轮子。通过这个过程，你不仅能彻底理解Future、Waker、Executor如何协同工作，更能掌握一种“量身定制”系统核心组件的思维方式，这对于追求极致性能的 Rust 开发者来说，是无价之宝。

2. 核心设计思路：自顶向下拆解一个 Runtime 的骨架

在动手写代码之前，我们必须想清楚一个 runtime 究竟要做什么。抛开网络、文件 I/O、定时器等高级功能，一个 runtime 最核心的职责只有两个：管理未来（Future）和调度任务（Task）。我们的设计将紧紧围绕这两个核心展开。

2.1 任务（Task）的本质：携带上下文的 Future

在 Rust 的异步世界里，Future是一个可能尚未完成的计算。但一个裸的Future并不能直接运行，它需要被包装成一个Task。Task是一个可以独立调度和执行的工作单元。在我们的极简 runtime 中，一个Task结构体需要包含以下核心信息：

1. Future 本身：需要被执行的计算逻辑。
2. 执行器状态：这个任务是否就绪（Ready）、进行中（Running）还是已完成（Completed）？
3. 唤醒器（Waker）：这是异步生态系统的“神经中枢”。当Future因等待（比如我们模拟的“射线检测计算”）而阻塞时，它会保存这个Waker。一旦条件就绪（比如计算完成），就通过Waker通知执行器：“嘿，我可以继续执行了！”

我们的设计关键在于简化。我们不实现复杂的多线程工作窃取队列，而是采用一个单线程、基于VecDeque的就绪任务队列。执行器（Executor）循环从这个队列中取出就绪的Task进行轮询（poll）。这种模型对于我们的物理射线检测场景是合适的，因为任务本身是纯计算，没有 I/O 阻塞，使用多线程带来的收益可能被同步开销抵消，而单线程模型简单、可预测，更适合作为教学原型和特定高性能场景的起点。

2.2 执行器（Executor）与唤醒（Waking）的协作

这是整个 runtime 最精妙的部分。流程如下：

1. 任务入队：用户通过spawn函数将一个async块（即一个Future）包装成Task，并推入就绪队列。
2. 主循环：执行器启动一个无限循环，从队列头部取出任务。
3. 轮询：执行器调用Task::poll方法，进而调用其内部Future的poll方法。
4. 等待与唤醒：
   • 如果Future::poll返回Poll::Pending，表示它需要等待（例如，将射线检测提交给一个计算池，但结果还没出来）。这个Future会克隆并存储我们传递给它的Waker。
   • 如果返回Poll::Ready(result)，则任务完成，清理资源。
5. 外部事件驱动：在我们的物理检测场景中，“外部事件”就是计算完成。我们假设有一个（模拟的）PhysicsEngine组件。当它完成一批射线检测计算后，它会调用对应Waker的wake()方法。
6. 重新调度：wake()方法的实现，就是将这个任务重新放回执行器的就绪队列尾部。这样，下一次主循环就会再次轮询它，此时Future很可能就会返回Poll::Ready了。

这个“轮询-等待-唤醒-再调度”的闭环，就是所有 async runtime 的核心灵魂。我们的 200 行代码，就是把这个灵魂用最简洁的形式具象化。

3. 核心细节解析与实现要点

让我们开始动手，将上述设计转化为具体的 Rust 数据结构与代码。我们会从最核心的Waker机制开始，因为它是最抽象但也最关键的一环。

3.1 实现一个极简的 ArcWake：唤醒器的核心

在标准库中，std::task::Waker是一个胖指针，内部使用了虚表（vtable）来动态分发wake等操作。为了极致简化，我们可以借鉴futures库中的ArcWake模式，但实现得更轻量。

use std::sync::{Arc, Mutex}; use std::task::{Wake, Context}; use std::collections::VecDeque; // 任务队列的类型别名，方便使用 type TaskQueue = Arc<Mutex<VecDeque<Arc<Task>>>>; // 我们自定义的 Waker 类型。它只需要持有一个指向任务队列的引用和任务的 ID（或直接引用）。 // 这里为了简单，我们让 Waker 直接持有需要被重新调度的 Task 的 Arc。 struct TaskWaker { task: Arc<Task>, queue: TaskQueue, } impl Wake for TaskWaker { fn wake(self: Arc<Self>) { // 唤醒操作：将任务重新推入就绪队列 let mut queue = self.queue.lock().unwrap(); queue.push_back(self.task.clone()); } // 通常也实现 wake_by_ref，避免不必要的克隆，这里为简化省略。 // fn wake_by_ref(self: &Arc<Self>) { ... } }

关键点解析：

• Waketrait 是 Rust 标准库提供的，用于定义唤醒行为。我们的TaskWaker实现了它。
• wake方法被调用时，其核心操作就是获取任务队列的锁，然后将关联的Task放回队列末尾。这就是“重新调度”。
• 我们使用了Arc<Mutex<...>>来共享任务队列。这是单线程环境下为了满足Waketrait 的Send + Sync约束而做的简化处理。在生产级多线程 runtime 中，这里会是更高效的无锁队列。

3.2 定义 Task：包装 Future 与状态

接下来，我们定义Task结构体。它需要存储 Future、其执行状态，并且能将自己转换为一个Waker。

use std::future::Future; use std::pin::Pin; use std::task::{Poll, Context}; struct Task { // 使用 Pin<Box<dyn Future<Output = ()> + Send>> 来存储一个 trait 对象。 // Pin 是必须的，因为 Future 在轮询期间内存地址必须稳定。 // Box 是动态分发所需。`+ Send` 约束是为了满足跨线程调度的可能性（虽然我们当前是单线程）。 future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>, // 执行状态：ReadyToPoll, Running, Completed。这里用一个简单的原子或 bool 标记即可。 // 为简化，我们省略状态机，仅通过 Future 的 Poll 结果来判断。 } impl Task { fn new(future: impl Future<Output = ()> + Send + 'static) -> Arc<Self> { Arc::new(Task { future: Mutex::new(Box::pin(future)), }) } fn poll(self: Arc<Self>, queue: &TaskQueue) { // 从 Mutex 中获取 Future 的可变引用 let mut future_guard = self.future.lock().unwrap(); let future = future_guard.as_mut(); // 为这个 Task 创建一个 Waker let waker = Arc::new(TaskWaker { task: self.clone(), queue: queue.clone(), }).into(); let mut cx = Context::from_waker(&waker); // 轮询 Future！ let _ = future.poll(&mut cx); // 注意：如果返回 Poll::Pending，Future 内部应该已经存储了我们的 waker。 // 如果返回 Poll::Ready(())，这个 Task 就完成了，我们什么也不需要做。 } }

注意事项与心得：

• Pin<Box<dyn Future...>>这个组合非常常见。Pin确保Future不会被移动，这对于自引用结构（很多Future都是）的安全至关重要。
• Task::poll方法接收&TaskQueue是为了传递给TaskWaker。这里有一个循环引用：Task通过Waker持有queue，而queue又存储着Task。Arc和Mutex帮助我们安全地管理这种共享状态。
• 在实际的复杂 runtime 中，Task结构会更复杂，可能包含一个状态机来明确跟踪是“就绪”、“休眠”还是“完成”，以避免无效的轮询。我们这里做了极大简化。

3.3 构建 Executor：驱动一切的主循环

执行器是粘合剂，它持有任务队列，并提供spawn接口，并运行主循环。

struct Executor { ready_queue: TaskQueue, } impl Executor { fn new() -> Self { Executor { ready_queue: Arc::new(Mutex::new(VecDeque::new())), } } fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) { let task = Task::new(future); let mut queue = self.ready_queue.lock().unwrap(); queue.push_back(task); } fn run(&self) { loop { // 每一轮循环，处理当前就绪队列中的所有任务 let task_opt = { let mut queue = self.ready_queue.lock().unwrap(); queue.pop_front() }; if let Some(task) = task_opt { task.poll(&self.ready_queue); } else { // 队列为空，可以短暂休眠以避免忙等待。 // 但在我们这个极简示例中，我们假设任务会不断被外部事件（物理引擎）唤醒。 // 为了演示，我们简单地进行 yield 或短暂睡眠。 std::thread::yield_now(); } } } }

核心逻辑剖析：

1. spawn：将用户提供的Future包装成Task，并立即放入就绪队列。这意味着任务创建后很快就会被轮询。
2. run：这是心脏。它不断从队列中取任务并调用poll。
   • 如果poll返回Pending，任务未来的唤醒就依赖于其内部存储的Waker被调用。
   • 如果poll返回Ready，任务结束，循环继续。
   • 如果队列为空，我们让出 CPU 时间片。在实际应用中，这里可能会阻塞在某个条件变量上，直到有新的任务被wake进队列。

4. 融入物理射线检测场景

现在，我们有了一个能跑起来的 runtime 骨架。但如何让它和“物理射线检测”结合起来呢？关键在于模拟一个会产生阻塞（Pending）并能在完成后唤醒任务的Future。

4.1 模拟物理引擎与异步检测 Future

我们创建一个模拟的PhysicsEngine，它有一个方法用于提交射线检测请求，这个请求是“异步”的。

// 模拟的物理引擎 struct PhysicsEngine; impl PhysicsEngine { // 这是一个“异步”方法，它返回一个 Future。 // 这个 Future 会模拟一个耗时的计算，并在计算完成后通过 Waker 通知。 async fn cast_ray(&self, origin: [f32; 3], direction: [f32; 3]) -> Option<[f32; 3]> { // 关键点：这里我们如何模拟异步等待？ // 我们不能真的在这里做耗时计算，否则会阻塞执行器线程。 // 我们需要将计算“提交”到某个地方，然后让 Future 进入 Pending 状态。 // 为了演示，我们创建一个特殊的 Future：RayCastFuture。 RayCastFuture::new(origin, direction).await } } // 代表一次射线检测的 Future struct RayCastFuture { origin: [f32; 3], direction: [f32; 3], // 一个标记，表示计算是否已完成。在实际中，这可能是与物理引擎通信的句柄。 completed: Arc<AtomicBool>, result: Arc<Mutex<Option<Option<[f32; 3]>>>>, } impl RayCastFuture { fn new(origin: [f32; 3], direction: [f32; 3]) -> Self { RayCastFuture { origin, direction, completed: Arc::new(AtomicBool::new(false)), result: Arc::new(Mutex::new(None)), } } } impl Future for RayCastFuture { type Output = Option<[f32; 3]>; // 命中点坐标，None 表示未命中 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { // 检查计算是否已完成 if self.completed.load(Ordering::Acquire) { // 已完成，取出结果并返回 Ready let result = self.result.lock().unwrap().take().unwrap(); Poll::Ready(result) } else { // 未完成，需要安排计算，并存储 Waker 以便完成后唤醒 // 这里是一个关键技巧：我们只在第一次 poll 时提交计算任务。 // 如何知道是第一次？我们可以用另一个标志位，或者这里我们简化：每次 poll 都尝试提交，但由外部逻辑保证只提交一次。 // 更典型的做法是，在 Future 被创建时，就立即将计算任务提交给一个工作池。 // 假设有一个全局的 PHYSICS_SIMULATOR 单例。 PHYSICS_SIMULATOR.submit_raycast( self.origin, self.direction, self.completed.clone(), self.result.clone(), cx.waker().clone(), // 将唤醒器传递给物理模拟器！ ); Poll::Pending } } }

4.2 模拟物理计算线程与唤醒机制

我们需要一个后台的“物理计算线程”来模拟耗时操作，并在完成后调用Waker。

use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; use std::time::Duration; // 全局物理模拟器（简化版，实际应用可能用更优雅的模式） struct PhysicsSimulator { // 任务接收通道 // 这里用 Vec 和 Mutex 简单模拟，生产环境应用 crossbeam-channel 等。 pending_jobs: Mutex<Vec<RayCastJob>>, } static PHYSICS_SIMULATOR: PhysicsSimulator = PhysicsSimulator { pending_jobs: Mutex::new(Vec::new()), }; struct RayCastJob { origin: [f32; 3], direction: [f32; 3], completed: Arc<AtomicBool>, result: Arc<Mutex<Option<Option<[f32; 3]>>>>, waker: Waker, } impl PhysicsSimulator { fn submit_raycast( &self, origin: [f32; 3], direction: [f32; 3], completed: Arc<AtomicBool>, result: Arc<Mutex<Option<Option<[f32; 3]>>>>, waker: Waker, ) { let job = RayCastJob { origin, direction, completed, result, waker, }; self.pending_jobs.lock().unwrap().push(job); } // 这个函数应该在另一个线程中循环运行 fn run_simulation_loop(&self) { loop { let maybe_job = { let mut jobs = self.pending_jobs.lock().unwrap(); jobs.pop() }; if let Some(job) = maybe_job { // 模拟耗时计算 thread::sleep(Duration::from_millis(10)); // 假设检测需要10毫秒 let hit_point = Some([1.0, 2.0, 3.0]); // 模拟计算结果 // 存储结果 *job.result.lock().unwrap() = Some(hit_point); // 标记完成 job.completed.store(true, Ordering::Release); // 唤醒关联的 Task！ job.waker.wake_by_ref(); } else { thread::sleep(Duration::from_millis(1)); } } } }

场景串联：

1. 用户调用physics_engine.cast_ray(...).await，这会创建一个RayCastFuture。
2. 执行器poll这个Future。
3. Future的poll方法发现计算未完成，于是将计算任务（包含Waker）提交给PHYSICS_SIMULATOR，并返回Poll::Pending。
4. 物理模拟线程在后台处理这个任务，计算完成后，设置结果标志，并调用job.waker.wake()。
5. wake()方法将对应的Task重新推入执行器的就绪队列。
6. 执行器主循环再次取出并poll这个Task。此时RayCastFuture::poll看到completed为true，便取出结果并返回Poll::Ready(hit_point)。
7. await表达式得到结果，异步任务继续执行后续逻辑。

5. 常见问题、调试技巧与性能考量

即使是这样一个小型 runtime，在实现和使用的过程中也会遇到不少坑。这里记录一些典型问题和思考。

5.1 为什么我的 Future 卡住了，再也不执行？

这是新手实现 runtime 时最常见的问题。根本原因通常是Waker没有正确存储或唤醒。

• 检查点1：Waker 是否被 Future 存储？在Future::poll返回Pending之前，必须确保传入的Context中的waker被 Future 以某种方式保存下来。通常是通过克隆（cx.waker().clone()）。如果没存，那么外部事件完成时就找不到通知对象。
• 检查点2：唤醒操作是否正确关联到任务？在我们的实现中，TaskWaker::wake需要将正确的Task推回队列。确保Arc引用没有在中间被意外丢弃，导致唤醒器唤醒了一个已经失效的任务句柄。
• 检查点3：执行器主循环是否在空转？如果队列为空，我们的示例简单使用了yield_now()。在真实场景中，如果所有任务都在等待 I/O（或像我们这里的物理计算），执行器线程应该被阻塞（例如，通过park/unpark或条件变量），直到有Waker被调用。否则会白白消耗 CPU。

调试技巧：在TaskWaker::wake和Executor::run的循环中加入简单的日志打印（例如println!("Waking task: {:p}", &*self.task)和println!("Polling task from queue")）。观察任务被唤醒后是否真的重新进入了队列并被轮询。

5.2 Pin 的必要性与自引用结构

我们的Task里用Pin<Box<dyn Future...>>不是偶然。很多复杂的Future（尤其是手写的状态机Future）可能是自引用的：即结构体的某个字段引用了另一个字段的地址。如果这个Future被移动了，这些内部引用就会失效，导致未定义行为。

注意：当你自己实现一个Future时，如果它内部需要持有指向自身数据的引用（例如，在.await点保存临时变量的地址），你必须使用Pin来保证内存稳定。对于async fn或async {}块，编译器会自动生成安全的、可能包含自引用的Future结构，这就是为什么它们必须被Pin住才能进行poll。

5.3 单线程 vs 多线程执行器

我们的极简 runtime 是单线程的。这对于 I/O 密集型或特定计算密集型任务（如果计算本身不能很好地并行化）可能是个瓶颈。如何扩展为多线程？

1. 工作窃取队列：这是tokio等 runtime 的做法。每个工作线程有自己的本地任务队列，也会从其他线程的队列“窃取”任务来平衡负载。这需要实现复杂的无锁数据结构。
2. 全局队列 + 锁：一个简单的多线程版本是让所有工作线程从一个共享的Mutex<VecDeque>中获取任务。但锁竞争会成为瓶颈。
3. 我们的物理检测场景思考：物理射线检测通常是令人尴尬的并行任务，即任务间几乎没有依赖。一个更高效的架构可能是：
   • 保留我们的单线程执行器作为任务生成与协调器。
   • 使用rayon这类并行迭代库或一个简单的线程池来处理批量的射线检测计算。
   • RayCastFuture的poll方法不再提交单次检测，而是将一批检测请求发送给计算池，并等待整批结果。这样能极大减少任务调度和跨线程通信的开销。

5.4 错误处理与资源清理

我们的示例忽略了错误处理和资源清理（如任务完成后的Task对象析构）。在生产环境中：

• Future的Output应该是Result<T, E>类型，错误需要能传播。
• Task在完成后应从所有数据结构中移除，避免内存泄漏。这需要更精细的状态管理。
• 当执行器被关闭时，需要优雅地终止所有任务和后台线程。

5.5 性能优化的关键点

对于物理检测这类高性能场景，即使是极简 runtime，也有优化空间：

• 避免动态分发：Box<dyn Future>涉及虚函数调用，有开销。可以尝试使用async块生成的具体Future类型，但这会增加类型系统的复杂性。另一种思路是使用enum来枚举有限的几种任务类型。
• Waker 复用：频繁创建和克隆Arc<Waker>有分配开销。成熟的 runtime 会实现Waker的对象池。
• 批处理唤醒：在物理引擎中，一帧可能完成成千上万次射线检测。如果每次检测完成都调用一次wake()，会导致执行器被频繁唤醒。更好的方式是批量处理：物理引擎将这一帧所有完成检测的Waker收集起来，在一帧结束时一次性全部唤醒。这需要更复杂的Waker设计（例如，将Waker与一个“完成标记”位图关联）。

通过这个 200 多行的极简项目，我们亲手搭建了 Rustasync/await生态的基石。它不完美，但清晰地揭示了Future、Executor和Waker三者如何通过协作，将看似神秘的异步编程转化为可控的底层机制。理解这些，不仅能让你更自信地使用tokio，更能让你在遇到性能瓶颈或需要定制并发模型时，拥有深入底层、动手改造的能力。这或许就是系统编程语言 Rust 带给我们的最硬核的乐趣之一。