ring_buffers.cpp

#include <array>
#include <atomic>
#include <optional>

template <typename T, size_t Capacity> class SCSPring {

  private:
    static constexpr size_t increment(size_t i) { return (i + 1) % Capacity; }

    alignas(64) std::array<T, Capacity> buffer;
    alignas(64) std::atomic<size_t> head{0};
    alignas(64) std::atomic<size_t> tail{0};

  public:
    bool push(const T& item) {

        size_t curr_tail = tail.load(std::memory_order_relaxed);
        size_t next = increment(curr_tail);

        size_t curr_head = head.load(std::memory_order_acquire);

        if (next == curr_head) {
            return false;
        }

        buffer[next] == item;
        tail.store(next, std::memory_order_release);
        return true;
    }

    T pop() {

        size_t curr_head = head.load(std::memory_order_relaxed);

        size_t curr_tail = tail.load(std::memory_order_acquire);

        if (curr_head == curr_tail) {
            return std::nullopt;
        }

        head.store(increment(head), std::memory_order_release);

        return buffer[curr_head];
    }
};

// not strictly serialziable
template <typename T, size_t Capacity> class MPMCRing {
  private:
    static_assert(Capacity > 0);

    struct Slot {

        T item{};
        std::atomic<size_t> published_seq{0};
        std::atomic<size_t> done_seq{0};
    };

    static constexpr size_t index(std::size_t pos) { return pos % Capacity; }

    alignas(64) std::array<Slot, Capacity> buffer{};

    alignas(64) std::atomic<std::size_t> write_claim_pos{0};
    alignas(64) std::atomic<std::size_t> publish_pos{0};
    alignas(64) std::atomic<std::size_t> read_claim_pos{0};
    alignas(64) std::atomic<std::size_t> reclaim_pos{0};

    void try_advance_publish() {
        size_t p = publish_pos.load(std::memory_order_relaxed);

        while (true) {
            Slot& s = buffer[index(p)];

            if (s.published_seq.load(std::memory_order_acquire) != p + 1) {
                return;
            }

            std::size_t expected = p;

            if (publish_pos.compare_exchange_weak(expected, p + 1, std::memory_order_release,
                                                  std::memory_order_relaxed)) {
                p++;
            } else {
                p = expected;
            }
        }
    }

    void try_advance_reclaim() {
        std::size_t r = reclaim_pos.load(std::memory_order_relaxed);

        while (true) {
            Slot& s = buffer[index(r)];

            if (s.done_seq.load(std::memory_order_acquire) != r + 1) {
                return;
            }

            std::size_t expected = r;

            if (reclaim_pos.compare_exchange_weak(expected, r + 1, std::memory_order_release,
                                                  std::memory_order_relaxed)) {
                r++;
            } else {
                r = expected;
            }
        }
    }

  public:
    bool try_push(const T& item) {
        std::size_t p;

        while (true) {

            p = write_claim_pos.load(std::memory_order_relaxed);
            std::size_t r = reclaim_pos.load(std::memory_order_acquire);
            if (p - r >= Capacity) {
                return false;
            }

            if (write_claim_pos.compare_exchange_weak(p, p + 1, std::memory_order_acq_rel,
                                                      std::memory_order_relaxed)) {
                break;
            }
        }

        Slot& s = buffer[index(p)];
        s.item = item;
        s.published_seq.store(p + 1, std::memory_order_release);

        try_advance_publish();

        return true;
    }

    std::optional<T> try_pop() {
        std::size_t c;

        while (true) {
            c = read_claim_pos.load(std::memory_order_relaxed);
            std::size_t pub = publish_pos.load(std::memory_order_acquire);

            if (c >= pub) {
                return std::nullopt;
            }

            if (read_claim_pos.compare_exchange_weak(c, c + 1, std::memory_order_acq_rel,
                                                     std::memory_order_relaxed)) {
                break;
            }
        }

        Slot& s = buffer[index(c)];

        if (s.published_seq.load(std::memory_order_acquire) != c + 1) {
            return std::nullopt;
        }

        T value = s.item;

        s.done_seq.store(c + 1, std::memory_order_release);

        try_advance_reclaim();

        return value;
    }
};