mpmc.hpp

/*
 * rtchan - Realtime-safe channels via lock-free ringbuffer queues
 * Copyright (c) 2025 Fawn <rubiefawn@gmail.com>
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED “AS IS” AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */
 
#pragma once
#include "rtchan.hpp"
#include "internal/busy_wait_hint.h"
#include <array>
#include <bit> // std::has_single_bit() to ensure the buffer length is a power of 2
#include <utility>
 
namespace rtchan {

template<typename T, size_t N>
requires (0 < N) && (std::has_single_bit(N))
class mpmc : public chan<T> {
public:
    auto try_send(T&& item) noexcept -> bool override {
        size_t index, next_index;
 
        size_t write_claimed_expected = write_claimed.load(std::memory_order_relaxed);
        do {
            size_t last_read_idx = read_committed.load(std::memory_order_relaxed);
            const ptrdiff_t occupied = write_claimed_expected - last_read_idx;
            if (occupied >= N) { return false; } // No space remaining
            index = write_claimed_expected; next_index = write_claimed_expected + 1;
        } while (!write_claimed.compare_exchange_strong(write_claimed_expected, next_index)); // TODO: Memory order for success/failure
 
        buf[index_wrapped(index)] = std::move(item);
 
        size_t write_committed_expected = index;
        while (!write_committed.compare_exchange_strong(write_committed_expected, next_index)) { // TODO: Memory order for success/failure
            write_committed_expected = index; // Reset this as the CAS will have changed it
            busy_wait_hint();
        }
 
        return true;
    }
 
    auto try_send(span<const T>&& items) noexcept -> span<const T> override {
        size_t count = 0;
        size_t index, next_index;
 
        size_t write_claimed_expected = write_claimed.load(std::memory_order_relaxed);
        do {
            size_t last_read_idx = read_committed.load(std::memory_order_relaxed);
            const ptrdiff_t occupied = write_claimed_expected - last_read_idx;
            if (occupied >= N) { return items; } // No space remaining
            count = std::min(items.size(), static_cast<size_t>(N - occupied));
            index = write_claimed_expected; next_index = write_claimed_expected + count;
        } while (!write_claimed.compare_exchange_strong(write_claimed_expected, next_index)); // TODO: Memory order for success/failure
 
        const size_t i = index_wrapped(index);
        const ptrdiff_t spill = i + count - N;
        if (spill > 0) {
            const auto fits = static_cast<size_t>(count - spill);
            std::memcpy(&buf[i], items.data(), sizeof(T) * fits);
            std::memcpy(buf.data(), &items[fits], sizeof(T) * spill);
        } else { std::memcpy(&buf[i], items.data(), sizeof(T) * count); }
 
        size_t write_committed_expected = index;
        while (!write_committed.compare_exchange_strong(write_committed_expected, next_index)) { // TODO: Memory order for success/failure
            write_committed_expected = index; // Reset this as the CAS will have changed it
            busy_wait_hint();
        }
 
        return items.last(items.size() - count);
    }
 
    void send(T&& item) noexcept override {
        
    }
 
    void send(span<const T>&& items) noexcept override {
        
    }
 
    auto try_recv(T& dest) noexcept -> bool override {
        size_t index, next_index;
 
        size_t read_claimed_expected = read_claimed.load(std::memory_order_relaxed);
        do {
            size_t last_written_idx = write_committed.load(std::memory_order_relaxed);
            const ptrdiff_t ready = last_written_idx - read_claimed_expected;
            if (0 >= ready) { return false; } // Nothing available to receive
            index = read_claimed_expected, next_index = read_claimed_expected + 1;
        } while (!read_claimed.compare_exchange_strong(read_claimed_expected, next_index)); // TODO: Memory order for success/failure
 
        dest = std::move(&buf[index_wrapped(index)]);
 
        size_t read_committed_expected = index;
        while (!read_committed.compare_exchange_strong(read_committed_expected, next_index)) { // TODO: Memory order for success/failure
            read_committed_expected = index; // Reset this as the CAS will have changed it
            busy_wait_hint();
        }
 
        return true;
    }
 
    auto try_recv(span<T>& dest) noexcept -> span<T> override {
        size_t count = 0;
        size_t index, next_index;
 
        size_t read_claimed_expected = read_claimed.load(std::memory_order_relaxed);
        do {
            size_t last_written_idx = write_committed.load(std::memory_order_relaxed);
            const ptrdiff_t ready = last_written_idx - read_claimed_expected;
            if (0 >= ready) { return 0; } // Nothing available to receive
            count = std::min(dest.size(), static_cast<size_t>(ready));
            index = read_claimed_expected, next_index = read_claimed_expected + count;
        } while (!read_claimed.compare_exchange_strong(read_claimed_expected, next_index)); // TODO: Memory order for success/failure
 
        const size_t i = index_wrapped(index);
        const ptrdiff_t spill = i + count - N;
        if (spill > 0) {
            const auto fits = static_cast<size_t>(count - spill);
            std::memcpy(dest.data(), &buf[i], sizeof(T) * fits);
            std::memcpy(&dest[fits], buf.data(), sizeof(T) * spill);
        } else { std::memcpy(dest.data(), &buf[i], sizeof(T) * count); }
 
        size_t read_committed_expected = index;
        while (!read_committed.compare_exchange_strong(read_committed_expected, next_index)) { // TODO: Memory order for success/failure
            read_committed_expected = index; // Reset this as the CAS will have changed it
            busy_wait_hint();
        }
 
        return dest.first(count);
    }
 
    void recv(T& dest) noexcept override {
        
    }
 
    void recv(span<T>& dest) noexcept override {
        
    }
    
    auto try_reserve_send(size_t count) noexcept -> span<T> override {
        
    }
 
    auto reserve_send_up_to(size_t count) noexcept -> span<T> override {
        
    }
 
    auto reserve_send(size_t count) noexcept -> span<T> override {
        
    }
 
    auto try_commit_send(span<const T>& reservation) noexcept -> bool override {
        
    }
 
    void commit_send(span<const T>& reservation) noexcept override {
        
    }
 
    auto try_reserve_recv(size_t count) noexcept -> span<T> override {
        
    }
 
    auto reserve_recv_up_to(size_t count) noexcept -> span<T> override {
        
    }
 
    auto reserve_recv(size_t count) noexcept -> span<T> override {
        
    }
 
    auto try_commit_recv(span<const T>& reservation) noexcept -> bool override {
        
    }
 
    void commit_recv(span<const T>& reservation) noexcept override {
        
    }
    
private:
    static constexpr auto index_wrapped(size_t index) noexcept -> size_t { return index & (N - 1); };
    std::array<T, N> buf = {};
    alignas(std::hardware_destructive_interference_size) std::atomic_size_t read_committed = 0;
    alignas(std::hardware_destructive_interference_size) std::atomic_size_t read_claimed = 0;
    alignas(std::hardware_destructive_interference_size) std::atomic_size_t write_committed = 0;
    alignas(std::hardware_destructive_interference_size) std::atomic_size_t write_claimed = 0;
};
 
} // namespace rtchan