Day 045 — JoinHandleをdropしてもタスクは止まらない。JoinSetのDropだけがタスク群を能動的にabortする

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

Day 041〜044 で JoinSet を土台に、動的な数のタスクの束ね方（041）、Semaphore による並行度制御（042）、mpsc による進捗通知（043）、watch による外部シャットダウン（044）と組み立ててきました。ここまで一貫して「JoinSet に集めて、abort_all() で打ち切る」という書き方をしてきましたが、実は一度も確認していない前提があります。それは——JoinHandle や JoinSet を「持っているだけの値」を drop したとき、その中のタスクはどうなるのか、という問いです。

Rustの所有権システムに慣れていると、「値を持つ変数が drop された = その値が管理していたリソースは解放される」という直感を持ちやすくなります。File は drop されればクローズされ、MutexGuard は drop されればロックが解放されます。この直感を tokio::spawn が返す JoinHandle<T> にそのまま適用すると、「JoinHandle を drop すればタスクも止まるはずだ」と考えてしまいますが、これは誤りです。tokio::spawn されたタスクは、呼び出し元とは独立してランタイムのスケジューラに直接ぶら下がる「デタッチされた」実行単位であり、JoinHandle はそのタスクの完了を観測するための窓口に過ぎません。窓口を閉じても、タスク本体（ランタイム上のスケジュール済みFuture）は生き続けます。

一方、Day 041〜044 で使ってきた JoinSet<T> は、Drop を実装する際に内部で保持している各タスクの AbortHandle を使って全タスクを能動的に abort するという設計になっています。つまり JoinHandle 単体と JoinSet は、見た目こそ「タスクをまとめて持つ入れ物」という点で似ていますが、drop 時の挙動が正反対です。この非対称性を正確に理解していないと、「グレースフルシャットダウンなんて JoinSet を drop すればいいのでは？」という誤った設計判断（Day 044 で明示的な watch + select! を実装した意味を見失う判断）につながります。

📝 問題

以下の3つのケースについて、「メインタスクを一定時間 sleep させて観測したとき、AtomicUsize カウンタはいくつになっているか」を先に予想してから、実際に cargo test で検証するコードを書いてください。

📋 コピーasync fn record_completion(counter: Arc, delay_ms: u64) {
    tokio::time::sleep(Duration::from_millis(delay_ms)).await;
    counter.fetch_add(1, Ordering::SeqCst);
}

• ケースA: tokio::spawn(record_completion(...)) の戻り値（JoinHandle）を、await せずに即座に drop() する。1個のタスクをspawnした場合、カウンタは最終的にいくつになるか？
• ケースB: 5個のタスクを tokio::spawn して Vec<JoinHandle<()>> に集める。1つもawaitせずに Vec ごと drop() する。カウンタは最終的にいくつになるか？
• ケースC: 5個のタスクを JoinSet<()>::spawn で JoinSet に集める。1つもawaitせずに JoinSet ごと drop() する。カウンタは最終的にいくつになるか？

要件:

• 上記3ケースそれぞれについて、#[tokio::test] で「drop後に十分な時間 sleep してからカウンタの値を assert する」テストを書いてください（各タスクの delay_ms は30ms程度、観測用の待機は100ms程度を目安にしてください）
• テストを書く前に、3ケースそれぞれの予想値をコメントとして残してください
• なぜ JoinHandle（またはそれを集めたVec）と JoinSet とで挙動が異なるのか、「所有権を持つ」ことと「実行を制御する権限を持つ」ことの違いという観点で、1〜2文で説明を加えてください

🔍 ヒント（段階的開示）

📋 コピー#[tokio::test]
async fn case_a_bare_spawn() {
    let counter = Arc::new(AtomicUsize::new(0));
    let handle = tokio::spawn(record_completion(Arc::clone(&counter), 30));
    drop(handle);

    tokio::time::sleep(Duration::from_millis(100)).await;
    assert_eq!(counter.load(Ordering::SeqCst), /* 予想値を入れる */);
}

// ケースB, ケースCも同様の形で、Vec/JoinSetをdropする箇所を差し替えて書く

✅ 模範解答

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

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::task::JoinSet;

/// sleep後にカウンタをインクリメントすることで、「このタスクが最後まで
/// 実行されたかどうか」を外部から観測できるようにするための道具。
async fn record_completion(counter: Arc, delay_ms: u64) {
    tokio::time::sleep(Duration::from_millis(delay_ms)).await;
    counter.fetch_add(1, Ordering::SeqCst);
}

#[tokio::main]
async fn main() {
    let counter_a = Arc::new(AtomicUsize::new(0));
    {
        let handle = tokio::spawn(record_completion(Arc::clone(&counter_a), 30));
        drop(handle); // JoinHandleをdropしてもタスクは走り続ける(デタッチ)
    }

    let counter_b = Arc::new(AtomicUsize::new(0));
    {
        let mut handles = Vec::new();
        for _ in 0..5 {
            handles.push(tokio::spawn(record_completion(Arc::clone(&counter_b), 30)));
        }
        drop(handles); // Vecをdropしても中身のタスクは止まらない
    }

    let counter_c = Arc::new(AtomicUsize::new(0));
    {
        let mut set: JoinSet<()> = JoinSet::new();
        for _ in 0..5 {
            set.spawn(record_completion(Arc::clone(&counter_c), 30));
        }
        drop(set); // JoinSetをdropすると、追跡している全タスクがabortされる
    }

    tokio::time::sleep(Duration::from_millis(150)).await;
    println!("A (bare spawn):        {}/1", counter_a.load(Ordering::SeqCst));
    println!("B (Vec):   {}/5", counter_b.load(Ordering::SeqCst));
    println!("C (JoinSet):           {}/5", counter_c.load(Ordering::SeqCst));
}

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

    #[tokio::test]
    async fn case_a_bare_spawn_keeps_running_after_handle_is_dropped() {
        // 予想: 1 (JoinHandleのdropはタスクをキャンセルしない)
        let counter = Arc::new(AtomicUsize::new(0));
        let handle = tokio::spawn(record_completion(Arc::clone(&counter), 30));
        drop(handle);

        tokio::time::sleep(Duration::from_millis(100)).await;
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "bare spawnはhandleをdropしても最後まで実行されるはず"
        );
    }

    #[tokio::test]
    async fn case_b_vec_of_join_handles_keeps_tasks_running_after_vec_is_dropped() {
        // 予想: 5 (Vecはただの入れ物。中身の各JoinHandleが
        // dropされる際の挙動はケースAと同じなので、全タスクが完走する)
        let counter = Arc::new(AtomicUsize::new(0));
        let mut handles = Vec::new();
        for _ in 0..5 {
            handles.push(tokio::spawn(record_completion(Arc::clone(&counter), 30)));
        }
        drop(handles);

        tokio::time::sleep(Duration::from_millis(100)).await;
        assert_eq!(
            counter.load(Ordering::SeqCst),
            5,
            "Vecをdropしても中のタスクは1つも止まらないはず"
        );
    }

    #[tokio::test]
    async fn case_c_join_set_aborts_all_tasks_when_dropped() {
        // 予想: 0 (JoinSetのDrop実装は内部のAbortHandle全てにabortを呼ぶ)
        let counter = Arc::new(AtomicUsize::new(0));
        let mut set: JoinSet<()> = JoinSet::new();
        for _ in 0..5 {
            set.spawn(record_completion(Arc::clone(&counter), 30));
        }
        drop(set);

        tokio::time::sleep(Duration::from_millis(100)).await;
        assert_eq!(
            counter.load(Ordering::SeqCst),
            0,
            "JoinSetをdropすると追跡中の全タスクがabortされ、1つも完走しないはず"
        );
    }
}

// 説明: JoinHandle(や、それをただ集めただけのVec)は「タスクの完了を
// 観測する権利」しか持っていない——所有権に似た形をしているが、実行を
// 止める権限までは持っていない。一方JoinSetは、単なる入れ物ではなく
// 「自分がspawnしたタスク群の生存管理そのものに責任を持つ」型として
// 設計されており、そのDrop実装がタスクの実行権限(abort)まで行使する。

🪜 Step-by-Step 解説

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

Rustの所有権システムは「値の生存期間とリソースの生存期間を一致させる」ことをRAII（Resource Acquisition Is Initialization）という形で徹底しますが、tokio::spawn されたタスクはこの原則から意図的に外れています。タスクを呼び出し元の生存期間から切り離す「デタッチされた並行性」こそが tokio::spawn の存在意義（呼び出し元の関数がreturnした後もタスクを走らせ続けられる）であり、これは所有権システムの型では表現しきれない設計上の選択です。一方 JoinSet は、この「デタッチされ放題」の世界に対して、ライブラリ側が独自に「構造化された並行性（structured concurrency）」——タスクの生存期間を、それを保持する値の生存期間に改めて縛り直す——という規律を型として提供したもの、と捉えることができます。

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

今日のテーマは概念理解が中心ですが、この実験コードを書く際によくあるミスとして、Arc::clone を忘れてループの2周目以降で同じ counter を再利用しようとするケースがあります。

📋 コピー// アンチパターン: cloneせずにループの中でcounterをそのまま使おうとする
let counter = Arc::new(AtomicUsize::new(0));
let mut handles = Vec::new();
for _ in 0..5 {
    handles.push(tokio::spawn(record_completion(counter, 30))); // 2周目でエラー
}

🌐 他言語との比較

面白いのは、Go・Javaの「デタッチが既定」という設計はRustのtokio::spawn単体の挙動と揃っている一方、C++20で導入されたstd::jthreadは「スコープを抜けたら自動的に停止要求を出してjoinする」というRAII的な設計を持ち、これはむしろJoinSetの挙動に近いということです。つまりRustは、言語の標準的な非同期タスク生成（tokio::spawn）自体は他言語同様「デタッチが既定」としつつ、JoinSetというライブラリレベルの型で構造化された代替手段を別途提供する、という2階建ての設計を取っています。

🏆 実務での使いどころ

• fire-and-forgetが正しい選択の場面: HTTPリクエストへのレスポンスを返した後に、アクセスログを非同期で外部の集計サービスに送信するようなタスク。リクエストハンドラの生存期間にログ送信タスクを縛る必要はなく、tokio::spawnで投げっぱなしにするのが自然
• 構造化が必須の場面: リクエストハンドラが内部で複数のタスクをspawnして並行処理し、その結果を集約してレスポンスに含める場合。ハンドラがタイムアウトやクライアント切断で早期returnする際、spawnした子タスクが野放しで走り続けるのは望ましくない（リソースリークやコネクションプールの枯渇に直結する）ため、JoinSetやtokio_util::task::TaskTracker + CancellationTokenでスコープを縛る
• グレースフルシャットダウンの設計判断: Day 044のように「打ち切った件数を集計する」要件があるなら、単純なdrop任せではなくabort_all() + join_next()による明示的な回収ループが必要になる、という判断根拠を今日の実験が裏付ける

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: tokio_util::task::TaskTracker と CancellationToken を使い、JoinSetよりもさらに構造化された（親子関係を持つキャンセルの伝播ができる）タスク管理を実装し、Day 044のwatchベースのシャットダウンと比較する
• 次回予告: async trait（async-trait / RPITIT） — 非同期メソッドを持つトレイトの設計

🎯 自己評価

自分の回答

気づき・メモ