Day 042 — Semaphore × JoinSetで最大並行度を制御する。「動的な数のタスク」と「同時に何本まで」は別の設計軸

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

Day 041 では tokio::task::JoinSet を使い、「動的な数のタスクを束ね、完了順に集計し、全体のタイムアウトで打ち切る」バッチ処理 run_batch_correct を実装しました。しかしこの実装には本番投入前に必ず突き当たる制約が1つ抜けています。外部APIやDBには同時接続数の上限がある、という現実です。

inputs.len() が10万件あるバッチを run_batch_correct にそのまま渡すと、JoinSet は10万個のタスクを一斉に spawn し、10万本の fetch がほぼ同時に外部APIへリクエストを送ります。相手のレート制限に弾かれるか、あるいは自分のプロセスのソケット数・メモリを圧迫するかのどちらかです。「動的な数のタスクを扱える」ことと「同時に何本まで実行してよいか」は別の問題であり、後者を設計せずに前者だけ実装すると本番で事故ります。

この「同時実行数の上限（=最大並行度）を設ける」という要求に対して、Rust/tokioのエコシステムには複数のアプローチがあります。

1. 入力をチャンクに分割し、チャンクごとにJoinSetで処理して順番に待つ — 実装は単純だが、1チャンク内で最も遅いタスクが次のチャンクの開始をブロックする（並行度の恩恵が段階的にしか出ない）
2. tokio::sync::Semaphoreでタスク本体の実行区間を許可制にする — spawn自体は全タスク分すぐに行うが、各タスクの中身はSemaphoreからpermit（許可証）を取得できるまで待たされる。JoinSetの「完了順集計」「abort_all()による打ち切り」といったDay 041で学んだ性質はそのまま使える
3. futures::stream::StreamExt::buffer_unordered(N)を使う — ストリームコンビネータとして簡潔に書けるが、JoinSet固有のabort_all()やタスク単位のJoinError（panic検知）とは別の抽象なので、Day 041の設計と混ぜるには変換コストがかかる

今日はこのうち2番目、SemaphoreをJoinSetと組み合わせて最大並行度を制御する設計を実装します。「タスクの数は動的」「タスクの同時実行数には上限がある」「全体にも個別にもタイムアウトがある」という3つの制約を同時に満たす設計判断がテーマです。

📝 問題

Day 041 の run_batch_correct を拡張し、最大並行度を指定できる run_batch_bounded を設計・実装してください。

• シグネチャ: async fn run_batch_bounded(inputs: Vec<u32>, max_concurrency: usize, per_task_timeout: Duration, total_timeout: Duration) -> BatchReport
• fetch 関数・BatchReport 構造体は Day 041 と同一のものを使用する（下記に再掲）
• 設計判断として: Semaphore の permit をどこで取得するか（spawn する前のループの中か、spawn された非同期ブロックの中か）を決め、その理由を説明できるようにしてください。この選択次第で「spawn 自体が max_concurrency でブロックされるか」「JoinSet に登録されるタスク数がいつ確定するか」が変わります
• 実際に同時実行数が max_concurrency を超えないことを、テストで（内部状態を覗くのではなく）観測可能な形で検証してください
• total_timeout を超えた場合、Semaphore の permit 待ちで止まっているタスクも含めて abort_all() で打ち切れることを確認してください

📋 コピーasync fn fetch(input: u32) -> Result {
    tokio::time::sleep(Duration::from_millis(((input % 5) * 40) as u64)).await;
    if input % 7 == 0 {
        return Err(format!("input {input} is not allowed (divisible by 7)"));
    }
    if input % 11 == 0 {
        panic!("input {input} triggered an unexpected panic");
    }
    Ok(input * input)
}

#[derive(Debug, Default, PartialEq, Eq)]
struct BatchReport {
    succeeded: Vec,
    application_errors: usize,
    panicked: usize,
    timed_out: usize,
    aborted_by_total_timeout: usize,
}

🔍 ヒント（段階的開示）

📋 コピーasync fn run_batch_bounded(
    inputs: Vec,
    max_concurrency: usize,
    per_task_timeout: Duration,
    total_timeout: Duration,
) -> BatchReport {
    let semaphore = Arc::new(Semaphore::new(max_concurrency));
    let mut set: JoinSet, tokio::time::error::Elapsed>> =
        JoinSet::new();

    for input in inputs {
        let semaphore = Arc::clone(&semaphore);
        set.spawn(async move {
            let _permit = semaphore.acquire_owned().await.expect("semaphore is never closed");
            tokio::time::timeout(per_task_timeout, fetch(input)).await
            // _permit はここでdropされ、Semaphoreに1つ返却される
        });
    }

    // 以降の集計ループ・total_timeout・abort_all() はDay 041のrun_batch_correctと同じ形
    // ...
}

✅ 模範解答

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

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;

async fn fetch(input: u32) -> Result {
    tokio::time::sleep(Duration::from_millis(((input % 5) * 40) as u64)).await;
    if input % 7 == 0 {
        return Err(format!("input {input} is not allowed (divisible by 7)"));
    }
    if input % 11 == 0 {
        panic!("input {input} triggered an unexpected panic");
    }
    Ok(input * input)
}

#[derive(Debug, Default, PartialEq, Eq)]
struct BatchReport {
    succeeded: Vec,
    application_errors: usize,
    panicked: usize,
    timed_out: usize,
    aborted_by_total_timeout: usize,
}

/// 最大並行度を`max_concurrency`に制限しつつ、Day 041の`run_batch_correct`と
/// 同じ完了順集計・per_task_timeout・total_timeoutによる打ち切りを行う。
///
/// 設計判断: Semaphoreのpermitは「spawnする前のループ」ではなく「spawnされた
/// 非同期ブロックの内側」で取得する。理由は2つ:
///   1. spawn自体をブロックしないため、JoinSetには常に全inputs分のタスクが
///      即座に登録される（Day 041の「動的な数のタスクを束ねる」性質を壊さない）
///   2. permitのライフタイムがspawnされたタスク（'static）の内部に閉じるため、
///      acquire_owned()でSemaphoreの所有権を共有すれば借用の問題が起きない
async fn run_batch_bounded(
    inputs: Vec,
    max_concurrency: usize,
    per_task_timeout: Duration,
    total_timeout: Duration,
) -> BatchReport {
    let semaphore = Arc::new(Semaphore::new(max_concurrency));
    let mut set: JoinSet, tokio::time::error::Elapsed>> =
        JoinSet::new();

    for input in inputs {
        let semaphore = Arc::clone(&semaphore);
        set.spawn(async move {
            // acquire_owned(): Arcの所有権を共有するため、この
            // 非同期ブロックが'staticであることと矛盾しない。
            // permitを保持したままfetchの.awaitをまたぐのは合法（tokioの
            // MutexGuardと違い、Semaphoreは.awaitをまたいで保持する使い方を
            // 前提に設計されている）。
            let _permit = semaphore
                .acquire_owned()
                .await
                .expect("semaphore is never closed");
            tokio::time::timeout(per_task_timeout, fetch(input)).await
            // _permitはここでdropされ、Semaphoreにpermitが1つ返却される。
            // 次にacquire_owned()で待っているタスクがあれば起床する。
        });
    }

    let mut report = BatchReport::default();

    let timed_out_overall = tokio::time::timeout(total_timeout, async {
        while let Some(join_result) = set.join_next().await {
            match join_result {
                Ok(Ok(Ok(value))) => report.succeeded.push(value),
                Ok(Ok(Err(_app_err))) => report.application_errors += 1,
                Ok(Err(_elapsed)) => report.timed_out += 1,
                Err(join_err) if join_err.is_panic() => report.panicked += 1,
                Err(_cancelled) => {}
            }
        }
    })
    .await
    .is_err();

    if timed_out_overall {
        // abort_all()は「permit待ちでまだfetchを開始していないタスク」も
        // 「fetch実行中のタスク」もまとめてキャンセル要求する。
        // permit待ち中のタスクはpermitを一度も取得していないのでdropしても
        // 何も返却する必要がなく、リークは起きない。
        set.abort_all();
        while let Some(join_result) = set.join_next().await {
            match join_result {
                Ok(Ok(Ok(value))) => report.succeeded.push(value),
                Ok(Ok(Err(_app_err))) => report.application_errors += 1,
                Ok(Err(_elapsed)) => report.timed_out += 1,
                Err(join_err) if join_err.is_panic() => report.panicked += 1,
                Err(_cancelled) => report.aborted_by_total_timeout += 1,
            }
        }
    }

    report
}

#[tokio::main]
async fn main() {
    let report = run_batch_bounded(
        (1..=50).collect(),
        10, // 同時実行数を10本に制限
        Duration::from_millis(500),
        Duration::from_secs(5),
    )
    .await;
    println!("{report:#?}");
}

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

    #[tokio::test]
    async fn correctness_matches_unbounded_behavior() {
        // max_concurrencyを十分大きく取れば、Day 041のrun_batch_correctと
        // 分類結果は一致するはず（7の倍数=Err, 11の倍数=panic, それ以外=成功）
        let report = run_batch_bounded(
            vec![1, 2, 7, 11, 14],
            100,
            Duration::from_secs(1),
            Duration::from_secs(5),
        )
        .await;

        assert_eq!(report.succeeded.len(), 2); // 1, 2
        assert!(report.succeeded.contains(&1));
        assert!(report.succeeded.contains(&4));
        assert_eq!(report.application_errors, 2); // 7, 14
        assert_eq!(report.panicked, 1); // 11
        assert_eq!(report.timed_out, 0);
        assert_eq!(report.aborted_by_total_timeout, 0);
    }

    #[tokio::test]
    async fn max_concurrency_caps_parallel_execution() {
        // 10件すべてが160ms sleepするタスク（input % 5 == 4）。
        // 7や11の倍数を意図的に避け、全件成功させて時間測定のノイズを消す。
        let inputs: Vec = vec![4, 9, 19, 24, 29, 34, 39, 54, 59, 64];

        let start = Instant::now();
        let report = run_batch_bounded(
            inputs,
            2, // 同時実行数を2本に制限
            Duration::from_secs(5),
            Duration::from_secs(5),
        )
        .await;
        let elapsed = start.elapsed();

        assert_eq!(report.succeeded.len(), 10);
        // 10タスク ÷ 同時実行数2 = 5ラウンド × 160ms = 800msが理論下限。
        // 無制限に並行実行されていれば160ms程度で終わるはずなので、
        // 十分な余裕を持って400ms以上かかっていることを確認する。
        assert!(
            elapsed >= Duration::from_millis(400),
            "max_concurrencyが効いていない可能性: elapsed = {elapsed:?}"
        );
    }

    #[tokio::test]
    async fn total_timeout_aborts_tasks_waiting_on_permit() {
        // 同時実行数1、10件のタスク。total_timeoutを1タスク分の実行時間より
        // 短く設定し、「permit待ちで一度もfetchを開始していないタスク」が
        // abort_allで正しく打ち切られることを確認する。
        let inputs: Vec = vec![4, 9, 19, 24, 29, 34, 39, 54, 59, 64];

        let report = run_batch_bounded(
            inputs,
            1, // 同時実行数1本 = ほぼ直列実行
            Duration::from_secs(5),
            Duration::from_millis(50), // 1件目の160ms sleepすら終わらせない
        )
        .await;

        // 誰も成功できないまま全件がabortされる
        assert_eq!(report.succeeded.len(), 0);
        assert_eq!(report.aborted_by_total_timeout, 10);
    }
}

🪜 Step-by-Step 解説

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

Semaphoreを「spawnの外」ではなく「spawnの中」に置くという判断は、Rustにおける並行処理設計が「リソースの多重度をどの粒度の非同期タスクに割り当てるか」を型とコードの構造で明示させるという思想の延長線上にあります。Goであればsync.WaitGroupとバッファ付きチャンネルを組み合わせて似たパターンを書けますが、「どこでブロックし、どこで解放されるか」はコメントを読まないと分かりません。RustではArc<Semaphore>の所有権がどのスコープに属しているか、OwnedSemaphorePermitがどの変数の生存期間に紐づいているかを見れば、解放タイミングをコードの構造そのものから読み取れます。

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

ヒント2で触れた「spawnする前のループの中で借用版acquire()してからspawnに持ち込む」誤った設計を試すと、以下のようなエラーになります。

📋 コピー// アンチパターン: ループの中で借用版acquire()を呼び、permitをspawnへ持ち込もうとする
for input in inputs {
    let permit = semaphore.acquire().await.unwrap(); // semaphore: &Semaphore を借用
    set.spawn(async move {
        let _permit = permit; // ここでエラー
        tokio::time::timeout(per_task_timeout, fetch(input)).await
    });
}

🌐 他言語との比較

Go・Javaいずれも「取得したら解放を書く」という規約でリークを防ぐのに対し、Rustは「解放は変数のスコープが尽きた時に自動で起きる」という言語機構でリークを防ぎます。これは所有権システムがロックやリソース管理全般に一貫して適用される設計思想の一例です。

🏆 実務での使いどころ

• 外部APIのレート制限順守: サードパーティAPIの「秒間N接続まで」という制約下で、大量のバッチリクエストを安全に流し込む
• DBコネクションプールとの整合: Semaphoreのpermit数をDBコネクションプールのサイズと合わせ、sqlx等のプールが枯渇してタイムアウトエラーを起こす前に、アプリケーション側で並行度を絞る
• CPUバウンドな処理のスレッドプール保護: tokio::task::spawn_blockingを使う処理で、OSスレッドプールのサイズ以上のブロッキングタスクを同時に走らせないようにする

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: max_concurrencyを実行時に動的に変更できるようにする（Semaphore::add_permits()を使い、外部からのバックプレッシャー信号に応じて許可数を増減させるサーキットブレーカー的な設計）
• 次回予告: async trait（async-trait / RPITIT） — 非同期メソッドを持つトレイトの設計

🎯 自己評価

自分の回答

気づき・メモ