Day 036 — Futureは「怠惰」な値。async fn内の同期ブロッキングは全タスクを止める

📚 背景知識（読んでから問題へ）

Day 031〜035ではstd::thread・Arc<Mutex<T>>・mpscチャンネル・Send/Syncという「OSスレッドによる並行処理」の土台を固めてきました。今日から数日は、Rustのもう一つの並行処理モデルである非同期処理（async/await）に入ります。

OSスレッドと何が違うのか。std::thread::spawnで作るスレッドはOS（カーネル）がスケジューリングする実体で、1つのOSスレッドはスタック領域だけで数百KB〜数MBを消費します。数千〜数百万の同時接続を捌くI/Oバウンドな処理でOSスレッドを1接続1スレッドで扱うと、生成コスト・メモリ使用量がボトルネックになります（いわゆる"C10K問題"）。Rustのasync fnが生み出すのはOSスレッドではなく、軽量な「タスク」です。1つのasyncタスクの状態は数十〜数百バイト程度に収まることが多く、数十万タスクを1つのOSスレッド上で並行実行することも現実的です。

Futureトレイト — asyncの正体。async fn foo() -> i32 { ... }という関数は、呼び出してもその場では何も実行されません。呼び出した瞬間に返るのはstd::future::Futureトレイトを実装した1つの状態機械（ステートマシン）の値です。

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

pub enum Poll<T> {
    Ready(T),
    Pending,
}

Futureは「未来のある時点で値が得られるかもしれない計算」を表す受動的な値です。誰か（ランタイム）がpollを呼んで初めて一歩前進します。Poll::Pendingを返す際、Futureは自分のWaker（cx.waker()）をI/O完了などの通知元に登録しておき、準備が整ったらwake()経由でランタイムに「もう一度pollしてよい」と伝えます。これによりランタイムは「まだ準備ができていないFutureをビジーループで何度もpollし続ける」という無駄を避けられます。

async fnの本体はコンパイラによって、変数の状態を保持するenum（ステートマシン）に変換されます。.awaitのたびに「ここまで進んだ」という状態がバリアントとして刻まれ、次にpollされたときはそこから再開します。これはヒープ確保や動的ディスパッチを一切必要としない、コルーチンのゼロコスト実装です。

ランタイムが必要な理由。Futureはただの値なので、誰かがpollし続けなければ永遠に何も起きません。標準ライブラリは意図的にpollし続ける「エグゼキュータ」を提供せず、tokio・async-std・smolなど複数のランタイムが並立できる設計になっています。本スキルではtokioを標準採用します。

📝 問題

要件1（概念説明）: 次の3つの問いに、コードを交えず日本語の文章で説明してください。

1. async fn foo() {}を呼び出しただけで（.awaitせずに）関数呼び出し文だけ書いた場合、fooの中身は実行されるか。理由も述べよ。
2. Future::pollがPoll::Pendingを返したあと、そのFutureが「いつ再度pollされるべきか」をランタイムはどうやって知るのか。
3. async fnの中でstd::thread::sleep(Duration::from_secs(1))（同期的なブロッキングsleep）を呼ぶと、tokioランタイム上で何が問題になるか。tokio::time::sleep(...).awaitとは何が違うか。

要件2（実装・比較実験）: 以下の2つの非同期タスクを同時に実行し、実行順序のログを観察するプログラムをtokioで書いてください。

• タスクA: std::thread::sleep（同期ブロッキング）で200ms待ってから"A done"を出力する処理を、あえてasync fnの中に書く
• タスクB: 100ms間隔で3回"B tick"を出力する処理をtokio::time::sleep(...).awaitで実装する

これを#[tokio::main(flavor = "current_thread")]（シングルスレッドランタイム）でtokio::join!を使って同時に実行し、「Bのtickが期待通りのタイミングで出力されない（Aのブロッキングにスレッドを奪われる）」ことを実際にログで確認してください。

要件3（修正）: 要件2のタスクAをtokio::task::spawn_blockingを使って正しく非同期タスクの外に逃がすように修正し、タスクBが期待通りのタイミングで動くようになることを確認してください。

🔍 ヒント（段階的開示）

📋 コピーuse std::time::{Duration, Instant};

async fn blocking_task_a(start: Instant) {
    std::thread::sleep(Duration::from_millis(200)); // ← ここが問題
    println!("[{:?}] A done", start.elapsed());
}

async fn ticking_task_b(start: Instant) {
    for i in 1..=3 {
        tokio::time::sleep(Duration::from_millis(100)).await;
        println!("[{:?}] B tick {}", start.elapsed(), i);
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let start = Instant::now();
    tokio::join!(blocking_task_a(start), ticking_task_b(start));
}

✅ 模範解答

📋 コピー// Cargo.toml に以下を追加して検証すること:
// [dependencies]
// tokio = { version = "1", features = ["full"] }

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

/// 【問題のあるバージョン】async fnの中で同期的にブロッキングするタスク。
/// std::thread::sleepは.awaitではないため、ランタイムに制御を返さず
/// 実行中のOSスレッドを200ms独占し続ける。
async fn blocking_task_a(start: Instant) {
    println!("[{:>5?}] A start (blocking sleep 200ms)", start.elapsed());
    std::thread::sleep(Duration::from_millis(200)); // 協調ポイントが無い
    println!("[{:>5?}] A done", start.elapsed());
}

/// 100ms間隔で3回tickするタスク。tokio::time::sleepは.awaitのたびに
/// 「準備ができるまでこのスレッドを他のタスクに譲る」という協調ポイントになる。
async fn ticking_task_b(start: Instant) {
    for i in 1..=3 {
        tokio::time::sleep(Duration::from_millis(100)).await;
        println!("[{:>5?}] B tick {}", start.elapsed(), i);
    }
}

/// 【修正バージョン】spawn_blockingで同期処理を専用スレッドプールへ逃がす。
/// これによりcurrent_threadランタイムのメインスレッドはブロックされず、
/// タスクBが本来のタイミング通りに進行できる。
async fn fixed_task_a(start: Instant) {
    println!("[{:>5?}] A(fixed) start", start.elapsed());
    tokio::task::spawn_blocking(move || {
        std::thread::sleep(Duration::from_millis(200));
    })
    .await
    .expect("blocking task panicked");
    println!("[{:>5?}] A(fixed) done", start.elapsed());
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    println!("=== 要件2: ブロッキングがBを止める様子を観察 ===");
    let start = Instant::now();
    tokio::join!(blocking_task_a(start), ticking_task_b(start));

    println!();
    println!("=== 要件3: spawn_blockingで修正した後 ===");
    let start2 = Instant::now();
    tokio::join!(fixed_task_a(start2), ticking_task_b(start2));
}

#[cfg(test)]
mod tests {
    use super::*;

    // current_threadランタイム上で、ブロッキングなAがある場合とspawn_blockingで
    // 逃がした場合とで、Bの最初のtickが出るまでの経過時間が大きく変わることを
    // 数値で確認するテスト。
    #[tokio::test(flavor = "current_thread")]
    async fn blocking_sleep_delays_sibling_task() {
        let start = Instant::now();
        // Aが先にpollされてブロッキングsleepに入ると、Bのタイマー登録自体が
        // 200ms遅れるため、最初のtickは 200ms + 100ms 前後になる。
        tokio::join!(blocking_task_a(start), async {
            tokio::time::sleep(Duration::from_millis(100)).await;
        });
        assert!(start.elapsed() >= Duration::from_millis(200));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn spawn_blocking_does_not_delay_sibling_task() {
        let start = Instant::now();
        // spawn_blockingは別スレッドプールで実行されるため、メインタスク側は
        // すぐにawait地点で制御を返し、100ms待ちのタスクは本来の時間で完了する。
        tokio::join!(fixed_task_a(start), async {
            tokio::time::sleep(Duration::from_millis(100)).await;
        });
        assert!(start.elapsed() < Duration::from_millis(400));
    }
}

🪜 Step-by-Step 解説

💡 設計思想・なぜこう書くのか

Futureが「怠惰な値」として設計されているのは、ゼロコスト抽象化の思想の延長です。もしasync fnを呼んだ瞬間に即座に実行が始まる（JavaScriptのPromiseのような"eager"設計）なら、コンパイラは「呼び出しと同時に何らかのタスクキューに登録する」処理を暗黙に埋め込まねばならず、常に何らかのランタイム・ヒープ確保が必要になります。Rustは「実行するかどうか、いつ・どのスレッドで実行するかはすべて呼び出し側（ランタイム）が制御する」という設計にすることで、標準ライブラリ自体はいかなるランタイムにも依存しない、ゼロオーバーヘッドなFutureという「値」だけを提供しています。

🛑 コンパイルエラーが出た場合

本問は概念理解が主題のためコンパイルエラーは想定していませんが、#[tokio::main]のfeatures指定を忘れると次のようなエラーが出ます。

🌐 他言語との比較

Go経験者にとって重要な注意点は、「goroutineは重い処理を書いてもGoランタイムがある程度自動的に緩和してくれる場面がある」のに対し、「Rustのasync fnは自分で.awaitしない限り絶対に協調しない」という点です。Goのつもりでasync fnの中に重い同期処理を書くと、要件2で観察した通りの詰まりが本番環境で発生します。

🏆 実務での使いどころ

• Webフレームワーク（axum, actix-web）のハンドラはasync fnで書くのが標準。DBクエリやHTTPクライアント呼び出しなど「待ち時間の長いI/O」を.awaitすることで、1つのOSスレッドで大量の同時リクエストを処理できる
• CPU集約的な処理（画像処理・暗号化・大きなJSONのパース等）を非同期ハンドラの中に直接書くと、本問のタスクAのように他のリクエスト処理を止めてしまう。実務ではtokio::task::spawn_blockingやrayonとの連携で退避させるのが定石
• gRPCサーバー・メッセージキューコンシューマなど大量の同時接続・購読を扱うシステムでは、asyncタスクの軽量性（1タスクあたり数百バイト）がOSスレッドモデルに対する決定的な優位点になる
• ライブラリ設計では「このクレートの重い関数は同期版と非同期版のどちらを提供すべきか」を意識的に判断する必要があり、判断を誤ると利用者が本問と同じ事故を踏むことになる

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: tokio::spawnで独立タスクとして切り出した場合とtokio::join!で1つのタスク内に併存させた場合とで、パニック時の挙動やキャンセルの伝播がどう変わるかを調べてみましょう
• 次回予告: Futureトレイトとasyncランタイム(tokio) — 自前で簡単なFutureを手書きしてpollモデルをさらに深く理解します

🎯 自己評価

自分の回答

気づき・メモ