rust-concurrency - 背景

歴史的経緯

並行処理の課題

• 共有メモリモデル（1960年代〜）

• メッセージパッシング（1970年代〜）

• 現代の並行処理（2000年代〜）

Rustのアプローチ

// コンパイル時にデータ競合を検出
// Send: 別スレッドに転送可能
// Sync: 複数スレッドから同時アクセス可能

// Mutex<T>: T: Send なら Mutex<T>: Sync
// → 異なるスレッドから安全にアクセス可能

コンピュータサイエンス的な意味

Send と Sync

// Send: 所有権をスレッド間で転送可能
// - ほとんどの型は Send
// - Rc は Send でない（参照カウントが非アトミック）

// Sync: &T を複数スレッドに共有可能
// - T: Sync ⟺ &T: Send
// - RefCell は Sync でない（借用チェックが非スレッドセーフ）

データ競合の定義

データ競合は以下の3条件が全て成立する場合に発生:
1. 2つ以上のスレッドが同じメモリにアクセス
2. 少なくとも1つが書き込み
3. アクセスが同期されていない

Rustの所有権システムはこれを静的に防止

メモリ順序

use std::sync::atomic::Ordering;

// Relaxed: 順序保証なし（カウンタ向け）
// Acquire/Release: 同期ペア
// SeqCst: 最も強い保証（デフォルト）

counter.fetch_add(1, Ordering::Relaxed);

実践での活用

Webサーバー

// リクエストごとにスレッドプールから処理
let pool = ThreadPool::new(num_cpus::get());

for stream in listener.incoming() {
    let stream = stream.unwrap();
    pool.execute(|| {
        handle_connection(stream);
    });
}

データ処理パイプライン

// チャネルで処理をつなぐ
let (tx1, rx1) = mpsc::channel();
let (tx2, rx2) = mpsc::channel();

// ステージ1: 読み込み
thread::spawn(move || {
    for line in reader.lines() {
        tx1.send(line.unwrap()).unwrap();
    }
});

// ステージ2: 変換
thread::spawn(move || {
    for line in rx1 {
        tx2.send(process(line)).unwrap();
    }
});

// ステージ3: 出力
for result in rx2 {
    println!("{}", result);
}

キャッシュシステム

use std::sync::RwLock;
use std::collections::HashMap;

struct Cache<K, V> {
    data: RwLock<HashMap<K, V>>,
}

impl<K: Eq + Hash, V: Clone> Cache<K, V> {
    fn get(&self, key: &K) -> Option<V> {
        self.data.read().unwrap().get(key).cloned()
    }

    fn set(&self, key: K, value: V) {
        self.data.write().unwrap().insert(key, value);
    }
}

実世界とのギャップ

デッドロックは防げない

// Rustでもデッドロックは発生しうる
let a = Arc::new(Mutex::new(1));
let b = Arc::new(Mutex::new(2));

// スレッド1: a → b の順でロック
// スレッド2: b → a の順でロック
// → デッドロック！

// 対策: ロック順序を統一

async vs スレッド

// スレッド: I/O ブロッキングOK
// - スレッド生成コスト（数MB）
// - コンテキストスイッチコスト

// async: I/O で yield
// - タスク生成コスト小
// - ランタイム必要（tokio等）

性能のトレードオフ