Rust异步运行时最小实现 - extreme 分享

Rust语言通过定义了Future Trait , 奠定了异步语法的基石，而Rust的异步代码时惰性的，必须有一个运行时来驱动，Rust本身还没提供这样的实现，社区中有不少开源方案，比如tokio等。
Tokio的运行时是一个事件循环，利用了不同平台的异步非阻塞特性，比如kqueue，epoll等。
我一直想要弄清楚runtime是怎么调度Future，而Future完成时又是怎么通知runtime，extreme 实现了一个最小运行时，可以让一窥究竟。

1. use std::sync::{Arc, Condvar, Mutex};
2. use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
4. #[derive(Default)]
5. struct Park(Mutex<bool>, Condvar);
7. fn unpark(park: &Park) {
8.     *park.0.lock().unwrap() = true;
9.     park.1.notify_one();
10. }
12. static VTABLE: RawWakerVTable = RawWakerVTable::new(
13.     |clone_me| unsafe {
14.         let arc = Arc::from_raw(clone_me as *const Park);
15.         std::mem::forget(arc.clone());
16.         RawWaker::new(Arc::into_raw(arc) as *const (), &VTABLE)
17.     },
18.     |wake_me| unsafe { unpark(&Arc::from_raw(wake_me as *const Park)) },
19.     |wake_by_ref_me| unsafe { unpark(&*(wake_by_ref_me as *const Park)) },
20.     |drop_me| unsafe { drop(Arc::from_raw(drop_me as *const Park)) },
21. );
23. /// Run a `Future`.
24. pub fn run<F: std::future::Future>(mut f: F) -> F::Output {
25.     let mut f = unsafe { std::pin::Pin::new_unchecked(&mut f) };
26.     let park = Arc::new(Park::default());
27.     let sender = Arc::into_raw(park.clone());
28.     let raw_waker = RawWaker::new(sender as *const _, &VTABLE);
29.     let waker = unsafe { Waker::from_raw(raw_waker) };
30.     let mut cx = Context::from_waker(&waker);
32.     loop {
33.         match f.as_mut().poll(&mut cx) {
34.             Poll::Pending => {
35.                 let mut runnable = park.0.lock().unwrap();
36.                 while !*runnable {
37.                     runnable = park.1.wait(runnable).unwrap();
38.                 }
39.                 *runnable = false;
40.             }
41.             Poll::Ready(val) => return val,
42.         }
43.     }
44. }

复制代码

这个简短的例子表达了实现一个运行时的最低需求

• 实现RawWakerVTable
  
• 如何通过Waker唤醒runtime继续调度，这里用了信号量
  

本质上运行时可以抽象成一个不断运行的循环体，在循环体内不断调用Future的poll方法。
(当Future返回Poll:ending时，此处简化为使用信号量的等待操作)
这个例子也说明了Future的调用能返回时，需要调用存储在ctx里面的Waker::waker()方法，唤醒运行时继续执行阻塞的异步任务

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！