Modern Embedded Recipes · 110/152

MPMC Queue 구현 — Multi-producer Multi-consumer Lock-Free

2026년 4월 18일 · Hawk · 5분 읽기

#한 줄 요약

“MPMC 큐는 lock-free의 최고 난이도 자료구조입니다.” SPSC는 head/tail 두 atomic만으로 가능하지만, MPMC는 매 slot에 sequence number가 필요합니다.

#어떤 상황에서 쓰나

worker pool, packet pipeline, telemetry aggregator처럼 여러 producer가 동시에 enqueue하고 여러 consumer가 동시에 dequeue하는 모든 곳에 MPMC가 필요합니다. mutex 기반 queue는 contention 시 throughput이 한계에 부딪힙니다. lock-free MPMC는 보통 5~10배의 throughput을 냅니다.

다만 진짜 lock-free MPMC는 구현이 까다롭습니다. 잘못된 구현은 ABA, ordering, 메모리 회수 모두 실패합니다. 검증된 algorithm(Vyukov, Disruptor)을 가져다 쓰는 것이 정석입니다.

#핵심 개념

SPSC	1 producer, 1 consumer	head + tail 두 atomic이면 충분
MPSC	N producer, 1 consumer	tail은 단일 owner, head는 CAS
SPMC	1 producer, N consumer	head는 단일 owner, tail은 CAS
MPMC	N producer, M consumer	slot마다 sequence number 필요

MPMC의 두 대표 패턴입니다.

Queue	특성
Vyukov bounded MPMC	slot마다 seq, CAS로 enqueue/dequeue 위치 예약
LMAX Disruptor	ring + cursor, batching 친화, 단일 producer 변종도
`boost::lockfree::queue`	Michael & Scott 기반, unbounded
`moodycamel::ConcurrentQueue`	매우 빠른 unbounded 구현 (open source)

bounded는 capacity가 고정이라 backpressure가 명확하고, unbounded는 capacity가 무한이지만 메모리 사용량을 통제하기 어렵습니다.

#코드 / 실제 사용 예

#Vyukov bounded MPMC

1
template <typename T, size_t N>
2
struct mpmc_queue {
3
    static_assert((N & (N - 1)) == 0, "N must be power of 2");
4

5
    struct slot {
6
        std::atomic<uint64_t> seq;
7
        T data;
8
    };
9

10
    alignas(64) slot buf[N];
11
    alignas(64) std::atomic<uint64_t> enq_pos{0};
12
    alignas(64) std::atomic<uint64_t> deq_pos{0};
13

14
    mpmc_queue() {
15
        for (size_t i = 0; i < N; i++)
16
            buf[i].seq.store(i, std::memory_order_relaxed);
17
    }
18

19
    bool enqueue(const T &v) {
20
        uint64_t pos = enq_pos.load(std::memory_order_relaxed);
21
        slot *s;
22
        for (;;) {
23
            s = &buf[pos & (N - 1)];
24
            uint64_t seq = s->seq.load(std::memory_order_acquire);
25
            intptr_t diff = (intptr_t)seq - (intptr_t)pos;
26
            if (diff == 0) {
27
                if (enq_pos.compare_exchange_weak(pos, pos + 1,
28
                        std::memory_order_relaxed)) break;
29
            } else if (diff < 0) {
30
                return false;    /* full */
31
            } else {
32
                pos = enq_pos.load(std::memory_order_relaxed);
33
            }
34
        }
35
        s->data = v;
36
        s->seq.store(pos + 1, std::memory_order_release);
37
        return true;
38
    }
39

40
    bool dequeue(T &v) {
41
        uint64_t pos = deq_pos.load(std::memory_order_relaxed);
42
        slot *s;
43
        for (;;) {
44
            s = &buf[pos & (N - 1)];
45
            uint64_t seq = s->seq.load(std::memory_order_acquire);
46
            intptr_t diff = (intptr_t)seq - (intptr_t)(pos + 1);
47
            if (diff == 0) {
48
                if (deq_pos.compare_exchange_weak(pos, pos + 1,
49
                        std::memory_order_relaxed)) break;
50
            } else if (diff < 0) {
51
                return false;    /* empty */
52
            } else {
53
                pos = deq_pos.load(std::memory_order_relaxed);
54
            }
55
        }
56
        v = s->data;
57
        s->seq.store(pos + N, std::memory_order_release);
58
        return true;
59
    }
60
};

핵심 아이디어는 각 slot의 sequence number가 현재 상태를 인코딩하는 것입니다. enqueuer가 자기 자리를 예약(seq == pos)하고, 데이터를 쓴 후 seq = pos + 1로 publish합니다. dequeuer는 seq == pos + 1을 확인하고 데이터를 읽은 후 seq = pos + N으로 다음 cycle을 안내합니다.

#SPSC 비교

1
template <typename T, size_t N>
2
struct spsc_queue {
3
    static_assert((N & (N - 1)) == 0);
4
    alignas(64) std::atomic<size_t> head{0};
5
    alignas(64) std::atomic<size_t> tail{0};
6
    alignas(64) T buf[N];
7

8
    bool push(const T &v) {
9
        size_t h = head.load(std::memory_order_relaxed);
10
        size_t n = (h + 1) & (N - 1);
11
        if (n == tail.load(std::memory_order_acquire)) return false;
12
        buf[h] = v;
13
        head.store(n, std::memory_order_release);
14
        return true;
15
    }
16

17
    bool pop(T &v) {
18
        size_t t = tail.load(std::memory_order_relaxed);
19
        if (t == head.load(std::memory_order_acquire)) return false;
20
        v = buf[t];
21
        tail.store((t + 1) & (N - 1), std::memory_order_release);
22
        return true;
23
    }
24
};

SPSC는 두 atomic이 거의 전부입니다. MPMC의 복잡성과 비교됩니다.

#LMAX Disruptor 아이디어

1
Ring buffer + cursor (sequence)
2
producer:
3
  next = cursor.fetch_add(1)
4
  buf[next & (N-1)] = item
5
  available.set(next)
6

7
consumer:
8
  while (cursor_pub < target) yield;
9
  process(buf[target & (N-1)])

Disruptor는 atomic cursor 하나에 fetch_add로 자리를 예약합니다. consumer batching이 매우 효율적이지만 구현이 큽니다.

#boost::lockfree::queue

1
#include <boost/lockfree/queue.hpp>
2

3
boost::lockfree::queue<int> q(128);     /* unbounded라도 초기 size hint */
4

5
void prod(void) { while (!q.push(42)); }
6
void cons(void) { int v; while (!q.pop(v)); }

unbounded MPMC가 필요하면 가장 안전한 선택입니다. Michael & Scott 알고리즘 + tagged pointer.

#moodycamel::ConcurrentQueue

1
#include "concurrentqueue.h"
2

3
moodycamel::ConcurrentQueue<int> q;
4
q.enqueue(42);
5
int v;
6
q.try_dequeue(v);
7

8
/* 또는 batch */
9
int items[16];
10
size_t n = q.try_dequeue_bulk(items, 16);

매우 빠른 unbounded MPMC 구현입니다. linux/windows/macos 모두 동작하고 header-only입니다.

#측정 / 성능 비교

8 코어 Intel Xeon, 메시지 1억 개 처리 throughput입니다.

1
구조                        4P/4C throughput
2
mutex + std::queue          8 M ops/s
3
boost::lockfree::queue      45 M ops/s
4
Vyukov MPMC                 95 M ops/s
5
moodycamel ConcurrentQueue  120 M ops/s
6
LMAX Disruptor (batched)    180 M ops/s
7

8
SPSC만 (비교용)              350 M ops/s

SPSC가 가장 빠르고, MPMC는 잘 짜도 SPSC의 1/3 이하입니다. 단일 thread queue가 가능한 경우 굳이 MPMC를 안 쓰는 것이 best입니다.

1
contention 영향 (Vyukov, 16 thread)
2
4 producer, 4 consumer      95 M ops/s
3
8 producer, 8 consumer      75 M ops/s
4
16 producer, 16 consumer    50 M ops/s

contention이 늘수록 cache line ping-pong이 늘어 throughput이 떨어집니다.

#자주 보는 함정

단순 fetch_add로 MPMC 구현

1
size_t pos = head.fetch_add(1);
2
buf[pos & mask] = v;    /* 다른 producer가 이미 같은 slot 쓸 수 있음 */

slot 단위 동기화가 없으면 race가 발생합니다. sequence number가 필수입니다.

Unbounded라고 capacity 무시

1
moodycamel::ConcurrentQueue<int> q;
2
while (true) q.enqueue(big_item);    /* 메모리 폭주 */

unbounded는 producer가 너무 빠르면 메모리가 폭주합니다. backpressure 정책을 둡니다.

false sharing 무시

1
struct queue {
2
    std::atomic<size_t> head;
3
    std::atomic<size_t> tail;     /* 같은 line — false sharing */
4
};

head와 tail은 반드시 별도 line에 둡니다. 9-09편 참고.

SPSC 코드를 MPMC로 사용

1
spsc_queue q;
2
producer1.push(...);   /* race */
3
producer2.push(...);

SPSC는 정확히 1 producer + 1 consumer만 보장합니다. type 단계에서 구별합니다.

큰 항목을 by-value

1
mpmc_queue<huge_struct, 1024> q;    /* slot마다 huge_struct 복사 */

큰 항목은 pointer만 큐에 넣고 pool에서 alloc합니다. 6-06편의 by-pointer 패턴 참고.

#정리

SPSC는 head/tail 두 atomic으로 가능하지만 MPMC는 slot마다 sequence number가 필요합니다.
Vyukov bounded MPMC가 가장 친숙한 구현 패턴입니다.
unbounded는 boost::lockfree나 moodycamel을 가져다 쓰는 것이 안전합니다.
LMAX Disruptor는 batching이 가능해 throughput이 최고입니다.
false sharing(head와 tail) 회피와 power-of-two capacity가 필수입니다.
SPSC가 가능한 경우 굳이 MPMC를 쓰지 않습니다.
producer 수 만큼 backpressure 정책(drop, block, replace)을 명시합니다.

이로써 Part 9 Concurrency 응용의 마지막 챕터입니다. 시리즈 전체를 마무리하며 처음부터 새로 만든 BSP/firmware의 모든 레이어가 한 자리에 모였습니다.

MPMC Queue 구현 — Multi-producer Multi-consumer Lock-Free

#한 줄 요약

#어떤 상황에서 쓰나

#핵심 개념

#코드 / 실제 사용 예

#Vyukov bounded MPMC

#SPSC 비교

#LMAX Disruptor 아이디어

#boost::lockfree::queue

#moodycamel::ConcurrentQueue

#측정 / 성능 비교

#자주 보는 함정

#정리

#관련 항목

Modern Embedded Recipes · 111 of 152

관련 글

False Sharing 해결 — Cache Line Padding·SoA 적용

ABA 문제 회피 — Tagged Pointer·Hazard·Generation Counter

Spinlock vs Mutex 결정 가이드 — Context Switch·Hold Time

이 글을 참조하는 글 (3)