【译】Async/Await(三)——Aysnc/Await模式

原文标题:Async/Await
原文链接:https://os.phil-opp.com/async-await/#multitasking
公众号: Rust 碎碎念
翻译 by: Praying

Async/Await 模式(The Async/Await Pattern)

async/await 背后的思想是让程序员能够像写普通的同步代码那样来编写代码,由编译器负责将其转为异步代码。它基于asyncawait两个关键字来发挥作用。async关键字可以被用于一个函数签名,负责把一个同步函数转为一个返回 future 的异步函数。

async fn foo() -> u32 {
    0
}

// the above is roughly translated by the compiler to:
fn foo() -> impl Future<Output = u32> {
    future::ready(0)
}

这个关键字是无法单独发挥作用的,但是在async函数内部,await关键字可以被用于取回(retrieve)一个 future 的异步值。

async fn example(min_len: usize) -> String {
    let content = async_read_file("foo.txt").await;
    if content.len() < min_len {
        content + &async_read_file("bar.txt").await
    } else {
        content
    }
}

尝试在 playground 上运行这段代码[1]

这个函数是对example函数的一个直接转换,example函数使用了上面提到的组合子函数(译注:在译文 Async/Await(二)中)。通过使用.await操作,我们能够在不需要任何闭包或者Either的情况下检索一个 future 的值。因此,我们可以像写普通的同步代码一样来写我们的代码,不同之处在于我们写的仍然是异步代码。

状态机转换

编译器在背后把async函数体转为一个状态机(state machine)[2],每一个.await调用表示一个不同的状态。对于上面的example函数,编译器创建了一个带有下面四种状态的状态机:

【译】Async/Await(三)——Aysnc/Await模式

每个状态表示函数中一个不同的暂停点。"Start"和"End"状态表示开始执行的函数和执行结束的函数。"Waiting on foo.txt"状态表示函数当前正在等待第一个async_read_file的结果。类似地,"Waiting on bar.txt"表示函数正在等待第二个async_read_file结果。

这个状态机通过让每一个poll调用成为一次状态转换来实现Future trait。

【译】Async/Await(三)——Aysnc/Await模式

上面这张图用箭头表示状态切换,用菱形表示分支路径。例如,如果foo.txt没有准备好,就会选择标记"no"的路径然后进入”Waiting on foo.txt“状态。否则,就会选择"yes"路径。中间较小的没有标题的红色菱形表示example函数的if content.len() < 100分支。

我们可以看到第一个poll调用启动了这个函数并使函数一直运行直到它到达一个尚未就绪的 future。如果这条路径上的所有 future 都已就绪,该函数就可以一直运行到"End"状态,这里它把自己的结果包装在Poll::Ready中然后返回。否则,状态机进入到一个等待状态并返回"Poll::Pending"。在下一个poll调用时,状态机从上次等待状态开始然后重试上次操作。

保存状态

为了能够从上次等待状态继续下去,状态机必须在内部记录当前状态。此外,它还必须要保存下次poll调用时继续执行需要的所有变量。这也正是编译器大展身手的地方:因为编译器知道哪个变量在何时被使用,所以它可以自动生成结构体,这些结构体准确地包含了所需要的变量。

例如,编译器可以针对上面的example函数生成类似下面的结构体:

//  再次放上`example` 函数 ,你就不用去上面找它了
async fn example(min_len: usize) -> String {
    let content = async_read_file("foo.txt").await;
    if content.len() < min_len {
        content + &async_read_file("bar.txt").await
    } else {
        content
    }
}

// 编译器生成的状态结构体:

struct StartState {
    min_len: usize,
}

struct WaitingOnFooTxtState {
    min_len: usize,
    foo_txt_future: impl Future<Output = String>,
}

struct WaitingOnBarTxtState {
    content: String,
    bar_txt_future: impl Future<Output = String>,
}

struct EndState {}

在"Start"和"Waiting on foo.txt"这两个状态(分别对应 StartState 和 WaitingOnFooTxtState 结构体)里,参数min_len需要被存储起来,因为在后面和content.len()进行比较时会需要用到它。"Waiting on foo.txt"状态还需要额外存储一个foo_txt_future,它表示由async_read_file调用返回的 future。这个 future 在当状态机继续的时候会被再次轮询(poll),所以它也需要被保存起来。

"Waiting on bar.txt"状态(译注:对应WaitingOnBarTxtState 结构体)包含了content变量,因为它会在bar.txt就绪后被用于字符串拼接。该状态还存储了一个bar_txt_future用以表示对bar.txt正在进行的加载。WaitingOnBarTxtState结构体不包含min_len变量因为它在和 content.len()比较后就不再被需要了。在"End"状态下,没有存储任何变量,因为函数在这里已经运行完成。

注意,这里只是编译器针对代码可能生成的一个示例。结构体的命名以及字段的布局都是实现细节并且可能有所不同。

完整的状态机类型

虽然具体的编译器生成代码是一个实现细节,但是它有助于我们理解example函数生成的状态机看起来是怎么样的?我们已经定义了表示不同状态的结构体并且包含需要的字段。为了能够在此基础上创建一个状态机,我们可以把它组合进enum

enum ExampleStateMachine {
    Start(StartState),
    WaitingOnFooTxt(WaitingOnFooTxtState),
    WaitingOnBarTxt(WaitingOnBarTxtState),
    End(EndState),
}

我们为每个状态定义一个单独的枚举变量,并且把对应的状态结构体添加到每个变量中作为一个字段。为了实现状态转换,编译器基于example函数生成了一个Future trait 的实现:

impl Future for ExampleStateMachine {
    type Output = String// return type of `example`

    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
        loop {
            match self { // TODO: handle pinning
                ExampleStateMachine::Start(state) => {…}
                ExampleStateMachine::WaitingOnFooTxt(state) => {…}
                ExampleStateMachine::WaitingOnBarTxt(state) => {…}
                ExampleStateMachine::End(state) => {…}
            }
        }
    }
}

future 的Output类型是String,因为它是example函数的返回类型。为了实现poll函数,我们在loop内部对当前的状态使用一个 match 语句。其思想在于只要有可能就切换到下一个状态,当无法继续的时候就使用一个显式的return Poll::Pending

简单起见,我们只能展示简化的代码且不对pinning[3]、所有权、生命周期等进行处理。所以,这段代码以及接下来的代码就当成是伪代码,不要直接使用。当然,实际上编译器生成的代码已经正确地处理好了一切,尽管可能是以另一种方式。

为了让代码片段尽可能地小,我们为每个 match 分支单独展示代码。让我们先从Start状态开始:

ExampleStateMachine::Start(state) => {
    // from body of `example`
    let foo_txt_future = async_read_file("foo.txt");
    // `.await` operation
    let state = WaitingOnFooTxtState {
        min_len: state.min_len,
        foo_txt_future,
    };
    *self = ExampleStateMachine::WaitingOnFooTxt(state);
}

状态机在函数开始时就处于Start状态,在这种情况下,我们从example函数体执行所有的代码,直至遇到第一个.await。为了处理.await操作,我们把self状态机的状态更改为WaitingOnFooTxt,该状态包括了对WaitingOnFooTxtState的构造。

因为match self {...} 状态是在一个循环里执行的,这个执行接下来跳转到WaitingOnFooTxt分支:

ExampleStateMachine::WaitingOnFooTxt(state) => {
    match state.foo_txt_future.poll(cx) {
        Poll::Pending => return Poll::Pending,
        Poll::Ready(content) => {
            // from body of `example`
            if content.len() < state.min_len {
                let bar_txt_future = async_read_file("bar.txt");
                // `.await` operation
                let state = WaitingOnBarTxtState {
                    content,
                    bar_txt_future,
                };
                *self = ExampleStateMachine::WaitingOnBarTxt(state);
            } else {
                *self = ExampleStateMachine::End(EndState));
                return Poll::Ready(content);
            }
        }
    }
}

在这个 match 分支,我们首先调用foo_txt_futurepoll函数。如果它尚未就绪,我们就退出循环然后返回Poll::Pending。因为这种情况下self仍处于WaitingOnFooTxt状态,下一次的poll调用将会进入到相同的 match 分支然后重试对foo_txt_future轮询。

foo_txt_future就绪后,我们把结果赋予content变量并且继续执行example函数的代码:如果content.len()小于保存在状态结构体里的min_lenbar.txt文件会被异步地读取。我们再次把.await操作转换为一个状态改变,这次改变为WaitingOnBarTxt状态。因为我们在一个循环里面正在执行match,执行流程直接跳转到新的状态对应的 match 分支,这个新分支对bar_txt_future进行了轮询。

一旦我们进入到else分支,后面就不再会进行.await操作。我们到达了函数结尾并返回包装在Poll::Ready中的content。我们还把当前的状态改为了End状态。

WaitingOnBarTxt状态的代码看起来像下面这样:

ExampleStateMachine::WaitingOnBarTxt(state) => {
    match state.bar_txt_future.poll(cx) {
        Poll::Pending => return Poll::Pending,
        Poll::Ready(bar_txt) => {
            *self = ExampleStateMachine::End(EndState));
            // from body of `example`
            return Poll::Ready(state.content + &bar_txt);
        }
    }
}

WaitingOnFooTxt状态类似,我们从轮询bar_txt_future开始。如果它仍然是 pending,我们退出循环然后返回Poll::Pending。否则,我们可以执行example函数最后的操作:将来自 future 的结果与content相连接。我们把状态机更新到End状态,然后将结果包装在Poll::Ready中进行返回。

最后,End状态的代码看起来像下面这样:

ExampleStateMachine::End(_) => {
    panic!("poll called after Poll::Ready was returned");
}

在返回Poll::Ready之后,future 不应该被再次轮询。因此,当我们已经处于End状态时,如果poll被调用我们将会 panic。

我们现在知道编译器生成的状态机以及它对Future trait 的实现是什么样子的了。实际上,编译器是以一种不同的方式来生成代码。(如果你感兴趣的话,当前的实现是基于生成器(generator)[4]的,但是这只是一个实现细节)。

最后一部分是生成的示例函数本身的代码。记住,函数签名是这样定义的:

async fn example(min_len: usize) -> String

因为完整的函数体实现是通过状态机来实现的,这个函数唯一需要做的事情是初始化状态机并将其返回。生成的代码看起来像下面这样:

fn example(min_len: usize) -> ExampleStateMachine {
    ExampleStateMachine::Start(StartState {
        min_len,
    })
}

这个函数不再有async修饰符,因为它现在显式地返回一个ExampleStateMachine类型,这个类型实现了Future trait。正如所期望的,状态机在Start状态被构造,并使用min_len参数初始化与之对应的状态结构体。

记住,这个函数没有开始状态机的执行。这是 Rust 中 future 的一个基本设计决定:在第一次轮询之前,它们什么都不做。

【译】Async/Await(三)——Aysnc/Await模式

参考资料

[1]

尝试在 playground 上运行这段代码: https://play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=d93c28509a1c67661f31ff820281d434


[2]

状态机(state machine): https://en.wikipedia.org/wiki/Finite-state_machine


[3]

pinning: https://doc.rust-lang.org/stable/core/pin/index.html


[4]

生成器(generator): https://doc.rust-lang.org/nightly/unstable-book/language-features/generators.html


发表评论

评论已关闭。

相关文章