Rust异步机制

发表于 阅读次数：

Future 抽象

Rust中对异步Task的核心抽象为Future trait，源码如下：

1
2
3
4
pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

Future trait描述状态机对外暴露的接口，用户或Runtime通过poll方法推动状态机，返回执行结果，

  1. 遇到了阻塞 Pending

  2. 执行完毕 Ready+返回值

async和await本质上是一个语法糖，我们可以手动实现Future 来实现异步

1
2
3
4
5
6
7
8
9
fn do_future -> MyFuture { MyFuture }

struct MyFuture;
impl Future for MyFuture {
    type Output = i32;
    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output>{
        Poll::Ready(1)
    }
}

我们注意到 poll 方法中有一个 Context，源码如下：

1
2
3
4
5
6
pub struct Context<'a> {
    waker: &'a Waker,
    // 标记字段
    _marker: PhantomData<fn(&'a ()) -> &'a ()>,
    _marker2: PhantomData<*mut ()>,
}

Context中的 waker 就是用于唤醒 Task，他是由 Runtime 构造并且由 Runtime 调用

Future状态机运转流程

rust-async

Rust异步任务的大致流程包括：

  1. 运行时组件内部首先将IO注册到Poller中（后面会讲到Poller）
  2. 之后将任务放入（spawn）运行时的任务队列（Task queue）中
  3. 运行时不停地从任务内部取出任务并且执行
  4. 任务执行poll方法，例如执行 read/receive/write 系统调用
    4.1. 如果IO没有准备好（Kernel返回WOUND_BLOCK），那么便会返回Pending
    4.2. IO准备好了（Kernel返回Ok（fd）），返回Ready < T >
  5. 接着执行Task queue中的其他任务，没有任务那么就进入wait
  6. 进入wait之后Poller会通过系统调用（例如epoll_wait），根据返回的IO找到之前放入的Waker，Waker执行wake方法将先前Pending的任务放回Task queue

Runtime核心组件

rust-async-2

Runtime的核心组件包括IO组件，Executor以及Reactor

IO组件主要提供异步接口，将自己的fd注册到Reactor上，并在IO未准备就绪时，将waker放到关联任务中。Executor取出并执行任务（poll），并在Task queue空时转向Reactor，Reactor与Kernel打交道，在IO就绪时将关联的任务唤醒（加入Task queue），并将执行权交给Executor。