From 38d5d43bb3a5aecc3b505dc016d60fe117441c32 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Tue, 16 Sep 2025 01:39:33 +0300 Subject: [PATCH 1/2] chore(detail): rename MPSC ring header Rename mpsc_ring_any.hpp to MpscRingAny.hpp to align with existing PascalCase headers in detail. --- .../logit_cpp/logit/detail/MpscRingAny.hpp | 143 ++++++++++++++++++ 1 file changed, 143 insertions(+) create mode 100644 include/logit_cpp/logit/detail/MpscRingAny.hpp diff --git a/include/logit_cpp/logit/detail/MpscRingAny.hpp b/include/logit_cpp/logit/detail/MpscRingAny.hpp new file mode 100644 index 0000000..e38881d --- /dev/null +++ b/include/logit_cpp/logit/detail/MpscRingAny.hpp @@ -0,0 +1,143 @@ +// 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 From fc4ecc8860fe52b65d5720605c6055c7bfd71044 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Tue, 16 Sep 2025 01:40:47 +0300 Subject: [PATCH 2/2] chore: update MpscRingAny.hpp --- .../logit_cpp/logit/detail/MpscRingAny.hpp | 218 +++++++++--------- 1 file changed, 106 insertions(+), 112 deletions(-) diff --git a/include/logit_cpp/logit/detail/MpscRingAny.hpp b/include/logit_cpp/logit/detail/MpscRingAny.hpp index e38881d..fba3fc6 100644 --- a/include/logit_cpp/logit/detail/MpscRingAny.hpp +++ b/include/logit_cpp/logit/detail/MpscRingAny.hpp @@ -12,131 +12,125 @@ 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 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 + + /// \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 (;;) { + + 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 free, seq == pos + + // When ready, seq == pos + 1 std::intptr_t diff = - static_cast(seq) - static_cast(pos); - + static_cast(seq) - static_cast(pos + 1); + if (diff == 0) { - if (m_enqueue_pos.compare_exchange_weak( + if (!m_dequeue_pos.compare_exchange_strong( 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; + return false; // Single consumer: should be rare. } - // 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); + + 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 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; + + /// \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; } - - 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; -}; + + 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