rust-concurrency

Rust の並行処理

標準ライブラリの std::thread、std::sync、std::sync::atomic モジュールと、Async Book に基づいています。スレッドや非同期処理を使う Rust コードの設計、デバッグ、負荷テスト、レビューに使用します。キャンセル、タスクの所有権、ロックのスコープ、ランタイムのサイズ設定、キュー、過負荷管理、メッセージパッシングを扱い、基本的な所有権の問題は rust-stable に、安全でないコードの不変条件は rust-unsafe-ffi に引き継ぎます。

対応範囲

✅ 得意分野

1. OS スレッド（thread::spawn、Builder、join、スコープ付きスレッド、move クロージャ）
2. 同期プリミティブ（Mutex、RwLock、Barrier、Condvar、OnceLock、LazyLock）
3. アトミック型（AtomicBool/Isize/Usize、load/store/fetch_add/swap/compare_exchange、Ordering）
4. チャネル（mpsc：マルチプロデューサー・シングルコンシューマー、Receiver、Sender）
5. Send / Sync トレイトシステム（自動導出と手動実装）
6. Future トレイトを使った async/await 構文
7. Tokio ランタイム（tokio::main、tokio::spawn、select!、JoinSet）
8. 非同期 I/O の基礎（tokio::fs、tokio::net、tokio::io）
9. 有界キュー、バックプレッシャー、低速なコンシューマー、並行処理の上限、過負荷への対策
10. タスクの監視、接続のライフサイクル、キャンセル安全性、グレースフルシャットダウン
11. CPU バウンドなデータ並列処理と専用 Rayon プール
12. Crossbeam チャネル、キュー、ワークスティーリングデック、スコープ付きスレッド
13. 計測で必要性が確認された場合の、読み取り中心のスナップショット、シャーディングされたマップ、キャッシュ、代替ロック
14. Loom モデル検査と Tokio ランタイム診断

⚠️ 前提条件

1. Rust の所有権モデルの理解（rust-stable）

❌ 対象外のシナリオ

1. unsafe コードの並行実行 → rust-unsafe-ffi スキルを使用してください
2. 基本的な所有権／借用 → rust-stable スキルを使用してください

使用する場面

• 「複数のスレッドでデータを処理する」
• 「async/await の書き方」
• 「Tokio ランタイムの使い方」
• 「スレッド間でデータを共有する」
• 「データ競合を回避する」
• 「高並行サービスでレート制限とグレースフルシャットダウンを行う」
• 「Tokio チャネルの滞留や低速なコンシューマー」

データプライバシー

このスキルは、ユーザーデータを収集、保存、送信しません。

---

I. OS スレッド

use std::thread;

let handle = thread::spawn(move || {
    println!("Hello from thread!");
});
handle.join().unwrap();

// Thread with configuration
let builder = thread::Builder::new()
    .name("worker".into())
    .stack_size(1024 * 1024);
let handle = builder.spawn(move || { /* ... */ }).unwrap();

// scoped threads (1.63+)
let mut v = vec![1, 2, 3];
thread::scope(|s| {
    s.spawn(|| {
        v.push(4); // borrow, no move required
    });
});
println!("{v:?}"); // v remains usable

II. 同期プリミティブ

use std::sync::{Arc, Mutex, RwLock, Barrier, OnceLock, LazyLock};

// Mutex (mutual exclusion lock)
let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];

for _ in 0..10 {
    let counter = Arc::clone(&counter);
    handles.push(thread::spawn(move || {
        let mut num = counter.lock().unwrap();
        *num += 1;
    }));
}

// RwLock (read-write lock)
let data = Arc::new(RwLock::new(vec![1, 2, 3]));
{
    let read = data.read().unwrap();
    assert_eq!(read.len(), 3);
} // Drop the read guard before taking the write lock.
data.write().unwrap().push(4);

// OnceLock (thread-safe lazy initialization)
static CONFIG: OnceLock<String> = OnceLock::new();
let config = CONFIG.get_or_init(|| load_config());

// LazyLock
static CACHE: LazyLock<HashMap<String, Data>> = LazyLock::new(HashMap::new);

III. アトミック操作

use std::sync::atomic::{
    AtomicBool, AtomicU64, Ordering
};

static COUNTER: AtomicU64 = AtomicU64::new(0);
COUNTER.fetch_add(1, Ordering::SeqCst);

static READY: AtomicBool = AtomicBool::new(false);
READY.store(true, Ordering::Release);
let ready = READY.load(Ordering::Acquire);

// Ordering levels
// Relaxed — no ordering guarantees (only atomicity)
// Release — write visibility
// Acquire — read visibility
// AcqRel — both reads and writes visible
// SeqCst — global sequential order (strongest, but not automatically default; explicit Ordering required for atomic operations)

IV. チャネル

use std::sync::mpsc;

let (tx, rx) = mpsc::channel();
thread::spawn(move || {
    tx.send(1).unwrap();
    tx.send(2).unwrap();
});
for received in rx {
    println!("Got: {received}");
}

// Multi-producer scenario
let (tx, rx) = mpsc::channel();
let tx1 = tx.clone();

V. async/await

use tokio::time;

async fn do_work(id: u32) -> &'static str {
    time::sleep(time::Duration::from_secs(1)).await;
    println!("Task {id} done");
    "ok"
}

#[tokio::main]
async fn main() {
    // Concurrent execution
    let (r1, r2) = tokio::join!(do_work(1), do_work(2));

    // select!
    tokio::select! {
        result = do_work(1) => println!("task1: {result}"),
        result = do_work(2) => println!("task2: {result}"),
    }

    // tokio::spawn
    let handle = tokio::spawn(do_work(3));
    handle.await.unwrap();
}

VI. Send / Sync

// T is Send if its ownership can be transferred across threads
// &T is Sync if it can be shared references across threads

// Types that are both Send + Sync: Arc<Mutex<T>>, i32, &'static str
// !Send types: Rc<T>, *const T
// !Sync types: RefCell<T>, Cell<T>

// Manual implementations are unsafe contracts. Do not add them merely to
// satisfy a compiler error; prove aliasing, lifetime, and thread-safety first.

VII. 実行モデルの選択

ワークロード	最初の選択肢	避けること
準備完了を基準とする多数のネットワーク操作	入力を有界に受け付ける Tokio タスク	無制限の入力ごとにタスクやバッファーを割り当てること
独立した CPU 負荷の高い項目	Rayon の並列イテレーターまたは専用プール	Tokio ワーカーで長時間の CPU 処理を実行すること
ブロッキングするファイルシステム、FFI、またはレガシー API	spawn_blocking の呼び出しを有界にするか、専用プールを使用する	Tokio のブロッキングキューをバックプレッシャーとして扱うこと
同期 MPMC メッセージングまたはワークスティーリング	Crossbeam チャネル、キュー、またはデック	計測せずにロックフリー構造を選択すること
クリティカルセクションが短い小規模な共有状態	std::sync ロック	ガードを .await やコールバックをまたいで保持すること
読み取り中心の不変スナップショット	プロファイリング後に ArcSwap を使用する	読み取り中心の値すべてに並行マップを使うこと
キー付きの可変状態の共有	競合テスト後に所有権をシャーディングするか、DashMap を使用する	アトミック性を設計せずに複数のキーを操作すること
有効期限付きの並行キャッシュ	容量と削除ポリシーを明示した Moka	無制限のマップをキャッシュと呼ぶこと

Tokio は主に I/O 並行処理向け、Rayon は CPU 並列処理向けです。両者を組み合わせるには、明示的な受け渡し、独立した並行処理の上限、シャットダウンの所有権が必要です。サードパーティーのプリミティブを導入する前に、並行処理ツールの選択をお読みください。

ワークフロー

1. ワークロードを分類する — ランタイムやプリミティブを選ぶ前に、準備完了駆動の I/O、CPU 並列処理、ブロッキング呼び出し、同期、永続メッセージングを分けます。
2. 並行処理の予算を定める — 最大接続数、処理中タスク数、キュー容量、項目サイズ、タイムアウト、メモリ、CPU プール、シャットダウン期限を定義します。
3. 状態の所有権を決める — パーティション化または単一ライターによる所有権を優先します。状態を共有する場合は、アトミック性とロック順序の契約を明示します。
4. 通信セマンティクスを選択する — 有界のポイントツーポイント、リクエスト／リプライ、最新値、ロスありブロードキャスト、永続リプレイを意図的に選びます。キュー満杯時と受信側の遅延時の動作を指定します。
5. 実行を監督する — タスクまたはスレッドのハンドルを保持し、失敗を伝播させ、パニックを封じ込め、孤立した処理を防ぎ、呼び出し元によるキャンセル時の動作を定義します。
6. グレースフルシャットダウンを設計する — 新たな受け入れを停止し、プロデューサーを閉じ、キャンセルを通知し、期限内に join し、必要な状態をフラッシュして、未解決の失敗を返します。
7. 調整の前に測定する — スループット、p50／p95／p99 レイテンシ、キューの深さ、飽和度、タスクのポーリング時間、ウェイクアップ、ロック待機時間、CPU、アロケーション、RSS を記録します。
8. モデルを検証する — 過負荷とキャンセルをテストし、小規模な同期ステートマシンには Loom を使い、ランタイムの停止には tokio-console または tracing を使います。並行処理のテストと診断を参照してください。

注意点

1. Mutex::lock() は MutexGuard を返します。デッドロックを避けるため、ドロップする前に await しないでください
2. tokio::spawn の Future は Send と 'static の両方を満たす必要があります。Send でない参照があるとコンパイルエラーになります
3. 非同期クロージャは通常のクロージャとは異なる方法で所有権をキャプチャします。状態を移譲するには move キーワードを明示的に使ってください
4. select! の分岐でキャンセルされた Future は、ドロップされる前にクリーンアップ処理を直接実行しません
5. Atomic Ordering は関係性のセマンティクスではありません。Relaxed を誤用すると、予期しないメモリ順序の問題が発生する可能性があります
6. ブロードキャストの遅延は通常の成功経路とは異なります。破棄する、スナップショットを再構築する、遅いコンシューマーを切断する、イベントを永続的にリプレイする、のいずれかを選ぶ必要があります
7. max_blocking_threads が制限するのはブロッキングスレッドの数だけであり、投入キューへのバックプレッシャーは提供しません。コストの高いタスクには Semaphore または有界キューが必要です
8. JoinSet は完了順に結果を返します。API が入力順を要求する場合は、集約時に順序を復元できるよう、インデックスを引き回してください
9. DashMap、parking_lot、ArcSwap、ロックフリーキューは、パフォーマンスだけでなくセマンティクスも変えます。ベンチマークで不変条件のレビューを代替することはできません
10. Loom が認識できるのは Loom 対応型を通じて行われる同期処理だけであり、状態空間の爆発が起きる場合もあります。モデルは小さく決定的に保ってください
11. Rayon の処理は、キャンセルとプールの所有権を明示的に設計しない限り、非同期呼び出し元の終了後も続く可能性があります

オンデマンドリソース

• 並行処理の例
• 型とツールのクイックリファレンス
• 並行処理ツールの選択：Tokio、Rayon、Crossbeam、ロック、シャーディングされたマップ、スナップショット、キャッシュを選ぶ際に参照してください。
• 並行処理のテストと診断：同期の正しさの証明、ランタイム停止の診断、過負荷時やシャットダウン時の負荷テストを行う際に参照してください。
• 本番向け非同期サービスのパターン：アクター、バックプレッシャー、遅いコンシューマー、タスクの監督、ランタイム設定、シャットダウンプロトコルを設計する際に参照してください。
• examples/golden-threads/：CI でビルドされるスコープ付きスレッドの例

Official References

• std::thread Documentation
• std::sync Documentation
• std::sync::atomic Documentation
• Async Book
• Tokio Guide
• Rayon
• Crossbeam
• Loom