EventDispatcher.cppm

// Copyright (C) 2026 mxreal64
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <https://gnu.org>.
//
// -----------------------------------------------------------------------
// NOTE:
// optimization notes vs. the last iteration:
//
// 1. ticket claiming is now fetch_add instead of a load+CAS retry loop.
//    fetch_add on producer_sequence_/consumer_sequence_ ALWAYS succeeds in
//    one atomic RMW -- every thread gets a distinct ticket immediately, no
//    retries possible. The original's CAS loop could fail and re-spin the
//    entire claim (reload + recheck slot + re-CAS) every time two threads
//    raced for the same ticket; fetch_add removes that failure mode
//    completely; this is the textbook Vyukov bounded MPMC design -- the
//    original had reinvented ticket claiming with CAS, which is strictly
//    more expensive under contention for no additional correctness benefit.
//    the only spin left is "wait for this slot to actually be free/full",
//    which is a plain load (no CAS), so a producer only ever spins when the
//    ring is genuinely full, and a consumer only when it's genuinely empty.
//
// 2. added a SpscDispatcher which is specialized fast path for the common
//    case of exactly one producer thread and one consumer thread. it drops the
//    per-slot atomic sequence counter entirely (nothing else can race for a
//    given slot in SPSC), keeping only one atomic per side plus a cached,
//    non-atomic copy of the other side's cursor that's refreshed only when
//    it looks like the ring might be full/empty. this is not a drop-in
//    replacement for NanosecondDispatcher ofc; it is UB if more than one
//    thread ever calls publish() or more than one thread ever calls
//    consume_next(). use it only when you can guarantee that shape.
//
// -----------------------------------------------------------------------

export module EventDispatcher;

import std;

export template <typename EventType>
struct alignas(64) EventSlot {
    alignas(alignof(EventType)) std::byte event_storage[sizeof(EventType)];
    alignas(64) std::atomic<uint64_t> sequence{0};
};

export template <typename EventType, std::size_t Capacity>
class NanosecondDispatcher {
private:
    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be a power of two.");
    static constexpr std::size_t Mask = Capacity - 1;

    alignas(64) EventSlot<EventType> ring_buffer_[Capacity];
    alignas(64) std::atomic<uint64_t> producer_sequence_{0};
    alignas(64) std::atomic<uint64_t> consumer_sequence_{0};

    static inline void cpu_relax() noexcept {
        #if defined(__x86_64__) || defined(_M_X64)
        __builtin_ia32_pause();
        #elif defined(__aarch64__)
        asm volatile("yield" ::: "memory");
        #endif
    }

public:
    NanosecondDispatcher() {
        for (std::size_t i = 0; i < Capacity; ++i) {
            ring_buffer_[i].sequence.store(i, std::memory_order_relaxed);
        }
    }

    ~NanosecondDispatcher() {
        uint64_t head = consumer_sequence_.load(std::memory_order_relaxed);
        uint64_t tail = producer_sequence_.load(std::memory_order_relaxed);
        for (uint64_t i = head; i < tail; ++i) {
            auto* slot = &ring_buffer_[i & Mask];
            std::destroy_at(reinterpret_cast<EventType*>(slot->event_storage));
        }
    }

    NanosecondDispatcher(const NanosecondDispatcher&) = delete;
    NanosecondDispatcher& operator=(const NanosecondDispatcher&) = delete;

    template <typename... Args>
    void publish(Args&&... args) noexcept {
        // static_assert(std::is_nothrow_constructible_v<EventType, Args...>,
        //     "publish() is noexcept; EventType construction must not throw.");

        // Unconditional reservation: always succeeds in exactly one RMW,
        // no retry loop, no lost-wakeup / re-check window.
        uint64_t ticket = producer_sequence_.fetch_add(1, std::memory_order_relaxed);
        EventSlot<EventType>* slot = &ring_buffer_[ticket & Mask];

        // Only spins if the ring is genuinely full (this slot's previous
        // occupant hasn't been consumed and freed yet).
        while (slot->sequence.load(std::memory_order_acquire) != ticket) {
            cpu_relax();
        }

        ::new (static_cast<void*>(slot->event_storage)) EventType(std::forward<Args>(args)...);
        slot->sequence.store(ticket + 1, std::memory_order_release);
    }

    template <typename EventHandler>
    void consume_next(EventHandler&& handler) noexcept {
        uint64_t ticket = consumer_sequence_.fetch_add(1, std::memory_order_relaxed);
        EventSlot<EventType>* slot = &ring_buffer_[ticket & Mask];

        // Only spins if the ring is genuinely empty.
        while (slot->sequence.load(std::memory_order_acquire) != ticket + 1) {
            cpu_relax();
        }

        auto* item = reinterpret_cast<EventType*>(slot->event_storage);
        std::forward<EventHandler>(handler)(*item);
        std::destroy_at(item);

        slot->sequence.store(ticket + Capacity, std::memory_order_release);
    }
};

// ---------------------------------------------------------------------
// spsc fast path: only valid when exactly 1 thread ever calls publish()
// and exactly one (possibly different) thread ever calls consume_next().
//                  Drops the per-slot atomic entirely.
// ---------------------------------------------------------------------
export template <typename EventType, std::size_t Capacity>
class SpscDispatcher {
private:
    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be a power of two.");
    static constexpr std::size_t Mask = Capacity - 1;

    alignas(64) std::byte storage_[Capacity][sizeof(EventType)];

    alignas(64) std::atomic<uint64_t> write_idx_{0};
    alignas(64) uint64_t cached_read_idx_{0};   // producer-owned cache, non-atomic

    alignas(64) std::atomic<uint64_t> read_idx_{0};
    alignas(64) uint64_t cached_write_idx_{0};  // consumer-owned cache, non-atomic

    static inline void cpu_relax() noexcept {
        #if defined(__x86_64__) || defined(_M_X64)
        __builtin_ia32_pause();
        #elif defined(__aarch64__)
        asm volatile("yield" ::: "memory");
        #endif
    }

public:
    SpscDispatcher() = default;

    ~SpscDispatcher() {
        uint64_t head = read_idx_.load(std::memory_order_relaxed);
        uint64_t tail = write_idx_.load(std::memory_order_relaxed);
        for (uint64_t i = head; i < tail; ++i) {
            std::destroy_at(reinterpret_cast<EventType*>(storage_[i & Mask]));
        }
    }

    SpscDispatcher(const SpscDispatcher&) = delete;
    SpscDispatcher& operator=(const SpscDispatcher&) = delete;

    // Producer thread only.
    template <typename... Args>
    void publish(Args&&... args) noexcept {
        uint64_t w = write_idx_.load(std::memory_order_relaxed);

        if (w - cached_read_idx_ == Capacity) {
            cached_read_idx_ = read_idx_.load(std::memory_order_acquire);
            while (w - cached_read_idx_ == Capacity) {
                cpu_relax();
                cached_read_idx_ = read_idx_.load(std::memory_order_acquire);
            }
        }

        ::new (static_cast<void*>(storage_[w & Mask])) EventType(std::forward<Args>(args)...);
        write_idx_.store(w + 1, std::memory_order_release);
    }

    // Consumer thread only.
    template <typename EventHandler>
    void consume_next(EventHandler&& handler) noexcept {
        uint64_t r = read_idx_.load(std::memory_order_relaxed);

        if (r == cached_write_idx_) {
            cached_write_idx_ = write_idx_.load(std::memory_order_acquire);
            while (r == cached_write_idx_) {
                cpu_relax();
                cached_write_idx_ = write_idx_.load(std::memory_order_acquire);
            }
        }

        auto* item = reinterpret_cast<EventType*>(storage_[r & Mask]);
        std::forward<EventHandler>(handler)(*item);
        std::destroy_at(item);

        read_idx_.store(r + 1, std::memory_order_release);
    }
};