From d1204867a7b725e6ff7f8a0eba0ac48178269d30 Mon Sep 17 00:00:00 2001 From: AneesPatel Date: Thu, 30 Apr 2026 02:19:04 -0400 Subject: [PATCH] add MPMC queue (Vyukov design) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-slot sequence counters for coordination — no single shared lock. Head/tail on separate cache lines to avoid false sharing between producers and consumers. Stress tested with 4+4 threads, 1M items. --- CMakeLists.txt | 2 + include/MPMCQueue.h | 99 ++++++++++++++++++++++++++++++++++++++++ tests/test_mpmcqueue.cpp | 77 +++++++++++++++++++++++++++++++ 3 files changed, 178 insertions(+) create mode 100644 include/MPMCQueue.h create mode 100644 tests/test_mpmcqueue.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index e12e3e6..f6649f7 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -8,3 +8,5 @@ set(CMAKE_EXPORT_COMPILE_COMMANDS ON) set(CMAKE_CXX_FLAGS_RELEASE "-O3 -march=native -DNDEBUG") include_directories(include) + +add_executable(test_mpmcqueue tests/test_mpmcqueue.cpp) diff --git a/include/MPMCQueue.h b/include/MPMCQueue.h new file mode 100644 index 0000000..921a57b --- /dev/null +++ b/include/MPMCQueue.h @@ -0,0 +1,99 @@ +#pragma once + +#include +#include +#include + +// Multi-Producer Multi-Consumer queue based on Dmitry Vyukov's design. +// Each slot has its own sequence counter so producers and consumers can +// coordinate without a single shared lock. +// +// The capacity must be a power of two (not enforced, but things will break if it isn't). +// Push/pop spin on their respective slots if the queue is full/empty — this is +// intentional for low-latency use, not a bug. +template +class MPMCQueue { +private: + struct Slot { + T data; + // seq == index means the slot is empty and ready to write + // seq == index + 1 means the slot has data ready to read + std::atomic seq; + }; + +public: + MPMCQueue() { + for (size_t i = 0; i < Capacity; ++i) { + slots_[i].seq.store(i, std::memory_order_relaxed); + } + } + + // Try to enqueue. Returns false immediately if the queue is full (non-blocking). + bool push(const T& val) { + size_t pos = head_.load(std::memory_order_relaxed); + + while (true) { + Slot& slot = slots_[pos % Capacity]; + size_t seq = slot.seq.load(std::memory_order_acquire); + intptr_t diff = static_cast(seq) - static_cast(pos); + + if (diff == 0) { + // Slot is ready. Try to claim it by advancing head. + if (head_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) { + slot.data = val; + // Signal to consumers that data is ready + slot.seq.store(pos + 1, std::memory_order_release); + return true; + } + // Another producer grabbed it, retry with updated pos + } else if (diff < 0) { + // Queue is full + return false; + } else { + // Another producer is mid-write on this slot, reload head and retry + pos = head_.load(std::memory_order_relaxed); + } + } + } + + // Try to dequeue. Returns nullopt if the queue is empty (non-blocking). + std::optional pop() { + size_t pos = tail_.load(std::memory_order_relaxed); + + while (true) { + Slot& slot = slots_[pos % Capacity]; + size_t seq = slot.seq.load(std::memory_order_acquire); + intptr_t diff = static_cast(seq) - static_cast(pos + 1); + + if (diff == 0) { + // Data is here. Try to claim this slot by advancing tail. + if (tail_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) { + T val = std::move(slot.data); + // Reset the slot for the next lap around the ring + slot.seq.store(pos + Capacity, std::memory_order_release); + return val; + } + // Another consumer grabbed it, retry + } else if (diff < 0) { + // Queue is empty + return std::nullopt; + } else { + pos = tail_.load(std::memory_order_relaxed); + } + } + } + + bool empty() const { + size_t h = head_.load(std::memory_order_acquire); + size_t t = tail_.load(std::memory_order_acquire); + return h == t; + } + +private: + // head_ and tail_ are on separate cache lines to avoid false sharing. + // The SPSC ring buffer doesn't bother with this, but with multiple threads + // hammering both ends it starts to matter. + alignas(64) std::atomic head_{0}; + alignas(64) std::atomic tail_{0}; + Slot slots_[Capacity]; +}; diff --git a/tests/test_mpmcqueue.cpp b/tests/test_mpmcqueue.cpp new file mode 100644 index 0000000..5ada2d6 --- /dev/null +++ b/tests/test_mpmcqueue.cpp @@ -0,0 +1,77 @@ +#include "MPMCQueue.h" +#include +#include +#include +#include + +int main() { + // 4 producers, 4 consumers, 1M total items + const int NUM_PRODUCERS = 4; + const int NUM_CONSUMERS = 4; + const int ITEMS_PER_PRODUCER = 250000; + const int TOTAL = NUM_PRODUCERS * ITEMS_PER_PRODUCER; + + MPMCQueue queue; + std::atomic sum_produced{0}; + std::atomic sum_consumed{0}; + std::atomic done_producers{0}; + + auto producer = [&](int id) { + long long local_sum = 0; + for (int i = 0; i < ITEMS_PER_PRODUCER; ++i) { + int val = id * ITEMS_PER_PRODUCER + i; + local_sum += val; + // Spin until we can push (queue might be momentarily full) + while (!queue.push(val)) { + std::this_thread::yield(); + } + } + sum_produced.fetch_add(local_sum, std::memory_order_relaxed); + done_producers.fetch_add(1, std::memory_order_release); + }; + + auto consumer = [&]() { + long long local_sum = 0; + int consumed = 0; + while (true) { + auto val = queue.pop(); + if (val) { + local_sum += *val; + ++consumed; + if (consumed == TOTAL / NUM_CONSUMERS) break; + } else { + // Queue empty, either producers are still running or we're done + if (done_producers.load(std::memory_order_acquire) == NUM_PRODUCERS && queue.empty()) { + break; + } + std::this_thread::yield(); + } + } + sum_consumed.fetch_add(local_sum, std::memory_order_relaxed); + }; + + std::vector threads; + for (int i = 0; i < NUM_PRODUCERS; ++i) + threads.emplace_back(producer, i); + for (int i = 0; i < NUM_CONSUMERS; ++i) + threads.emplace_back(consumer); + + for (auto& t : threads) + t.join(); + + // Drain any remaining items (edge case with uneven distribution) + auto val = queue.pop(); + while (val) { + sum_consumed.fetch_add(*val, std::memory_order_relaxed); + val = queue.pop(); + } + + if (sum_produced.load() == sum_consumed.load()) { + std::cout << "MPMCQueue test passed! Sum: " << sum_produced.load() << std::endl; + return 0; + } else { + std::cout << "MPMCQueue test FAILED! Produced sum: " << sum_produced.load() + << " Consumed sum: " << sum_consumed.load() << std::endl; + return 1; + } +}