スレッドと同期プリミティブ

スレッドは並列実行のコンテキストを提供し、同期プリミティブはどの操作が同時に発生可能か、どの結果が他の操作より先に可視化されるべきかを制約します。Rust の型システムは不正な共有の多くを防げますが、ロック順序、終了プロトコル、マルチステップなビジネスロジックの原子性については依然として設計が必要です。

通常のスレッドとスコープスレッド

thread::spawn のクロージャーは通常 Send + 'static を要求し、JoinHandle を返します。join は結果を待機して観測しますが、ハンドルをドロップしてもスレッドの待機やキャンセルは自動的に行われません。スレッドスタックサイズのデフォルト値はプラットフォーム実装に依存し、必要に応じて Builder::stack_size で設定します。これはクロスプラットフォームで固定される定数ではありません。

スレッドが確実に現在のスコープ内で終了する場合は、外側のオブジェクトを借用できます：

fn main() {
    let values = [1_u32, 2, 3];
    let sum = std::thread::scope(|scope| {
        let worker = scope.spawn(|| values.iter().sum::<u32>());
        worker.join().unwrap()
    });
    assert_eq!(sum, 6);
}

scope は、スコープが返される前に関連するスレッドが必要な join を完了することを保証するため、非 'static の借用を使用できます。依然としてスレッドの panic を処理する必要があり、Send や借用ルールを回避するためにスコープ形式を使うことはできません。thread::scope

Mutex と RwLock が保護するのはアクセスの区間

MutexGuard が生存している間はロックが保持され、通常のデストラクションやアンワインド時に解放されます。panic=abort やプロセスの強制終了などでは、すべてのデストラクタが実行されることが保証されません。クリティカルセクションの境界を明確にし、ロック保持中にコールバックの実行、長時間の I/O、または逆方向の依存関係にあるロックの待機を避けてください。

標準ライブラリの Mutex のポイズニングは、未完成の更新が存在する可能性を示すヒントであり、回避不可能な安全バリアではありません。PoisonError からガードを取り戻すことは可能ですが、不変条件を評価するまでは回復しないでください。panic の検出も完全ではないため、unsafe 抽象は UB を防ぐためにポイズンだけに依存することはできません。Mutex

RwLock は複数のリーダーまたは 1 つのライターを許可しますが、公平性はプラットフォーム実装に依存し、読み取りが多い書き込みが少ない場合に Mutex より必ずしも高速になるとは限りません。読み込みロックの保持中にアップグレードを待ったり、ライターが待機している間に再帰的に読み込みロックを取得したりすると、予測不可能なブロッキングを引き起こす可能性があるため、明確な操作順序を採用してください。

チャンネル：所有権の移動とクローズ設計

std::sync::mpsc はブロッキングインターフェースで、マルチプロデューサ・シングルコンシューマです。channel は無制限のキューイングセマンティクスを提供し、sync_channel は制限付きバッファを提供します。容量 0 は合流（join）に使用できます。非同期タスクでは、ワーカーをブロックしないよう適切な非同期チャンネルを使用してください。

最後のメッセージを送信しても、チャンネルがクローズされたわけではありません。すべての sender が解放された後、受信側はキューに残っているメッセージをすべて読み終えてから、recv は切断エラーを返します。使用しない sender のクローンが残っていると、待機が終了しなくなります。受信側がクローズされると送信は失敗する可能性があるため、永遠にコンシューマーが存在すると仮定するのではなく、ワークを回収してください。

メッセージのやり取りは、内部で「メモリが完全に共有されていない」ことを意味しません。これは、アプリケーションが値と所有権に基づいて協調を整理するためのインターフェースを提供するものであり、基盤には依然として共有キューと同期機構が使われている可能性があります。

送信の成功がビジネス処理の完了を意味しない

use std::{sync::mpsc, thread};
fn main() {
    let (work_tx, work_rx) = mpsc::sync_channel(0);
    let (done_tx, done_rx) = mpsc::channel();
    let worker = thread::spawn(move || {
        let value: String = work_rx.recv().unwrap();
        let length = value.len();
        done_tx.send(length).unwrap();
    });
    work_tx.send(String::from("job")).unwrap();
    // 合流は recv がペアになったことを確認するだけ；処理結果は別のチャンネルで返される。
    assert_eq!(done_rx.recv().unwrap(), 3);
    worker.join().unwrap();
    let unsent = work_tx.send(String::from("next")).unwrap_err().0;
    assert_eq!(unsent, "next");
}

容量 0 の send は受信操作と合流しますが、受信後にも処理作業が残ります。ビジネスの完了を知りたい場合は、確認応答または結果を返さなければなりません。バッファ付きチャンネルの send を処理の成功と見なしてはいけません。SendError<T> は送信できなかった T を呼び出し元に返すため、呼び出し元は保存、再ルーティング、またはレポートを行うことができ、失敗パスでの所有権の不明確な消失を防ぐことができます。sync_channel

制限付きチャンネルもデッドロックに関与する可能性があります。送信者が空きを待っている間に、コンシューマーがメッセージを処理するために必要なロックを保持している場合、コンシューマーはキューの圧力を解放できません。同期設計では、ロックとチャンネルの依存関係を一緒に確認する必要があり、ロックの取得順序だけを調べているだけでは不十分です。

Condvar：まず述語があり、次に通知がある

use std::sync::{Condvar, Mutex};
use std::thread;

fn main() {
    let shared = (Mutex::new(false), Condvar::new());
    thread::scope(|scope| {
        let waiter = scope.spawn(|| {
            let (lock, condition) = &shared;
            let guard = lock.lock().unwrap();
            let ready = condition.wait_while(guard, |ready| !*ready).unwrap();
            assert!(*ready);
        });
        let (lock, condition) = &shared;
        *lock.lock().unwrap() = true;
        condition.notify_one();
        waiter.join().unwrap();
    });
}

通知が待機スレッドが正式にスリープする前に発生した場合でも、その後 true を確認すると待機しないため、「その通知をたまたま受信した」ことに依存しません。wait_while はループチェックを記述し、疑似起床や他の待機スレッドが先に状態を消費する競合に対処します。Condvar

通知はセマフォの許可ではなく、将来 wait を呼び出すスレッドのために回数を保存しません。1 つの Condvar は、インターフェースの要件に従って同じ関連ロックと組み合わせて使用する必要があります。状態のチェックと変更は共通のプロトコルに従わなければなりません。

「今後結果がない」ことも述語に含める

ready フラグのみの待機プログラムでは、プロデューサが事前に失敗した場合、コンシューマーは永遠に true を待たされる可能性があります。以下では、待機、完了、クローズの 3 つの状態を区別します。クローズ自体も待機スレッドに通知する必要があります。

use std::{sync::{Condvar, Mutex}, thread};
#[derive(Debug)]
enum State { Waiting, Ready(u32), Closed }
fn receive(shared: &(Mutex<State>, Condvar)) -> Option<u32> {
    let (lock, condition) = shared;
    let mut state = condition.wait_while(lock.lock().unwrap(),
        |state| matches!(state, State::Waiting)).unwrap();
    match std::mem::replace(&mut *state, State::Closed) {
        State::Ready(value) => Some(value),
        State::Closed => None,
        State::Waiting => unreachable!("述語は待機条件を満たしていない"),
    }
}
fn run(result: Option<u32>) -> Option<u32> {
    let shared = (Mutex::new(State::Waiting), Condvar::new());
    thread::scope(|scope| {
        let receiver = scope.spawn(|| receive(&shared));
        *shared.0.lock().unwrap() = match result {
            Some(value) => State::Ready(value), None => State::Closed,
        };
        shared.1.notify_one();
        receiver.join().unwrap()
    })
}
fn main() {
    assert_eq!(run(Some(42)), Some(42));
    assert_eq!(run(None), None);
}

これはシングルプロデューサ・シングルコンシューマ・1 回限りの配信のローカルモデルであり、汎用チャンネルではありません。複数の待機スレッドに対するクローズでは通常 notify_all が必要であり、状態遷移後に誰がまだ待機可能かを再考する必要があります。プロデューサの panic は自動的に Closed にはならないため、所有者がクローズを担当するか、切断セマンティクスを持つチャンネルを採用する必要があります。

タイムアウト後もロックで保護された状態を確認する

wait_timeout が返す際にはロックが再取得され、起床の原因は通知、タイムアウト、または疑似起床のいずれかです。手動でループを書く場合、毎回 5 秒の完全な待機を再実行すると、連続する起床により総待機時間が大幅に延びる可能性があります。wait_timeout_while を使用して総時間を管理するか、固定された期限を保存して残り時間を計算してください。

use std::{sync::{Condvar, Mutex}, time::Duration};
fn main() {
    let ready = Mutex::new(false);
    let condition = Condvar::new();
    let (state, timeout) = condition.wait_timeout_while(
        ready.lock().unwrap(), Duration::ZERO, |ready| !*ready,
    ).unwrap();
    assert!(!*state);
    assert!(timeout.timed_out());
}

ここでは、スレッドのスケジューリング速度に依存せずに、満たされていない条件のタイムアウトブランチを検証するために 0 予算を使用しています。実際のタイムアウトも、正確な返却時刻を保証するものではありません。待機スレッドはロックの再取得と CPU の獲得を必要とするためです。最終的に結果が利用可能かどうかを決定するのは、述語であり、「通知を受信した」ことや「期限に達した」ことだけでは不十分です。Condvar のタイムアウト

アトミック変数：アトミック性は他のデータの公開を自動的に行わない

Relaxed はそのアトミック操作自体を保証するだけで、必要なクロスオブジェクトの公開関係を確立しません。Acquire は、対応する Release 書き込みまたは適用される解放シーケンスを観察した時点で初めて同期を確立します。2 つのスレッドでそれぞれ Acquire/Release キーワードを書くだけでは不十分です。

以下では、2 つのアトミック変数を使用して 1 回の公開を示します。ペイロード自体もアトミック型を使用し、教学コードに同期されていない通常のメモリアクセスが含まれないようにしています：

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::thread;

fn main() {
    let payload = AtomicUsize::new(0);
    let ready = AtomicBool::new(false);
    thread::scope(|scope| {
        scope.spawn(|| {
            payload.store(42, Ordering::Relaxed);
            ready.store(true, Ordering::Release);
        });
        while !ready.load(Ordering::Acquire) {
            thread::yield_now();
        }
        assert_eq!(payload.load(Ordering::Relaxed), 42);
    });
}

ready=true を読み取る Acquire は、それを公開した Release とペアになり、それ以前の payload の書き込みは後の読み取りに先行します。この例では 1 回の公開のみ行っています。同じフラグを繰り返し再利用する場合は、世代管理或其他のプロトコルが必要であり、単純にリセットして競合がないと仮定することはできません。

SeqCst は、対応する順序一貫性のあるアトミック操作に単一の全順序の要件を追加しますが、任意の通常の変数を自動的にアトミックにするものではなく、複数の操作をトランザクションとしてまとめるものでもありません。compare_exchange の成功と失敗のメモリ順序は異なります。失敗は読み取りに過ぎないため、書き込み方向の順序を任意に使用することはできません。Ordering

順序一貫性は複合操作の分割不能性を意味しない

use std::{sync::{Barrier, atomic::{AtomicUsize, Ordering}}, thread};
fn main() {
    let count = AtomicUsize::new(0);
    let barrier = Barrier::new(2);
    thread::scope(|scope| {
        for _ in 0..2 {
            scope.spawn(|| {
                let old = count.load(Ordering::SeqCst);
                barrier.wait();
                count.store(old + 1, Ordering::SeqCst);
            });
        }
    });
    assert_eq!(count.load(Ordering::SeqCst), 1);
    count.store(0, Ordering::Relaxed);
    thread::scope(|scope| {
        for _ in 0..2 {
            scope.spawn(|| { count.fetch_add(1, Ordering::Relaxed); });
        }
    });
    assert_eq!(count.load(Ordering::Relaxed), 2);
}

Barrier は、書き込み前に両側が読み取りを完了することを保証するため、丢失更新は確定的であり、実行を繰り返して運に頼る必要はありません。2 番目の fetch_add は、インクリメントを読み取り・変更・書き込みのアトミック操作として行います。ここではカウントが他のデータを公開するために使用されておらず、最終的な観測がスコープスレッドの完了後に行われるため、Relaxed は必要なカウントの原子性を表現するのに十分です。fetch_add の整数オーバーフローはラップアラウンドとして処理されます。カウントがラップアラウンドできない場合は、チェック付きの更新プロトコルを使用してください。AtomicUsize

Barrier は、参加者の数とフェーズが明確なシナリオにのみ適しています。ある参加者が到着前に早期に返却したり panic したりした場合、他のスレッドは永遠に待機する可能性があります。Barrier 自体はビジネスの終了回復を提供しません。また、ミューテックスでもありません。バリアを通過した後、各スレッドは同時に継続できます。

各操作が原子的でも更新を失う理由

分離した各読み書きが SeqCst の原子操作でも、両スレッドが同じ旧値を先に読めます。その順序を作ってから fetch_add に変えると、解決策は強いメモリ順序ではなく単一の原子的な読み取り・変更・書き込みだと分かります。このカウンタは他データの公開には使いません。

Acquire が読んだ書き込みはどれか

Acquire だけでなく、どの書き込みを読んだかを追います。単発公開の同期辺を示し、false なら分岐せず、true でも Release または Acquire が欠ければ必要な保証はありません。payload も原子なので、同期がない比較も定義された動作内です。

同期方法を選択する際は、待機と所有権を先に問う

カウントのみが必要で他の状態を公開しない場合は、Relaxed アトミックを検討してください。複数のフィールドが共通の不変条件を維持している場合、Mutex が証明しやすいことが多いです。複雑な述語を待機する場合は、ロックと Condvar を使用します。リソースの排他を 1 つの実行者に委ねる場合は、チャンネルを使用します。

スピンアップの例は順序を示すためのものであり、汎用的な待機戦略ではありません。長時間の待機には、適切なブロッキングまたは非同期通知メカニズムを使用してください。並行プログラムの検証では、正常な出力だけでなく、早期終了、クローズ、エラー、タスクの回収、ロックの依存関係も確認してください。

停止条件をプロトコルの一部とする

AtomicBool の停止フラグだけでは、recv や Condvar でブロックされているスレッドを唤醒 できません。フラグの書き込みに加え、チャンネルのクローズ、通知の送信、またはタイムアウト付きの待機を使用して、スレッドが停止条件を再確認する機会を与える必要があります。join は単に終了を待つだけで、スレッドに対してキャンセル要求を発行するものではありません。

デッドロックのトラブルシューティングでは、まず待機関係を図にしてください。誰が誰がデータを提供するかやリソースを解放するかを待っているか、誰が相手の必要なリソースを保持しているか。単に sleep を繰り返したり、スレッド数を増やしたり、より強力なメモリ順序に変えたりしても、このプロトコル層の循環依存関係は通常修正できません。