Rustのasync/awaitの特徴4つ

async/awaitの最小限の機能が、 Rust 1.38.0 リリースを目標に準備されています。Rust1.36.0のリリースが2019-07-04で、Rustは6週間ごとにリリースされるので、順調にいけば 2019-09-26 頃にリリースされると思われます。もちろんnightlyではすでに試せます。

さてこのasync/awaitですが、他の言語のasync/awaitと基本コンセプトは近いものの、いくつか異なる点があります。個人的には以下のことを把握しておくとよいと思いました。

後置await構文
戻り値型 (内部戻り値型・実装ごとに異なる型)
駆動 (awaitまたはspawnしないと進まない)
キャンセル (awaitポイントは中止ポイント)

本稿は現象の説明にとどめ、そうなっている理由には基本的に言及しませんが、どれもきちんと理由があってそうなっています。その点はご承知おきください。

1. 後置await構文

Rustのasync/awaitでは、後置await構文が採用されました。これについて1つだけ知っておいてほしいのは、これは気まぐれで決定されたわけではなく、多くの提案と議論があったなかで、Rustの非同期エコシステムを前進させるためになされた苦渋の決断だということです。Rustユーザーやメンテナの全員が後置awaitに賛成というわけではなく、すべての選択肢に利点と欠点がある中で選ばれたものであることをご承知おきください。詳しくはたとえばIRLO: A final proposal for await syntaxなどを参照してください。

(playground)

pub async fn post(pool: &Pool, id: PostId) -> Result<Post, Error> {
    let mut conn = pool.connect().await?;
    let post = Post::find(&mut conn, id)
        .await?
        .ok_or_else(|| NotFoundError::new(id))?;
    Ok(post)
}

この .await がRustにおけるawait記法です。多くの場合、エラーハンドリングのための ? 演算子と併用されます。

2. 戻り値型

Rustのasync/awaitは、戻り値型を内部型方式 (関数の内側から見たときの戻り値型) で記述します。たとえば、上の

pub async fn post(pool: &Pool, id: PostId) -> Result<Post, Error> {
    ...
}

というシグネチャは、外から見ると

pub fn post(pool: &Pool, id: PostId) -> impl Future<Output = Result<Post, Error>> {
    ...
}

と(ほぼ)等価です。

また、実際の戻り値型は impl Future とされていますが、これは単一の型ではないことに注意が必要です。実装ごとに異なる匿名型が割り当てられているので、特定の状況下ではそのまま代入できない場合があります。(このあたりの事情はRustのクロージャやイテレーターと同じなので省略します。)

同一の型にまとめるときは、 Pin<Box<dyn Future<Output = _>>> という型を使います。

(playground)

#![feature(async_closure)]

use futures::prelude::*;
use std::pin::Pin;

// 非同期コールバックを登録できる構造体
pub struct Callbacks {
    callbacks: Vec<Box<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>>,
}

impl Callbacks {
    pub fn new() -> Self {
        Self {
            callbacks: Vec::new(),
        }
    }

    pub fn push<F, Fut>(&mut self, f: F)
    where
        F: Fn() -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        self.callbacks.push(Box::new(move || f().boxed()))
    }
}

#[test]
fn test_callback() {
    let mut cbs = Callbacks::new();
    cbs.push(async || ());
}

3. 駆動

Futureはawaitまたはspawnしないと進みません。たとえば、次のコードは何も起きません。

(playground)

#![feature(async_await)]

fn main() {
    let _ = foo(); // 何も起きない
}

async fn foo() {
    eprintln!("foo");
}

ただし、 async fn を使わずに自作した非同期関数では、前処理が実行される可能性があります。たとえば次のコードは foo だけを出力します。

(playground)

fn main() {
    let _ = foo();
}

fn foo() -> impl Future<Output = ()> {
    eprintln!("foo");
    async {
        eprintln!("bar");
    }
}

上の例では let _ = で明示的に捨てていますが、Futureを暗黙的に捨てようとした場合は警告が出るため、これ自体でハマることは少ないと思います。しかし、たとえば以下の例は注意が必要です。

(playground)

// fut1とfut2を並列的に実行したかったかもしれないパターン
// (実際には直列的に実行される)
async fn foo() {
    let fut1 = task1();
    let fut2 = task2();
    fut1.await; // ここで初めてfut1が駆動される
    fut2.await; // ここで初めてfut2が駆動される
}

並列に実行したい場合は tokio::spawn や runtime::spawn などでタスクをspawnする必要があります。 (spawnの方法は使っているランタイムによって異なるので一概には言えません)

4. キャンセル

以下のような空想上の例を考えます。

(playground)

async fn foo() {
    eprintln!("task1 started.");
    task1().await; // ここで実行が終わることがありえる
    eprintln!("task1 done.");
}

これを実行したさい、 task1().await; で処理が終わってしまい、続く eprintln! が実行されないケースがありえます。たとえば、以下のような状況だったとします

foo 自体が、HTTPサーバーのリクエストに対する応答処理の一部である。
サーバー側でタイムアウトが設定されている。
task1 はDBアクセスである。

この状況で、DBアクセス中にデッドラインを超過した場合、 foo の処理はそこで中止されます。

await の後に掃除系の処理をしていた場合などは問題になる可能性があります。必ず実行したい後処理がある場合、以下のような対策が可能です。

まとまって実行したい部分を別タスクに spawn してしまう。この場合、タイムアウトが発生しても「spawnしたタスクの終了を待つ部分」が中止されるだけで、別タスクに分離した部分は影響ありません。
掃除系の処理を Drop によって行うようにする。ファイルや接続のクローズ処理は今でもこの方法が使われています。

まとめ

Rustのasync/awaitを使う上で以下のような違いを覚えておくとハマりにくいと思います。

後置await構文
戻り値型 (内部戻り値型・実装ごとに異なる型)
駆動 (awaitまたはspawnしないと進まない)
キャンセル (awaitポイントは中止ポイント)

これらに加えて、スレッド安全性や所有権・借用まわりでもたくさん苦労すると思いますが、これはコンパイラが教えてくれるので予め覚えておく必要はそこまでないでしょう。