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结构体需要包含以下核心信息Future 本身需要被执行的计算逻辑。执行器状态这个任务是否就绪Ready、进行中Running还是已完成Completed唤醒器Waker这是异步生态系统的“神经中枢”。当Future因等待比如我们模拟的“射线检测计算”而阻塞时它会保存这个Waker。一旦条件就绪比如计算完成就通过Waker通知执行器“嘿我可以继续执行了”我们的设计关键在于简化。我们不实现复杂的多线程工作窃取队列而是采用一个单线程、基于VecDeque的就绪任务队列。执行器Executor循环从这个队列中取出就绪的Task进行轮询poll。这种模型对于我们的物理射线检测场景是合适的因为任务本身是纯计算没有 I/O 阻塞使用多线程带来的收益可能被同步开销抵消而单线程模型简单、可预测更适合作为教学原型和特定高性能场景的起点。2.2 执行器Executor与唤醒Waking的协作这是整个 runtime 最精妙的部分。流程如下任务入队用户通过spawn函数将一个async块即一个Future包装成Task并推入就绪队列。主循环执行器启动一个无限循环从队列头部取出任务。轮询执行器调用Task::poll方法进而调用其内部Future的poll方法。等待与唤醒如果Future::poll返回Poll::Pending表示它需要等待例如将射线检测提交给一个计算池但结果还没出来。这个Future会克隆并存储我们传递给它的Waker。如果返回Poll::Ready(result)则任务完成清理资源。外部事件驱动在我们的物理检测场景中“外部事件”就是计算完成。我们假设有一个模拟的PhysicsEngine组件。当它完成一批射线检测计算后它会调用对应Waker的wake()方法。重新调度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 ArcMutexVecDequeArcTask; // 我们自定义的 Waker 类型。它只需要持有一个指向任务队列的引用和任务的 ID或直接引用。 // 这里为了简单我们让 Waker 直接持有需要被重新调度的 Task 的 Arc。 struct TaskWaker { task: ArcTask, queue: TaskQueue, } impl Wake for TaskWaker { fn wake(self: ArcSelf) { // 唤醒操作将任务重新推入就绪队列 let mut queue self.queue.lock().unwrap(); queue.push_back(self.task.clone()); } // 通常也实现 wake_by_ref避免不必要的克隆这里为简化省略。 // fn wake_by_ref(self: ArcSelf) { ... } }关键点解析Waketrait 是 Rust 标准库提供的用于定义唤醒行为。我们的TaskWaker实现了它。wake方法被调用时其核心操作就是获取任务队列的锁然后将关联的Task放回队列末尾。这就是“重新调度”。我们使用了ArcMutex...来共享任务队列。这是单线程环境下为了满足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 { // 使用 PinBoxdyn FutureOutput () Send 来存储一个 trait 对象。 // Pin 是必须的因为 Future 在轮询期间内存地址必须稳定。 // Box 是动态分发所需。 Send 约束是为了满足跨线程调度的可能性虽然我们当前是单线程。 future: MutexPinBoxdyn FutureOutput () Send, // 执行状态ReadyToPoll, Running, Completed。这里用一个简单的原子或 bool 标记即可。 // 为简化我们省略状态机仅通过 Future 的 Poll 结果来判断。 } impl Task { fn new(future: impl FutureOutput () Send static) - ArcSelf { Arc::new(Task { future: Mutex::new(Box::pin(future)), }) } fn poll(self: ArcSelf, 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::PendingFuture 内部应该已经存储了我们的 waker。 // 如果返回 Poll::Ready(())这个 Task 就完成了我们什么也不需要做。 } }注意事项与心得PinBoxdyn 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 FutureOutput () 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(); } } } }核心逻辑剖析spawn将用户提供的Future包装成Task并立即放入就绪队列。这意味着任务创建后很快就会被轮询。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 状态。 // 为了演示我们创建一个特殊的 FutureRayCastFuture。 RayCastFuture::new(origin, direction).await } } // 代表一次射线检测的 Future struct RayCastFuture { origin: [f32; 3], direction: [f32; 3], // 一个标记表示计算是否已完成。在实际中这可能是与物理引擎通信的句柄。 completed: ArcAtomicBool, result: ArcMutexOptionOption[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: Pinmut Self, cx: mut Context_) - PollSelf::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: MutexVecRayCastJob, } static PHYSICS_SIMULATOR: PhysicsSimulator PhysicsSimulator { pending_jobs: Mutex::new(Vec::new()), }; struct RayCastJob { origin: [f32; 3], direction: [f32; 3], completed: ArcAtomicBool, result: ArcMutexOptionOption[f32; 3], waker: Waker, } impl PhysicsSimulator { fn submit_raycast( self, origin: [f32; 3], direction: [f32; 3], completed: ArcAtomicBool, result: ArcMutexOptionOption[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)); } } } }场景串联用户调用physics_engine.cast_ray(...).await这会创建一个RayCastFuture。执行器poll这个Future。Future的poll方法发现计算未完成于是将计算任务包含Waker提交给PHYSICS_SIMULATOR并返回Poll::Pending。物理模拟线程在后台处理这个任务计算完成后设置结果标志并调用job.waker.wake()。wake()方法将对应的Task重新推入执行器的就绪队列。执行器主循环再次取出并poll这个Task。此时RayCastFuture::poll看到completed为true便取出结果并返回Poll::Ready(hit_point)。await表达式得到结果异步任务继续执行后续逻辑。5. 常见问题、调试技巧与性能考量即使是这样一个小型 runtime在实现和使用的过程中也会遇到不少坑。这里记录一些典型问题和思考。5.1 为什么我的 Future 卡住了再也不执行这是新手实现 runtime 时最常见的问题。根本原因通常是Waker没有正确存储或唤醒。检查点1Waker 是否被 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里用PinBoxdyn Future...不是偶然。很多复杂的Future尤其是手写的状态机Future可能是自引用的即结构体的某个字段引用了另一个字段的地址。如果这个Future被移动了这些内部引用就会失效导致未定义行为。注意当你自己实现一个Future时如果它内部需要持有指向自身数据的引用例如在.await点保存临时变量的地址你必须使用Pin来保证内存稳定。对于async fn或async {}块编译器会自动生成安全的、可能包含自引用的Future结构这就是为什么它们必须被Pin住才能进行poll。5.3 单线程 vs 多线程执行器我们的极简 runtime 是单线程的。这对于 I/O 密集型或特定计算密集型任务如果计算本身不能很好地并行化可能是个瓶颈。如何扩展为多线程工作窃取队列这是tokio等 runtime 的做法。每个工作线程有自己的本地任务队列也会从其他线程的队列“窃取”任务来平衡负载。这需要实现复杂的无锁数据结构。全局队列 锁一个简单的多线程版本是让所有工作线程从一个共享的MutexVecDeque中获取任务。但锁竞争会成为瓶颈。我们的物理检测场景思考物理射线检测通常是令人尴尬的并行任务即任务间几乎没有依赖。一个更高效的架构可能是保留我们的单线程执行器作为任务生成与协调器。使用rayon这类并行迭代库或一个简单的线程池来处理批量的射线检测计算。RayCastFuture的poll方法不再提交单次检测而是将一批检测请求发送给计算池并等待整批结果。这样能极大减少任务调度和跨线程通信的开销。5.4 错误处理与资源清理我们的示例忽略了错误处理和资源清理如任务完成后的Task对象析构。在生产环境中Future的Output应该是ResultT, E类型错误需要能传播。Task在完成后应从所有数据结构中移除避免内存泄漏。这需要更精细的状态管理。当执行器被关闭时需要优雅地终止所有任务和后台线程。5.5 性能优化的关键点对于物理检测这类高性能场景即使是极简 runtime也有优化空间避免动态分发Boxdyn Future涉及虚函数调用有开销。可以尝试使用async块生成的具体Future类型但这会增加类型系统的复杂性。另一种思路是使用enum来枚举有限的几种任务类型。Waker 复用频繁创建和克隆ArcWaker有分配开销。成熟的 runtime 会实现Waker的对象池。批处理唤醒在物理引擎中一帧可能完成成千上万次射线检测。如果每次检测完成都调用一次wake()会导致执行器被频繁唤醒。更好的方式是批量处理物理引擎将这一帧所有完成检测的Waker收集起来在一帧结束时一次性全部唤醒。这需要更复杂的Waker设计例如将Waker与一个“完成标记”位图关联。通过这个 200 多行的极简项目我们亲手搭建了 Rustasync/await生态的基石。它不完美但清晰地揭示了Future、Executor和Waker三者如何通过协作将看似神秘的异步编程转化为可控的底层机制。理解这些不仅能让你更自信地使用tokio更能让你在遇到性能瓶颈或需要定制并发模型时拥有深入底层、动手改造的能力。这或许就是系统编程语言 Rust 带给我们的最硬核的乐趣之一。
一键保存高清图表到本地（附完整参数详解）)