Skip to content
Published on

非同期Rust：async/awaitとTokioランタイム

🇰🇷KO🇺🇸EN🇯🇵JA
Split View필사 모드
Authors

はじめに — なぜRustのasyncは違うのか

多くの言語では、非同期はランタイムに組み込まれています。JavaScriptには最初からイベントループが回っており、Goにはゴルーチンをスケジューリングするランタイムが既定で付いてきます。しかしRustは違います。Rustはasync構文を言語レベルで提供しますが、それを実際に動かすランタイムは標準ライブラリに入れませんでした。 ランタイムは別のクレート（多くはTokio）から選んで使う構造です。

この選択は最初こそ奇妙に感じますが、理由があります。Rustは組み込み機器から大規模サーバーまで幅広く使われますが、あらゆる状況に合う単一のランタイムは存在しません。そこで言語は最小限の骨組み（Future トレイトと async/await 構文）だけを提供し、スケジューリング方針は利用者に委ねます。

この記事はその骨組みから始めます。Future とは何か、.await が実際に何をするのか、その上でTokioがどうタスクを動かすのか、そして実務で必ずぶつかる罠（Send制約、ブロッキングコード）まで、例で押さえます。

Futureは怠惰な状態機械だ

Rustで async fn を呼ぶと関数本体がすぐ実行されそうですが、何も起こりません。 async fn は呼び出し時点で本体を実行する代わりに、Future という値を一つ作って返すだけです。この Future は「まだ完了していない計算」を表す値です。

async fn say_hello() {
    println!("hello");
}

fn main() {
    let fut = say_hello(); // ここでは"hello"は出力されない！
    // futはただのFuture値で、まだ実行されていない
}

これがRust asyncの核心の性質、**怠惰さ（laziness）です。Futureは誰かがそれをポーリング（poll）**してくれるまで一歩も進みません。ポーリングとは「いま進められる？」と尋ねることです。Futureはこの問いに二通りで答えます。計算が終わっていれば Poll::Ready(値) を、まだ待つものがあれば Poll::Pending を返します。

Future トレイトの骨組みはおおよそこう見えます。

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

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

コンパイラは async ブロックを**状態機械（state machine）**へ変換します。各 .await 地点が一つの状態になり、ポーリングされるたびに前に止まった地点から再開して次の .await まで進みます。この状態機械のおかげで、スタックを使わずに関数の途中状態を保存でき、だからRustのasyncはヒープ割り当てなしでも動くほど軽量です。

.await — 途中で譲る地点

.await は他のFutureが完了するのを待つ地点です。しかし「待つ」といってもスレッドを掴んで止まるわけではありません。.await の本当の意味は、**「このFutureがまだPendingなら、制御権を実行器へ譲って他の仕事をさせろ」**です。

async fn fetch_and_process() {
    let data = fetch_data().await;     // ここで待機中なら他のタスクへ譲る
    let result = process(data).await;  // 完了したら続けて進む
    println!("{result}");
}

fetch_data().await でデータがまだ準備できていなければ、この関数はその場で止まって制御を手放します。すると実行器はその間、他のタスクを動かします。あとでデータが準備できると、実行器はこのFutureを再びポーリングし、関数は止まっていた .await 地点から再開します。この協調的な譲り合いが、一つのスレッドで膨大な並行処理をこなせる秘訣です。

ここで重要な規則が一つ。.await は async 文脈（async fn や async ブロック）の中でしか使えません。通常の同期関数からFutureを .await することはできません。では最初のFutureは誰がポーリングを始めるのでしょうか。ランタイムです。

実行器とランタイム — Tokio

Futureは怠惰なので、誰かがポーリングを始めて動かし続けて初めて実際に実行されます。その役割を担うのが実行器（executor）で、実行器にタイマーやネットワークI/Oといった付加機能を加えて仕上げたものがランタイム（runtime）です。Rustのエコシステムで事実上の標準はTokioです。

もっとも簡単な始め方は #[tokio::main] マクロです。このマクロが main 関数を包んでランタイムを立て、あなたの最上位のFutureをその上で完了までポーリングして動かします。

#[tokio::main]
async fn main() {
    println!("start");
    say_hello().await; // ここで実際に実行される
    println!("end");
}

async fn say_hello() {
    println!("hello");
}

#[tokio::main] は実際、次のようなコードへ展開されます。同期の main がランタイムを作り、block_on でasyncブロックを完了まで動かすのです。

fn main() {
    tokio::runtime::Runtime::new()
        .unwrap()
        .block_on(async {
            println!("start");
            say_hello().await;
            println!("end");
        });
}

Tokioは既定でマルチスレッドスケジューラを使います。CPUコア数ぶんのワーカースレッドを持ち、タスクをその上に分配して実行します。あるワーカーが暇なら別のワーカーのタスクを奪ってくる**ワークスティーリング（work-stealing）**方式で負荷を均等に分けます。軽量な単一スレッドランタイムが必要なら #[tokio::main(flavor = "current_thread")] に切り替えることもできます。

タスクとspawn — 並行の単位

ここまでは一つのFutureを順番に .await してきました。しかし非同期の真の力は、複数の作業を同時に動かすときに出ます。その単位が**タスク（task）**で、tokio::spawn で作ります。

tokio::spawn はFutureを受け取り、ランタイムに独立したタスクとして登録します。このタスクは即座にバックグラウンドで実行を始め、spawn はそのタスクのハンドル（JoinHandle）をすぐに返します。タスクはOSスレッドよりずっと軽量なので、数万個立ち上げても負担は小さいです。

#[tokio::main]
async fn main() {
    let mut handles = vec![];

    for i in 0..5 {
        let handle = tokio::spawn(async move {
            // 各タスクが独立して、同時に実行される
            some_async_work(i).await
        });
        handles.push(handle);
    }

    // すべてのタスクが終わるのを待つ
    for handle in handles {
        let result = handle.await.unwrap();
        println!("got: {result}");
    }
}

async fn some_async_work(i: u32) -> u32 {
    i * 2
}

spawn したタスクは親が .await しなくても自ら進みますが、結果を受け取るには JoinHandle を .await します。タスクがパニックしていれば .await の結果が Err で返るので、ここで確認できます。

メッセージキューやワーカープールのように、複数の生産者・消費者がタスクとして行き交って通信する構造が気になるなら、当サイトのメッセージキュープレイグラウンドでキューの動作を視覚的に試せます。非同期タスクがチャネルを通じて仕事を受け渡す様子を直感的に理解するのに役立ちます。

Sendと静的ライフタイム — spawnの制約

tokio::spawn に渡すFutureには二つの制約が付きます。シグネチャを見れば理由が分かります。渡すFutureは Send でなければならず、'static ライフタイムでなければなりません。

まず**Send 制約**から見ましょう。マルチスレッドランタイムでは、タスクはどのワーカースレッドでも実行され得て、さらに .await を挟んで別のスレッドへ移されることもあります。そのためにはFutureが抱えるすべての値がスレッド間で安全に移動できる、つまり Send である必要があります。だから .await をまたいで Send でない値を保持しているとコンパイルエラーになります。代表例が、先のスマートポインタの記事で見た Rc です。

use std::rc::Rc;

async fn bad() {
    let data = Rc::new(5); // RcはSendではない
    some_async_work().await; // エラー: .awaitをまたいでRcを保持している
    println!("{}", data);
}

この場合の解決策は Rc の代わりに Arc を使うことです。あるいは Send でない値を .await の前にスコープから落とし、.await 地点をまたがせない方法もあります。

'static 制約は、タスクのライフタイムが親スコープより長くなり得るためです。spawn したタスクは親がすでに戻った後も生きている可能性があるので、親スコープの参照を借りると危険です。だからFutureは借りた参照ではなく値を所有しなければなりません。実務ではこれは、クロージャに move を付け、必要なデータを Arc で共有する形でよく解決されます。

非同期トレイト — async fn in traits

しばらくの間、Rustではトレイトの中に async fn を直接置くことができませんでした。async fn が返すFutureのサイズがトレイトレベルで分からなかったからです。そこで長らく async-trait というクレートがこの隙間を埋めてきました。このクレートはマクロで戻り値の型を Box<dyn Future> に変え、ヒープ割り当てを引き換えにトレイトへ非同期メソッドを入れられるようにします。

use async_trait::async_trait;

#[async_trait]
trait Repository {
    async fn find(&self, id: u64) -> Option<String>;
}

struct MyRepo;

#[async_trait]
impl Repository for MyRepo {
    async fn find(&self, id: u64) -> Option<String> {
        // 実際にはDBを非同期で照会する
        Some(format!("item {id}"))
    }
}

良い知らせは、比較的最近のRustでトレイト内の async fn が言語レベルで対応され始めたことです。いまや多くの場合、async-trait マクロなしでもトレイトに async fn を直接書けます。ただしトレイトオブジェクト（dyn）として使う場合や、戻り値のFutureに Send 境界を付ける場合など、一部の細部では依然として気を配る点があり、フレームワークによっては async-trait が引き続き使われることもあります。

select! — 複数のFutureのうち先に終わったもの

複数のFutureを同時に待って、そのうちもっとも先に終わった一つに反応したいときがあります。代表的には「作業を待つが、タイムアウトが先に来たら取り消す」場合です。こういうとき tokio::select! マクロを使います。

select! は列挙した複数のFutureを同時にポーリングし、そのうち一つが先に完了したらそのアームを実行し、残りは取り消します。

use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    tokio::select! {
        result = do_work() => {
            println!("作業完了: {result}");
        }
        _ = sleep(Duration::from_secs(5)) => {
            println!("タイムアウト！作業を諦める");
        }
    }
}

async fn do_work() -> u32 {
    sleep(Duration::from_secs(10)).await;
    42
}

上の例では do_work は10秒かかりますがタイムアウトは5秒なので、5秒後にタイマーのアームが先に完了してそちらが実行され、do_work は取り消されます。ここでRust asyncの重要な性質が表れます。Futureをポーリングしなければそのまま取り消されるという点です。進行中だった do_work は単にもうポーリングされず、自然に捨てられます。別途の取り消し信号を送る必要はありません。

select! をループの中で使うと、複数のイベントソース（チャネル受信、タイマー、終了信号など）を一つのループで同時に監視するイベントループを作れます。これが非同期サーバーの心臓部でよく見られるパターンです。

ブロッキング vs 非同期 — 混ぜてはいけない理由

非同期プログラミングでもっともよくあり致命的な間違いは、非同期文脈の中でブロッキングコードを呼ぶことです。ここで「ブロッキング」とは、CPUを長く使う重い計算や、std::thread::sleep、同期ファイルI/O、同期ネットワーク呼び出しのようにスレッドを実際に止める作業を指します。

なぜ危険なのでしょうか。非同期タスクはワーカースレッドを少しの間借りてポーリングされ、.await で譲ってスレッドを手放します。ところがタスクがブロッキング呼び出しをすると、譲らずにワーカースレッドを丸ごと掴んでしまいます。そのワーカーに割り当てられた他のすべてのタスクはその間まったく進めず飢えます。最悪の場合、いくつかのブロッキングタスクがすべてのワーカーを占有し、ランタイム全体が止まります。

// 悪い例: 非同期文脈でスレッドを丸ごとブロッキング
async fn bad() {
    std::thread::sleep(std::time::Duration::from_secs(5)); // ワーカースレッドを5秒間掴む！
}

// 良い例: 非同期タイマーで譲る
async fn good() {
    tokio::time::sleep(std::time::Duration::from_secs(5)).await; // スレッドを手放す
}

まとめると二つの原則です。一つ目、眠ったり待ったりする作業は必ず非同期版を使いましょう。std::thread::sleep の代わりに tokio::time::sleep、同期HTTPの代わりに非同期HTTPクライアントのように。二つ目、CPUを長く使う重い計算やどうしても同期なライブラリは tokio::task::spawn_blocking で専用のブロッキングスレッドプールへ切り離しましょう。こうすれば重い作業がasyncワーカーを飢えさせません。

async fn heavy() {
    let result = tokio::task::spawn_blocking(|| {
        // CPUを長く使う計算や同期I/Oはここで
        expensive_sync_computation()
    })
    .await
    .unwrap();
    println!("{result}");
}

fn expensive_sync_computation() -> u64 {
    (0..1_000_000).sum()
}

おわりに

Rustのasyncは最初こそ見慣れませんが、いくつかの核心を掴めば一貫した絵につながります。

  • async fn は怠惰なFutureを作るだけで、ポーリングされるまで何もしません。
  • .await はスレッドを掴む代わりに制御権を譲る地点です。
  • Futureを実際に動かすには**ランタイム（Tokio）**が必要で、tokio::spawn で並行タスクを立ち上げます。
  • spawn したFutureは Send + 'static でなければならず、スレッド間の移動とライフタイムを意識します。
  • select! で複数のFutureを競わせ、ポーリングを止めればFutureは自然に取り消されます。
  • 非同期文脈ではブロッキングを避け、重い同期作業は spawn_blocking で隔離しましょう。

これらの原則を身につければ、Rustのasyncは安全性と性能を同時に与える強力な道具になります。言語が与えるのは骨組みだけですが、その骨組みが明確だからこそ、あなたは何が起きているかを正確に理解しながらコードを書けます。

参考資料

Discuss on Twitter • View on GitHub