参考资料

async runtime

[tokio:main] 实际是一个过程宏，将 async fn main 展开成

fn main() {
    tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap().block_on(async {
        println!("Hello, world!");
    });
}

等价代码

#[tokio::main]
async fn main() {
    println!("Hello, world!");
}

先构造一个 Async Runtime 的实例，再调用类似 block_on 的方法来驱动并发计算执行。

std::future::Future 的定义

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

pub trait Future {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context)
        -> Poll<Self::Output>;
}

Output 是关联类型，是 Future 完成后结果的类型
Future 的所有者负责通过 poll 来推进 Future 完成

示例

Async Rust 的实现 | 夜天之书

struct MyFuture {
    id: u32,
    start: Instant,
    duration: Duration,
}

fn block_on<F: Future>(fut: F) -> F::Output {
    let mut fut = pin!(fut);
    let t = thread::current();
    let waker = Arc::new(ThreadWaker(t)).into();
    let mut cx = Context::from_waker(&waker);
    loop {
        match fut.as_mut().poll(&mut cx) {
            Poll::Ready(res) => return res,
            Poll::Pending => thread::park(),
        }
    }
}
struct ThreadWaker(Thread);
impl Wake for ThreadWaker {
    fn wake(self: Arc<Self>) {
        self.0.unpark();
    }
}

impl Future for MyFuture {
    type Output = ();
    fn poll(&mut self, cx: &mut Context) -> Poll<Self::Output> {
        println!("Polling {} {}", self.id, chrono::Utc::now());
        let now = Instant::now();
        let id = self.id;
        let expect = self.start + self.duration;
        if expect > now {
            let duration = expect - now;
            let waker = cx.waker().clone();
            thread::spawn(move || {
                thread::sleep(duration);
                // 用来唤醒
                waker.wake();
            });
            println!("Waking {id} {}", chrono::Utc::now());
            Poll::Pending
        } else {
            println!("Done {} {}", id, chrono::Utc::now());
            Poll::Ready(())
        }

    }
}

fn main() {
    println!("Start {}", chrono::Utc::now());
    block_on(MyFuture {
        id: 1,
        start: Instant::now(),
        duration: Duration::from_millis(1000 * 10),
    });
    println!("End {}", chrono::Utc::now());
}

block_on 函数第一次调用 poll 方法，这次 poll 返回了 Poll::Pending 结果，
block_on 阻塞在 thread::park() 上。该方法用同步原语信号量阻塞当前线程，并等待其他线程调用 unpark 唤醒内部信号量。

多个 futures

前面的 future 只有一个能运行，改动 block 的逻辑，让其能同时运行多个 future，改变 block_on 的函数名称

fn spawn<F>(fut: F) -> Receiver<F::Output>
where F: Future + Send + 'static,
      F::Output: Send,
{
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let mut fut = pin!(fut);
        let t = thread::current();
        let waker = Arc::new(ThreadWaker(t)).into();
        let mut cx = Context::from_waker(&waker);
        loop {
            match fut.as_mut().poll(&mut cx) {
                Poll::Ready(res) => {
                    let _ = tx.send(res);
                    break;
                }
                Poll::Pending => thread::park(),
            }
        }
    });
    rx
}

fn main() {
    println!("Start {}", chrono::Utc::now());
    let r1 = spawn(MyFuture {
        id: 1,
        start: Instant::now(),
        duration: Duration::from_secs(10),
    });
    let r2 = spawn(MyFuture {
        id: 2,
        start: Instant::now(),
        duration: Duration::from_secs(10),
    });
    let h1 = thread::spawn(move || {
        r1.recv().unwrap();
        println!("Finish 1 {}", chrono::Utc::now());
    });
    let h2 = thread::spawn(move || {
        r2.recv().unwrap();
        println!("Finish 2 {}", chrono::Utc::now());
    });
    h1.join().unwrap();
    h2.join().unwrap();
}

Send + 'static

Future 多了一个 Send + 'static，让其能在多个线程中传递

如果想 Future 的产生和计算都在同一个线程中

let runtime = tokio::runtime::Runtime::new().unwrap();
runtime.block_on(async {
    let local = tokio::task::LocalSet::new();
    let rc = Rc::new(42);
    local.run_until(async move {
        println!("rc={rc}");
    }).await;
});

能使用 LocalSet
如果把 local.run_until 换成 tokio::spawn 那么 Rust 编译器就会抱怨被 move 进去的 Rc 实例不是 Send 的。
LocalSet 的底层实现是持有一个 LocalState 实例，其中指示了持有该 LocalSet 的线程的 ThreadId 是什么。
tokio 运行时调度过程里，如果拿出来的作业对应的 ThreadId 跟当前 ThreadId 不同，则不执行该作业，而是放回 LocalSet 的任务队列里。

Waker API

pub fn waker_fn<F: Fn() + Send + Sync + 'static>(f: F) -> Waker {
    Waker::from(Arc::new(Helper(f)))
}

struct Helper<F>(F);

impl<F: Fn() + Send + Sync + 'static> Wake for Helper<F> {
    fn wake(self: Arc<Self>) {
        (self.0)();
    }

    fn wake_by_ref(self: &Arc<Self>) {
        (self.0)();
    }
}

pub struct Waker {
    waker: RawWaker,
}

pub struct RawWaker {
    data: *const (),
    vtable: &'static RawWakerVTable,
}

pub struct RawWakerVTable {
    clone: unsafe fn(*const ()) -> RawWaker,
    wake: unsafe fn(*const ()),
    wake_by_ref: unsafe fn(*const ()),
    drop: unsafe fn(*const ()),
}

impl Waker {
    pub fn wake_by_ref(&self) {
        unsafe {
            (self.waker.vtable.wake_by_ref)(self.waker.data);
        }
    }
    pub fn wake(self) {
        let waker = self.waker.vtable.wake;
        let data = self.waker.data;
        crate::mem::forget(self);
        unsafe {
            (waker)(data)
        };
    }
}

在使用 async fn main 下生成的 Future 如下

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

enum MainFuture {
    // Initialized, never polled
    State0,
    // Waiting on `Delay`, i.e. the `future.await` line.
    State1(Delay),
    // The future has completed.
    Terminated,
}

impl Future for MainFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<()>
    {
        use MainFuture::*;

        loop {
            match *self {
                State0 => {
                    let when = Instant::now() +
                        Duration::from_millis(10);
                    let future = Delay { when };
                    *self = State1(future);
                }
                State1(ref mut my_future) => {
                    match Pin::new(my_future).poll(cx) {
                        Poll::Ready(out) => {
                            assert_eq!(out, "done");
                            *self = Terminated;
                            return Poll::Ready(());
                        }
                        Poll::Pending => {
                            return Poll::Pending;
                        }
                    }
                }
                Terminated => {
                    panic!("future polled after completion")
                }
            }
        }
    }
}

MainFuture 是一个状态机，从 State0 开始，通过调用 poll 来获悉 Future 是否完成
如果 Future 无法完成，通常是由于它正在等待的资源没有准备好，则返回 Poll::Pending 。
接收 Poll::Pending 向调用者表明 Future 将在稍后完成，调用者应稍后再次调用 poll 。
谁来执行这个 poll 呢？

执行者 (Executors)

Futures 必须有 poll 调用它们来转换它们的状态。
Futures 是由其他依赖 Future 组成的。
问题是，在最外层 Future，是什么调用 poll ？
要运行异步函数，它们必须传递给 tokio::spawn 或者是用 #[tokio::main] 注释的主函数。这导致将生成的最外层 Future 提交给Tokio 执行者。

执行者负责在外部未来 Future 调用 Future::poll ，驱动异步计算完成。

use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use futures::task;

fn main() {
    let mut mini_tokio = MiniTokio::new();

    mini_tokio.spawn(async {
        let when = Instant::now() + Duration::from_millis(10);
        let future = Delay { when };

        let out = future.await;
        assert_eq!(out, "done");
    });

    mini_tokio.run();
}

struct MiniTokio {
    tasks: VecDeque<Task>,
}

type Task = Pin<Box<dyn Future<Output = ()> + Send>>;

impl MiniTokio {
    fn new() -> MiniTokio {
        MiniTokio {
            tasks: VecDeque::new(),
        }
    }

    /// Spawn a future onto the mini-tokio instance.
    fn spawn<F>(&mut self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        self.tasks.push_back(Box::pin(future));
    }

    fn run(&mut self) {
        let waker = task::noop_waker();
        let mut cx = Context::from_waker(&waker);

        while let Some(mut task) = self.tasks.pop_front() {
            if task.as_mut().poll(&mut cx).is_pending() {
                self.tasks.push_back(task);
            }
        }
    }
}

spawn 比较简单，就 push 这个 task 进入队列
run 方法，这里 executor 从不 sleep，忙等。效率不高
理想情况下，只有 future 在取得进展的时候，才能轮询 futures。
任务想要从 TCP 套接字读取数据，那么只想在 TCP 套接字接收到数据时轮询任务
理想情况下，mini-tokio 只会在该时刻过去后才轮询任务

Wakers

这是一个资源能够通知等待任务该资源已准备好继续某些操作
poll 的 Context 参数有一个 waker() 方法。此方法返回绑定到当前任务的 Waker 。
Waker 有一个 wake() 方法。
调用此方法会向执行程序 (exectutors) 发出信号，表明应安排执行相关任务。
资源在转换到就绪状态时调用 wake() 以通知执行者轮询任务将能够取得进展。
Waker 实际是一个回调闭包，使用 wake 和 wake_by_ref 来通知异步运行时再次当前的 Future 计算

最开始的实现：

impl Future for Delay {
    type Output = &'static str;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<&'static str>
    {
        if Instant::now() >= self.when {
            println!("Hello world");
            Poll::Ready("done")
        } else {
            // Ignore this line for now.
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

在返回 Poll::Pending 之前，调用了 cx.waker().wake_by_ref()
通过返回 Poll::Pending，负责向唤醒者发出信号。
- 因为还没有实现计时器线程，所以向内联唤醒器发出信号。
- 这样做会导致 future 被立即重新安排、再次执行，并且可能还没有准备好完成。
下一步更新 Mini Tokio 以接收唤醒通知。
希望 Executors 只在任务被唤醒时运行任务，为此，MiniTokio 将提供自己的 waker。

当 waker 被调用时，其相关任务被排队等待执行

use crossbeam::channel;
use std::sync::Arc;

struct MiniTokio {
    scheduled: channel::Receiver<Arc<Task>>,
    sender: channel::Sender<Arc<Task>>,
}

struct Task {
    // The `Mutex` is to make `Task` implement `Sync`. Only
    // one thread accesses `future` at any given time. The
    // `Mutex` is not required for correctness. Real Tokio
    // does not use a mutex here, but real Tokio has
    // more lines of code than can fit in a single tutorial
    // page.
    future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
    executor: channel::Sender<Arc<Task>>,
}

impl Task {
    fn schedule(self: &Arc<Self>) {
        self.executor.send(self.clone());
    }
}

// 引入 futures
use futures::task::{self, ArcWake};
use std::sync::Arc;
impl ArcWake for Task {
    fn wake_by_ref(arc_self: &Arc<Self>) {
        arc_self.schedule();
    }
}

当上面的计时器线程调用 waker.wake() 时，任务被推送到通道中。

接下来在 MiniTokio::run() 函数中实现任务的接收和执行。

impl MiniTokio {
    fn run(&self) {
        while let Ok(task) = self.scheduled.recv() {
            task.poll();
        }
    }

    /// Initialize a new mini-tokio instance.
    fn new() -> MiniTokio {
        let (sender, scheduled) = channel::unbounded();

        MiniTokio { scheduled, sender }
    }

    /// Spawn a future onto the mini-tokio instance.
    ///
    /// The given future is wrapped with the `Task` harness and pushed into the
    /// `scheduled` queue. The future will be executed when `run` is called.
    fn spawn<F>(&self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        Task::spawn(future, &self.sender);
    }
}

impl Task {
    fn poll(self: Arc<Self>) {
        // Create a waker from the `Task` instance. This
        // uses the `ArcWake` impl from above.
        let waker = task::waker(self.clone());
        let mut cx = Context::from_waker(&waker);

        // No other thread ever tries to lock the future
        let mut future = self.future.try_lock().unwrap();

        // Poll the future
        let _ = future.as_mut().poll(&mut cx);
    }

    // Spawns a new task with the given future.
    //
    // Initializes a new Task harness containing the given future and pushes it
    // onto `sender`. The receiver half of the channel will get the task and
    // execute it.
    fn spawn<F>(future: F, sender: &channel::Sender<Arc<Task>>)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        let task = Arc::new(Task {
            future: Mutex::new(Box::pin(future)),
            executor: sender.clone(),
        });

        let _ = sender.send(task);
    }
}

BruceChen7 / gitblog

async和await的概念 #55

参考资料

async runtime

[tokio:main] 实际是一个过程宏，将 async fn main 展开成

std::future::Future 的定义

示例

多个 futures

Send + 'static

如果想 Future 的产生和计算都在同一个线程中

Waker API

在使用 async fn main 下生成的 Future 如下

执行者 (Executors)

Wakers

type/rust #public