MPMC bounded queue

Deep dive

How it works

In: a value to push, or a pop request · Out: Ok/Err(full) for push, Some/None for pop · State: enqueue and dequeue counters, one sequence number per slot · Cost: one CAS, two loads and one store per operation; order of 20-50 ns uncontended

cap = 4, slot seq after init: [0, 1, 2, 3]
push at pos 0: seq[0]==0 -> CAS enq 0->1 -> write -> seq[0]=1
pop  at pos 0: seq[0]==1 (pos+1) -> CAS deq 0->1 -> read -> seq[0]=0+4=4 (free for lap 2)

1. Push: pos = enq.load(). Look at slot = buf[pos & mask], dif = slot.seq - pos.
   • dif == 0: try CAS(enq, pos, pos+1). On success write the value and slot.seq.store(pos+1, Release). On failure retry with the returned value.
   • dif < 0: the slot still holds the previous lap's item: queue full.
   • dif > 0: another producer already claimed pos: reload enq.
2. Pop: same with dif = slot.seq - (pos+1); on success read the value and store seq = pos + cap.
3. The slot's acquire load of seq pairs with the release store by the previous owner, transferring both the value and the "slot is free" fact. The CAS itself can be relaxed.

Code

use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::sync::atomic::{AtomicUsize, Ordering::{Acquire, Relaxed, Release}};

#[repr(align(128))]
struct Pad<T>(T);

struct Slot<T> { seq: AtomicUsize, val: UnsafeCell<MaybeUninit<T>> }

pub struct MpmcQueue<T> { buf: Box<[Slot<T>]>, mask: usize, enq: Pad<AtomicUsize>, deq: Pad<AtomicUsize> }
unsafe impl<T: Send> Send for MpmcQueue<T> {}
unsafe impl<T: Send> Sync for MpmcQueue<T> {}

impl<T> MpmcQueue<T> {
    pub fn new(cap: usize) -> Self {
        let cap = cap.max(2).next_power_of_two();
        let buf = (0..cap)
            .map(|i| Slot { seq: AtomicUsize::new(i), val: UnsafeCell::new(MaybeUninit::uninit()) })
            .collect();
        MpmcQueue { buf, mask: cap - 1, enq: Pad(AtomicUsize::new(0)), deq: Pad(AtomicUsize::new(0)) }
    }

    pub fn push(&self, v: T) -> Result<(), T> {
        let mut pos = self.enq.0.load(Relaxed);
        loop {
            let slot = &self.buf[pos & self.mask];
            let dif = slot.seq.load(Acquire).wrapping_sub(pos) as isize;
            if dif == 0 {
                match self.enq.0.compare_exchange_weak(pos, pos.wrapping_add(1), Relaxed, Relaxed) {
                    Ok(_) => {
                        // SAFETY: winning the CAS for `pos` gives exclusive access to this slot for this lap.
                        unsafe { (*slot.val.get()).write(v); }
                        slot.seq.store(pos.wrapping_add(1), Release); // hand to the consumer of `pos`
                        return Ok(());
                    }
                    Err(cur) => pos = cur,
                }
            } else if dif < 0 {
                return Err(v); // previous lap's item not consumed yet: full
            } else {
                pos = self.enq.0.load(Relaxed); // someone else took `pos`
            }
        }
    }

    pub fn pop(&self) -> Option<T> {
        let mut pos = self.deq.0.load(Relaxed);
        loop {
            let slot = &self.buf[pos & self.mask];
            let dif = slot.seq.load(Acquire).wrapping_sub(pos.wrapping_add(1)) as isize;
            if dif == 0 {
                match self.deq.0.compare_exchange_weak(pos, pos.wrapping_add(1), Relaxed, Relaxed) {
                    Ok(_) => {
                        let v = unsafe { (*slot.val.get()).assume_init_read() };
                        slot.seq.store(pos.wrapping_add(self.mask + 1), Release); // free for next lap
                        return Some(v);
                    }
                    Err(cur) => pos = cur,
                }
            } else if dif < 0 {
                return None; // not yet filled: empty (or a producer is mid-write)
            } else {
                pos = self.deq.0.load(Relaxed);
            }
        }
    }
}

impl<T> Drop for MpmcQueue<T> {
    fn drop(&mut self) { while self.pop().is_some() {} }
}

Complexity and constants

• Each op: 1 load of the shared counter, 1 slot seq load, 1 CAS, 1 slot seq store. The counter line is the hot spot; the slot lines move once per item.
• Uncontended: order of 20-50 ns per push+pop pair cross-core. With N producers on one counter, each CAS is a line transfer (order of 50-100 ns) and failures retry, so total throughput flattens or drops beyond a few cores.
• Memory: cap × (sizeof(T) + 8) rounded to alignment; no allocation after construction.

Efficiency

• Cost: a push+pop pair is order of 20-50 ns cross-core when uncontended. Under heavy contention total throughput is order of 10-50 M ops/s, because every CAS on the shared counter is a line transfer (order of 50-100 ns) and failures retry. Memory: cap × (sizeof(T) + 8), no allocation after construction. Bound: coherence traffic on the two counters.
• Levers:
  • enq and deq on separate 128-byte lines — the two hot counters stop invalidating each other — always
  • fetch_add instead of CAS to claim positions (Rigtorp, Folly) — no CAS retries, but push then waits for its slot — when blocking is acceptable
  • claim k positions with one CAS — amortizes the counter transfer over k items — batch producers or consumers
  • one queue per NUMA node or per core, plus work stealing — removes the shared counters — many cores
  • pad slots to a line — no sharing between neighbouring slots — small T, lockstep producers and consumers
• Naive → standard → state of the art:
  • mutex-guarded ring · a lock handoff per operation, order of µs when contended (see futex mutex)
  • Vyukov sequence-number ring, one CAS per operation · order of 20-50 ns uncontended
  • fetch_add-based ring queues (LCRQ, SCQ, wCQ), sharded queues · scale better under contention (Morrison and Afek; Nikolaev)

Implementation notes

• Pad enq and deq onto separate lines; consider padding slots to a line if T is small and producers/consumers run in lockstep on neighbouring slots.
• Blocking variants: spin briefly, then park on a futex/eventfd per queue or per slot (Folly MPMCQueue uses a turn sequencer per slot, Rigtorp's queue spins on the slot's turn).
• fetch_add instead of CAS to claim positions (Rigtorp, Folly): no CAS retries, but the operation must then wait for its slot, making push blocking rather than try_push.
• Capacity must be a power of two for the mask; the queue holds exactly cap items (no wasted slot).
• Batch: claim k positions with one CAS (pos -> pos + k) after checking k slots, to amortize the counter.

Pitfalls

• Comparing sequence numbers with plain </- instead of wrapping_sub(..) as isize: breaks when the counter wraps, which on 32-bit targets (2^32 ops) takes minutes to hours.
• pop returning None while the queue is not empty: a stalled producer holds slot pos; later slots may already be full. Callers must treat None as "nothing available now", not "empty forever".
• Making the slot seq load Relaxed: the consumer can read the value before it is written.
• Dropping the queue while items remain without draining: leaks or double drops (handled by Drop above).
• Many producers + many consumers on many sockets: the shared counters dominate; one queue per NUMA node or per core scales better.

Variants and alternatives

In production

Language libraries: crossbeam ArrayQueue (Rust; bounded MPMC based on Vyukov's design with per-slot stamps), Rust std::sync::mpsc bounded channel (array flavour derived from crossbeam), Folly MPMCQueue, Rigtorp MPMCQueue. Networking: DPDK rte_ring (also used by its memory pool allocator). Trading and messaging: LMAX Disruptor.

Implementations

Sources

• Dmitry Vyukov, Bounded MPMC queue, 1024cores.net. post
• Maged M. Michael, Michael L. Scott, Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms, PODC 1996. paper
• Adam Morrison, Yehuda Afek, Fast Concurrent Queues for x86 Processors, PPoPP 2013. paper
• Ruslan Nikolaev, A Scalable, Portable, and Memory-Efficient Lock-Free FIFO Queue, DISC 2019. paper