#pragma once #include #include #include #include #include #include #include "cache_line.hpp" namespace quant::core { /** * Lock-Free Multi-Producer Single-Consumer (MPSC) Bounded Ring Buffer Queue. * Based on Dmitry Vyukov's bounded queue algorithm. * * Allows multiple concurrent network / gateway client threads to push orders * into a single matching engine thread without mutexes. */ template class alignas(CACHELINE_SIZE) LockFreeMPSCQueue { static_assert(Capacity >= 2, "Capacity must be at least 2"); static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be a power of two"); struct Cell { std::atomic sequence; T data; }; public: LockFreeMPSCQueue() : tail_(0), head_(0) { buffer_ = std::make_unique(Capacity); for (size_t i = 0; i < Capacity; ++i) { buffer_[i].sequence.store(i, std::memory_order_relaxed); } } ~LockFreeMPSCQueue() = default; LockFreeMPSCQueue(const LockFreeMPSCQueue&) = delete; LockFreeMPSCQueue& operator=(const LockFreeMPSCQueue&) = delete; /** * Push an item into the queue. Safe for multiple concurrent producers. */ template bool emplace(Args&&... args) { Cell* cell; size_t pos = tail_.load(std::memory_order_relaxed); for (;;) { cell = &buffer_[pos & MASK]; size_t seq = cell->sequence.load(std::memory_order_acquire); intptr_t diff = static_cast(seq) - static_cast(pos); if (diff == 0) { if (tail_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) { break; } } else if (diff < 0) { return false; // Queue is full } else { pos = tail_.load(std::memory_order_relaxed); } } cell->data = T(std::forward(args)...); cell->sequence.store(pos + 1, std::memory_order_release); return true; } bool push(const T& item) { return emplace(item); } bool push(T&& item) { return emplace(std::move(item)); } /** * Pop an item from the queue. Strictly single-consumer. */ bool pop(T& out_item) { Cell* cell = &buffer_[head_ & MASK]; size_t seq = cell->sequence.load(std::memory_order_acquire); intptr_t diff = static_cast(seq) - static_cast(head_ + 1); if (diff == 0) { out_item = std::move(cell->data); cell->sequence.store(head_ + Capacity, std::memory_order_release); ++head_; return true; } return false; // Queue is empty } std::optional pop() { T item; if (pop(item)) return item; return std::nullopt; } [[nodiscard]] size_t size() const noexcept { size_t t = tail_.load(std::memory_order_relaxed); size_t h = head_; return (t >= h) ? (t - h) : 0; } [[nodiscard]] bool empty() const noexcept { return size() == 0; } [[nodiscard]] constexpr size_t capacity() const noexcept { return Capacity; } private: static constexpr size_t MASK = Capacity - 1; // Tail: shared and modified by multiple producers via CAS alignas(CACHELINE_SIZE) std::atomic tail_; // Head: accessed and modified strictly by the single consumer alignas(CACHELINE_SIZE) size_t head_; std::unique_ptr buffer_; }; } // namespace quant::core