plotter/include/plot/data/ring_buffer.hpp

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#pragma once
#include <atomic>
#include <bit>
#include <vector>

#include <plot/core/types.hpp>

namespace plot::data {

/// Single-producer / single-consumer ring of at most `capacity` elements.
///
/// The capacity is rounded up to a power of two -- `capacity()` reports the
/// *rounded* value, not the requested one, and `size()` saturates there -- so
/// slot addressing is a mask instead of a modulo. Callers that need the
/// retained window's exact bound must read `capacity()`, never their own
/// constructor argument (see `StreamSeries`, which reserves the rounded value).
///
/// The producer writes a slot and then publishes the new
/// sample count with a release store; a consumer that observes that count with an
/// acquire load is guaranteed to see the slot contents written before it.
///
/// Slots are std::atomic and every slot access is relaxed. Because the ring
/// *overwrites*, the producer's next push reuses the slot of the snapshot's
/// oldest retained element while the consumer may still be reading it, so a plain
/// load/store there would be a data race (UB), not merely a stale value. A
/// relaxed atomic load/store is a plain mov/ldr/str on the supported targets, so
/// the sample path costs the same. Relaxed is enough because the ordering lives in
/// the count handshake above; the count fixes *which* indices a snapshot reads,
/// not that the slots still hold those samples (see StreamSeries).
///
/// A push into a full ring silently discards the oldest element: for a live
/// plotting stream the newest sample is the interesting one and the retained
/// window slides forward.
///
/// Two indexing conventions, deliberately named apart:
///  - retained-window (operator[], index 0 = oldest element still held) is a
///    single-threaded view: it re-derives the window from the live count, so a
///    producer pushing concurrently would move it under the reader's feet.
///  - absolute (base(), copy_from()) fixes one count for a whole read, which is
///    what a consumer holding a snapshot must use. StreamSeries::acquire() takes
///    such a snapshot and then reads only absolute indices.
template <typename T>
class RingBuffer {
  static_assert(std::atomic<T>::is_always_lock_free,
                "RingBuffer slots are atomic and must be lock-free: the sample path may not lock");
 public:
  explicit RingBuffer(u64 capacity)
      : cap_(std::bit_ceil(capacity < 1 ? u64{1} : capacity)),
        mask_(cap_ - 1),
        buf_(static_cast<usize>(cap_)) {}

  /// Producer thread only.
  void push(T v) {
    const u64 i = produced_.load(std::memory_order_relaxed);
    buf_[static_cast<usize>(i & mask_)].store(v, std::memory_order_relaxed);
    produced_.store(i + 1, std::memory_order_release);
  }

  /// Producer thread only. Equivalent to n calls to push().
  void push(const T* vs, u64 n) {
    for (u64 i = 0; i < n; ++i) push(vs[i]);
  }

  /// Producer thread only: rewrites the slot at absolute index `absolute`, without pushing.
  /// Returns false when that index is not retained -- already overwritten, or not yet
  /// produced -- in which case nothing is written.
  ///
  /// This is for *backfill*: a slot pushed as a placeholder and corrected when the data that
  /// belongs there arrives late. It is the same race class as `push()` -- one relaxed store
  /// into a slot a consumer may be reading, and T is lock-free by the class's static_assert,
  /// so a reader sees the old value or the new one and never a torn one. It deliberately does
  /// not touch `produced_`: the count fixes which indices a snapshot reads, and an amend
  /// changes a value inside that window rather than extending it.
  bool amend(u64 absolute, T v) {
    const u64 p = produced_.load(std::memory_order_relaxed);
    if (absolute >= p) return false;
    if (p > cap_ && absolute < p - cap_) return false;
    buf_[static_cast<usize>(absolute & mask_)].store(v, std::memory_order_relaxed);
    return true;
  }

  /// Consumer: retained element count, saturating at capacity().
  u64 size() const noexcept {
    const u64 p = produced_.load(std::memory_order_acquire);
    return p < cap_ ? p : cap_;
  }

  /// Consumer: total number of elements ever pushed (monotonic).
  u64 produced() const noexcept { return produced_.load(std::memory_order_acquire); }

  u64 capacity() const noexcept { return cap_; }

  /// Consumer: absolute index of the oldest element still retained. 0 until the
  /// ring has wrapped.
  u64 base() const noexcept {
    const u64 p = produced_.load(std::memory_order_acquire);
    return p > cap_ ? p - cap_ : 0;
  }

  /// Single-threaded: retained-window index (0 = oldest retained). `i` must be
  /// < size(). Returns by value: the slot is an atomic, so a relaxed load is the
  /// only way to observe it.
  T operator[](u64 i) const noexcept {
    return buf_[static_cast<usize>((base() + i) & mask_)].load(std::memory_order_relaxed);
  }

  /// Consumer: copies the `n` elements at absolute indices [absolute, absolute+n)
  /// into `out`, masking the index to wrap. The caller must hold a snapshot
  /// that retains those indices; this never consults the live count, so
  /// a pushing producer cannot shift the read -- but it fixes only *which*
  /// indices are read, not that the slots still hold those samples: a producer
  /// that overtakes capacity() has already rewritten the oldest ones (see the
  /// class note).
  void copy_from(u64 absolute, u64 n, T* out) const {
    if (n == 0) return;
    // Explicit relaxed loads: std::copy cannot read atomic slots, and the mask
    // makes the wrap split unnecessary.
    for (u64 i = 0; i < n; ++i) {
      out[i] = buf_[static_cast<usize>((absolute + i) & mask_)].load(std::memory_order_relaxed);
    }
  }

  /// Consumer thread only: discards every element (not safe while a producer is
  /// pushing).
  void clear() noexcept { produced_.store(0, std::memory_order_relaxed); }

  /// Consumer thread only: discards every element and resumes the absolute count
  /// at `absolute`, so the next push() lands there (not safe while a producer is
  /// pushing).
  ///
  /// This is for a *recompute*: a ring whose contents are derived from another is
  /// emptied and refilled from a given absolute origin, and it must come back
  /// numbered as its source is. clear() would restart it at 0 and leave every
  /// absolute index -- base(), produced(), copy_from() -- off by that origin.
  void clear_to(u64 absolute) noexcept {
    produced_.store(absolute, std::memory_order_relaxed);
  }

 private:
  u64 cap_;
  u64 mask_;
  std::vector<std::atomic<T>> buf_;
  std::atomic<u64> produced_{0};
};

}  // namespace plot::data