Day 029 — std::threadによる並行処理基礎（スレッド生成とjoin）

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

Phase 2まではシングルスレッドの世界で所有権・借用・トレイトを扱ってきました。今日からPhase 3、「恐れなき並行性（fearless concurrency）」の入口であるstd::threadを扱います。

std::thread::spawnはOSスレッドを1つ生成し、渡したクロージャを別スレッドで実行します。戻り値はJoinHandle<T>で、.join()を呼ぶとそのスレッドの終了を待ち、クロージャの戻り値をResult<T, Box<dyn Any + Send + 'static>>として受け取れます（Errはそのスレッドがpanic!した場合）。

use std::thread;

let handle = thread::spawn(|| {
    println!("別スレッドで実行中");
    42
});
let result: i32 = handle.join().unwrap();

ここで重要なのがthread::spawnのシグネチャです。

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
    F: FnOnce() -> T,
    F: Send + 'static,
    T: Send + 'static,

F: Send + 'staticという境界がRustの並行処理設計の核心です。

• 'static: 生成されたスレッドは、それを生成した親スレッドよりも長生きする可能性があります（join()を呼び忘れれば親が先に関数を抜けるかもしれません）。もしクロージャがローカル変数への参照を借用していたら、その参照先が先に破棄されダングリングポインタになり得ます。Rustはこれをコンパイル時に防ぐため、クロージャが借用する値はすべて'static（プログラム全体で有効、または所有権ごと移動された値）でなければならないと要求します。
• Send: その型の所有権を別スレッドに安全に移動できることを示すマーカートレイトです（詳細はDay 032「Send/Syncトレイト」で扱います）。

📝 問題

関数A: parallel_sum(data: Vec<i32>, num_threads: usize) -> i32
dataをnum_threads個のチャンク（塊）に分割し、それぞれを別スレッドで並列に部分和を計算させ、全スレッドのjoin()結果を合算した合計値を返す関数を実装してください。num_threadsがdata.len()を割り切らない場合も正しく動作すること。

関数B（デバッグ問題）: 以下のコードは「並列和を計算しよう」という意図で書かれていますが、コンパイルが通りません。実際に出るエラーを読み、原因を説明した上で、正しく動作するように修正してください。

📋 コピー// このコードはコンパイルエラーになる
use std::thread;

fn parallel_sum_wrong(data: Vec<i32>, num_threads: usize) -> i32 {
    let chunk_size = (data.len() + num_threads - 1) / num_threads;
    let mut handles = Vec::new();

    for chunk in data.chunks(chunk_size) {
        handles.push(thread::spawn(|| {
            chunk.iter().sum::<i32>()
        }));
    }

    handles.into_iter().map(|h| h.join().unwrap()).sum()
}

🔍 ヒント（段階的開示）

📋 コピーuse std::thread;

fn parallel_sum(data: Vec<i32>, num_threads: usize) -> i32 {
    let chunk_size = (data.len() + num_threads - 1) / num_threads;
    let mut handles = Vec::new();

    for chunk in data.chunks(chunk_size) {
        let owned_chunk: Vec<i32> = chunk.to_vec(); // 所有権を持つコピーを作る
        handles.push(thread::spawn(move || {
            // owned_chunk はこのクロージャにムーブされ、'static を満たす
            owned_chunk.iter().sum::<i32>()
        }));
    }

    handles
        .into_iter()
        .map(|h| /* ??? join()して中身を取り出す */)
        .sum()
}

✅ 模範解答

📋 コピー// このコードは cargo new day029_thread_basics の後、
// src/main.rs をこの内容で置き換えて `cargo run` で検証できます。

use std::thread;

// 関数A: data を num_threads 個のチャンクに分割し、
// 各チャンクの部分和を別スレッドで並列計算してから合算する。
fn parallel_sum(data: Vec<i32>, num_threads: usize) -> i32 {
    if data.is_empty() || num_threads == 0 {
        return 0;
    }
    // チャンクサイズは切り上げ計算にすることで、割り切れない場合も
    // 全要素を漏れなくカバーできる(最後のチャンクだけ小さくなる)。
    let chunk_size = (data.len() + num_threads - 1) / num_threads;
    let mut handles: Vec<thread::JoinHandle<i32>> = Vec::new();

    for chunk in data.chunks(chunk_size) {
        // &[i32] スライスは data を借用しているだけなので、
        // to_vec() で所有権を持つ Vec<i32> にコピーしてから move する。
        // これでクロージャは data への参照を一切持たなくなり、'static を満たす。
        let owned_chunk: Vec<i32> = chunk.to_vec();
        let handle = thread::spawn(move || owned_chunk.iter().sum::<i32>());
        handles.push(handle);
    }

    // join() は Result<i32, Box<dyn Any + Send>> を返す。
    // Err はそのスレッドが panic した場合。今回は panic しない前提なので unwrap で良い。
    handles
        .into_iter()
        .map(|h| h.join().unwrap())
        .sum()
}

// 関数Bの修正版(関数Aと同じ考え方: 借用ではなく所有権を渡す)
fn parallel_sum_fixed(data: Vec<i32>, num_threads: usize) -> i32 {
    let chunk_size = (data.len() + num_threads - 1) / num_threads;
    let mut handles = Vec::new();

    for chunk in data.chunks(chunk_size) {
        let owned_chunk = chunk.to_vec();
        handles.push(thread::spawn(move || owned_chunk.iter().sum::<i32>()));
    }

    handles.into_iter().map(|h| h.join().unwrap()).sum()
}

fn main() {
    let data: Vec<i32> = (1..=100).collect(); // 1〜100の合計 = 5050

    let a = parallel_sum(data.clone(), 4);
    assert_eq!(a, 5050);
    println!("関数A(4スレッド): {}", a);

    // 割り切れないケース(100個を7スレッド)
    let a_uneven = parallel_sum(data.clone(), 7);
    assert_eq!(a_uneven, 5050);
    println!("関数A(7スレッド・割り切れない): {}", a_uneven);

    let b = parallel_sum_fixed(data, 3);
    assert_eq!(b, 5050);
    println!("関数B修正版(3スレッド): {}", b);

    println!("OK");
}

$ cargo run
関数A(4スレッド): 5050
関数A(7スレッド・割り切れない): 5050
関数B修正版(3スレッド): 5050
OK

🪜 Step-by-Step 解説

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

Rustのstd::thread::spawnがSend + 'staticを要求するのは、「スレッド間でデータを安全にやり取りする責任を、実行時のロックや慣習ではなく、コンパイル時の型システムに移す」というRustの一貫した設計哲学の表れです。

C++のstd::threadはラムダが参照キャプチャしたローカル変数を平気で受け取れてしまい、スレッドがまだ実行中に親スコープを抜けるとダングリング参照が発生します（これは実行時のクラッシュか、最悪の場合サイレントな未定義動作として現れます）。Rustでは同じミスをコンパイルエラーとして即座に検出します。今回の「関数B」で見たエラーはまさにこの安全機構が働いた結果です。

所有権システムは単一スレッド内の安全性のためだけではなく、この並行処理の文脈でこそ最大の価値を発揮します——「あるデータに同時に何人がアクセスできるか」を型で表現する仕組みそのものが、スレッド境界をまたいでも自然に機能するのです。

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

問題文の「関数B」の元のコードをそのままコンパイルすると、次のようなエラーになります。

読み方: E0373は「クロージャが借用した値より長生きする可能性がある」ことを示すエラーコードです。thread::spawnのnote:が「引数の型は'staticを満たす必要がある」とはっきり教えてくれています。help:は「moveキーワードを付けて所有権を奪えば直るかもしれない」と提案してきますが、注意点として、単にmoveを付けただけではchunk: &[i32]という参照そのものがムーブされるだけで、それが指す先のdataのライフタイム問題は解決しません（chunkはスライス参照であってVec<i32>の実体ではないため）。今回のケースではmoveを付けるだけでなく、chunk.to_vec()で実体をコピーして所有権付きの値に変換する一手間が必須です。

🌐 他言語との比較

RustはC++に近い「メモリ共有型の並行処理」を提供しながら、Go/Java/TypeScriptが実行時の規律や設計上の制約で回避している問題を、コンパイル時の型チェックとして解決している点が最大の特徴です。

🏆 実務での使いどころ

• CPUバウンドな重い計算（画像処理・数値計算・大規模データの集計など）を複数コアに分散させ、std::threadで単純な並列化を行う場面（今回のparallel_sumはこのパターンの最小形）
• I/Oバウンドな処理（多数のHTTPリクエストなど）はstd::threadよりもDay 105以降で学ぶasync/await + tokioの方が軽量で適していることが多く、「CPUバウンド → OSスレッド」「I/Oバウンド → 非同期タスク」という使い分けの判断軸を持つことが実務上重要
• バッチ処理パイプラインで、入力データを固定数のワーカースレッドに分割し、join()で全ワーカーの完了を待ってから次工程に進む、という素朴な並列化パターンは今でもデータエンジニアリング系のツールで頻繁に使われる

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: parallel_sumをstd::thread::scope（Rust 1.63〜安定化）を使って書き直してみましょう。スコープ付きスレッドを使うと、to_vec()によるコピーなしで&dataを直接複数スレッドから安全に借用できます（スコープがjoinを保証するため'static制約が緩和される）
• 次回予告: 次のテーマはArc<Mutex<T>>による共有状態管理。今日は「各スレッドが独立したデータを持つ」設計でしたが、次回は複数スレッドが同じデータを安全に共有・変更する方法を学びます

🎯 自己評価

自分の回答

気づき・メモ