並行プログラミング

目次

• 概要
• 主なモデル
• 並行と並列
• Threadとlock
• Atomicとmemory ordering
• async/await
• 構造化並行性
• キャンセルとtimeout
• Channelとmessage passing
• Actor model
• 並行モデルの選び方
• 典型的な問題
• Backpressure
• テスト
• 観測とデバッグ
• 並行処理レビューの観点
• Mutexとlock-free データ構造
• メモリ順序と Happens-Before 関係
• async/await の詳細
• Actor Model の実装
• Backpressure の実装
• 並行処理のテスト と デバッグ
• 並行処理レビューの詳細チェックリスト
• イベント駆動モデルと非同期 I/O
• Python における GIL と並行性
• メッセージパッシングと分散システム
• 性能分析: スループット vs レイテンシ
• マルチプロセッシング と asyncio の選択
• 並行プログラミングのパターンと実装
• 並行データ構造
• スレッドプール と Task Queue
• Deadlock と Prevention
• Lock-free Programming
• Java Memory Model と Volatile
• 高度な同期メカニズム
• まとめ
• 参考文献

概要

同時に進む処理を、どう安全に書くか

並行プログラミングでは、複数の処理が同時進行することで性能や応答性を高めます。しかし同時実行は、競合、可視性、デッドロック、順序依存といった新しい難しさを持ち込みます。

この章で重視すること

• 並列と並行を区別する
• shared mutable stateを減らす方向で考える
• lock、channel、async、actorなどの選択肢を整理する

主なモデル

• thread + lock
• async/await
• message passing
• actor model
• immutability中心の設計

どれが絶対に優れているというより、共有状態をどう扱うかの違いとして見ると整理しやすいです。

並行モデルを選ぶときは、最初に「何を共有する必要があるか」を確認します。共有状態が少ないならmessage passingやimmutable dataで単純にできます。低レイヤで高性能な共有データ構造が必要ならlockやatomicが候補になりますが、その分レビューの難度は上がります。

この判断は固定ではありません。最初はmessage passingで単純に作り、性能要件が見えてからshared memoryへ寄せる方が安全な場合もあります。

並行と並列

並行は、複数の作業を同時進行として扱う構造です。並列は、実際に複数の計算資源で同時に実行することです。

1コアでも並行処理はできます。I/O待ちの間に別の処理へ切り替えれば、応答性は上がります。一方、CPU boundな計算を本当に速くするには並列実行が必要です。

Threadとlock

threadは古典的で強力な方法です。複数threadが同じメモリを共有するため、共有状態を守るlockが必要になります。

lockを使うときの注意は次です。

• lockの範囲を小さくする
• lock順序を固定する
• I/O中にlockを持ち続けない
• 共有mutable stateを減らす

Atomicとmemory ordering

低レイヤの並行処理では、lock以外にatomic操作を使うことがあります。atomicは単一の読み書きやcompare-and-swapを安全に行うための道具です。

ただしatomicは簡単ではありません。memory orderingを誤ると、あるthreadから見える順序と別のthreadから見える順序がずれることがあります。通常のアプリケーションでは、まずlock、channel、actor、既存のconcurrent collectionを使い、atomicは本当に必要な場所へ閉じ込めます。

async/await

async/awaitは、I/O待ちの多い処理で効きます。OS threadを大量に増やさず、event loop上で待ちを管理できます。

ただし、asyncはCPU boundな処理を自動で速くするものではありません。重い計算をevent loop上で走らせると、他の処理を止めます。

構造化並行性

構造化並行性は、開始した並行タスクの寿命を親のスコープに結びつける考え方です。タスクを起動したまま忘れると、キャンセル漏れ、エラーの握りつぶし、終了待ち漏れが起きます。

悪い兆候:
  startBackgroundJob() だけ呼び、誰も完了を待たない

よい方向:
  親スコープが子タスクを待つ
  失敗時に子タスクをキャンセルする
  timeoutとcleanupをまとめて扱う

Kotlin coroutine、Swift structured concurrency、Python TaskGroupなどは、この考え方を言語や標準ライブラリへ取り込んでいます。

キャンセルとtimeout

並行処理では、成功時だけでなく「やめる」設計が重要です。

• ユーザーが画面を閉じた
• リクエストがtimeoutした
• 上流処理が失敗した
• shutdownが始まった

このとき、子タスク、DB query、外部API呼び出し、queue処理が止まらないと、資源を使い続けます。キャンセル可能なAPIを選び、finallyやdeferで後始末を行います。

timeoutは単に短くすればよいわけではありません。上流のtimeoutより下流のtimeoutが長いと、ユーザーには失敗を返したあとも裏側で処理が残ります。逆に短すぎるtimeoutは不要なretryを増やし、障害時の負荷を悪化させます。

キャンセル設計では、止めることと、途中まで進んだ処理をどう戻すかを分けて考えます。副作用を持つ処理では、冪等性や補償処理も必要になります。

Channelとmessage passing

message passingでは、共有メモリを直接触るのではなく、メッセージでやり取りします。状態を所有する処理を決め、他の処理はメッセージで依頼する形にすると、競合を減らせます。

Actor model

actorは、自分の状態を持ち、メッセージを受け取って処理する単位です。各actorは基本的に独立して動くため、状態の所有者が分かりやすくなります。

分散システムやUI、ゲーム、バックグラウンドジョブなどで考え方が使われます。

並行モデルの選び方

「asyncにすれば速い」「threadを増やせば速い」ではなく、待ち時間、CPU使用、共有状態、キャンセルの形で選びます。

典型的な問題

• race condition
• deadlock
• livelock
• starvation
• visibility問題

Deadlock

deadlockは、複数の処理が互いに相手の解放を待って進めなくなる状態です。

防ぐには、lockの取得順序を固定し、timeoutやtry-lockを使い、そもそも複数lockを同時に持たない設計にします。

Backpressure

処理する側より作る側が速いと、queueが溜まり続けます。backpressureは、下流が詰まったときに上流へその状態を伝える考え方です。

• queue lengthを監視する
• rate limitをかける
• drop / retry / shed loadを決める
• timeoutを明示する

backpressureがないシステムでは、遅延が静かに蓄積し、最後にmemory不足やtimeoutの連鎖として表面化します。queueがあるだけでは不十分で、queueが詰まったときに上流がどう振る舞うかを決めます。

重要なのは、すべてを処理しようとしない判断です。リアルタイム性が重要なデータなら古いものを捨てる、決済のように失えない処理なら受け付けを制限する、というように業務の意味で選びます。

テスト

並行バグは再現しにくいので、テストでは次を意識します。

• deterministicなスケジューラを使えるか
• race detectorを使う
• timeoutを短くしすぎない
• 共有状態の境界を小さくする
• property-based testingを検討する

観測とデバッグ

並行バグはログだけでは見えにくいことがあります。次を記録すると調査しやすくなります。

• task id / request id
• queue length
• lock wait time
• timeout理由
• retry回数
• cancellationの発生箇所
• thread pool / event loopの使用率

特に「遅い」の原因がCPUなのか、I/O待ちなのか、lock待ちなのかを分けることが重要です。

並行処理レビューの観点

レビューでは、次を確認します。

• 共有mutable stateはどこか
• その所有者は誰か
• lock順序は固定されているか
• timeoutとキャンセルがあるか
• retryで二重実行にならないか
• queueが詰まったときにどうなるか
• shutdown時に処理中タスクを待つか

並行処理は「正常に動く」より「壊れ方が制御されている」ことが重要です。

共有メモリを使うときの注意

共有メモリは高速ですが、正しさの説明が難しくなります。MDNのAtomicsやSharedArrayBufferの説明でも、共有されたメモリに複数threadが読み書きする場合は、atomic操作や待機・通知の意味を理解する必要があるとされています。

共有メモリを選ぶ前に、次を確認します。

WebではSharedArrayBufferに追加のセキュリティ要件があるように、並行性は実行環境の制約とも結びつきます。言語機能だけでなく、runtime、OS、ブラウザ、deployment環境まで含めて設計します。

Mutexとlock-free データ構造

Mutex 設計

Mutexを使うときの基本的な安全性パターン：

// Rustの例（RAII）
use std::sync::Mutex;

let data = Mutex::new(vec![1, 2, 3]);

{
    let mut guard = data.lock().unwrap();
    guard.push(4);
} // ここで自動的にunlock

// 一般的な間違い：lockを長く持ちすぎ
let mut guard = data.lock().unwrap();
let result = expensive_io_operation();  // I/O中も lock保持
expensive_cpu_operation(result);
drop(guard);  // 後でunlock

Lock-free アルゴリズムの基本：

• Compare-And-Swap (CAS): 原子的に「期待値と一致したら新値へ」操作
• Memory order (Sequentially Consistent, Release-Acquire, Relaxed)
• ABA問題：値が戻ってくる間に別の変更が起きる可能性

実装例（Treiber stack - lock-freeスタック）：

Top = [A] -> [B] -> [C]

push(X):
  new_node = new Node(X)
  loop:
    expected_top = Top
    new_node.next = expected_top
    if CAS(Top, expected_top, new_node):
      break

pop():
  loop:
    expected_top = Top
    if expected_top == null:
      return null
    next_node = expected_top.next
    if CAS(Top, expected_top, next_node):
      return expected_top.value

Readers-Writer Lock

読み取り多数、書き込み少数のシナリオ：

• 複数読み取りは同時実行
• 書き込みは排他（読み取りとも排他）

実装イメージ：

state = { reader_count: 0, writer_flag: false }

read_lock():
  lock(state)
  while state.writer_flag:
    wait()
  state.reader_count += 1
  unlock(state)

write_lock():
  lock(state)
  while state.reader_count > 0 or state.writer_flag:
    wait()
  state.writer_flag = true
  unlock(state)

read_unlock():
  lock(state)
  state.reader_count -= 1
  if state.reader_count == 0:
    notify_all()
  unlock(state)

メモリ順序と Happens-Before 関係

マルチスレッド環境でのメモリ操作の順序は、CPUの投機実行やコンパイラ最適化により、プログラムで書いた順序と異なるかもしれません。

// Thread 1
x = 1;                  // (A)
std::atomic_store(&y, 1, std::memory_order_release);  // (B)

// Thread 2
while (!std::atomic_load(&y, std::memory_order_acquire)) {}  // (C)
print(x);               // (D) - 常に 1 が見える

メモリオーダー：

• Relaxed: 順序を保証しない、原子性だけ
• Release-Acquire: リリース（書き込み）とアクイア（読み取り）ペアで順序を保証
• Sequentially Consistent: すべてのスレッドで同じ全順序（最も厳しい）

async/await の詳細

非同期処理を簡潔に書ける言語機能：

# Python asyncio
async def fetch_user(user_id):
    user = await db.query(f"SELECT * FROM users WHERE id = {user_id}")
    return user

async def fetch_multiple():
    users = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3)
    )
    return users

asyncio.run(fetch_multiple())

Rustの例：

async fn fetch_user(client: &Client, id: u64) -> Result<User> {
    let response = client.get(format!("/users/{}", id)).send().await?;
    response.json().await
}

#[tokio::main]
async fn main() {
    let results = futures::future::join3(
        fetch_user(&client, 1),
        fetch_user(&client, 2),
        fetch_user(&client, 3)
    ).await;
}

Asyncの制約：

• CPU bound タスクはブロッキング（event loopを止める）
• 通常はworker pool（tokio::task::spawn_blocking）で分離
• デッドロックに注意（.await 中にlock保持）

Actor Model の実装

Erlang/Akkaなど：

// Akka (Scala)
sealed trait CounterMsg
case object Increment extends CounterMsg
case object GetCount extends CounterMsg
case class CountResponse(n: Int) extends CounterMsg

class CounterActor extends Actor {
  var count = 0

  override def receive: Receive = {
    case Increment =>
      count += 1
    case GetCount =>
      sender() ! CountResponse(count)
  }
}

// Usage
val system = ActorSystem("CounterSystem")
val counter = system.actorOf(Props[CounterActor], "counter")

counter ! Increment
counter ! Increment

implicit val timeout = Timeout(1 second)
val future = (counter ? GetCount).mapTo[CountResponse]
val result = Await.result(future, timeout.duration)

Actor設計の利点：

• スレッド間のメッセージで状態変化を追跡しやすい
• スケーリングが容易（分散へ拡張可能）
• エラー処理の枠組みが明確

Backpressure の実装

上流の生産速度が下流の処理速度を上回るとき、バッファオーバーフローやメモリ枯渇を防ぐため、上流を制御します。

# Producer - Consumer with backpressure
import asyncio

async def producer(queue):
    for i in range(100):
        print(f"Producing {i}")
        await queue.put(i)  # キューが満杯なら自動待機
        await asyncio.sleep(0.01)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consuming {item}")
        await asyncio.sleep(0.1)  # 遅い処理

async def main():
    queue = asyncio.Queue(maxsize=5)  # backpressure point
    
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

asyncio.run(main())

RxJS (Reactive Extensions) での表現：

const { interval } = require('rxjs');
const { throttleTime, bufferCount } = require('rxjs/operators');

interval(10)  // 100ms ごとに値を出す
  .pipe(
    throttleTime(500),  // 500ms に制限
    bufferCount(10)     // 10個単位で処理
  )
  .subscribe(batch => {
    console.log('Processing batch:', batch);
  });

並行処理のテスト と デバッグ

Race condition の検出

// Go: race detector
// コンパイル時に -race フラグを使用
// go test -race ./...

package main

import "sync"

func main() {
    var x int
    var wg sync.WaitGroup
    
    wg.Add(2)
    
    go func() {
        defer wg.Done()
        x = 1  // data race
    }()
    
    go func() {
        defer wg.Done()
        print(x)  // data race
    }()
    
    wg.Wait()
}

// 実行時に検出：
// WARNING: DATA RACE
// Write at 0x00c0001b0010 by goroutine 6:
//     main.main.func1()
//         race.go:13 +0x34

Thread sanitizer (C++)

// Compile with: clang++ -fsanitize=thread -g -O1
#include <thread>

int x = 0;
std::mutex m;

void increment() {
    // m.lock();  // この行をコメントアウトすると race
    x++;
    // m.unlock();
}

int main() {
    std::thread t1(increment);
    std::thread t2(increment);
    t1.join();
    t2.join();
}

デバッグ困難性への対策

並行バグは非決定的で再現が難しい：

1. 詳細ログ: スレッドID、タイムスタンプ付きログ
2. 繰り返し実行: ストレステスト（大量スレッド、繰り返し）
3. 静的解析: linter、型システム
4. 監視: デッドロック検知、デッドロック timeout

# Python の threading.stack_size() や sys.settrace() で監視
import sys
import threading

def trace_calls(frame, event, arg):
    if event == 'call':
        code = frame.f_code
        if 'interesting_module' in code.co_filename:
            print(f"{threading.current_thread().name}: {code.co_filename}:{frame.f_lineno} {code.co_name}")
    return trace_calls

sys.settrace(trace_calls)

並行処理レビューの詳細チェックリスト

コードレビュー時：

イベント駆動モデルと非同期 I/O

ネットワーク I/O の大部分は「待ち時間」です。スレッドベースではなく、イベントループで処理するモデルが効率的:

Node.js の非同期パラダイム

// Blocking (悪い例)
const file1 = readFileSync('file1.txt');
const file2 = readFileSync('file2.txt');  // file1 完了まで待機
console.log(file1.length + file2.length);

// Non-blocking (良い例)
fs.readFile('file1.txt', (err, data1) => {
  fs.readFile('file2.txt', (err, data2) => {
    console.log(data1.length + data2.length);
  });
});

// Async/Await (最新)
async function process() {
  const [file1, file2] = await Promise.all([
    fs.promises.readFile('file1.txt'),
    fs.promises.readFile('file2.txt')
  ]);
  console.log(file1.length + file2.length);
}

イベントループは単一スレッドで大量接続を処理。CPU コアは複数プロセスで活用。

Go の Goroutine と CSP

Go の goroutine は OS スレッドと異なり、軽量（スタック ~2KB）です:

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    ch := make(chan int, 10)
    
    // 1,000,000 個の goroutine を起動
    for i := 0; i < 1000000; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            ch <- n * 2
        }(i)
    }
    
    // メインスレッドが全タスクの完了を待機
    wg.Wait()
    close(ch)
}

OS スレッドなら数千個が限界。goroutine なら100万個も軽い。

Channel による通信

Goroutine 間は Channel で通信:

func fetch(url string, ch chan string) {
    resp, _ := http.Get(url)
    body, _ := ioutil.ReadAll(resp.Body)
    ch <- string(body)
}

func main() {
    ch := make(chan string)
    
    go fetch("http://example.com/1", ch)
    go fetch("http://example.com/2", ch)
    
    result1 := <-ch  // 最初の完了を待機
    result2 := <-ch
    
    fmt.Println(result1, result2)
}

Channel は同期化とデータ転送を同時に実現。

Python における GIL と並行性

Python の Global Interpreter Lock (GIL) は、複数スレッドがバイトコードを同時実行できません。

CPU バウンドは マルチプロセッシング

# Bad: スレッド（GIL により順次実行）
import threading
import time

def cpu_task():
    total = 0
    for i in range(100_000_000):
        total += i
    return total

# 2スレッドで 50% 高速化期待
start = time.time()
threads = [threading.Thread(target=cpu_task) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
# 実際: ほぼ単一スレッド並 (GIL のため順次実行)

# Good: マルチプロセッシング（各プロセスが独立 GIL）
from multiprocessing import Pool

start = time.time()
with Pool(2) as p:
    results = p.map(cpu_task, range(2))
# 約2倍高速化

I/O バウンドは asyncio

# スレッド： I/O 待機中も他タスク実行できない可能性
import requests
import threading

def fetch(url):
    response = requests.get(url)  # ブロック
    return response.text

threads = [threading.Thread(target=fetch, args=(url,)) 
           for url in urls]

# asyncio: I/O 待機中に他タスクに制御を譲る
import aiohttp
import asyncio

async def fetch_async(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    tasks = [fetch_async(url) for url in urls]
    return await asyncio.gather(*tasks)  # 全並行実行

asyncio.run(main())

I/O 待機時間が大きければ asyncio が効率的。

メッセージパッシングと分散システム

スレッド間通信は複雑（共有メモリ）。代わりにメッセージパッシングを活用:

Erlang/Elixir のプロセスモデル

-module(counter).
-export([start/0, increment/1]).

start() ->
    spawn(fun loop/0).

loop() ->
    receive
        {increment, From} ->
            From ! {ok},
            loop();
        {value, From} ->
            From ! {value, N},
            loop()
    end.

各プロセスが独立。メッセージキューで通信。障害時の再起動も簡単。

この設計は、マイクロサービスアーキテクチャと親和性が高い。

Actor Model のスケーリング

Akka（Java/Scala）はActor Modelを実装:

class CounterActor extends Actor {
    var count = 0
    
    def receive = {
        case Increment =>
            count += 1
        case GetValue =>
            sender() ! count
    }
}

val system = ActorSystem("MySystem")
val counter = system.actorOf(Props[CounterActor], "counter")

counter ! Increment  // 非同期メッセージ送信
counter ! GetValue

複数マシンに分散可能。Node 障害時も他 Actor は継続。

性能分析: スループット vs レイテンシ

並行性設定によって特性が大きく変わります:

Thread pool size が小さい:
  - レイテンシ: 高（キュー待機）
  - スループット: 低

Thread pool size が大きい:
  - レイテンシ: 低
  - スループット: 中程度（コンテキストスイッチ増加）

最適値: CPU コア数 × 1～2 倍（I/O 待機率により調整）

実際のアプリケーションでは負荷テストで最適値を決定します。

マルチプロセッシング と asyncio の選択

Python の GIL では CPU バウンド にマルチプロセッシング。I/O バウンド に asyncio。

Go の Goroutine は OS スレッドより軽量（1~2KB）。百万個も効率的。Channel で安全な通信。

同期化プリミティブ：Mutex は排他的ロック。RWMutex は読み取り複数並行。WaitGroup で複数ワーカーの同期。Semaphore でリソース制御。

デッドロック防止：4つの必要条件のいずれかを破る。ロック順序を統一。プロセス間通信でロックを避ける。

並行プログラミングのパターンと実装

Mutex と Semaphore

Mutex（相互排除）: クリティカルセクションへの排他的アクセス。acquire / release。

Semaphore: カウンター付きロック。特定数のスレッドまで並行実行許可。リソースプール管理（DB コネクション等）。

Binary Semaphore = Mutex（ほぼ同義）。Counting Semaphore は複数リソース管理。

Condition Variable

スレッド間の同期信号。一方がイベント待機（wait）、他方が通知（signal/notify）。

例: Producer-Consumer パターン。Buffer が空の場合 Consumer は wait。Producer がデータ追加時に notify。

Atomic 操作と Memory Barrier

Atomic 変数: Compare-And-Swap（CAS）で lock-free 同期。複数スレッドでメモリ競合なし。

Memory Barrier: CPU キャッシュ一貫性を強制。volatile キーワードで barrier 付与（Java）。

Actor Model

メッセージパッシングで スレッド間通信。共有メモリではなく actor メールボックス。Akka（Java/Scala）、Erlang で採用。

スケーラビリティ: 数百万 actor をスケーリング可能（スレッドより軽量）。

並行データ構造

Thread-Safe Queue: java.util.concurrent.ConcurrentLinkedQueue。Lock-free 実装で高性能。

Read-Write Lock: 読者複数、書者排他。読み込みが大多数の場合効率的。

Copy-On-Write Array: 変更時にコピー。読み込み多数で効率的（garbage overhead あり）。

ers_API/Using_web_workers)

書籍

• [Rust async

スレッドプール と Task Queue

Fixed Thread Pool: スレッド数を制限。コンテキストスイッチ削減。例：ExecutorService.newFixedThreadPool(n)。

Work Stealing: 空いたスレッドが他のスレッドのキューから仕事を取る。負荷分散。

Deadlock と Prevention

Deadlock の 4 条件: Mutual exclusion、Hold and wait、No preemption、Circular wait。いずれか 1 つ除去で deadlock 回避。

Prevention: Lock 取得順序の統一。Timeout 付き lock acquisition。

Detection: Wait-for graph で循環検出。Victim selection で recovery。

Lock-free Programming

Compare-and-swap（CAS）による原子的更新。Spin lock vs Blocking lock。

Hazard pointer: メモリ解放のタイミング管理。false sharing 回避。

Java Memory Model と Volatile

Visibility: volatile 変数の変更はすべてのスレッドに即座に見える。

Happens-before relationship: 操作の順序を保証。Lock acquisition → release で同期。

高度な同期メカニズム

Barrier

複数スレッドが特定地点に到達するまで待機。全員到達後に進行。

例：Parallel loop の各iteration が Barrier で同期。

Read-Write Lock（RWLock）

読者複数、書者排他。読み込みが大多数の場合効率的。

Java: java.util.concurrent.locks.ReadWriteLock

Rust: Arc<RwLock>

Semaphore（セマフォ）

Resource pool 管理。接続数制限（Database connection pool）。

初期値 N で N 個 acquire 可能。全て acquire されるとブロック。

Atomic Operations

Lock-free programming で高性能。Compare-And-Swap（CAS）。

Java: java.util.concurrent.atomic.AtomicInteger など。

C++: std::atomic

Coroutines（コルーチン）

Light-weight 並行。スレッドより少ないコンテキストスイッチ。

Python asyncio、Kotlin suspend functions、C++20 co_await。

book](https://www.amazon.com/s?k=Rust+async+book)

関連技術とエコシステム

ここで紹介した各技術には、活発なコミュニティ・エコシステムが存在。 継続的な学習とアップデートを推奨。実装言語・フレームワークの選択は プロジェクト要件に基づいて判断。性能、保守性、開発速度のバランスが重要。

まとめ

並行プログラミングでは、計算そのものよりも共有状態の扱いが難所になります。まずは共有を減らし、必要な同期を明示するところから考えるのが安全です。

参考文献

公式・標準

• Python asyncio
• Java Concurrency tutorial
• Go Concurrency Patterns
• MDN: Atomics
• [MDN: Using Web Workers](https://developer.mozilla.org/en-US/docs/Web/API/Web_Work