mpmc_ring.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright 2026 Johnson Ogundeji
//
// O(1) amortized push/pop, no allocation after construction, ABA-safe via
// per-slot sequence numbers.  Runtime capacity (always rounded up to next
// power of two, minimum 2).
//
// Phase 9b: MPMC ColumnBatch Event Bus.
 
#pragma once
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <stdexcept>
#include <type_traits>
 
namespace signet::forge {
 
// ============================================================================
// MpmcRing<T> — Vyukov bounded MPMC queue
// ============================================================================
 
template<typename T>
class MpmcRing {
public:
    // -------------------------------------------------------------------------
    // Construction
    // -------------------------------------------------------------------------
 
    explicit MpmcRing(size_t capacity) {
        if (capacity == 0) {
            throw std::invalid_argument("MpmcRing capacity must be >= 1");
        }
        size_t cap = 2;
        while (cap < capacity) cap <<= 1;
        // Invariant: cap is a power of two (enforced by construction above)
        capacity_ = cap;
        mask_     = cap - 1;
 
        slots_ = std::make_unique<Slot[]>(cap);
        for (size_t i = 0; i < cap; ++i)
            slots_[i].seq.store(i, std::memory_order_relaxed);
 
        enqueue_.store(0, std::memory_order_relaxed);
        dequeue_.store(0, std::memory_order_relaxed);
    }
 
    MpmcRing(const MpmcRing&)            = delete;
    MpmcRing& operator=(const MpmcRing&) = delete;
    MpmcRing(MpmcRing&&)                 = delete;
    MpmcRing& operator=(MpmcRing&&)      = delete;
 
    // -------------------------------------------------------------------------
    // push — non-blocking; returns false if ring is full
    // -------------------------------------------------------------------------
 
    [[nodiscard]] bool push(T val) noexcept(std::is_nothrow_move_assignable_v<T>) {
        size_t pos = enqueue_.load(std::memory_order_relaxed);
        for (;;) {
            Slot&        slot = slots_[pos & mask_];
            const size_t seq  = slot.seq.load(std::memory_order_acquire);
            const auto   diff = static_cast<intptr_t>(seq)
                              - static_cast<intptr_t>(pos);
 
            if (diff == 0) {
                // Slot free — try to claim it.
                if (enqueue_.compare_exchange_weak(
                        pos, pos + 1,
                        std::memory_order_relaxed,
                        std::memory_order_relaxed)) {
                    slot.data = std::move(val);
                    slot.seq.store(pos + 1, std::memory_order_release);
                    return true;
                }
                // Another producer won the CAS; reload and retry.
            } else if (diff < 0) {
                return false;  // ring is full
            } else {
                pos = enqueue_.load(std::memory_order_relaxed);
            }
        }
    }
 
    // -------------------------------------------------------------------------
    // pop — non-blocking; returns false if ring is empty
    // -------------------------------------------------------------------------
 
    [[nodiscard]] bool pop(T& out) noexcept(std::is_nothrow_move_assignable_v<T>) {
        size_t pos = dequeue_.load(std::memory_order_relaxed);
        for (;;) {
            Slot&        slot = slots_[pos & mask_];
            const size_t seq  = slot.seq.load(std::memory_order_acquire);
            const auto   diff = static_cast<intptr_t>(seq)
                              - static_cast<intptr_t>(pos + 1);
 
            if (diff == 0) {
                // Slot full — try to claim it.
                if (dequeue_.compare_exchange_weak(
                        pos, pos + 1,
                        std::memory_order_relaxed,
                        std::memory_order_relaxed)) {
                    out = std::move(slot.data);
                    // Mark slot free for the NEXT full cycle.
                    slot.seq.store(pos + mask_ + 1, std::memory_order_release);
                    return true;
                }
                // Another consumer won the CAS; reload and retry.
            } else if (diff < 0) {
                return false;  // ring is empty
            } else {
                pos = dequeue_.load(std::memory_order_relaxed);
            }
        }
    }
 
    // -------------------------------------------------------------------------
    // Metadata
    // -------------------------------------------------------------------------
 
    [[nodiscard]] size_t capacity() const noexcept { return capacity_; }
 
    [[nodiscard]] size_t size() const noexcept {
        const size_t head = enqueue_.load(std::memory_order_relaxed);
        const size_t tail = dequeue_.load(std::memory_order_relaxed);
        return (head >= tail) ? (head - tail) : 0u;
    }
 
    [[nodiscard]] bool empty() const noexcept { return size() == 0u; }
 
private:
    struct alignas(64) Slot {
        std::atomic<size_t> seq{0};
        T                   data{};
    };
 
    // Separate cache lines for producer and consumer cursors.
    static constexpr size_t kCacheLine = 64;
 
    alignas(kCacheLine) std::atomic<size_t> enqueue_{0};
    alignas(kCacheLine) std::atomic<size_t> dequeue_{0};
 
    std::unique_ptr<Slot[]> slots_;
    size_t                  capacity_{0};
    size_t                  mask_{0};
};
 
} // namespace signet::forge