Day 032 — チャンネル（mpsc）によるワーカープールとバックプレッシャー設計

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

昨日（Day 031）はArc<Mutex<T>>で「1つの状態を複数スレッドが共有し、ロックで排他制御する」というアプローチを扱いました。今日はそれとは対照的な発想であるチャンネル（channel）を扱います。Rustコミュニティでよく引用される標語に、Go言語由来のこんな言葉があります。

Arc<Mutex<T>>方式では、複数スレッドが同じTに生の読み書きアクセス権を持ち、その競合をロックで手動調整します。チャンネル方式では、値の所有権そのものをメッセージとして送信側から受信側へムーブします。送信した瞬間、送信側はその値へのアクセス権を失う（ムーブセマンティクスにより二重アクセスがコンパイル時に排除される）ため、「同じデータに複数スレッドが同時にアクセスする」という状況自体が構造的に起こりません。

std::sync::mpsc（Multi-Producer Single-Consumer、複数の送信者・単一の受信者）が標準ライブラリの提供するチャンネルです。

• mpsc::channel::<T>() — (Sender<T>, Receiver<T>)のペアを返す。バッファは実質無制限（メモリが許す限りキューできる）
• Sender<T>はClone可能 → 複数スレッドがそれぞれclone()したSenderを持てる（Multi-Producer）
• Receiver<T>はClone不可能 → 受信できるのは常に1箇所だけ（Single-Consumer）。これが「複数ワーカーで1つの仕事キューを分け合う」ような設計でArc<Mutex<Receiver<T>>>が必要になる理由です
• 全てのSenderがドロップされると、Receiver::recv()はErrを返し、for x in receiverのようなイテレータ利用では自動的にループが終了する（「チャンネルが閉じた」ことが型システムのレベルではなく実行時のResultで伝わる）
• mpsc::sync_channel::<T>(bound) — バッファサイズをbound件に制限したチャンネル。バッファが満杯のときsend()は受信側が1件取り出すまでブロックする。これは「送信側が受信側より速すぎる」状況を強制的に減速させるバックプレッシャーの仕組みになる

📝 問題

以下の2つの機能を実装してください。

要件1: WorkerPool（複数ワーカーがジョブチャンネルを共有して消費する）

• type Job = Box<dyn FnOnce() + Send + 'static>;を定義する
• WorkerPool::new(num_workers: usize) -> Self: 内部でmpsc::channel::<Job>()を作り、ReceiverをArc<Mutex<Receiver<Job>>>で包んでnum_workers本のワーカースレッドに共有させる。各ワーカーはループで「ロックを獲得 → 1件recv() → ロック解放 → ジョブを実行」を繰り返し、チャンネルが閉じたら（recv()がErrを返したら）ループを抜けてスレッド終了する
• submit(&self, job: Job): ジョブをチャンネルへ送信する
• shutdown(self): 内部のSenderをドロップしてワーカーへ終了を知らせ、全ワーカーのjoin()を待つ
• WorkerPool::new(4)でプールを作り、1..=20の各nについてn * nを計算して結果チャンネルへ送るジョブをsubmitする。メインスレッドは結果を全て受信し、件数が20件・合計が2870（1^2 + 2^2 + ... + 20^2）であることを確認する

要件2: sync_channelによるバックプレッシャーの体感

• mpsc::sync_channel::<u32>(2)（バッファ2件）でチャンネルを作る
• 送信側スレッドは0..5を順番にsend()し、各send()の直後の経過時間（Instant::elapsed()）を記録する
• メインスレッドは150ミリ秒待ってから受信を開始し、1件受信するごとに80ミリ秒スリープする（＝受信側をわざと遅くする）
• 全て受信し終えたら、0,1,2,3,4が順番通り受信できたこと、および3件目の送信（インデックス2）がバッファが空くまでブロックされて100ミリ秒以上かかっていることを確認する

🔍 ヒント（段階的開示）

📋 コピーtype Job = Box<dyn FnOnce() + Send + 'static>;

struct WorkerPool {
    job_tx: Sender<Job>,
    handles: Vec<thread::JoinHandle<()>>,
}

impl WorkerPool {
    fn new(num_workers: usize) -> Self {
        let (job_tx, job_rx) = mpsc::channel::<Job>();
        let job_rx = Arc::new(Mutex::new(job_rx));
        let mut handles = Vec::new();
        for _ in 0..num_workers {
            let job_rx = Arc::clone(&job_rx);
            handles.push(thread::spawn(move || loop {
                let job = { /* ??? ロックしてrecv() */ };
                match job {
                    Ok(job) => job(),
                    Err(_) => break,
                }
            }));
        }
        WorkerPool { job_tx, handles }
    }
    // submit, shutdown は自分で設計する
}

✅ 模範解答

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

use std::sync::mpsc::{self, Receiver, Sender, SyncSender};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

// ── 要件1: WorkerPool（複数ワーカーがジョブチャンネルを共有して消費する）──

type Job = Box<dyn FnOnce() + Send + 'static>;

struct WorkerPool {
    job_tx: Sender<Job>,
    handles: Vec<thread::JoinHandle<()>>,
}

impl WorkerPool {
    fn new(num_workers: usize) -> Self {
        let (job_tx, job_rx) = mpsc::channel::<Job>();
        let job_rx = Arc::new(Mutex::new(job_rx));

        let mut handles = Vec::new();
        for _id in 0..num_workers {
            let job_rx = Arc::clone(&job_rx);
            let handle = thread::spawn(move || loop {
                // ロックは「1件受信する」間だけ保持し、ジョブの実行はロック外で行う。
                // recv()自体はロックを保持したままブロックしうるが、
                // ジョブが届けば即座にロックを解放するため他ワーカーを長時間妨げない。
                let job = {
                    let rx = job_rx.lock().unwrap();
                    rx.recv()
                };
                match job {
                    Ok(job) => job(),
                    Err(_) => break, // 全Senderがdropされ、チャンネルが閉じた
                }
            });
            handles.push(handle);
        }

        WorkerPool { job_tx, handles }
    }

    fn submit(&self, job: Job) {
        self.job_tx.send(job).expect("worker pool is shut down");
    }

    fn shutdown(self) {
        // job_tx をドロップすることで全ワーカーの recv() に Err を返させ、終了させる
        drop(self.job_tx);
        for handle in self.handles {
            handle.join().unwrap();
        }
    }
}

fn demo_worker_pool() {
    let pool = WorkerPool::new(4);
    let (result_tx, result_rx) = mpsc::channel::<i64>();

    for n in 1..=20i64 {
        let result_tx = result_tx.clone();
        pool.submit(Box::new(move || {
            let squared = n * n;
            result_tx.send(squared).expect("main receiver dropped");
        }));
    }
    drop(result_tx); // 元のSenderをdropしないと下のfor文がブロックし続ける

    let mut total = 0i64;
    let mut count = 0;
    for squared in result_rx {
        total += squared;
        count += 1;
    }

    assert_eq!(count, 20);
    assert_eq!(total, 2870); // 1^2+2^2+...+20^2
    println!("demo_worker_pool: count={}, total={}", count, total);

    pool.shutdown();
}

// ── 要件2: sync_channelによるバックプレッシャー ──

fn demo_backpressure() {
    // sync_channel(2): バッファは2件まで。3件目のsendは、
    // 受信側が最低1件受信してバッファに空きができるまでブロックする。
    let (tx, rx): (SyncSender<u32>, Receiver<u32>) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        let start = Instant::now();
        let mut send_times = Vec::new();
        for i in 0..5u32 {
            tx.send(i).unwrap();
            send_times.push(start.elapsed());
        }
        send_times
    });

    // 受信側は意図的にゆっくり受信し、送信側がブロックされる様子を作る
    thread::sleep(Duration::from_millis(150));
    let mut received = Vec::new();
    for _ in 0..5 {
        received.push(rx.recv().unwrap());
        thread::sleep(Duration::from_millis(80));
    }

    let send_times = producer.join().unwrap();

    assert_eq!(received, vec![0, 1, 2, 3, 4]);
    // 送信0,1はバッファ(容量2)に即座に収まるが、送信2以降は
    // 受信側の消費を待つため明確に遅くなる
    assert!(send_times[2] >= Duration::from_millis(100));
    println!("demo_backpressure: received={:?}", received);
    println!("demo_backpressure: send_times={:?}", send_times);
}

fn main() {
    demo_worker_pool();
    demo_backpressure();
    println!("OK");
}

$ rustc -O src/main.rs -o day032 && ./day032   # (1回目)
demo_worker_pool: count=20, total=2870
demo_backpressure: received=[0, 1, 2, 3, 4]
demo_backpressure: send_times=[23.333µs, 23.583µs, 153.54375ms, 238.518583ms, 321.007083ms]
OK

# 2回目・3回目も send_times[2] は 151ms 前後で安定して assert を満たした

🪜 Step-by-Step 解説

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

チャンネルの本質は「所有権の受け渡しを型システムに乗せる」ことです。send(value)はvalueをムーブします。送信した瞬間、送信側のスコープにはその値へのアクセス手段が残りません（借用チェッカーが以降の使用を拒否します）。つまり「送信側と受信側が同じデータに同時にアクセスしてデータ競合する」というクラスのバグは、Mutexのようにロックという実行時の規律で防ぐのではなく、そもそも同時に2箇所からアクセスできる状態が作れないという所有権システムの静的な保証によって排除されます。

これはRustの「ゼロコスト抽象化」思想とも一致します。チャンネルはロックのような明示的な排他制御プリミティブを使わずに安全な並行処理を実現しますが、内部実装（mpsc）自体はロックやアトミック操作を使っており、魔法ではありません。重要なのは、利用者側のコードからはロックの取得・解放・順序といった関心事が消え、「値を送る」「値を受け取る」という高レベルな操作だけが残る点です。Day 031のArc<Mutex<T>>が「共有された状態をどう安全に守るか」という問いに答えるのに対し、チャンネルは「そもそも状態を共有しない設計にできないか」という問いへの答えです。

🛑 ハングしてしまう場合

今回は実装問題ですが、チャンネル周りで非常によく踏むのは「コンパイルエラー」ではなく「ハング（永遠にブロックする）バグ」です。次のコードはコンパイルは通りますが、実行するとforループが終わらずプログラムが停止しなくなります。

📋 コピー// このコードはコンパイルは通るが、実行すると永遠にハングする
fn buggy_collect() {
    let (result_tx, result_rx) = mpsc::channel::<i64>();

    for n in 1..=20i64 {
        let result_tx = result_tx.clone();
        thread::spawn(move || {
            result_tx.send(n * n).unwrap();
        });
    }
    // ここで result_tx (元のSender) が drop されずに残ったまま
    // for文へ突入してしまう

    let mut total = 0;
    for squared in result_rx {
        total += squared; // 20件受信し終えても、result_txが生きているので
                           // チャンネルは「閉じていない」と判断され続け、ここで永遠に待機する
    }
    println!("{}", total); // 到達しない
}

気づき方: プログラムが（パニックもエラーも出さずに）応答しなくなったら、mpsc::Receiverをイテレータやrecv()ループで待っている箇所を疑い、対応する全てのSender（元・複製すべて）が意図したタイミングでドロップされているかを確認します。修正方針は模範解答のdrop(result_tx);のように、不要になったSenderを明示的にdropするか、Senderを専用のスコープ（{ }ブロック）に閉じ込めてスコープアウトで自動ドロップさせることです。

🌐 他言語との比較

🏆 実務での使いどころ

• ワーカープール/ジョブキュー: HTTPリクエスト処理・画像変換・バッチ集計など、「独立した単位のタスクを複数ワーカーで消費する」パターンは今回のWorkerPoolがほぼそのまま実務コードの骨格になる（Rust公式の"The Rust Programming Language"の最終章「マルチスレッドWebサーバー」もほぼ同じ設計）
• プロデューサー/コンシューマーによるパイプライン処理: ログの収集→整形→送信のように、複数ステージを別スレッド（またはtokioタスク）に分割し、ステージ間をチャンネルでつなぐ設計はストリーム処理・ETLパイプラインの定番構成
• バックプレッシャー設計: sync_channelの有界バッファは、下流の処理が詰まったときに上流を自動的に減速させる仕組みとして使われる。無制限のchannel()を使うと、下流が詰まった場合にメモリ上に未処理タスクが際限なく積み上がり、OOM（メモリ枯渇）につながるリスクがあるため、実務では意図的にバッファサイズを絞ったsync_channel（やtokio::sync::mpsc::channel(bound)）を使うことが多い

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: WorkerPoolにジョブの実行結果（成功/パニック）を集計する仕組みを追加してみましょう。ジョブがパニックした場合、thread::spawnしたワーカースレッド自体がパニックで終了してしまいます（catch_unwindで捕捉しない限り）。ワーカーが1つ落ちてもプール全体が機能し続けるようにするには、どう設計すればよいか考えてみてください
• 次回予告: 次回はSend/Syncトレイトの理解を扱います。今日と昨日で「なぜArc<Mutex<T>>やmpsc::Sender<T>がスレッド間を安全に行き来できるのか」を支えている、コンパイラのSend/Syncマーカートレイトの仕組みを正面から見ていきます

🎯 自己評価

自分の回答

気づき・メモ