並行性と並列性完全攻略：マルチスレッド、非同期、イベントループのすべて

"並行性(Concurrency)は独立して実行されるタスクを扱うこと、
 並列性(Parallelism)は複数のタスクを同時に実行すること。
 並行性は構造(structure)に関すること、
 並列性は実行(execution)に関すること。"
                                          — Rob Pike

並行性 (Concurrency):            並列性 (Parallelism):
一人が二つの仕事を交互に         二人がそれぞれ一つの仕事を同時に

   Task A ━━━━━━╸               Task A ━━━━━━━━━━━━
                 ╻               Task B ━━━━━━━━━━━━
   Task B ━━━━━━╸
                 ╻               CPU Core 1 ━━━━━━━━
   Task A ━━━━━━╸               CPU Core 2 ━━━━━━━━
                 ╻
   CPU Core 1 ━━━━━━━━
   (タイムスライシング)          (物理的に同時実行)

ユーザースレッド: T1    T2    T3    T4
                 │      │      │      │
カーネルスレッド: KT1   KT2   KT3   KT4
                 │      │      │      │
CPUコア:        Core1  Core2  Core1  Core2

ユーザースレッド: T1  T2  T3  T4  T5  T6
                 └──┬──┘  └──┬──┘  └──┬──┘
                    │        │        │
カーネルスレッド:   KT1      KT1      KT1
                    │
CPUコア:           Core1

ユーザースレッド: G1  G2  G3  G4  G5  G6  G7  G8
                 └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘
                    │       │       │       │
カーネルスレッド:   KT1     KT2     KT3     KT4
                    │       │       │       │
CPUコア:           Core1   Core2   Core3   Core4

┌─────────────────┬──────────────┬──────────────┬──────────────┐
│ 項目            │ 1:1          │ N:1          │ M:N          │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│ スレッドコスト  │ ~1MBスタック │ ~KBスタック  │ ~KBスタック  │
│ 生成速度        │ 遅い(~ms)    │ 速い(~us)    │ 速い(~us)    │
│ コンテキスト    │ 高い(カーネル)│ 低い(ユーザー)│ 低い(ユーザー)│
│ マルチコア      │ O            │ X            │ O            │
│ I/Oブロッキング │ スレッドのみ │ 全体ブロック │ スレッドのみ │
│ 最大スレッド数  │ ~数千        │ ~数百万      │ ~数百万      │
│ 例              │ Java, C++    │ 初期Ruby     │ Go, Erlang   │
└─────────────────┴──────────────┴──────────────┴──────────────┘

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    results := make(chan string, 10)

    urls := []string{
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    }

    for _, url := range urls {
        wg.Add(1)
        go func(u string) {
            defer wg.Done()
            results <- fmt.Sprintf("Fetched: %s", u)
        }(url)
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    for result := range results {
        fmt.Println(result)
    }
}

import java.util.concurrent.*;
import java.util.List;

public class VirtualThreadExample {
    public static void main(String[] args) throws Exception {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = List.of(
                executor.submit(() -> fetchURL("https://api.example.com/users")),
                executor.submit(() -> fetchURL("https://api.example.com/orders")),
                executor.submit(() -> fetchURL("https://api.example.com/products"))
            );

            for (var future : futures) {
                System.out.println(future.get());
            }
        }

        // Structured Concurrency (Preview)
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            var user = scope.fork(() -> fetchUser(1));
            var order = scope.fork(() -> fetchOrder(1));

            scope.join();
            scope.throwIfFailed();

            System.out.println(user.get() + ", " + order.get());
        }
    }
}

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

        for url, result in zip(urls, results):
            print(f"Fetched {url}: {len(result)} bytes")

asyncio.run(main())

use tokio;
use reqwest;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let urls = vec![
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    ];

    let mut handles = vec![];

    for url in urls {
        let handle = tokio::spawn(async move {
            let resp = reqwest::get(url).await?;
            let body = resp.text().await?;
            Ok::<String, reqwest::Error>(body)
        });
        handles.push(handle);
    }

    for handle in handles {
        match handle.await? {
            Ok(body) => println!("Fetched: {} bytes", body.len()),
            Err(e) => eprintln!("Error: {}", e),
        }
    }

    Ok(())
}

async function fetchAll() {
    const urls = [
        'https://api.example.com/users',
        'https://api.example.com/orders',
        'https://api.example.com/products',
    ];

    // Promise.allで同時実行
    const results = await Promise.all(
        urls.map(url => fetch(url).then(r => r.json()))
    );

    // Promise.allSettledで失敗許容
    const settled = await Promise.allSettled(
        urls.map(url => fetch(url).then(r => r.json()))
    );

    settled.forEach((result, i) => {
        if (result.status === 'fulfilled') {
            console.log(`Success: ${urls[i]}`);
        } else {
            console.log(`Failed: ${urls[i]} - ${result.reason}`);
        }
    });
}

%% Erlangプロセス（軽量アクター）
-module(counter).
-export([start/0, increment/1, get/1]).

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

loop(Count) ->
    receive
        {increment, From} ->
            From ! {ok, Count + 1},
            loop(Count + 1);
        {get, From} ->
            From ! {value, Count},
            loop(Count);
        stop ->
            ok
    end.

increment(Pid) ->
    Pid ! {increment, self()},
    receive {ok, NewCount} -> NewCount end.

get(Pid) ->
    Pid ! {get, self()},
    receive {value, Count} -> Count end.

┌─────────────┬─────────────┬───────────────┬──────────────┬──────────────┐
│ 言語        │ モデル      │ 単位          │ スケジュール │ 特徴         │
├─────────────┼─────────────┼───────────────┼──────────────┼──────────────┤
│ Go          │ CSP (M:N)   │ ゴルーチン    │ ランタイム   │ チャネル通信 │
│ Java 21+    │ M:N         │ Virtual Thread│ JVM          │ 後方互換     │
│ Python      │ イベントルー│ コルーチン    │ asyncio      │ GIL制約      │
│ Rust        │ async/await │ Future/Task   │ tokio等      │ ゼロコスト   │
│ JavaScript  │ イベントルー│ Promise       │ libuv/V8     │ シングル     │
│ Erlang      │ Actor(M:N)  │ プロセス      │ BEAM VM      │ 耐障害性     │
│ Kotlin      │ コルーチン  │ Coroutine     │ Dispatcher   │ 構造的同時性 │
└─────────────┴─────────────┴───────────────┴──────────────┴──────────────┘

┌───────────────────────────┐
│         timers            │  ← setTimeout, setInterval
│    (タイマーCB実行)       │
├───────────────────────────┤
│     pending callbacks     │  ← システムエラーCB
├───────────────────────────┤
│       idle, prepare       │  ← 内部使用
├───────────────────────────┤
│          poll             │  ← I/O CB（ほとんどのCB）
│    (新しいI/Oイベント待機)│
├───────────────────────────┤
│          check            │  ← setImmediate CB
├───────────────────────────┤
│     close callbacks       │  ← socket.on('close')
└───────────────────────────┘
   各フェーズ間: マイクロタスクキュー処理
   (Promise.then, queueMicrotask, process.nextTick)

console.log('1: sync');

setTimeout(() => console.log('2: macro (setTimeout)'), 0);

Promise.resolve().then(() => console.log('3: micro (Promise)'));

queueMicrotask(() => console.log('4: micro (queueMicrotask)'));

console.log('5: sync');

// 出力順序:
// 1: sync
// 5: sync
// 3: micro (Promise)
// 4: micro (queueMicrotask)
// 2: macro (setTimeout)

┌──────────────────────────────────────────┐
│              Node.js                     │
│  ┌─────────────────────────────────┐     │
│  │     JavaScript (V8 Engine)      │     │
│  │  ┌───────────┐  ┌────────────┐  │     │
│  │  │  Async    │  │ Event      │  │     │
│  │  │  APIs     │  │ Queue      │  │     │
│  │  └─────┬─────┘  └──────┬─────┘  │     │
│  └────────┼────────────────┼────────┘     │
│           │   Event Loop   │              │
│  ┌────────▼────────────────▼────────┐     │
│  │           libuv                   │     │
│  │  ┌──────────┐  ┌──────────────┐  │     │
│  │  │ Thread   │  │ epoll/kqueue │  │     │
│  │  │ Pool     │  │ /IOCP        │  │     │
│  │  │ (4 thds) │  │ (non-block)  │  │     │
│  │  └──────────┘  └──────────────┘  │     │
│  └───────────────────────────────────┘     │
│           │                │               │
│      File I/O          Network I/O         │
│      DNS lookup        TCP/UDP             │
└──────────────────────────────────────────┘

import asyncio

async def process_item(item: str) -> str:
    await asyncio.sleep(1)
    return f"Processed: {item}"

async def main():
    items = ["a", "b", "c", "d", "e"]

    # 同時実行
    tasks = [process_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    print(results)  # 1秒で全完了

    # タイムアウト
    try:
        result = await asyncio.wait_for(
            process_item("slow"),
            timeout=0.5
        )
    except asyncio.TimeoutError:
        print("Timeout!")

    # セマフォで同時実行数制限
    semaphore = asyncio.Semaphore(3)

    async def limited_process(item):
        async with semaphore:
            return await process_item(item)

    results = await asyncio.gather(*[limited_process(i) for i in items])

asyncio.run(main())

import kotlinx.coroutines.*

suspend fun fetchUser(id: Int): User {
    delay(1000)
    return User(id, "User $id")
}

fun main() = runBlocking {
    // 構造的同時性: 親キャンセルは子に伝播
    coroutineScope {
        val user = async { fetchUser(1) }
        val orders = async { fetchOrders(1) }

        println("User: ${user.await()}")
        println("Orders: ${orders.await()}")
    } // 全子コルーチン完了まで待機

    // SupervisorScope: 子の失敗が他の子に影響しない
    supervisorScope {
        val job1 = launch {
            delay(100)
            throw RuntimeException("Failed!")
        }
        val job2 = launch {
            delay(200)
            println("Job2 completed") // 正常実行
        }
    }
}

従来の並行性:                   構造的同時性:
┌─────────────┐                ┌─────────────┐
│    Parent    │                │    Parent    │
│  ┌───┐ ┌───┐│                │  ┌───┐ ┌───┐│
│  │ A │ │ B ││                │  │ A │ │ B ││
│  └─┬─┘ └─┬─┘│                │  └─┬─┘ └─┬─┘│
└────┼─────┼───┘                └────┼─────┼───┘
     │     │ ← 誰が管理?            │     │ ← Parentが管理
     ▼     ▼   リーク可能!          ▼     ▼   自動クリーンアップ!

ルール:
1. すべての並行作業はスコープ内で開始
2. スコープは全子タスク完了まで待機
3. 親キャンセル時に全子もキャンセル
4. 子エラーは親に伝播

┌──────────┐    メッセージ    ┌──────────┐
│  Actor A │ ───────────────→ │  Actor B │
│ State: x │    メッセージ    │ State: y │
│ Mailbox  │ ←─────────────── │ Mailbox  │
└──────────┘                  └──────────┘
     │
     │ メッセージ
     ▼
┌──────────┐
│  Actor C │
│ State: z │
│ Mailbox  │
└──────────┘

核心: 共有状態なし、ロックなし、メッセージ受け渡しのみ

                  ┌──────────────┐
                  │  Supervisor  │
                  │ (one_for_one)│
                  └──────┬───────┘
                    ┌────┼────┐
                    ▼    ▼    ▼
              ┌─────┐ ┌─────┐ ┌─────┐
              │ W1  │ │ W2  │ │ W3  │
              └─────┘ └─────┘ └─────┘

W2がクラッシュした場合:
1. Supervisorが検知
2. W2のみ再起動（one_for_one戦略）
3. W1、W3は影響なし
4. 5分以内に5回以上再起動ならSupervisorも終了

適している場合:
- 数百万の独立した状態管理（チャットルーム、IoTデバイス）
- 分散システム（ノード間通信）
- 耐障害性が必須のシステム（通信、金融）
- リアルタイムシステム（ゲームサーバー、トレーディング）

適していない場合:
- シンプルなCRUDアプリケーション
- 強い整合性(strong consistency)が必要
- トランザクション処理が核心
- 順序保証が重要なパイプライン

type SafeMap struct {
    mu   sync.Mutex
    data map[string]int
}

func (m *SafeMap) Set(key string, val int) {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.data[key] = val
}

func (m *SafeMap) Get(key string) (int, bool) {
    m.mu.Lock()
    defer m.mu.Unlock()
    val, ok := m.data[key]
    return val, ok
}

type Cache struct {
    mu   sync.RWMutex
    data map[string]string
}

// 複数のゴルーチンが同時に読み取り可能
func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    val, ok := c.data[key]
    return val, ok
}

// 書き込みは排他ロック
func (c *Cache) Set(key, val string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = val
}

import asyncio

async def worker(name, semaphore):
    async with semaphore:
        print(f"{name} acquired semaphore")
        await asyncio.sleep(1)
        print(f"{name} released semaphore")

async def main():
    semaphore = asyncio.Semaphore(3)  # 最大3つ同時実行
    tasks = [worker(f"Worker-{i}", semaphore) for i in range(10)]
    await asyncio.gather(*tasks)

asyncio.run(main())

type BoundedQueue struct {
    mu       sync.Mutex
    notEmpty *sync.Cond
    notFull  *sync.Cond
    items    []interface{}
    maxSize  int
}

func NewBoundedQueue(maxSize int) *BoundedQueue {
    q := &BoundedQueue{maxSize: maxSize}
    q.notEmpty = sync.NewCond(&q.mu)
    q.notFull = sync.NewCond(&q.mu)
    return q
}

func (q *BoundedQueue) Put(item interface{}) {
    q.mu.Lock()
    defer q.mu.Unlock()

    for len(q.items) == q.maxSize {
        q.notFull.Wait() // 空きが出るまで待機
    }
    q.items = append(q.items, item)
    q.notEmpty.Signal() // コンシューマーを起こす
}

func (q *BoundedQueue) Take() interface{} {
    q.mu.Lock()
    defer q.mu.Unlock()

    for len(q.items) == 0 {
        q.notEmpty.Wait() // アイテムが来るまで待機
    }
    item := q.items[0]
    q.items = q.items[1:]
    q.notFull.Signal() // プロデューサーを起こす
    return item
}

import "sync/atomic"

type LockFreeCounter struct {
    value int64
}

func (c *LockFreeCounter) Increment() int64 {
    for {
        old := atomic.LoadInt64(&c.value)
        new := old + 1
        if atomic.CompareAndSwapInt64(&c.value, old, new) {
            return new
        }
        // CAS失敗、リトライ（他のゴルーチンが先に変更）
    }
}

func (c *LockFreeCounter) Get() int64 {
    return atomic.LoadInt64(&c.value)
}

スレッドA: 値読み取り → A
                          スレッドB: A → B 変更
                          スレッドB: B → A 変更（元の値に復元）
スレッドA: CAS(A → C) 成功!（しかし途中で値が変わっていた）

解決策:
- バージョン番号追加 (A,v1 → B,v2 → A,v3)
- タグ付きポインタ (JavaのAtomicStampedReference)
- Hazard Pointer

Lock-Freeが適している場合:
- 非常に高い競合(contention)状況
- リアルタイムシステム（優先度逆転防止）
- 単純なデータ構造（カウンター、スタック、キュー）

Lock-Freeを避けるべき場合（ほとんどの場合）:
- 一般的なアプリケーション（Mutexで十分）
- 複雑なデータ構造（ツリー、グラフ）
- 正確性検証が困難な状況
- チームがLock-Freeに慣れていない場合

"Mutexで動くならMutexを使え。"

// バグ: 複数のゴルーチンが同時にcounterを変更
var counter int

func buggyIncrement() {
    for i := 0; i < 1000; i++ {
        go func() {
            counter++ // 読み取り-変更-書き込みがアトミックでない
        }()
    }
}

// 修正: atomicまたはmutexを使用
var safeCounter int64

func safeIncrement() {
    for i := 0; i < 1000; i++ {
        go func() {
            atomic.AddInt64(&safeCounter, 1)
        }()
    }
}

4つの必要条件（すべて満たすとデッドロック）:
1. 相互排除: リソースを一度に1つだけ使用
2. 保持と待機: リソースを保持しながら別のリソースを待機
3. 非横取り: 保持されたリソースを強制的に奪えない
4. 循環待機: A→B→C→A 循環待機

// デッドロック例
var mu1, mu2 sync.Mutex

func goroutine1() {
    mu1.Lock()
    defer mu1.Unlock()
    time.Sleep(time.Millisecond)
    mu2.Lock()       // mu2待機（goroutine2が保持）
    defer mu2.Unlock()
}

func goroutine2() {
    mu2.Lock()
    defer mu2.Unlock()
    time.Sleep(time.Millisecond)
    mu1.Lock()       // mu1待機（goroutine1が保持）
    defer mu1.Unlock()
}

// 修正: 常に同じ順序でロック取得
func fixed1() {
    mu1.Lock()
    defer mu1.Unlock()
    mu2.Lock()
    defer mu2.Unlock()
}

func fixed2() {
    mu1.Lock()       // mu1を先に（同じ順序）
    defer mu1.Unlock()
    mu2.Lock()
    defer mu2.Unlock()
}

AとBが狭い廊下で出会う:
A: 左に避ける ← → B: 右に避ける（同じ方向!）
A: 右に避ける ← → B: 左に避ける（また同じ方向!）
...（無限ループ、誰も進めない）

vs デッドロック: 何もしない
vs ライブロック: 動き続けるが進展なし

1997年火星探査機Mars Pathfinder事件:

Low-priorityタスクがミューテックスを保持
   |
Medium-priorityタスクがLowを横取り（ミューテックスと無関係）
   |
High-priorityタスクがミューテックス待ち... 永遠に!
（Lowが実行されないとミューテックスが解放されないが、Mediumが常に横取り）

解決策: 優先度継承プロトコル
Lowがミューテックス保持中はHighの優先度を一時的に継承

# Go: レースディテクター
go test -race ./...
go run -race main.go

# Java: Thread Sanitizer
java -XX:+UseThreadSanitizer MyApp

# Rust: コンパイラがコンパイル時に検出!
# Send/Syncトレイト違反 = コンパイルエラー

# Python: asyncioデバッグモード
PYTHONASYNCIODEBUG=1 python app.py

# 一般: Helgrind (Valgrindツール)
valgrind --tool=helgrind ./program

func producerConsumer() {
    ch := make(chan int, 100)

    var prodWg sync.WaitGroup
    for i := 0; i < 3; i++ {
        prodWg.Add(1)
        go func(id int) {
            defer prodWg.Done()
            for j := 0; j < 10; j++ {
                ch <- id*100 + j
            }
        }(i)
    }

    go func() {
        prodWg.Wait()
        close(ch)
    }()

    var consWg sync.WaitGroup
    for i := 0; i < 2; i++ {
        consWg.Add(1)
        go func(id int) {
            defer consWg.Done()
            for item := range ch {
                fmt.Printf("Consumer %d: %d\n", id, item)
            }
        }(i)
    }

    consWg.Wait()
}

従来のスレッドプール:          ワークスティーリング:
┌─────────────┐              ┌─────────────┐
│ 共有キュー   │              │ Worker 1    │
│ [T1|T2|T3]  │              │ ローカル:[T1]│
│             │              ├─────────────┤
│ 全ワーカーが │              │ Worker 2    │
│ 競合!       │              │ ローカル:[T2]│ ← 空ならW1から窃取
│             │              ├─────────────┤
└─────────────┘              │ Worker 3    │
                             │ ローカル:[T3]│
                             └─────────────┘

利点: キャッシュ局所性向上、競合削減
使用: Java ForkJoinPool、Goランタイム、Tokio(Rust)

高速化 = 1 / ((1 - P) + P/N)

P = 並列化可能な割合
N = プロセッサ数

例: 95%並列化可能、64コアの場合
高速化 = 1 / (0.05 + 0.95/64) = 1 / 0.0648 = 15.4倍

核心: コアをいくら増やしても直列部分(5%)がボトルネック!
最大理論的高速化 = 1/0.05 = 20倍

高速化 = N - alpha * (N - 1)

alpha = 直列部分の割合
N = プロセッサ数

核心的な違い:
- アムダール: 「問題サイズ固定、コアを増やすと?」
- グスタフソン: 「コア固定、問題サイズを増やすと?」

現実ではデータが増えると並列部分も増える
→ グスタフソンがより楽観的で現実的

コンテキストスイッチコスト比較:
┌───────────────────────────┬──────────────┐
│ スイッチ種類              │ おおよそ     │
├───────────────────────────┼──────────────┤
│ 関数呼び出し              │ ~1-5 ns      │
│ ゴルーチンスイッチ        │ ~100-200 ns  │
│ コルーチンスイッチ(Python)│ ~1-5 us      │
│ スレッドスイッチ(同プロセス)│ ~1-10 us   │
│ プロセススイッチ          │ ~10-50 us    │
│ 仮想マシンスイッチ        │ ~100+ us     │
└───────────────────────────┴──────────────┘

問題: 異なる変数が同じキャッシュラインに位置

CPU Core 1          CPU Core 2
    |                   |
+---v-------------------v---+
| Cache Line (64 bytes)     |
| [counter_A] [counter_B]   |  ← 異なる変数だが
|                           |     同じキャッシュライン!
+---------------------------+

Core 1がcounter_Aを変更 → キャッシュライン無効化 → Core 2もリロード必要
Core 2がcounter_Bを変更 → キャッシュライン無効化 → Core 1もリロード必要

解決: パディングでキャッシュラインを分離

type PaddedCounter struct {
    value int64
    _pad  [56]byte // 64バイトキャッシュラインパディング
}

type Counters struct {
    c1 PaddedCounter
    c2 PaddedCounter
}