Day 040 — select!が「dropする」のはどこからどこまでか。キャンセル安全性を条件式と後続ブロックの境界で見極める

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

Day 038ではtokio::spawn・tokio::join!・tokio::select!の3つが「タスクの独立性」「パニック伝播」「キャンセル」の観点でまったく違う挙動をすることを学びました。今日はその中でも実務で最も事故りやすいselect!のキャンセル安全性（cancel safety）を、もう一段掘り下げます。

tokio::select! は複数のFutureを同時にpollし、最初に完了した1つの結果を返します。選ばれなかった残りのFutureはその場でdropされる——つまりキャンセルされます。RustのFutureは「途中まで実行してdropされても、型システムは何も保証してくれない」ため、あるasync処理が「処理の途中でdropされても安全か」は、その関数のドキュメントを読んで自分で確認するしかありません。キャンセル安全でない操作（例: バッファに読み込み中のread_lineのように、部分的に読んだデータが呼び出し元に返らないまま消えるもの）をselect!のループ内に置くと、「データが静かに消える」という再現困難なバグになります。

Go のselect文はチャネル操作にしか使えず、「選ばれなかった方は最初から実行されていない」ため、この種の問題は原理的に起きません。tokioのselect!は任意のFutureを対象にできる代わりに、「実行途中のFutureを打ち切る」という強力さとリスクを併せ持っています。

📝 問題

以下は、複数のワーカーからのメッセージをチャネルで受け取りつつ、5秒ごとに現在の受信件数をログ出力する「よくある」実装です。

📋 コピーuse tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

async fn process(count: &mut u32, msg: String) {
    // 重い処理を模倣（この処理の途中でタスクがキャンセルされると
    // countがインクリメントされないままメッセージが失われる、と思われがち）
    sleep(Duration::from_millis(50)).await;
    *count += 1;
    println!("processed: {msg} (total={count})");
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::(100);

    tokio::spawn(async move {
        for i in 0..20 {
            let _ = tx.send(format!("msg-{i}")).await;
            sleep(Duration::from_millis(30)).await;
        }
    });

    let mut count = 0u32;
    loop {
        tokio::select! {
            Some(msg) = rx.recv() => {
                process(&mut count, msg).await;
            }
            _ = sleep(Duration::from_secs(5)) => {
                println!("--- heartbeat: {count} messages processed so far ---");
            }
        }
    }
}

以下を説明してください。

問1: このコードは一見「メッセージロスト」の典型的な誤解パターンに見えますが、実はこのコード自体にはキャンセル安全性の問題はありません。なぜ安全と言えるのか、select!の各分岐で何が起きているかを踏まえて説明してください。

問2: では、どのように書き換えると「本当に」キャンセル安全性の問題（処理途中のメッセージ消失）が発生するか、具体的な書き換えパターンを1つ示してください。

問3: tokio::spawn / tokio::join! / tokio::select!の3つについて、「タスクの生存期間」「呼び出し元との構造的な親子関係」「エラー・パニック時の伝播」の3観点で違いを整理してください。

🔍 ヒント（段階的開示）

📋 コピーtokio::select! {
    result = recv_and_process(&mut rx, &mut count) => { /* ... */ }
    _ = sleep(Duration::from_secs(5)) => { /* heartbeat */ }
}

✅ 模範解答

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

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

// --- 安全な元のコード（再掲・比較用） ---
async fn process(count: &mut u32, msg: String) {
    sleep(Duration::from_millis(50)).await;
    *count += 1;
    println!("processed: {msg} (total={count})");
}

// --- キャンセル安全性が壊れる書き換え例 ---
// 「メッセージを受信し切ってから処理する」のではなく、
// 「受信して処理するところまで」を1つのFutureにまとめてしまうと、
// そのFuture全体がselect!の条件式側に置かれた場合にキャンセル対象になる。
async fn recv_and_process(
    rx: &mut mpsc::Receiver,
    count: &mut u32,
) -> Option<()> {
    let msg = rx.recv().await?;
    // ここがポイント: recv済みのmsgを保持したまま重い処理に入る。
    // この行の途中でFuture自体がdropされると、
    // 「チャネルからは取り出したのに処理されなかった」メッセージが発生する。
    sleep(Duration::from_millis(50)).await;
    *count += 1;
    println!("processed: {msg} (total={count})");
    Some(())
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::(100);

    tokio::spawn(async move {
        for i in 0..20 {
            let _ = tx.send(format!("msg-{i}")).await;
            sleep(Duration::from_millis(30)).await;
        }
    });

    let mut count = 0u32;
    loop {
        tokio::select! {
            // 危険パターン: recvと重い処理をまとめて条件式側に置いている。
            // sleep(5秒)側が先に完了した瞬間、このFutureごとdropされ、
            // 内部で保持中だったmsgが処理されずに消える。
            _ = recv_and_process(&mut rx, &mut count) => {}
            _ = sleep(Duration::from_secs(5)) => {
                println!("--- heartbeat: {count} messages processed so far ---");
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::sync::Arc;

    /// 安全版: recvだけを条件式に置き、重い処理は後続ブロックで行う。
    /// heartbeat側が先に完了しても、recv自体は一瞬で終わっているため
    /// 「受信済みなのに処理されない」メッセージは発生しない。
    #[tokio::test]
    async fn recv_only_in_condition_is_cancel_safe() {
        let (tx, mut rx) = mpsc::channel::(10);
        let processed = Arc::new(AtomicU32::new(0));

        tx.send("hello".to_string()).await.unwrap();
        drop(tx);

        let processed_clone = Arc::clone(&processed);
        tokio::select! {
            Some(msg) = rx.recv() => {
                sleep(Duration::from_millis(10)).await;
                processed_clone.fetch_add(1, Ordering::SeqCst);
                assert_eq!(msg, "hello");
            }
            _ = sleep(Duration::from_secs(5)) => unreachable!(),
        }
        assert_eq!(processed.load(Ordering::SeqCst), 1);
    }
}

🪜 Step-by-Step 解説

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

RustのFutureは「pollされて初めて進む」怠惰な値であり、.awaitは「このFutureが完了するまでここで止まる」という糖衣構文にすぎません。select!はこの性質を利用して「複数のFutureを同時にpollし、最初にPoll::Readyを返した1つだけ採用する」という薄いレイヤーとして実装できます。逆に言うと、言語やランタイムが「キャンセルしても大丈夫か」を検証してくれる仕組みはどこにもありません。

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

今回は概念理解問題であり意図的なコンパイルエラーは含めていませんが、select!関連でよく遭遇するエラーとして、条件式で束縛した変数を=>ブロックの外（他の分岐）から参照しようとした場合があります。

🌐 他言語との比較

Go のselect文はチャネル操作にしか使えないため、「実行途中でキャンセルされる」というバグのクラス自体が存在しません。tokioのselect!は任意のFutureを対象にできる分だけ表現力が高い代わりに、「実行途中で打ち切られる」というGoには存在しないリスクを抱えます。C#のTask.WhenAnyも類似の位置づけですが、明示的にCancellationTokenを伝播させる文化がある点で、tokioの「暗黙のdropキャンセル」とは設計がやや異なります。

🏆 実務での使いどころ

• シャットダウンシグナル待ち: tokio::select! { _ = shutdown_signal.recv() => { /* graceful shutdown */ } result = handle_request(req) => { /* 通常処理 */ } }のような形は、非同期サーバーの定番パターン
• タイムアウト付き処理: 外部APIへの問い合わせとsleepを競わせ、遅い方を打ち切る
• コードレビューの観点: 多くのチームでは「select!の条件式に置くFutureは、必ずそのドキュメントでcancel safetyを確認する」というレビュー観点をチェックリストに入れている。今回の例のように、条件式に置かれたFutureが「DBトランザクションの途中」や「複合操作の途中」を含んでいないかは要チェック

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: tokio_util::sync::CancellationTokenを使って、spawnした非構造化タスクにも明示的なキャンセル伝播を持たせる設計を実装してみましょう
• 次回予告: async trait（async-trait / RPITIT） — 非同期メソッドを持つトレイトの設計

🎯 自己評価

自分の回答

気づき・メモ