Day 033 — 2つのチャンネルを順番に待つと片方が飢餓状態になるバグを直す

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

昨日（Day 032）はArc<Mutex<Receiver<Job>>>で「1つのチャンネルを複数ワーカーが分け合う」パターンと、sync_channelによるバックプレッシャーを扱いました。今日は同じmpscを題材に、「複数のチャンネルを同時に待つ」場面で踏みやすい罠を扱います。

Go言語にはselect文があり、複数のチャンネルのうち「先に準備ができた方」を待つことができます。

select {
case alert := <-alertCh:
    handleAlert(alert)
case metric := <-metricCh:
    handleMetric(metric)
}

一方、std::sync::mpscにはこのselectに相当する機能がありません。Receiver<T>が持つブロッキング手段は基本的に次の3つです。

• recv() — 次の値が来るまで無期限にブロックする
• try_recv() — 即座にOk/Err(Empty)/Err(Disconnected)のいずれかを返す（ブロックしない）
• recv_timeout(Duration) — 指定時間だけブロックし、来なければErr(Timeout)を返す

複数のチャンネルを「両方とも取りこぼしなく処理したい」と思ったとき、recv()を素朴に複数のチャンネルへ順番に呼び出すコードを書いてしまいがちですが、これは片方のチャンネルが片方を飢餓状態（starvation）にするバグの温床になります。

📝 問題

以下は、監視システムの一部として書かれた「アラート（低頻度・重要）」と「メトリクス（高頻度）」を1つのループで処理するコードです。このコードには重大なバグがあります。

📋 コピーuse std::sync::mpsc::{self, Receiver};
use std::thread;
use std::time::Duration;

struct Alert { message: String }
struct Metric { name: String, value: f64 }

fn monitor(alert_rx: Receiver<Alert>, metric_rx: Receiver<Metric>) {
    loop {
        match alert_rx.recv() {
            Ok(alert) => println!("ALERT: {}", alert.message),
            Err(_) => break,
        }
        match metric_rx.try_recv() {
            Ok(metric) => println!("metric: {}={}", metric.name, metric.value),
            Err(_) => {}
        }
    }
}

fn main() {
    let (alert_tx, alert_rx) = mpsc::channel::<Alert>();
    let (metric_tx, metric_rx) = mpsc::channel::<Metric>();

    thread::spawn(move || {
        thread::sleep(Duration::from_millis(300));
        alert_tx.send(Alert { message: "disk usage 95%".to_string() }).unwrap();
        // alert_tx はここで drop される
    });

    thread::spawn(move || {
        for i in 0..5 {
            metric_tx.send(Metric { name: format!("cpu_{i}"), value: i as f64 * 1.5 }).unwrap();
        }
        // metric_tx はここで drop される
    });

    monitor(alert_rx, metric_rx);
}

要件1（コードレビュー）: metric_txは起動直後に5件のメトリクスを送信し終えているにもかかわらず、monitorは実際には何件のメトリクスを処理し、それは何ミリ秒後か。理由も含めて説明してください。

要件2（修正）: 「メトリクスは届き次第ほぼ即座に処理され、アラートも届き次第（取りこぼしなく）処理される」という本来の意図通りに動くよう、monitorを（crossbeam等の外部クレートを使わずに）再設計してください。Receiverを複数持ち続ける設計を維持したままrecv_timeoutでポーリングする方法でも動きますが、より筋の良い設計が他にあります。

要件3（検証）: 修正版で、以下をassertで検証するテストコード（main関数でよい）を書いてください。

• 5件のメトリクスは全てInstant::elapsed()が100ミリ秒未満のタイミングで処理される
• アラートはInstant::elapsed()が250ミリ秒以上のタイミングで処理される
• 処理件数の合計が6件（メトリクス5件＋アラート1件）である

🔍 ヒント（段階的開示）

📋 コピーenum Event {
    Alert(Alert),
    Metric(Metric),
}

fn main() {
    let (event_tx, event_rx) = mpsc::channel::<Event>();

    let alert_tx = event_tx.clone();
    thread::spawn(move || {
        // alert_tx.send(Event::Alert(...))
    });

    let metric_tx = event_tx.clone();
    thread::spawn(move || {
        // metric_tx.send(Event::Metric(...))
    });

    drop(event_tx); // 元のSenderもdropを忘れずに（Day 032の教訓）

    for event in event_rx {
        match event {
            Event::Alert(a) => { /* ... */ }
            Event::Metric(m) => { /* ... */ }
        }
    }
}

✅ 模範解答

📋 コピー// cargo new day033_mpsc_select の後、src/main.rs をこの内容で置き換えて
// `cargo run` で検証できます（外部クレート依存なし、標準ライブラリのみ）。

use std::sync::mpsc::{self, Receiver};
use std::thread;
use std::time::{Duration, Instant};

struct Alert { message: String }
struct Metric { name: String, value: f64 }

// ── 要件1で調査対象にした「バグ版」（比較用にそのまま残す）──

fn monitor_buggy(
    alert_rx: Receiver<Alert>,
    metric_rx: Receiver<Metric>,
    start: Instant,
) -> Vec<(String, Duration)> {
    let mut log = Vec::new();
    loop {
        match alert_rx.recv() {
            Ok(alert) => log.push((format!("ALERT: {}", alert.message), start.elapsed())),
            Err(_) => break,
        }
        match metric_rx.try_recv() {
            Ok(metric) => log.push((format!("metric: {}={}", metric.name, metric.value), start.elapsed())),
            Err(_) => {}
        }
    }
    log
}

fn demo_buggy() {
    let (alert_tx, alert_rx) = mpsc::channel::<Alert>();
    let (metric_tx, metric_rx) = mpsc::channel::<Metric>();
    let start = Instant::now();

    thread::spawn(move || {
        thread::sleep(Duration::from_millis(300));
        alert_tx.send(Alert { message: "disk usage 95%".to_string() }).unwrap();
    });
    thread::spawn(move || {
        for i in 0..5 {
            metric_tx.send(Metric { name: format!("cpu_{i}"), value: i as f64 * 1.5 }).unwrap();
        }
    });

    let log = monitor_buggy(alert_rx, metric_rx, start);
    let metric_count = log.iter().filter(|(m, _)| m.starts_with("metric")).count();
    println!("=== buggy ===");
    for (msg, elapsed) in &log {
        println!("{:?} {}", elapsed, msg);
    }
    println!("送信したメトリクス5件のうち、実際に処理されたのは{}件", metric_count);
}

// ── 要件2・3: 修正版（単一チャンネル + enum Event）──

enum Event {
    Alert(Alert),
    Metric(Metric),
}

fn monitor_fixed(rx: Receiver<Event>, start: Instant) -> Vec<(String, Duration)> {
    let mut log = Vec::new();
    for event in rx {
        match event {
            Event::Alert(a) => log.push((format!("ALERT: {}", a.message), start.elapsed())),
            Event::Metric(m) => log.push((format!("metric: {}={}", m.name, m.value), start.elapsed())),
        }
    }
    log
}

fn demo_fixed() {
    let (event_tx, event_rx) = mpsc::channel::<Event>();
    let start = Instant::now();

    let alert_tx = event_tx.clone();
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(300));
        alert_tx
            .send(Event::Alert(Alert { message: "disk usage 95%".to_string() }))
            .unwrap();
        // alert_tx（cloneされた方）はここでdrop
    });

    let metric_tx = event_tx.clone();
    thread::spawn(move || {
        for i in 0..5 {
            metric_tx
                .send(Event::Metric(Metric { name: format!("cpu_{i}"), value: i as f64 * 1.5 }))
                .unwrap();
        }
        // metric_tx（cloneされた方）はここでdrop
    });

    // Day 032の教訓: 元のevent_txを握ったままだとforループが終わらない
    drop(event_tx);

    let log = monitor_fixed(event_rx, start);
    println!("=== fixed ===");
    for (msg, elapsed) in &log {
        println!("{:?} {}", elapsed, msg);
    }

    // ── 要件3: assertによる検証 ──
    let metric_count = log.iter().filter(|(m, _)| m.starts_with("metric")).count();
    let alert_count = log.iter().filter(|(m, _)| m.starts_with("ALERT")).count();
    assert_eq!(log.len(), 6, "合計6件（メトリクス5件+アラート1件）処理されるはず");
    assert_eq!(metric_count, 5, "メトリクスは1件も取りこぼされないはず");
    assert_eq!(alert_count, 1);

    for (msg, elapsed) in log.iter().take(5) {
        assert!(msg.starts_with("metric"), "先頭5件はメトリクスのはず: {msg}");
        assert!(
            *elapsed < Duration::from_millis(100),
            "メトリクスはほぼ即座に処理されるはず: {:?}",
            elapsed
        );
    }
    let (last_msg, last_elapsed) = &log[5];
    assert!(last_msg.starts_with("ALERT"));
    assert!(
        *last_elapsed >= Duration::from_millis(250),
        "アラートは送信された約300ms後に処理されるはず: {:?}",
        last_elapsed
    );

    println!("OK: メトリクス5件は即座に、アラートは約300ms後に処理された（取りこぼしなし）");
}

fn main() {
    demo_buggy();
    println!();
    demo_fixed();
}

$ rustc -O src/main.rs -o day033 && ./day033   # (1回目)
=== buggy ===
301.185958ms ALERT: disk usage 95%
301.201417ms metric: cpu_0=0
送信したメトリクス5件のうち、実際に処理されたのは1件

=== fixed ===
180.958µs metric: cpu_0=0
185.541µs metric: cpu_1=1.5
186.583µs metric: cpu_2=3
187.791µs metric: cpu_3=4.5
188.5µs metric: cpu_4=6
305.129875ms ALERT: disk usage 95%
OK: メトリクス5件は即座に、アラートは約300ms後に処理された（取りこぼしなし）

# 2回目・3回目も buggy版は「metric_count = 1」で安定して再現し、
# fixed版はメトリクス5件が数百マイクロ秒以内・アラートが300ms前後で
# 安定してassertを満たした

🪜 Step-by-Step 解説

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

このバグの根本原因は「異なる種類のイベントを、異なる型の別々のチャンネルで表現した」ことにあります。Rustではenumが「複数の種類のうちどれか1つ」を表す直和型として非常に軽量に使えます（Phase 1のOption/Resultと同じ発想です）。イベント駆動なコードを設計するとき、「Aというイベントの種類」と「Bというイベントの種類」を型レベルで別々の経路（別々のチャンネル、別々のコールバック）に分けてしまうと、それらを同時に待ち受ける処理が本質的に難しくなります。

一方、enum Event { A(...), B(...) }のように1つの型へ統合してしまえば、「イベントを待つ」という関心事はfor event in rxという1箇所のコードに収束し、順序保証・取りこぼしなし・ブロッキングなしがチャンネルの標準動作としてタダで手に入ります。これは「所有権システムで安全性を型に落とし込む」というRustの思想を、並行処理の設計判断にまで一段引き上げた考え方です——「2つの独立した状態をどう同期させるか」を悩む代わりに、「そもそも1つの状態（1つのキュー）で表現できないか」を先に検討するのがRustらしい筋の良い設計です。

🛑 静かなバグの見つけ方

今回はコンパイルエラーではなく、コンパイルは通るのに意図通り動かない「静かなバグ」が題材でした。この種のバグは借用チェッカーが検出できる範囲の外にあり、実行して初めて気づけます。

気づき方: 「複数のチャンネル/キューを1つのループでポーリングしている」コードを見つけたら、「片方へのrecv()（ブロッキング呼び出し）が、もう片方の処理タイミングを遅延させていないか」を疑ってください。特に「低頻度なチャンネルへのrecv()を先に呼び、その後で高頻度なチャンネルをtry_recv()する」という順序は、低頻度チャンネル待ちの間ずっと高頻度チャンネルの処理が止まる典型的なアンチパターンです。

修正方針: 今回のように「そもそも合流できないか」を最初に検討し、型で表現できない構造的な理由がある場合に限り、recv_timeoutによるポーリングやcrossbeam-channelのselect!マクロを検討します。

🌐 他言語との比較

🏆 実務での使いどころ

• イベントループの設計: WebSocketメッセージ・タイマーのtick・シャットダウン信号など、由来の異なる複数のイベントソースを1つのループで処理したい場面では、今回のようにenum Eventで統合し単一のチャンネル（または単一のtokio::select!）で受けるのが定石
• アクターモデル: 1つのアクター（スレッドやタスク）が複数の呼び出し元から異なる種類のリクエストを受け取る設計では、enum Message { Get(...), Set(...), Shutdown }のように1つのメッセージ型へ統合し、単一のmpscチャンネルで受信するパターンが非常に広く使われる（tokioの公式ドキュメントの"actors"パターンもこの形）
• 監視・オブザーバビリティ基盤: 今回のようにアラート（低頻度）とメトリクス（高頻度）が混在するシステムでは、優先度や取りこぼし要件が異なるイベントを安易に同じポーリングループへ詰め込むと本番でサイレントに欠落が起きる。優先度が本当に異なる場合は「合流」ではなくsync_channelのbound設計や別スレッド化を検討する判断も必要

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: 今回のEventenumにShutdownバリアントを追加し、外部からevent_tx.send(Event::Shutdown)することでmonitor_fixedのループを能動的に終了できるようにしてみましょう。Senderを全てdropするまで待つのではなく、明示的な終了シグナルで止める設計との使い分けを考えてみてください
• 次回予告: 次回はSend/Syncトレイトの理解を扱います。Day 031・Day 032・今日で使ってきたArc<Mutex<T>>やmpsc::Sender<T>がなぜスレッド間を安全に行き来できるのか、その裏側でコンパイラが検査しているSend/Syncマーカートレイトの仕組みを正面から見ていきます

🎯 自己評価

自分の回答

気づき・メモ