asyncの実体

この章でわかること:

async fnがコンパイルで何になるか — 状態機械とそのサイズの実測
pollとWakerによる駆動モデル。ランタイム(tokio)が何をしているか
タスクとOSスレッドのコスト差の実測(生成コストで約160倍)
asyncの最大の罠 — ランタイムをブロックする(実測つき)
なぜ体系に必須か — Webアプリケーションの並行処理の主役はasyncです。 5章で「並行はasyncの領分」と割り振ったまま中身を開かなければ、本書の並行の地図は半分白紙のままです

async fnは何にコンパイルされるか

まず最重要の事実から。async fnは、呼んでも実行されません。 8章の遅延評価の実験で見たイテレータと同じく、async fnの呼び出しは「実行の計画書」を作って返すだけです。この計画書がFutureです。

ではFutureの実体は何か。コンパイラはasync fnの本体を、 .awaitの位置で分割した状態機械(state machine)に変換します。 enumで例えるなら、「開始前」「1つ目のawaitで中断中」「2つ目のawaitで中断中」…という状態を持ち、 poll(後述)が呼ばれるたびに続きを実行して次の状態へ進む構造体です。

決定的に重要なのはサイズです。中断中に生きている局所変数は、すべてこの構造体の中に保存されます。実験で確かめます。

use std::mem::size_of_val;

async fn tiny() -> u64 {
    1 + 1
}

// 4KBのバッファを .await をまたいで持つ
async fn holds_buffer() -> u64 {
    let buf = [7u8; 4096];
    tokio::task::yield_now().await; // ここで中断しうる
    buf.iter().map(|&b| b as u64).sum()
}

// 同じバッファでも .await の前に使い終わる
async fn drops_before_await() -> u64 {
    let sum = {
        let buf = [7u8; 4096];
        buf.iter().map(|&b| b as u64).sum()
    };
    tokio::task::yield_now().await;
    sum
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let f1 = tiny();
    let f2 = holds_buffer();
    let f3 = drops_before_await();
    println!("tiny               のFuture: {:>5} bytes", size_of_val(&f1));
    println!("バッファをまたぐ   のFuture: {:>5} bytes", size_of_val(&f2));
    println!("またぐ前に手放す   のFuture: {:>5} bytes", size_of_val(&f3));
    println!("結果: {} {} {}", f1.await, f2.await, f3.await);
}

筆者の実測(Playground)では:

何も持たないasync fnのFuture: 1バイト(状態番号だけ)
4KBの配列を.awaitをまたいで持つFuture: 4120バイト
同じ配列でも.awaitの前に使い終わるFuture: 32バイト

.awaitをまたぐ変数はFutureに埋め込まれる——この規則を知らないと、「大きなバッファを持ったままawaitする関数」を重ねただけで Futureが肥大化します(入れ子のasync fnのFutureは外側のFutureに入れ子で埋まるため、深い呼び出し連鎖では合算されます)。 tokioのspawnはFutureをヒープに置くので、これはアロケーションのサイズ(18章)に直結します。対策も実験の3つ目が示しています——大きな一時データは awaitをまたぐ前に手放す(またはBoxに逃がす)。

poll、Waker、ランタイム

計画書(Future)を実行に移すのがランタイム(async runtime)です。 Rust本体はFutureの形だけを定め、実行機構はtokioなどのライブラリに委ねています。その「形」がFutureトレイトです。

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

見慣れないPin<&mut Self>は「このFutureをもうメモリ上で動かさない」という約束です。状態機械の中には「自分自身の別のフィールドを指す参照」が保存されうるため(awaitをまたいで局所変数への参照が生きる場合)、構造体ごとメモリ上を移動するとその内部参照が壊れます。Pinは移動をコンパイル時に禁じることでこれを防ぎます。Contextは後述のWakerの入れ物です。駆動モデルはプル型です。

ランタイムがFutureのpollを呼ぶ —「続きを進めてください」
Futureは進めるところまで進み、完了ならReady(結果)、待ちが必要ならPendingを返す。このとき「待っているもの (ソケットなど)が準備できたら教えてね」とWakerという通知ハンドルを登録しておく
OSのI/O通知(Linuxならepoll)が来たら、ランタイムはWaker経由でそのタスクを実行待ち行列に戻し、またpollする

つまりasyncの並行性とは、「何万もの状態機械を、少数のOSスレッドが pollして回る」構造です。tokioの既定(マルチスレッドランタイム)はコア数分のワーカースレッドを持ち、5章のrayonと同じワークスティーリングでタスクを分配します。rayonとの違いは仕事の性質だけ——rayonは「計算が終わるまで走る」仕事を、 tokioは「すぐPendingで戻ってくる」仕事を回すよう調律されています。

タスクは何が「軽い」のか — 実測

「asyncのタスクはOSスレッドより軽い」とよく言われます。数字にします。

use std::time::Instant;

fn main() {
    // (1) OSスレッドを1万本
    let start = Instant::now();
    let handles: Vec<_> = (0..10_000)
        .map(|i| std::thread::spawn(move || i as u64 * 2))
        .collect();
    let sum: u64 = handles.into_iter().map(|h| h.join().unwrap()).sum();
    println!("OSスレッド   1万: {:>9.3?} (sum={sum})", start.elapsed());

    // (2) tokioタスクを10万個
    let rt = tokio::runtime::Runtime::new().unwrap();
    let start = Instant::now();
    let sum: u64 = rt.block_on(async {
        let handles: Vec<_> = (0..100_000)
            .map(|i| tokio::spawn(async move { i as u64 * 2 }))
            .collect();
        let mut sum = 0u64;
        for h in handles {
            sum += h.await.unwrap();
        }
        sum
    });
    println!("tokioタスク 10万: {:>9.3?} (sum={sum})", start.elapsed());
}

筆者の実測(Playground)では、OSスレッド1万本の生成と合流が約760ミリ秒(1本あたり約76マイクロ秒)、tokioタスク10万個が約47ミリ秒(1個あたり約0.5マイクロ秒)。1個あたり約160倍の差です (規模も条件も同一ではない粗い比較です——スレッドを10万本作る実験自体が現実的でない、という差も含めて読んでください)。

差の源泉は明快です。OSスレッドはカーネルへのシステムコール (27章)と、スタック用のメモリ領域 (既定で数MBの仮想空間+実体は14章のデマンドページング)を必要とします。tokioのタスクはユーザー空間でのヒープ確保1回(Futureのサイズ分)と待ち行列への追加だけです。「1接続1タスク」を10万接続でやれるのはこの軽さゆえです。

もう1つの含意: タスクの切り替え(あるタスクがPendingになり別のタスクをpollする)もユーザー空間の関数呼び出しにすぎず、 OSのコンテキストスイッチ(27章で実測します)より桁違いに軽いのです。

最大の罠 — ランタイムをブロックする

asyncの協調モデルには前提があります。 タスクはpollされたら速やかにPendingかReadyを返すこと。 pollの中で長時間戻らないタスクは、そのワーカースレッドを占拠し、同じスレッドで回るはずだった他の全タスクを道連れにします。

これを実測します。1スレッド構成のランタイムで、「300ミリ秒の同期スリープ(=CPUを返さないブロッキング処理の代役)」をするタスクAと、「50ミリ秒後に完了するはずの」タスクBを同居させます。

use std::time::{Duration, Instant};

// 1スレッド構成のランタイムで、
// 「重い同期処理」が同居するタスクにどう影響するかを見る
fn main() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_time()
        .build()
        .unwrap();

    // (1) タスクAが std::thread::sleep でスレッドごと止める
    rt.block_on(async {
        let start = Instant::now();
        let b = tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            start.elapsed()
        });
        // タスクBが先にタイマーを登録できるよう、一度実行を譲る
        tokio::task::yield_now().await;
        let a = tokio::spawn(async {
            std::thread::sleep(Duration::from_millis(300)); // ブロッキング!
        });
        a.await.unwrap();
        let b_done = b.await.unwrap();
        println!("(1) ブロッキング同居: 50msのはずのタスクBの完了 = {b_done:?}");
    });

    // (2) 重い処理を spawn_blocking で専用スレッドへ逃がす
    rt.block_on(async {
        let start = Instant::now();
        let b = tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            start.elapsed()
        });
        tokio::task::yield_now().await;
        let a = tokio::task::spawn_blocking(|| {
            std::thread::sleep(Duration::from_millis(300));
        });
        a.await.unwrap();
        let b_done = b.await.unwrap();
        println!("(2) spawn_blocking : 50msのはずのタスクBの完了 = {b_done:?}");
    });
}

use std::time::{Duration, Instant};

// 1スレッド構成のランタイムで、
// 「重い同期処理」が同居するタスクにどう影響するかを見る
fn main() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_time()
        .build()
        .unwrap();

// (1) タスクAが std::thread::sleep でスレッドごと止める
    rt.block_on(async {
        let start = Instant::now();
        let b = tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            start.elapsed()
        });
        // タスクBが先にタイマーを登録できるよう、一度実行を譲る
        tokio::task::yield_now().await;
        let a = tokio::spawn(async {
            std::thread::sleep(Duration::from_millis(300)); // ブロッキング!
        });
        a.await.unwrap();
        let b_done = b.await.unwrap();
        println!("(1) ブロッキング同居: 50msのはずのタスクBの完了 = {b_done:?}");
    });

// (2) 重い処理を spawn_blocking で専用スレッドへ逃がす
    rt.block_on(async {
        let start = Instant::now();
        let b = tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            start.elapsed()
        });
        tokio::task::yield_now().await;
        let a = tokio::task::spawn_blocking(|| {
            std::thread::sleep(Duration::from_millis(300));
        });
        a.await.unwrap();
        let b_done = b.await.unwrap();
        println!("(2) spawn_blocking : 50msのはずのタスクBの完了 = {b_done:?}");
    });
}

筆者の実測(Playground)では、タスクBの完了は約300ミリ秒—— 自分は50ミリ秒のタイマーなのに、タスクAが実行を返さない間一度もpollされず、Aのブロックが明けるまで巻き込まれました。 spawn_blocking(ブロッキング専用のスレッドプールへ逃がすtokioのAPI)に変えると、Bは予定どおり51ミリ秒で完了します。

実務での規則はこうです。

同期I/O、重いCPU計算、ロック待ちの長い処理を、asyncタスクの中で直接やらない。spawn_blocking(または重い計算ならrayonへの受け渡し)で隔離します
マルチスレッドランタイムでも問題は消えません。ワーカーが8本なら、 8個のブロッキングで全停止します。負荷が上がったときだけレイテンシが跳ねる、という発見しにくい形で現れます
「重い」の目安としては、10〜100マイクロ秒を超えてpollを占有する処理、という値がtokioの開発者たちの解説でよく引かれます

使い分けの地図 — async / スレッド / rayon

5章の表を完成させます。

道具	向く仕事	単位コスト
async (tokio)	大量の待ち(I/O)の多重化	タスク約0.5µs、待ちはほぼ無料
OSスレッド	少数の長寿命な並行作業、ブロッキングの隔離	生成約76µs+スタック
rayon	CPUを使い切るデータ並列計算	5章のとおり

Webサーバの典型構成はこの3層の合成です——接続の待ち受けと I/Oはtokio、その中で見つかった重い計算はrayonか spawn_blockingへ。この分業を崩したとき(計算をasyncで直接回す、 I/Oをrayonで待つ)に、この章の罠が発動します。

まとめ

async fnは.awaitで分割された状態機械を返します。 awaitをまたぐ変数はFutureに埋め込まれ、サイズを実測で 1バイト〜4120バイトまで見ました
ランタイムはpollとWakerで状態機械を駆動します。tokioはコア数のワーカー+ワークスティーリングです
タスクの生成はスレッドの約160分の1のコストでした。「1接続1タスク」の根拠です
pollを長時間占有するとランタイムごと止まります(実測: 50msのタスクが約300msに)。同期I/Oと重い計算はspawn_blockingやrayonへ
async・スレッド・rayonは競合ではなく、待ち/長寿命/計算の分業です

次章は、最適化の最終手段としてたびたび登場したunsafeを正面から扱います。未定義動作とは何か、何が許され何が許されないか、そしてそれを機械的に検査する道具(Miri)まで。