majiayu000 Concurrency Observation（並行性観測）
目的
並行性バグは「通常テストが通るのに本番で死ぬ」の代表。 このスキルは、非決定性を再現可能な失敗に変える。
観測の恩恵
- 非決定性を"再現可能な失敗"に変える
- 「怖いから触れない」を減らし、変更可能性が上がる
- 事故が起きたときの原因切り分けが可能になる
典型的な並行性バグ
| バグ種別 | 症状 | 検出方法 |
|---|---|---|
| データレース | 同時書き換え／読み書き競合 | レースデテクタ |
| デッドロック | ロック順、待ち合わせ | デッドロック検出 |
| ライブロック | 特定条件で進まない | タイムアウト監視 |
| await忘れ | 非同期順序の取り違え | 静的解析 |
| キャンセル漏れ | リソースリーク | ストレステスト |
Procedure
Step 1: 共有可変状態の特定
並行アクセスされる共有可変状態を列挙：
[ ] グローバル変数
[ ] シングルトンの可変フィールド
[ ] キャッシュ（in-memory）
[ ] 接続プール
[ ] ファイルシステム
Step 2: レースデテクタの適用
言語別の実行方法：
# Go
go test -race ./...
# Rust
cargo +nightly miri test
# C/C++
clang -fsanitize=thread -g program.c
# Python (ThreadSanitizer with C extensions)
# 直接サポートなし → ストレステストで補完
# Node.js
# 直接サポートなし → ストレステストで補完
Step 3: ストレステストの実施
"たまたま通る"を潰すための増幅器：
import concurrent.futures
import threading
def test_counter_under_stress():
counter = SharedCounter()
def increment_many():
for _ in range(1000):
counter.increment()
# 100スレッドで同時実行
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
futures = [executor.submit(increment_many) for _ in range(100)]
concurrent.futures.wait(futures)
# 期待値: 100 * 1000 = 100,000
assert counter.value == 100000
Step 4: タイムアウトと飽和の監視
# キュー飽和の監視
def test_queue_does_not_saturate():
queue = TaskQueue(max_size=1000)
# 大量タスク投入
for i in range(10000):
queue.put(Task(i))
# 一定時間内に処理完了
start = time.time()
queue.join()
elapsed = time.time() - start
assert elapsed < 60 # 60秒以内に完了
assert queue.dropped_count == 0 # ドロップなし
Step 5: 運用メトリクスの設定
# 並行性関連メトリクス
metrics:
- name: thread_pool_active_threads
type: gauge
labels: [pool_name]
- name: queue_depth
type: gauge
labels: [queue_name]
- name: request_timeout_total
type: counter
labels: [endpoint]
- name: lock_wait_seconds
type: histogram
labels: [lock_name]
最小セット（条件付き）
並行性がある領域のみ適用：
- (E1) レースデテクタ/サニタイザをCIで回す（対象モジュールだけでも）
- (E2) ストレス（反復）を1本
- (E3) タイムアウト＋飽和メトリクス（運用で"詰まり"を見える化）
設計パターン（予防）
並行性バグを予防する設計パターン：
| パターン | 説明 |
|---|---|
| スレッド閉じ込め | 特定スレッドのみがデータにアクセス |
| イミュータブル | 変更不可なデータ構造 |
| メッセージパッシング | チャネル/アクター経由の通信 |
| 読み書きロック | 読み取りは並行、書き込みは排他 |
Outputs
- 共有可変状態の一覧
- レースデテクタ設定（CI統合）
- ストレステストコード
- 並行性メトリクス設定
Examples
Go のレースデテクタ CI設定
# .github/workflows/race.yml
name: Race Detection
on: [push, pull_request]
jobs:
race:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v4
with:
go-version: '1.21'
- name: Run tests with race detector
run: go test -race -v ./...
Node.js のストレステスト
import { Worker } from 'worker_threads';
describe('Concurrency Stress', () => {
it('handles concurrent writes without data loss', async () => {
const writes: Promise<void>[] = [];
// 100並列で書き込み
for (let i = 0; i < 100; i++) {
writes.push(db.insert({ id: i, value: `data-${i}` }));
}
await Promise.all(writes);
// 全件書き込まれていること
const count = await db.count();
expect(count).toBe(100);
});
});