// ================================================================
// Rust: rayon によるデータ並列
// ================================================================
 
use rayon::prelude::*;
use std::collections::HashMap;
 
// --- 基本: 逐次 → 並列への変換 ---
 
// 逐次バージョン
fn sequential_sum(data: &[f64]) -> f64 {
    data.iter()
        .map(|x| x.powi(2))
        .sum()
}
 
// 並列バージョン（iter → par_iter に変更するだけ）
fn parallel_sum(data: &[f64]) -> f64 {
    data.par_iter()
        .map(|x| x.powi(2))
        .sum()
}
 
// --- 並列フィルタリングと集約 ---
 
#[derive(Debug, Clone)]
struct LogEntry {
    level: String,
    message: String,
    timestamp: u64,
}
 
fn analyze_logs(logs: &[LogEntry]) -> HashMap<String, usize> {
    // 並列でログレベルごとの件数を集計
    logs.par_iter()
        .fold(
            || HashMap::new(),  // 各スレッドのローカルmap
            |mut acc, entry| {
                *acc.entry(entry.level.clone()).or_insert(0) += 1;
                acc
            },
        )
        .reduce(
            || HashMap::new(),  // 初期値
            |mut a, b| {        // 2つのmapをマージ
                for (key, value) in b {
                    *a.entry(key).or_insert(0) += value;
                }
                a
            },
        )
}
 
// --- 並列ソートとチャンク処理 ---
 
fn parallel_sort_and_process(mut data: Vec<i64>) -> Vec<i64> {
    // 並列ソート（不安定ソート:高速だが同値の順序不定）
    data.par_sort_unstable();
 
    // 並列チャンク処理
    data.par_chunks(1024)
        .flat_map(|chunk| {
            chunk.iter()
                .filter(|&&x| x > 0)
                .map(|&x| x * 2)
                .collect::<Vec<_>>()
        })
        .collect()
}
 
// --- スレッドプール制御 ---
 
fn configure_thread_pool() {
    // カスタムスレッドプール（デフォルトは論理コア数）
    rayon::ThreadPoolBuilder::new()
        .num_threads(4)           // スレッド数を4に制限
        .stack_size(8 * 1024 * 1024)  // スタックサイズ8MB
        .build_global()
        .unwrap();
}
 
// --- 実行例 ---
 
fn main() {
    let data: Vec<f64> = (0..10_000_000)
        .map(|x| x as f64)
        .collect();
 
    let result = parallel_sum(&data);
    println!("並列合計: {}", result);
 
    // 並列イテレータの連鎖
    let processed: Vec<String> = (0..1_000_000_u64)
        .into_par_iter()
        .filter(|&x| x % 3 == 0)
        .map(|x| format!("item_{}", x))
        .collect();
    println!("処理件数: {}", processed.len());
}

# ================================================================
# Python: multiprocessing によるデータ並列
# ================================================================
 
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, as_completed
import numpy as np
from typing import List, Tuple
import time
 
# --- 基本: Pool.map ---
 
def square(x: float) -> float:
    """CPU密集型処理の例"""
    return sum(i ** 2 for i in range(int(x)))
 
def basic_parallel():
    """基本的な並列処理"""
    data = list(range(100, 200))
 
    # 逐次実行
    results_seq = [square(x) for x in data]
 
    # 並列実行
    with mp.Pool(processes=mp.cpu_count()) as pool:
        results_par = pool.map(square, data)
 
    assert results_seq == results_par
 
# --- チャンク処理でオーバーヘッド削減 ---
 
def process_chunk(chunk: List[float]) -> List[float]:
    """チャンク単位で処理（プロセス間通信を削減）"""
    return [x ** 2 + 2 * x + 1 for x in chunk]
 
def chunked_parallel():
    """チャンク分割による効率的な並列処理"""
    data = list(range(1_000_000))
    n_workers = mp.cpu_count()
 
    # データを均等にチャンク分割
    chunk_size = len(data) // n_workers
    chunks = [
        data[i:i + chunk_size]
        for i in range(0, len(data), chunk_size)
    ]
 
    with mp.Pool(processes=n_workers) as pool:
        results = pool.map(process_chunk, chunks)
 
    # 結果をフラット化
    flat_results = [item for chunk in results for item in chunk]
    print(f"処理件数: {len(flat_results)}")
 
# --- ProcessPoolExecutor と進捗追跡 ---
 
def heavy_computation(args: Tuple[int, List[float]]) -> dict:
    """IDとデータを受け取り、結果を辞書で返す"""
    task_id, data = args
    result = sum(x ** 2 for x in data)
    return {"task_id": task_id, "result": result}
 
def parallel_with_progress():
    """進捗追跡付き並列処理"""
    tasks = [
        (i, list(range(i * 1000, (i + 1) * 1000)))
        for i in range(50)
    ]
 
    completed = 0
    total = len(tasks)
    results = []
 
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {
            executor.submit(heavy_computation, task): task[0]
            for task in tasks
        }
 
        for future in as_completed(futures):
            task_id = futures[future]
            try:
                result = future.result(timeout=30)
                results.append(result)
                completed += 1
                print(f"進捗: {completed}/{total} "
                      f"(Task {task_id} 完了)")
            except Exception as e:
                print(f"Task {task_id} 失敗: {e}")
 
    return sorted(results, key=lambda r: r["task_id"])
 
# --- NumPy による暗黙的な並列化 ---
 
def numpy_parallel():
    """NumPyのベクトル化演算（内部でSIMD/マルチスレッド）"""
    # NumPyはBLAS/LAPACKを通じて暗黙的に並列化される
    a = np.random.rand(10000, 10000)
    b = np.random.rand(10000, 10000)
 
    # 行列積（内部で並列実行）
    c = np.dot(a, b)
 
    # 要素ごとの演算（SIMD最適化）
    d = np.sin(a) + np.cos(b) * np.exp(-a)
 
    return c, d
 
if __name__ == "__main__":
    basic_parallel()
    chunked_parallel()
    results = parallel_with_progress()
    print(f"全タスク完了: {len(results)} 件")

// ================================================================
// Java: Parallel Streams と ForkJoinPool
// ================================================================
 
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;
 
public class DataParallelExample {
 
    // --- Parallel Streams ---
    public static long parallelSum(List<Long> numbers) {
        return numbers.parallelStream()
            .filter(n -> n > 0)
            .mapToLong(n -> n * n)
            .sum();
    }
 
    // --- カスタムForkJoinPool（並列度制御） ---
    public static <T> List<T> parallelWithCustomPool(
            List<T> data,
            java.util.function.Function<T, T> transform,
            int parallelism) throws Exception {
 
        ForkJoinPool pool = new ForkJoinPool(parallelism);
        try {
            return pool.submit(() ->
                data.parallelStream()
                    .map(transform)
                    .collect(Collectors.toList())
            ).get();
        } finally {
            pool.shutdown();
        }
    }
 
    // --- ForkJoinTask の直接利用 ---
    static class ParallelMergeSort extends RecursiveAction {
        private final int[] array;
        private final int lo, hi;
        private static final int THRESHOLD = 1024;
 
        ParallelMergeSort(int[] array, int lo, int hi) {
            this.array = array;
            this.lo = lo;
            this.hi = hi;
        }
 
        @Override
        protected void compute() {
            if (hi - lo < THRESHOLD) {
                // 小さいサイズは逐次ソート
                Arrays.sort(array, lo, hi);
                return;
            }
            int mid = lo + (hi - lo) / 2;
            // 左右を並列にソート
            invokeAll(
                new ParallelMergeSort(array, lo, mid),
                new ParallelMergeSort(array, mid, hi)
            );
            merge(array, lo, mid, hi);
        }
 
        private void merge(int[] arr, int lo, int mid, int hi) {
            int[] temp = Arrays.copyOfRange(arr, lo, mid);
            int i = 0, j = mid, k = lo;
            while (i < temp.length && j < hi) {
                arr[k++] = (temp[i] <= arr[j]) ? temp[i++] : arr[j++];
            }
            while (i < temp.length) arr[k++] = temp[i++];
        }
    }
 
    public static void main(String[] args) throws Exception {
        // Parallel Stream の例
        List<Long> numbers = LongStream.rangeClosed(1, 10_000_000)
            .boxed()
            .collect(Collectors.toList());
        long sum = parallelSum(numbers);
        System.out.println("Sum: " + sum);
 
        // ForkJoinPool でのソート
        int[] data = new Random().ints(10_000_000).toArray();
        ForkJoinPool pool = ForkJoinPool.commonPool();
        pool.invoke(new ParallelMergeSort(data, 0, data.length));
        System.out.println("Sorted: " + Arrays.toString(
            Arrays.copyOf(data, 10)));
    }
}

// ================================================================
// Go: goroutine + errgroup によるタスク並列
// ================================================================
 
package main
 
import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"
 
    "golang.org/x/sync/errgroup"
)
 
// --- 基本: errgroup による並列タスク ---
 
type Dashboard struct {
    User    User
    Posts   []Post
    Stats   Stats
    Friends []Friend
}
 
func loadDashboard(ctx context.Context, userID int) (*Dashboard, error) {
    g, ctx := errgroup.WithContext(ctx)
    var (
        user    User
        posts   []Post
        stats   Stats
        friends []Friend
    )
 
    // 4つのタスクを並列実行
    g.Go(func() error {
        var err error
        user, err = fetchUser(ctx, userID)
        return err
    })
 
    g.Go(func() error {
        var err error
        posts, err = fetchPosts(ctx, userID)
        return err
    })
 
    g.Go(func() error {
        var err error
        stats, err = fetchStats(ctx, userID)
        return err
    })
 
    g.Go(func() error {
        var err error
        friends, err = fetchFriends(ctx, userID)
        return err
    })
 
    // 全タスクの完了を待機（1つでもエラーなら即キャンセル）
    if err := g.Wait(); err != nil {
        return nil, fmt.Errorf("dashboard load failed: %w", err)
    }
 
    return &Dashboard{user, posts, stats, friends}, nil
}
 
// --- 並列度制限付きタスク実行 ---
 
func processURLs(ctx context.Context, urls []string) ([]Result, error) {
    g, ctx := errgroup.WithContext(ctx)
    // 同時実行数を10に制限
    g.SetLimit(10)
 
    results := make([]Result, len(urls))
    var mu sync.Mutex
 
    for i, url := range urls {
        i, url := i, url  // ループ変数キャプチャ
        g.Go(func() error {
            result, err := fetch(ctx, url)
            if err != nil {
                return err
            }
            mu.Lock()
            results[i] = result
            mu.Unlock()
            return nil
        })
    }
 
    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}
 
// --- タイムアウト付きタスク並列 ---
 
func loadWithTimeout(userID int) (*Dashboard, error) {
    ctx, cancel := context.WithTimeout(
        context.Background(),
        5*time.Second,
    )
    defer cancel()
 
    return loadDashboard(ctx, userID)
}

// ================================================================
// Rust: タスク並列の2つのアプローチ
// ================================================================
 
// --- 非同期タスク並列（I/Oバウンド） ---
use tokio;
 
async fn load_dashboard(user_id: u64) -> Result<Dashboard, Error> {
    // tokio::join! で複数の非同期タスクを並列実行
    let (user, posts, stats) = tokio::join!(
        fetch_user(user_id),
        fetch_posts(user_id),
        fetch_stats(user_id),
    );
 
    Ok(Dashboard {
        user: user?,
        posts: posts?,
        stats: stats?,
    })
}
 
// try_join! でエラー時に早期終了
async fn load_dashboard_with_error_handling(
    user_id: u64,
) -> Result<Dashboard, Error> {
    let (user, posts, stats) = tokio::try_join!(
        fetch_user(user_id),
        fetch_posts(user_id),
        fetch_stats(user_id),
    )?;
 
    Ok(Dashboard { user, posts, stats })
}
 
// --- CPUバウンドなタスク並列（rayon::join） ---
use rayon;
 
fn parallel_analysis(data: &[f64]) -> (f64, f64, Vec<f64>) {
    let (mean_result, (variance_result, sorted)) = rayon::join(
        // 左タスク: 平均計算
        || {
            let sum: f64 = data.iter().sum();
            sum / data.len() as f64
        },
        // 右タスク: さらに分割
        || rayon::join(
            // 分散計算
            || {
                let mean = data.iter().sum::<f64>() / data.len() as f64;
                data.iter()
                    .map(|x| (x - mean).powi(2))
                    .sum::<f64>() / data.len() as f64
            },
            // ソート
            || {
                let mut sorted = data.to_vec();
                sorted.par_sort_unstable_by(|a, b| {
                    a.partial_cmp(b).unwrap()
                });
                sorted
            },
        ),
    );
 
    (mean_result, variance_result, sorted)
}

// ================================================================
// C++17/20: 並列アルゴリズムと std::async
// ================================================================
 
#include <algorithm>
#include <execution>  // C++17 並列実行ポリシー
#include <future>
#include <numeric>
#include <vector>
 
// --- C++17 並列アルゴリズム ---
 
void parallel_algorithms_example() {
    std::vector<int> data(10'000'000);
    std::iota(data.begin(), data.end(), 0);
 
    // 並列ソート
    std::sort(std::execution::par,
              data.begin(), data.end());
 
    // 並列変換
    std::transform(std::execution::par_unseq,
                   data.begin(), data.end(),
                   data.begin(),
                    { return x * 2; });
 
    // 並列リデュース
    long long sum = std::reduce(
        std::execution::par,
        data.begin(), data.end(),
        0LL);
 
    // 並列 for_each
    std::for_each(std::execution::par,
                  data.begin(), data.end(),
                   { x += 1; });
}
 
// --- std::async によるタスク並列 ---
 
struct AnalysisResult {
    double mean;
    double variance;
    int max_val;
};
 
AnalysisResult parallel_analysis(const std::vector<int>& data) {
    // 3つのタスクを非同期実行
    auto mean_future = std::async(std::launch::async, [&]() {
        double sum = std::reduce(
            std::execution::par,
            data.begin(), data.end(), 0.0);
        return sum / data.size();
    });
 
    auto var_future = std::async(std::launch::async, [&]() {
        double mean = std::reduce(
            std::execution::par,
            data.begin(), data.end(), 0.0) / data.size();
        double sq_sum = std::transform_reduce(
            std::execution::par,
            data.begin(), data.end(),
            0.0,
            std::plus<>{},
            mean {
                return (x - mean) * (x - mean);
            });
        return sq_sum / data.size();
    });
 
    auto max_future = std::async(std::launch::async, [&]() {
        return *std::max_element(
            std::execution::par,
            data.begin(), data.end());
    });
 
    return {
        mean_future.get(),
        var_future.get(),
        max_future.get()
    };
}
 
// --- 実行ポリシーの比較 ---
//
// std::execution::seq       逐次実行（デフォルト）
// std::execution::par       並列実行（スレッド使用）
// std::execution::par_unseq 並列+ベクトル化（SIMD併用）
// std::execution::unseq     ベクトル化のみ（C++20）

// ================================================================
// Rust: アトミック操作の実践パターン
// ================================================================
 
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
 
// --- パターン1: アトミックカウンタ ---
 
struct AtomicCounter {
    count: AtomicU64,
}
 
impl AtomicCounter {
    fn new() -> Self {
        Self { count: AtomicU64::new(0) }
    }
 
    fn increment(&self) -> u64 {
        // fetch_add は加算前の値を返す
        self.count.fetch_add(1, Ordering::Relaxed)
    }
 
    fn get(&self) -> u64 {
        self.count.load(Ordering::Relaxed)
    }
}
 
fn counter_example() {
    let counter = Arc::new(AtomicCounter::new());
    let mut handles = vec![];
 
    for _ in 0..8 {
        let counter = Arc::clone(&counter);
        handles.push(thread::spawn(move || {
            for _ in 0..100_000 {
                counter.increment();
            }
        }));
    }
 
    for h in handles {
        h.join().unwrap();
    }
    // 常に正確に 800,000 になる
    assert_eq!(counter.get(), 800_000);
}
 
// --- パターン2: スピンロック ---
 
struct SpinLock {
    locked: AtomicBool,
}
 
impl SpinLock {
    fn new() -> Self {
        Self { locked: AtomicBool::new(false) }
    }
 
    fn lock(&self) {
        // compare_exchange で false → true に変更を試みる
        while self.locked.compare_exchange_weak(
            false,              // 期待値
            true,               // 新しい値
            Ordering::Acquire,  // 成功時のOrdering
            Ordering::Relaxed,  // 失敗時のOrdering
        ).is_err() {
            // スピン待ち（ビジーウェイト）
            std::hint::spin_loop();
        }
    }
 
    fn unlock(&self) {
        self.locked.store(false, Ordering::Release);
    }
}
 
// --- パターン3: 一度だけ初期化（lazy init） ---
 
use std::sync::atomic::AtomicPtr;
use std::ptr;
 
struct LazyValue<T> {
    ptr: AtomicPtr<T>,
}
 
impl<T> LazyValue<T> {
    fn new() -> Self {
        Self { ptr: AtomicPtr::new(ptr::null_mut()) }
    }
 
    fn get_or_init(&self, init: impl FnOnce() -> T) -> &T {
        let mut p = self.ptr.load(Ordering::Acquire);
        if p.is_null() {
            let new = Box::into_raw(Box::new(init()));
            match self.ptr.compare_exchange(
                ptr::null_mut(),
                new,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => p = new,
                Err(existing) => {
                    // 他のスレッドが先に初期化した
                    unsafe { drop(Box::from_raw(new)); }
                    p = existing;
                }
            }
        }
        unsafe { &*p }
    }
}

// ================================================================
// Go: sync/atomic の実践パターン
// ================================================================
 
package main
 
import (
    "fmt"
    "sync"
    "sync/atomic"
)
 
// --- アトミックカウンタ ---
 
type AtomicCounter struct {
    count atomic.Int64
}
 
func (c *AtomicCounter) Increment() int64 {
    return c.count.Add(1)
}
 
func (c *AtomicCounter) Get() int64 {
    return c.count.Load()
}
 
// --- アトミック値（設定の動的更新） ---
 
type Config struct {
    MaxWorkers int
    Timeout    int
    Debug      bool
}
 
type ConfigHolder struct {
    config atomic.Value // *Config を格納
}
 
func (h *ConfigHolder) Load() *Config {
    return h.config.Load().(*Config)
}
 
func (h *ConfigHolder) Store(cfg *Config) {
    h.config.Store(cfg)
}
 
// 使用例: 設定のホットリロード
func configExample() {
    holder := &ConfigHolder{}
    holder.Store(&Config{MaxWorkers: 4, Timeout: 30, Debug: false})
 
    // ワーカーは常に最新の設定を参照
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            cfg := holder.Load()
            fmt.Printf("Workers: %d\n", cfg.MaxWorkers)
        }()
    }
 
    // 設定を動的に更新（ロック不要）
    holder.Store(&Config{MaxWorkers: 8, Timeout: 60, Debug: true})
    wg.Wait()
}

// ================================================================
// Rust: MapReduce の実装
// ================================================================
 
use rayon::prelude::*;
use std::collections::HashMap;
 
fn word_count(documents: &[String]) -> HashMap<String, usize> {
    documents
        .par_iter()
        // Map: 各ドキュメントの単語をカウント
        .map(|doc| {
            let mut local_counts = HashMap::new();
            for word in doc.split_whitespace() {
                let word = word.to_lowercase();
                *local_counts.entry(word).or_insert(0) += 1;
            }
            local_counts
        })
        // Reduce: ローカルカウントをマージ
        .reduce(
            || HashMap::new(),
            |mut acc, local| {
                for (word, count) in local {
                    *acc.entry(word).or_insert(0) += count;
                }
                acc
            },
        )
}
 
// 汎用 MapReduce フレームワーク
fn map_reduce<T, K, V, MapFn, ReduceFn>(
    data: &[T],
    map_fn: MapFn,
    reduce_fn: ReduceFn,
) -> HashMap<K, V>
where
    T: Sync,
    K: Eq + std::hash::Hash + Send,
    V: Send,
    MapFn: Fn(&T) -> Vec<(K, V)> + Sync,
    ReduceFn: Fn(V, V) -> V + Sync + Copy,
{
    data.par_iter()
        .flat_map(|item| map_fn(item))
        .fold(
            || HashMap::new(),
            |mut acc, (key, value)| {
                acc.entry(key)
                    .and_modify(|v| *v = reduce_fn(
                        std::mem::replace(v, unsafe {
                            std::mem::zeroed()
                        }),
                        value,
                    ))
                    .or_insert(value);
                acc
            },
        )
        .reduce(
            || HashMap::new(),
            |mut a, b| {
                for (key, value) in b {
                    a.entry(key)
                        .and_modify(|v| *v = reduce_fn(
                            std::mem::replace(v, unsafe {
                                std::mem::zeroed()
                            }),
                            value,
                        ))
                        .or_insert(value);
                }
                a
            },
        )
}

// ================================================================
// Go: Pipeline パターンの実装
// ================================================================
 
package main
 
import (
    "context"
    "fmt"
    "strings"
    "sync"
)
 
// パイプラインのステージを関数として定義
 
// Stage 1: データ生成
func generate(ctx context.Context, items ...string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, item := range items {
            select {
            case out <- item:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}
 
// Stage 2: 変換（大文字化）
func toUpper(ctx context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for s := range in {
            select {
            case out <- strings.ToUpper(s):
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}
 
// Stage 3: フィルタリング
func filterLong(ctx context.Context, in <-chan string,
                minLen int) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for s := range in {
            if len(s) >= minLen {
                select {
                case out <- s:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}
 
// Fan-Out: 1つのチャネルを複数のワーカーで読む
func fanOut(ctx context.Context, in <-chan string,
            n int, process func(string) string) <-chan string {
    out := make(chan string)
    var wg sync.WaitGroup
 
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for item := range in {
                select {
                case out <- process(item):
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
 
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}
 
func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
 
    // パイプライン構築
    stage1 := generate(ctx, "hello", "world", "go", "pipeline")
    stage2 := toUpper(ctx, stage1)
    stage3 := filterLong(ctx, stage2, 3)
 
    // 結果を消費
    for result := range stage3 {
        fmt.Println(result)
    }
    // 出力: HELLO, WORLD, PIPELINE
}

// ================================================================
// Rust: 偽共有の回避
// ================================================================
 
use std::sync::atomic::{AtomicU64, Ordering};
 
// BAD: 偽共有が発生する
struct BadCounters {
    counter_a: AtomicU64,  // 同じキャッシュラインに載る可能性
    counter_b: AtomicU64,
}
 
// GOOD: パディングで分離
#[repr(C)]
struct GoodCounters {
    counter_a: AtomicU64,
    _pad_a: [u8; 56],     // 64 - 8 = 56バイトのパディング
    counter_b: AtomicU64,
    _pad_b: [u8; 56],
}
 
// Rust の crossbeam には CachePadded が用意されている
use crossbeam_utils::CachePadded;
 
struct BestCounters {
    counter_a: CachePadded<AtomicU64>,
    counter_b: CachePadded<AtomicU64>,
}
 
impl BestCounters {
    fn new() -> Self {
        Self {
            counter_a: CachePadded::new(AtomicU64::new(0)),
            counter_b: CachePadded::new(AtomicU64::new(0)),
        }
    }
}

// ================================================================
// Java: @Contended による偽共有回避
// ================================================================
 
import java.util.concurrent.atomic.AtomicLong;
 
// Java 8+: @Contended アノテーション
// JVM起動時に -XX:-RestrictContended オプションが必要
public class PaddedCounters {
    // BAD: 偽共有
    static class BadCounters {
        volatile long counterA;
        volatile long counterB;
    }
 
    // GOOD: @Contended で自動パディング
    static class GoodCounters {
        @jdk.internal.vm.annotation.Contended
        volatile long counterA;
 
        @jdk.internal.vm.annotation.Contended
        volatile long counterB;
    }
 
    // GOOD: 手動パディング（@Contended が使えない場合）
    static class ManualPaddedCounters {
        volatile long counterA;
        long p1, p2, p3, p4, p5, p6, p7;  // 56バイトのパディング
        volatile long counterB;
        long q1, q2, q3, q4, q5, q6, q7;
    }
}

# ================================================================
# ANTI-PATTERN: 小さなタスクの過度な並列化
# ================================================================
 
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
 
# BAD: 1要素ずつプロセスに分配
# プロセス生成コスト >> 計算コスト
def bad_parallel():
    with ProcessPoolExecutor(max_workers=4) as executor:
        # 各要素をプロセス間通信で送受信 → 激遅
        results = list(executor.map(
            lambda x: x * 2,  # 極めて軽い計算
            range(10000)
        ))
    return results
 
# GOOD: チャンク単位で分配
def good_parallel():
    def process_chunk(chunk):
        return [x * 2 for x in chunk]
 
    data = list(range(10000))
    n_workers = mp.cpu_count()
    chunk_size = len(data) // n_workers
    chunks = [data[i:i+chunk_size]
              for i in range(0, len(data), chunk_size)]
 
    with ProcessPoolExecutor(max_workers=n_workers) as executor:
        results = list(executor.map(process_chunk, chunks))
    return [x for chunk in results for x in chunk]
 
# BEST: そもそも並列化しない（この規模なら逐次が最速）
def best_sequential():
    return [x * 2 for x in range(10000)]

// ================================================================
// ANTI-PATTERN: 共有可変状態への無防備なアクセス
// ================================================================
 
package main
 
import (
    "fmt"
    "sync"
    "sync/atomic"
)
 
// BAD: データ競合（go run -race で検出可能）
func badSharedState() {
    counter := 0
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter++ // DATA RACE: 非原子的な read-modify-write
        }()
    }
    wg.Wait()
    // counter は 1000 にならない可能性がある
    fmt.Println("BAD counter:", counter)
}
 
// GOOD: Mutex で保護
func goodWithMutex() {
    counter := 0
    var mu sync.Mutex
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            mu.Lock()
            counter++
            mu.Unlock()
        }()
    }
    wg.Wait()
    fmt.Println("GOOD counter (mutex):", counter) // 常に 1000
}
 
// BEST: アトミック操作（カウンタには最適）
func bestWithAtomic() {
    var counter atomic.Int64
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Add(1)
        }()
    }
    wg.Wait()
    fmt.Println("BEST counter (atomic):", counter.Load()) // 常に 1000
}
 
// ALSO GOOD: チャネルで集約（Go のイディオム）
func goodWithChannel() {
    results := make(chan int, 1000)
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            results <- 1
        }()
    }
 
    go func() {
        wg.Wait()
        close(results)
    }()
 
    counter := 0
    for v := range results {
        counter += v
    }
    fmt.Println("GOOD counter (channel):", counter) // 常に 1000
}

// ================================================================
// Rust: 画像フィルタの並列適用
// ================================================================
 
use rayon::prelude::*;
 
struct Image {
    width: usize,
    height: usize,
    pixels: Vec<u8>,  // RGBA: 4 bytes per pixel
}
 
impl Image {
    /// ガウシアンブラーを並列適用
    fn parallel_blur(&self, radius: usize) -> Image {
        let mut output = vec![0u8; self.pixels.len()];
        let kernel = create_gaussian_kernel(radius);
 
        // 行ごとに並列処理（データ並列）
        output
            .par_chunks_mut(self.width * 4)
            .enumerate()
            .for_each(|(y, row)| {
                for x in 0..self.width {
                    let (r, g, b, a) = apply_kernel(
                        &self.pixels,
                        self.width,
                        self.height,
                        x, y,
                        &kernel,
                        radius,
                    );
                    let idx = x * 4;
                    row[idx] = r;
                    row[idx + 1] = g;
                    row[idx + 2] = b;
                    row[idx + 3] = a;
                }
            });
 
        Image {
            width: self.width,
            height: self.height,
            pixels: output,
        }
    }
 
    /// 複数のフィルタをパイプラインで適用
    fn apply_filters(&self) -> Image {
        // 各フィルタは前のフィルタの結果に依存するため
        // フィルタ内をデータ並列化する
        let blurred = self.parallel_blur(3);
        let sharpened = blurred.parallel_sharpen();
        let adjusted = sharpened.parallel_brightness(1.2);
        adjusted
    }
}

// ================================================================
// Go: API レスポンスの並列集約
// ================================================================
 
package main
 
import (
    "context"
    "fmt"
    "sync"
    "time"
)
 
type Product struct {
    ID          int
    Name        string
    Price       float64
    Reviews     []Review
    Inventory   int
    Recommended []Product
}
 
// 並列度制限付きバッチ処理
func enrichProducts(ctx context.Context,
                    products []Product) ([]Product, error) {
    const maxConcurrency = 20
    sem := make(chan struct{}, maxConcurrency)
 
    var (
        mu       sync.Mutex
        enriched = make([]Product, len(products))
        errs     []error
    )
 
    var wg sync.WaitGroup
    for i, p := range products {
        i, p := i, p
        wg.Add(1)
 
        sem <- struct{}{} // セマフォ獲得
 
        go func() {
            defer wg.Done()
            defer func() { <-sem }() // セマフォ解放
 
            // 各商品に対して3つの外部呼び出しを並列実行
            result, err := enrichSingleProduct(ctx, p)
 
            mu.Lock()
            if err != nil {
                errs = append(errs, err)
            } else {
                enriched[i] = result
            }
            mu.Unlock()
        }()
    }
 
    wg.Wait()
 
    if len(errs) > 0 {
        return nil, fmt.Errorf("%d errors occurred", len(errs))
    }
    return enriched, nil
}
 
func enrichSingleProduct(ctx context.Context,
                         p Product) (Product, error) {
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()
 
    type reviewResult struct {
        reviews []Review
        err     error
    }
    type inventoryResult struct {
        count int
        err   error
    }
    type recommendResult struct {
        products []Product
        err      error
    }
 
    reviewCh := make(chan reviewResult, 1)
    inventoryCh := make(chan inventoryResult, 1)
    recommendCh := make(chan recommendResult, 1)
 
    go func() {
        reviews, err := fetchReviews(ctx, p.ID)
        reviewCh <- reviewResult{reviews, err}
    }()
    go func() {
        count, err := checkInventory(ctx, p.ID)
        inventoryCh <- inventoryResult{count, err}
    }()
    go func() {
        recs, err := getRecommendations(ctx, p.ID)
        recommendCh <- recommendResult{recs, err}
    }()
 
    rr := <-reviewCh
    ir := <-inventoryCh
    rcr := <-recommendCh
 
    if rr.err != nil {
        return p, rr.err
    }
 
    p.Reviews = rr.reviews
    p.Inventory = ir.count
    p.Recommended = rcr.products
    return p, nil
}