diff --git a/include/logit_cpp/logit/detail/MpscRingAny.hpp b/include/logit_cpp/logit/detail/MpscRingAny.hpp new file mode 100644 index 0000000..fba3fc6 --- /dev/null +++ b/include/logit_cpp/logit/detail/MpscRingAny.hpp @@ -0,0 +1,137 @@ +// detail/MpscRingAny.hpp +#ifndef _LOGIT_DETAIL_MPSC_RING_ANY_HPP_INCLUDED +#define _LOGIT_DETAIL_MPSC_RING_ANY_HPP_INCLUDED + +#include +#include +#include +#include +#include +#include +#include + +namespace logit { namespace detail { + + /// \brief Bounded MPSC ring buffer with arbitrary capacity (C++11). + /// \tparam T Stored type. + template + class MpscRingAny { + private: + /// \brief Single cell storing sequence number and raw storage for T. + struct Cell { + std::atomic m_seq; + typename std::aligned_storage::type m_storage; + }; + + public: + /// \brief Construct ring with given capacity (>= 2). + explicit MpscRingAny(std::size_t capacity) + : m_cap(capacity < 2 ? 2 : capacity), + m_cells(new Cell[m_cap]), + m_enqueue_pos(0), + m_dequeue_pos(0) { + for (std::size_t i = 0; i < m_cap; ++i) { + m_cells[i].m_seq.store(i, std::memory_order_relaxed); + } + } + + /// \brief Destroy remaining elements, if any. + ~MpscRingAny() { + T tmp; + while (try_pop(tmp)) { + // Element destroyed via move-from tmp + } + } + + MpscRingAny(const MpscRingAny&) = delete; + MpscRingAny& operator=(const MpscRingAny&) = delete; + + /// \brief Capacity of the ring. + std::size_t capacity() const noexcept { return m_cap; } + + /// \brief Try to enqueue value. Non-blocking. + /// \return true on success; false if queue is full. + template + bool try_push(U&& v) noexcept { + std::size_t pos = m_enqueue_pos.load(std::memory_order_relaxed); + for (;;) { + Cell& c = m_cells[pos % m_cap]; + std::size_t seq = c.m_seq.load(std::memory_order_acquire); + + // When free, seq == pos + std::intptr_t diff = + static_cast(seq) - static_cast(pos); + + if (diff == 0) { + if (m_enqueue_pos.compare_exchange_weak( + pos, pos + 1, + std::memory_order_relaxed, + std::memory_order_relaxed)) { + // We own the cell; construct T in-place. + new (&c.m_storage) T(std::forward(v)); + // Publish element. + c.m_seq.store(pos + 1, std::memory_order_release); + return true; + } + // CAS failed; 'pos' updated, retry. + } else if (diff < 0) { + // Full. + return false; + } else { + // Another producer advanced 'pos'; reload. + pos = m_enqueue_pos.load(std::memory_order_relaxed); + } + } + } + + /// \brief Try to dequeue value into out. Non-blocking. + /// \return true on success; false if queue is empty. + bool try_pop(T& out) noexcept { + std::size_t pos = m_dequeue_pos.load(std::memory_order_relaxed); + Cell& c = m_cells[pos % m_cap]; + std::size_t seq = c.m_seq.load(std::memory_order_acquire); + + // When ready, seq == pos + 1 + std::intptr_t diff = + static_cast(seq) - static_cast(pos + 1); + + if (diff == 0) { + if (!m_dequeue_pos.compare_exchange_strong( + pos, pos + 1, + std::memory_order_relaxed, + std::memory_order_relaxed)) { + return false; // Single consumer: should be rare. + } + + T* p = reinterpret_cast(&c.m_storage); + out = std::move(*p); + p->~T(); + + // Mark cell free for next cycle. + c.m_seq.store(pos + m_cap, std::memory_order_release); + return true; + } + + return false; // Empty or not yet published. + } + + /// \brief Lightweight emptiness check for current consumer position. + bool empty() const noexcept { + std::size_t pos = m_dequeue_pos.load(std::memory_order_acquire); + const Cell& c = m_cells[pos % m_cap]; + std::size_t seq = c.m_seq.load(std::memory_order_acquire); + std::intptr_t diff = + static_cast(seq) - static_cast(pos + 1); + return diff != 0; + } + + private: + std::size_t m_cap; + std::unique_ptr m_cells; + alignas(64) std::atomic m_enqueue_pos; + alignas(64) std::atomic m_dequeue_pos; + }; + +}} // namespace logit::detail + +#endif // _LOGIT_DETAIL_MPSC_RING_ANY_HPP_INCLUDED