#pragma once #include #include #include #include namespace plot::data { /// Single-producer / single-consumer ring of at most `capacity` elements. /// /// The capacity is rounded up to a power of two so slot addressing is a mask /// instead of a modulo. 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 class RingBuffer { static_assert(std::atomic::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(cap_)) {} /// Producer thread only. void push(T v) { const u64 i = produced_.load(std::memory_order_relaxed); buf_[static_cast(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]); } /// 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((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((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); } private: u64 cap_; u64 mask_; std::vector> buf_; std::atomic produced_{0}; }; } // namespace plot::data