CAS 기반 락프리 링 버퍼와 Mutex Lock 비교

Mutex Lock = 공유 데이터를 직접 보호한다.
Lock-Free Ring Buffer = 공유 데이터를 직접 건드리지 않고 이벤트를 전달한다.

Mutex 방식
여러 스레드 -> lock 획득 -> balance 직접 수정 -> unlock

락프리 링 버퍼 방식
여러 스레드 -> 입금 이벤트 push -> 단일 소비자 -> balance 수정

공유 데이터를 모두가 직접 수정해야 한다면 Mutex가 단순하다.
공유 데이터 수정을 한 곳으로 모을 수 있다면 락프리 링 버퍼가 유리할 수 있다.

Ring Buffer = 고정 크기 배열을 원형으로 재사용하는 자료구조
Lock-Free   = Mutex 없이 진행을 보장하려는 동시성 방식
CAS         = Lock-Free 구현에 자주 쓰이는 원자적 연산

0 -> 1 -> 2 -> 3 -> 0 -> 1 -> ...

CAS(메모리 주소, 예상 값, 새 값)

if 현재 메모리 값 == 예상 값:
    현재 메모리 값을 새 값으로 변경
    성공 반환
else:
    실패 반환

tail이 아직도 10이면 11로 바꿔줘.

lock 획득 -> 공유 데이터 수정 -> lock 해제

CAS로 보호한 것 = 링 버퍼 슬롯 예약
보호되지 않은 것 = balance 같은 외부 공유 변수

여러 생산자 = 입금 요청을 만드는 스레드
단일 소비자 = 입금 요청을 순서대로 처리하는 스레드

Producer A -> DepositEvent(500)  -> Ring Buffer
Producer B -> DepositEvent(1000) -> Ring Buffer

Ring Buffer -> Consumer 하나 -> balance += amount

#include <array>
#include <atomic>
#include <cstddef>
#include <iostream>
#include <thread>
 
template <typename T, std::size_t Capacity>
class MpscRingBuffer {
    static_assert((Capacity & (Capacity - 1)) == 0,
                  "Capacity must be a power of two");
 
    struct alignas(64) Slot {
        std::atomic<std::size_t> sequence;
        T value;
    };
 
    std::array<Slot, Capacity> buffer_;
    std::atomic<std::size_t> tail_{0}; // producer들이 CAS로 증가시킴
    std::size_t head_{0};              // consumer 하나만 접근
 
public:
    MpscRingBuffer() {
        for (std::size_t i = 0; i < Capacity; ++i) {
            buffer_[i].sequence.store(i, std::memory_order_relaxed);
        }
    }
 
    bool try_push(const T& item) {
        std::size_t pos = tail_.load(std::memory_order_relaxed);
 
        for (;;) {
            Slot& slot = buffer_[pos & (Capacity - 1)];
            std::size_t seq = slot.sequence.load(std::memory_order_acquire);
            std::ptrdiff_t diff =
                static_cast<std::ptrdiff_t>(seq) -
                static_cast<std::ptrdiff_t>(pos);
 
            if (diff == 0) {
                if (tail_.compare_exchange_weak(
                        pos,
                        pos + 1,
                        std::memory_order_relaxed,
                        std::memory_order_relaxed)) {
                    slot.value = item;
                    slot.sequence.store(pos + 1, std::memory_order_release);
                    return true;
                }
            } else if (diff < 0) {
                // consumer가 아직 이 슬롯을 비우지 않았으므로 링 버퍼가 가득 찬 상태
                return false;
            } else {
                pos = tail_.load(std::memory_order_relaxed);
            }
        }
    }
 
    bool try_pop(T& out) {
        Slot& slot = buffer_[head_ & (Capacity - 1)];
        std::size_t seq = slot.sequence.load(std::memory_order_acquire);
        std::ptrdiff_t diff =
            static_cast<std::ptrdiff_t>(seq) -
            static_cast<std::ptrdiff_t>(head_ + 1);
 
        if (diff != 0) {
            return false;
        }
 
        out = slot.value;
        slot.sequence.store(head_ + Capacity, std::memory_order_release);
        ++head_;
        return true;
    }
};
 
struct DepositEvent {
    int amount;
};
 
int main() {
    constexpr int producer_count = 2;
    constexpr int repeat_count = 100000;
 
    MpscRingBuffer<DepositEvent, 1024> queue;
    std::atomic<int> producers_done{0};
    int balance = 0; // consumer 스레드만 수정하므로 atomic이나 mutex가 필요 없음
 
    auto producer = [&](int amount) {
        for (int i = 0; i < repeat_count; ++i) {
            DepositEvent event{amount};
 
            while (!queue.try_push(event)) {
                std::this_thread::yield();
            }
        }
 
        producers_done.fetch_add(1, std::memory_order_release);
    };
 
    auto consumer = [&] {
        DepositEvent event{};
 
        while (producers_done.load(std::memory_order_acquire) < producer_count) {
            while (queue.try_pop(event)) {
                balance += event.amount;
            }
 
            std::this_thread::yield();
        }
 
        while (queue.try_pop(event)) {
            balance += event.amount;
        }
    };
 
    std::thread consumer_thread(consumer);
    std::thread p1(producer, 500);
    std::thread p2(producer, 1000);
 
    p1.join();
    p2.join();
    consumer_thread.join();
 
    std::cout << "최종 잔액: " << balance << "원\n";
    return 0;
}

tail_.compare_exchange_weak(pos, pos + 1, ...);

std::size_t head_{0};

sequence == producer가 기대한 값     -> 쓸 수 있는 슬롯
sequence < producer가 기대한 값      -> 아직 소비되지 않은 슬롯
sequence == consumer가 기대한 값 + 1 -> 읽을 수 있는 슬롯

while (!queue.try_push(event)) {
    std::this_thread::yield();
}

이벤트를 버릴 수 있는가?
생산자를 잠시 늦출 것인가?
버퍼 크기를 얼마로 잡을 것인가?
재시도 횟수나 대기 시간을 어떻게 둘 것인가?

#include <iostream>
#include <mutex>
#include <thread>
 
int balance = 0;
std::mutex mtx;
 
void deposit(int amount) {
    // 임계 구역 시작: 스코프 생성 시 자동 락 획득
    std::lock_guard<std::mutex> lock(mtx);
 
    balance = balance + amount;
 
    // 임계 구역 끝: 스코프 종료 시 자동 락 해제
}

컨텍스트 스위칭 = 실행 중이던 스레드를 멈추고 다른 스레드로 전환하는 비용
커널 모드 전환 = 운영체제 스케줄러가 개입하는 비용
캐시 미스       = 다시 깨어난 스레드가 이전 캐시 상태를 잃어서 생기는 비용
호송 현상       = 여러 스레드가 하나의 락 앞에서 줄을 서는 현상

작업을 이벤트로 표현할 수 있다.
공유 상태를 단일 소비자에게 모을 수 있다.
이벤트 처리 시간이 짧고 예측 가능하다.
버퍼가 가득 찰 때의 정책을 정할 수 있다.

공유 객체를 여러 스레드가 직접 읽고 써야 한다.
임계 구역 안의 작업이 복잡하다.
DB, 파일, 네트워크처럼 대기 시간이 긴 작업이 섞여 있다.
락프리 알고리즘의 정확성을 검증할 여력이 부족하다.

CAS는 슬롯 예약을 안전하게 만든다.
Ring Buffer는 고정 메모리를 반복해서 재사용한다.
Single Consumer는 balance 같은 공유 상태의 데이터 경합을 없앤다.