Lock-Free 並行プログラミング完全ガイド 2025: Atomics, Memory Model, CAS, Lock-Freeデータ構造

// Thread 1
x = 1;
y = 2;

// Thread 2
print(y);
print(x);

#include <atomic>

std::atomic<int> x{0};

// 最強 (Sequential Consistency, デフォルト)
x.store(1, std::memory_order_seq_cst);

// Release: このstore以前のすべてのメモリ作業が完了してから可視化
x.store(1, std::memory_order_release);

// Acquire: このload以降のすべてのメモリ作業はこのload以降に実行
int v = x.load(std::memory_order_acquire);

// Relaxed: atomic性のみ保証、順序なし (最速)
x.store(1, std::memory_order_relaxed);

std::atomic<bool> ready{false};
int data = 0;

// Thread 1 (Producer)
data = 42;
ready.store(true, std::memory_order_release);

// Thread 2 (Consumer)
while (!ready.load(std::memory_order_acquire)) {}
assert(data == 42);  // 保証される

bool compare_and_swap(int* ptr, int expected, int new_value) {
    if (*ptr == expected) {
        *ptr = new_value;
        return true;
    }
    return false;
}

std::atomic<int> counter{0};

void increment() {
    int old_value, new_value;
    do {
        old_value = counter.load();
        new_value = old_value + 1;
    } while (!counter.compare_exchange_weak(old_value, new_value));
}

counter.fetch_add(1, std::memory_order_relaxed);

use std::sync::atomic::{AtomicUsize, Ordering};

let counter = AtomicUsize::new(0);

// atomic increment
counter.fetch_add(1, Ordering::Relaxed);

// CAS
let result = counter.compare_exchange(
    0,                     // expected
    1,                     // new
    Ordering::Acquire,     // success ordering
    Ordering::Relaxed      // failure ordering
);

template<typename T>
class LockFreeStack {
    struct Node {
        T value;
        Node* next;
    };
    std::atomic<Node*> head{nullptr};

public:
    void push(T value) {
        Node* new_node = new Node{value, nullptr};
        Node* old_head;
        do {
            old_head = head.load();
            new_node->next = old_head;
        } while (!head.compare_exchange_weak(old_head, new_node));
    }

    bool pop(T& value) {
        Node* old_head;
        Node* new_head;
        do {
            old_head = head.load();
            if (!old_head) return false;
            new_head = old_head->next;
        } while (!head.compare_exchange_weak(old_head, new_head));
        value = old_head->value;
        delete old_head;  // 危険! 他スレッドが読んでいる可能性
        return true;
    }
};

初期: head -> A -> B -> C

Thread 1:                    Thread 2:
old_head = A
new_head = B
                              pop A -> delete A
                              pop B -> delete B
                              push A (再割り当てされたメモリ)
                              head -> A -> C

CAS(old_head=A, new_head=B)  <- 成功! しかし誤り
head -> B (すでに free 済みメモリ)

struct TaggedPtr {
    Node* ptr;
    uint64_t tag;  // 毎回インクリメント
};
std::atomic<TaggedPtr> head;

struct Node {
    T value;
    std::atomic<Node*> next{nullptr};
};

std::atomic<Node*> head;
std::atomic<Node*> tail;

void enqueue(T value) {
    Node* new_node = new Node{value};
    Node* old_tail;
    Node* old_next;
    while (true) {
        old_tail = tail.load();
        old_next = old_tail->next.load();
        if (old_tail == tail.load()) {  // 一貫性チェック
            if (old_next == nullptr) {
                // CAS で next 設定を試みる
                if (old_tail->next.compare_exchange_weak(old_next, new_node)) {
                    break;
                }
            } else {
                // tail が遅れている、helping
                tail.compare_exchange_weak(old_tail, old_next);
            }
        }
    }
    // tail を更新 (失敗しても OK、他スレッドが helping してくれる)
    tail.compare_exchange_weak(old_tail, new_node);
}

// 擬似コード
struct config* config = rcu_dereference(global_config);
int value = config->some_value;  // ロックなしで読む

// 更新
struct config* new_config = malloc(sizeof(*new_config));
memcpy(new_config, old_config, sizeof(*new_config));
new_config->some_value = 42;
rcu_assign_pointer(global_config, new_config);
synchronize_rcu();  // 全 reader 完了待ち
free(old_config);

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

// 単純カウンタ (低競合)
AtomicLong counter = new AtomicLong(0);
counter.incrementAndGet();

// 高競合カウンタ (Java 8+)
LongAdder adder = new LongAdder();
adder.increment();  // 分割セルで競合分散
long value = adder.sum();  // 集計

#include <atomic>

std::atomic<int> counter{0};
counter.fetch_add(1, std::memory_order_relaxed);

// Memory Orderingを正確に指定する必要がある
std::atomic<bool> flag{false};
flag.store(true, std::memory_order_release);
bool v = flag.load(std::memory_order_acquire);

use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering};

static COUNTER: AtomicUsize = AtomicUsize::new(0);
COUNTER.fetch_add(1, Ordering::Relaxed);

import "sync/atomic"

var counter int64
atomic.AddInt64(&counter, 1)
value := atomic.LoadInt64(&counter)

class LockFreeLogger {
    static constexpr size_t SIZE = 1024;
    std::atomic<size_t> head{0};
    std::array<std::string, SIZE> buffer;

public:
    void log(const std::string& msg) {
        size_t idx = head.fetch_add(1, std::memory_order_relaxed) % SIZE;
        buffer[idx] = msg;  // 注意: 同時書き込みが起こり得る、実コードではより慎重に
    }
};

use std::sync::atomic::{AtomicU64, Ordering};

struct IdGenerator {
    next_id: AtomicU64,
}

impl IdGenerator {
    fn next(&self) -> u64 {
        self.next_id.fetch_add(1, Ordering::Relaxed)
    }
}

class SpinLock {
    std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
    void lock() {
        while (flag.test_and_set(std::memory_order_acquire)) {
            // busy wait
        }
    }
    void unlock() {
        flag.clear(std::memory_order_release);
    }
};

while (flag.test_and_set(std::memory_order_acquire)) {
    _mm_pause();  // CPU ヒント
}

struct Counters {
    std::atomic<int> a;
    std::atomic<int> b;  // 同じキャッシュライン!
};

struct Counters {
    alignas(64) std::atomic<int> a;
    alignas(64) std::atomic<int> b;
};