Day 038 — spawn・join!・select!。同じ「並行実行」でも独立性・パニック伝播・キャンセルが違う

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

Day 037ではFutureを手書きし、「pollされて初めて前進する」という怠惰な値としての性質を学びました。今日はそのFutureを複数同時に走らせる3つの方法——tokio::spawn・tokio::join!・tokio::select!——の違いを、実際に手を動かして体で覚えます。この3つはどれも「並行実行」を実現しますが、タスクの独立性・パニック時の挙動・キャンセルの仕方がまったく異なります。この違いを理解せずに使い分けを誤ると、実務では「エラー処理が漏れる」「意図せずタスクがキャンセルされてリソースリークする」といった事故につながります。

tokio::spawn — 独立したタスクを生成する。tokio::spawn(some_future)は渡したFutureを新しいタスクとしてランタイムに登録します。呼び出し元とは完全に独立して実行され、渡すFutureは'staticかつSendである必要があります。タスク内でパニックが起きても呼び出し元には伝播せず、JoinHandle::awaitの戻り値Result<T, JoinError>のErrとして捕捉できます。

tokio::select! — 最初に完了したものだけを採用し、残りを打ち切る。複数のFutureを同時にpollし、最初にReadyになった1つだけを採用します。選ばれなかった残りのFutureはその場でドロップされます（キャンセルされます）。これはタイムアウト処理など必須のパターンですが、ドロップされたFuture内で途中まで進んでいた処理は失われるため、各Futureが「途中でキャンセルされても状態が壊れない」——キャンセル安全（cancel-safe）——かどうかを設計者が意識する必要があります。

📝 問題

疑似的な3つの「APIエンドポイントへの取得処理」fetch_endpoint(id, delay_ms, should_panic)（tokio::time::sleepで遅延をシミュレートするasync fn）を題材に、以下3つの要件を実装してください。

要件1（spawn + JoinHandle）: 3つのfetch_endpoint呼び出しをtokio::spawnでそれぞれ独立したタスクとして起動し、Vec<JoinHandle<String>>で管理してください。3つ目のタスクはshould_panic=trueにしてパニックさせ、全タスク完了後にパニックしたタスクだけErrとして捕捉でき、他の2つは正常にOkが得られることを確認してください。

要件2（join!）: パニックしない2つのfetch_endpoint呼び出しをtokio::join!で同時に実行し、両方の結果をタプルで受け取ってください。要件1との違い（新規タスクを作らない・パニックが伝播する）をコメントで説明してください。

要件3（select! とキャンセル安全性の検証）: delay_ms=300のfetch_endpointとdelay_ms=100のtokio::time::sleep（タイムアウト役）をtokio::select!で競わせてください。fetch_endpointの内部ではsleep前後でprintln!によるログを出し、タイムアウトが先に成立した場合、fetch_endpoint側の「完了ログ」が出力されないこと（＝キャンセルされて最後まで実行されなかったこと）を実行結果で確認してください。

🔍 ヒント（段階的開示）

📋 コピーuse std::time::Duration;
use tokio::time::sleep;

async fn fetch_endpoint(id: u32, delay_ms: u64, should_panic: bool) -> String {
    println!("[fetch {id}] start (delay={delay_ms}ms)");
    sleep(Duration::from_millis(delay_ms)).await;
    if should_panic {
        panic!("endpoint {id} returned malformed response");
    }
    println!("[fetch {id}] completed"); // select!でキャンセルされるとこの行は出力されない
    format!("endpoint {id} response")
}

async fn fetch_all_spawn() -> Vec<Result<String, tokio::task::JoinError>> {
    let handles = vec![
        tokio::spawn(fetch_endpoint(1, 100, false)),
        tokio::spawn(fetch_endpoint(2, 50, false)),
        tokio::spawn(fetch_endpoint(3, 80, true)),
    ];
    // ここで各handleを順にawaitし、Vec<Result<..>>に集める
    todo!()
}

✅ 模範解答

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

use std::time::Duration;
use tokio::time::sleep;

/// 疑似APIエンドポイント。delay_ms分だけ待ってから結果を返す。
/// should_panic=trueの場合、待機後にpanic!する（要件1の異常系検証用）。
async fn fetch_endpoint(id: u32, delay_ms: u64, should_panic: bool) -> String {
    println!("[fetch {id}] start (delay={delay_ms}ms)");
    sleep(Duration::from_millis(delay_ms)).await;
    if should_panic {
        panic!("endpoint {id} returned malformed response");
    }
    // select!でこのFutureがキャンセルされた場合、以降の行は実行されない。
    println!("[fetch {id}] completed");
    format!("endpoint {id} response ({delay_ms}ms)")
}

/// 要件1: tokio::spawnで3つを独立タスクとして起動し、
/// 全タスクの結果(Result<String, JoinError>)をVecで集約する。
/// spawnされたタスクはこの関数(呼び出し元のタスク)とは完全に独立して走るため、
/// 3つ目がpanicしても、この関数自体はpanicしない。
/// JoinError::is_panic()でパニックによる失敗かどうかを判定できる。
async fn fetch_all_spawn() -> Vec<Result<String, tokio::task::JoinError>> {
    let handles: Vec<tokio::task::JoinHandle<String>> = vec![
        tokio::spawn(fetch_endpoint(1, 100, false)),
        tokio::spawn(fetch_endpoint(2, 50, false)),
        tokio::spawn(fetch_endpoint(3, 80, true)), // 意図的にpanicさせる
    ];

    let mut results = Vec::with_capacity(handles.len());
    for handle in handles {
        results.push(handle.await);
    }
    results
}

/// 要件2: tokio::join!で2つを同じタスク内で同時に進める。
/// spawnと違い新しいタスクは生成されない(OSスレッドを追加消費しない)。
/// もしどちらかがpanicした場合、そのpanicはこの関数の呼び出し元まで
/// そのまま伝播する(JoinErrorのようなラップは存在しない)。
async fn fetch_all_join() -> (String, String) {
    let (a, b) = tokio::join!(
        fetch_endpoint(1, 100, false),
        fetch_endpoint(2, 50, false),
    );
    (a, b)
}

/// 要件3: tokio::select!でfetch_endpoint(300ms)とtimeout(100ms)を競わせる。
/// 100ms経過時点でtimeoutブランチが先にReadyになるため、
/// fetch_endpoint側のFutureはpollの途中(sleep(300ms)の待機中)でドロップされる。
/// その結果「[fetch 1] completed」のログは出力されない。
async fn fetch_with_timeout() {
    tokio::select! {
        result = fetch_endpoint(1, 300, false) => {
            println!("  -> fetch側が先に完了: {result}");
        }
        _ = sleep(Duration::from_millis(100)) => {
            println!("  -> タイムアウトが先に成立。fetch(1)はキャンセルされた");
        }
    }
}

#[tokio::main]
async fn main() {
    println!("=== 要件1: tokio::spawnによる独立タスク実行 ===");
    let results = fetch_all_spawn().await;
    for (i, r) in results.iter().enumerate() {
        match r {
            Ok(s) => println!("  task[{i}] -> Ok: {s}"),
            Err(e) => println!("  task[{i}] -> Err(JoinError): is_panic={}", e.is_panic()),
        }
    }
    assert!(results[0].is_ok());
    assert!(results[1].is_ok());
    assert!(results[2].is_err());
    assert!(results[2].as_ref().unwrap_err().is_panic());

    println!("\n=== 要件2: tokio::join!による構造化並行実行 ===");
    let (a, b) = fetch_all_join().await;
    println!("  join! results: {a} / {b}");

    println!("\n=== 要件3: tokio::select!によるタイムアウト競争とキャンセル ===");
    fetch_with_timeout().await;
}

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

    #[tokio::test]
    async fn spawn_isolates_panic_as_join_error() {
        let results = fetch_all_spawn().await;
        assert!(results[0].is_ok());
        assert!(results[1].is_ok());
        assert!(results[2].is_err());
        assert!(results[2].as_ref().unwrap_err().is_panic());
    }

    #[tokio::test]
    async fn join_runs_both_concurrently_not_sequentially() {
        let start = Instant::now();
        let (a, b) = fetch_all_join().await;
        // 逐次実行なら150ms、並行実行なら最大値の100msに近い時間で終わるはず
        assert!(start.elapsed() < Duration::from_millis(140));
        assert!(a.contains("endpoint 1"));
        assert!(b.contains("endpoint 2"));
    }

    #[tokio::test]
    async fn select_cancels_the_losing_branch_before_300ms() {
        let start = Instant::now();
        fetch_with_timeout().await;
        // 300ms(fetch側)を待たず、100ms(timeout側)で返ってくるはず
        assert!(start.elapsed() < Duration::from_millis(200));
    }
}

🪜 Step-by-Step 解説

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

spawn・join!・select!という3つの異なるAPIが用意されているのは、「並行実行」という一つの言葉が実際にはタスクの独立性の強さという1本の軸の上にある異なる要求を指しているためです。spawnは最も独立性が強く（他タスクの障害から隔離したい）、join!は独立性を持たせず全結果を確実に得たい、select!はそもそも一部の結果を「要らない」と早期に判断したい——という3つの異なるニーズに、Rust/tokioはそれぞれ専用の型・マクロで応えています。

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

fetch_endpointの引数に、呼び出し元のスタック上にある値への参照（例: &String）を渡してtokio::spawnしようとすると、次のようなエラーになります。

🌐 他言語との比較

特に注目すべきはキャンセル時の挙動の違いです。JavaScriptのPromise.raceでは「負けた」Promiseもバックグラウンドで実行され続けます（キャンセルという概念自体がない）。一方Rustのselect!は、負けたFutureを実際にDropするため、確保していたリソース（ファイルハンドル・ロックなど）はDrop実装に従って確実に解放されます。これは「値の生存期間が終わったら何が起きるか」を型システムとDropトレイトで保証するRustならではの利点であり、同時に「Dropされる瞬間に中途半端な状態を残さないこと（キャンセル安全性）」をプログラマが自分で設計する責任にもなります。

🏆 実務での使いどころ

• spawn: Webサーバーで各リクエストを個別タスクとして処理する（1リクエストのpanicが他のリクエスト処理を巻き込まないようにする）、バックグラウンドでのログ送信・メトリクス収集など「失敗しても本処理に影響させたくない」処理
• join!: 1つのHTTPレスポンスを作るために、複数の下流サービス（ユーザー情報API・在庫API・価格API等）へ同時に問い合わせ、全部の結果が揃ってからレスポンスを組み立てる場合
• select!: HTTPクライアントでのタイムアウト処理（実際の通信 vs sleepのレース）、複数のデータソース（キャッシュ vs DB）のうち速い方を採用する処理、シャットダウンシグナル（ctrl_c()）とメインループの競合待ち

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: tokio::select!に3つ目の枝としてtokio::signal::ctrl_c()を追加し、Ctrl+Cで安全にシャットダウンする実装に挑戦してみましょう
• 次回予告: async trait（async-trait / RPITIT） — トレイトに非同期メソッドを持たせる設計と、async-traitクレートとRPITIT（Return Position Impl Trait In Trait）の使い分けに進みます

🎯 自己評価

自分の回答

気づき・メモ