#pragma once #include #include #include #include #include #include #include "cache_line.hpp" namespace quant::core { /** * Lock-Free Single-Producer Single-Consumer (SPSC) Circular Ring Buffer. * Designed for microsecond inter-thread event streaming in High-Frequency Trading. * * Features: * - Power-of-two bitwise modulo wrapping: index & (Capacity - 1) * - Head and Tail stored on isolated 64-byte cache lines to eliminate CPU false sharing * - Explicit acquire-release memory order semantics (No locks, zero OS syscalls) * - Cache-friendly contiguous ring storage */ template class alignas(CACHELINE_SIZE) LockFreeSPSCQueue { static_assert(Capacity >= 2, "Capacity must be at least 2"); static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be a power of two for bitwise wrapping"); public: LockFreeSPSCQueue() : head_(0), cached_tail_(0), tail_(0), cached_head_(0) { ring_ = std::make_unique(Capacity); } ~LockFreeSPSCQueue() = default; LockFreeSPSCQueue(const LockFreeSPSCQueue&) = delete; LockFreeSPSCQueue& operator=(const LockFreeSPSCQueue&) = delete; LockFreeSPSCQueue(LockFreeSPSCQueue&&) noexcept = default; LockFreeSPSCQueue& operator=(LockFreeSPSCQueue&&) noexcept = default; /** * Push an item into the queue. * Called strictly by the Producer thread. */ template bool emplace(Args&&... args) { const size_t current_tail = tail_.load(std::memory_order_relaxed); // Check against cached head first to avoid cross-core cache invalidation if (current_tail - cached_head_ >= Capacity) { cached_head_ = head_.load(std::memory_order_acquire); if (current_tail - cached_head_ >= Capacity) { return false; // Queue full } } ring_[current_tail & MASK] = T(std::forward(args)...); tail_.store(current_tail + 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. * Called strictly by the Consumer thread. */ bool pop(T& out_item) { const size_t current_head = head_.load(std::memory_order_relaxed); // Check against cached tail first to reduce atomic loads if (current_head == cached_tail_) { cached_tail_ = tail_.load(std::memory_order_acquire); if (current_head == cached_tail_) { return false; // Queue empty } } out_item = std::move(ring_[current_head & MASK]); head_.store(current_head + 1, std::memory_order_release); return true; } std::optional pop() { T item; if (pop(item)) { return item; } return std::nullopt; } [[nodiscard]] size_t size() const noexcept { const size_t t = tail_.load(std::memory_order_relaxed); const size_t h = head_.load(std::memory_order_relaxed); return (t >= h) ? (t - h) : 0; } [[nodiscard]] bool empty() const noexcept { return head_.load(std::memory_order_relaxed) == tail_.load(std::memory_order_relaxed); } [[nodiscard]] constexpr size_t capacity() const noexcept { return Capacity; } private: static constexpr size_t MASK = Capacity - 1; // Head line (read/written by consumer) alignas(CACHELINE_SIZE) std::atomic head_; size_t cached_tail_; // Consumer's private shadow copy of tail char pad1_[CACHELINE_SIZE - sizeof(std::atomic) - sizeof(size_t)]; // Tail line (read/written by producer) alignas(CACHELINE_SIZE) std::atomic tail_; size_t cached_head_; // Producer's private shadow copy of head char pad2_[CACHELINE_SIZE - sizeof(std::atomic) - sizeof(size_t)]; // Storage buffer std::unique_ptr ring_; }; } // namespace quant::core