fix(data): level-per-emission pyramid cover, relaxed-atomic ring slots

cover_buckets chose one level per pass, so an odd first whole bucket (half of all

measurement ranges) never reached a level above 0 and folded every remaining

bucket at level 0 -- 15624 entries for a 1e6-sample pyramid. Pick the largest

aligned, in-range, materialised level at each cursor instead; the new cost test

counts folded entries through a PLOT_PYRAMID_COUNT_COVERS-only hook (absent from

normal builds) and fails on the old body.

Ring slots are now relaxed std::atomic: overwrite-on-full means the producer

reuses the slot of a snapshot's oldest sample while the consumer reads it, which

was a data race, not just a stale value. copy_from loads explicitly instead of

std::copy; push stays lock-free. A new producer/consumer case is the one TSan

flags without this.

Also extends the wrapped-window test with x0/bounds/pyramid/sum/sumsq/gather

checks, and corrects the StreamSeries threading note to the sample-count bound.

CMakeLists.txt

@@ -22,6 +22,11 @@ target_link_libraries(plot_core PRIVATE ${CMAKE_DL_LIBS})
 plot_apply_warnings(plot_core)
 
 if(PLOT_BUILD_TESTS)
+  # Test-only: counts the pyramid's covering-decomposition entries so a test can
+  # pin its O(log n) cost, which no query answer reveals. Off in every non-test
+  # build (see include/plot/data/detail/cover_count.hpp).
+  target_compile_definitions(plot_core PRIVATE PLOT_PYRAMID_COUNT_COVERS)
+
   enable_testing()
   add_subdirectory(tests)
 endif()

include/plot/data/detail/cover_count.hpp

@@ -0,0 +1,29 @@
+#pragma once
+
+#ifdef PLOT_PYRAMID_COUNT_COVERS
+
+#include <atomic>
+
+#include <plot/core/types.hpp>
+
+namespace plot::data::detail {
+
+/// Test-only instrumentation. Counts the pyramid entries the covering
+/// decomposition has emitted since the counter was last reset.
+///
+/// The decomposition's *cost* is invisible in every answer -- a decomposition
+/// that walks the range one level-0 bucket at a time returns exactly the same
+/// min/max/sum as the O(log n) one -- so a test that wants to pin "at most two
+/// entries per level" has to count entries. The counter exists only when
+/// PLOT_PYRAMID_COUNT_COVERS is defined (the test build defines it), so every
+/// normal build carries no counter and no increment on the query path.
+///
+/// Not part of the query surface and not meant for production use.
+inline std::atomic<u64>& cover_entry_count() {
+  static std::atomic<u64> count{0};
+  return count;
+}
+
+}  // namespace plot::data::detail
+
+#endif  // PLOT_PYRAMID_COUNT_COVERS

include/plot/data/ring_buffer.hpp

@@ -1,8 +1,6 @@
 #pragma once
-#include <algorithm>
 #include <atomic>
 #include <bit>
-#include <cstddef>
 #include <vector>
 
 #include <plot/core/types.hpp>
@@ -16,6 +14,15 @@ namespace plot::data {
 /// 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.
@@ -29,6 +36,8 @@ namespace plot::data {
 ///    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)),
@@ -38,7 +47,7 @@ class RingBuffer {
   /// Producer thread only.
   void push(T v) {
     const u64 i = produced_.load(std::memory_order_relaxed);
-    buf_[static_cast<usize>(i & mask_)] = v;
+    buf_[static_cast<usize>(i & mask_)].store(v, std::memory_order_relaxed);
     produced_.store(i + 1, std::memory_order_release);
   }
 
@@ -66,24 +75,25 @@ class RingBuffer {
   }
 
   /// Single-threaded: retained-window index (0 = oldest retained). `i` must be
-  /// < size().
+  /// < size(). Returns by value: the slot is an atomic, so a relaxed load is the
-  const T& operator[](u64 i) const noexcept {
+  /// only way to observe it.
-    return buf_[static_cast<usize>((base() + i) & mask_)];
+  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`, splitting the copy at the ring's wrap point. The caller must hold
+  /// into `out`, masking the index to wrap. The caller must hold a snapshot
-  /// a snapshot that retains those indices; this never consults the live count, so
+  /// that retains those indices; this never consults the live count, so
-  /// a pushing producer cannot shift the read.
+  /// 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;
-    const u64 s = absolute & mask_;
+    // Explicit relaxed loads: std::copy cannot read atomic slots, and the mask
-    const u64 run = std::min(n, cap_ - s);
+    // makes the wrap split unnecessary.
-    const auto first = buf_.begin() + static_cast<std::ptrdiff_t>(s);
+    for (u64 i = 0; i < n; ++i) {
-    std::copy(first, first + static_cast<std::ptrdiff_t>(run), out);
+      out[i] = buf_[static_cast<usize>((absolute + i) & mask_)].load(std::memory_order_relaxed);
-    if (run < n) {
-      std::copy(buf_.begin(), buf_.begin() + static_cast<std::ptrdiff_t>(n - run),
-                out + static_cast<std::ptrdiff_t>(run));
     }
   }
 
@@ -94,7 +104,7 @@ class RingBuffer {
  private:
   u64 cap_;
   u64 mask_;
-  std::vector<T> buf_;
+  std::vector<std::atomic<T>> buf_;
   std::atomic<u64> produced_{0};
 };
 

include/plot/data/stream_series.hpp

@@ -30,15 +30,23 @@ namespace plot::data {
 ///
 /// THREADING NOTE. The ring overwrites the oldest samples -- that is what keeps a
 /// live stream showing the newest data, and what the drop test pins down -- so a
-/// producer racing past capacity() can write the very slot the consumer is reading
+/// producer can overwrite the slots a consumer is reading even while it holds a
-/// at the window's oldest edge. That read is not ordered against the write, so a
+/// snapshot. Those slots are relaxed atomics (see RingBuffer), so this is not a
-/// race detector flags it: the value is still a single f32 (no tearing on
+/// data race and no value tears; what it is not is a frozen window.
-/// supported targets) and the worst case is one stale or too-new sample at the
+///
-/// oldest edge, corrected by the next acquire() once it sees the base moved. The
+/// The bound is the sample count, not the geometry. A snapshot count fixes *which
-/// pyramid itself never mixes windows, because acquire() derives the whole window
+/// absolute indices* a read walks, not that the slots still hold those samples:
-/// from one count. Ingest that is race-free even at the edge needs the producer to
+/// every push after the snapshot overwrites exactly one slot, the one belonging to
-/// stop before unread data (drop-on-full) or a claim protocol, i.e. a different
+/// the sample capacity() positions behind it. So if the producer pushes k samples
-/// contract from the one this class specifies.
+/// while a read is in flight, the first k positions of that read window can hold
+/// one of those k newer samples (or a still-newer sample, once the producer has
+/// lapped again) instead of the sample the snapshot names. Nothing else moves: the
+/// count, the absolute index -> x mapping and the newest end of the window are
+/// derived from the one published count, and the next acquire() rebuilds from the
+/// base it observes, so the pyramid never mixes two windows. Ingest with no reuse
+/// underneath a read at all needs the producer to stop before unread data
+/// (drop-on-full) or a claim protocol, i.e. a different contract from the one this
+/// class specifies.
 class StreamSeries final : public Series {
  public:
   explicit StreamSeries(u64 capacity, f64 x0, f64 dx);

src/data/minmax_pyramid.cpp

@@ -1,3 +1,4 @@
+#include <plot/data/detail/cover_count.hpp>
 #include <plot/data/minmax_pyramid.hpp>
 
 #include <algorithm>
@@ -26,22 +27,40 @@ constexpr u64 ceil_to_bucket(u64 v) {
 /// A level-l entry is emitted only when the whole level-l bucket exists (all
 /// kBucket << l samples are present, so its aggregate is final). Callers
 /// guarantee b1 <= ceil(size() / kBucket), so the only possible remainder is the
-/// single trailing partial level-0 bucket, which is emitted at level 0. The
+/// single trailing partial level-0 bucket, which is emitted at level 0.
-/// decomposition has O(log) entries, so every query below is O(log n).
+///
+/// The level is chosen per *emission*, not once per pass: at each cursor the
+/// largest level is taken whose bucket is aligned with the cursor, lies wholly
+/// inside [b0, b1) and has already been materialised. A cursor that starts
+/// misaligned (an odd first whole bucket, i.e. half of all measurement ranges)
+/// therefore burns at most one level-0 bucket and then resumes the coarse
+/// levels, so the decomposition is at most two entries per level and every query
+/// below is O(log n) for any start. Choosing the level once per pass instead left
+/// the level-0 loop to consume the whole remainder one bucket at a time, because
+/// no coarser level is ever reachable from an odd cursor.
 template <typename Fn>
 void cover_buckets(const MinMaxPyramid& p, u64 b0, u64 b1, Fn&& fn) {
   const u64 complete = p.size() / kBucket;
+  const u64 limit = std::min(b1, complete);
+  const u32 top = p.level_count() - 1;         // level_count() is at least 1
   u64 b = b0;
-  for (u32 li = p.level_count(); li-- > 0 && b < b1;) {
+  while (b < b1) {
-    const u64 lb = u64{1} << li;               // level-0 buckets per level-li bucket
+    u32 li = top;
-    const u64 avail = p.level(li).size();
+    for (;;) {
-    while (b < b1 && (b % lb) == 0 && b + lb <= b1 && b + lb <= complete &&
+      const u64 lb = u64{1} << li;
-           (b / lb) < avail) {
+      if ((b % lb) == 0 && b + lb <= limit &&
-      fn(li, b / lb);
+          (b / lb) < static_cast<u64>(p.level(li).size())) {
-      b += lb;
+        break;
       }
+      if (li == 0) break;
+      --li;
+    }
+#ifdef PLOT_PYRAMID_COUNT_COVERS
+    detail::cover_entry_count().fetch_add(1, std::memory_order_relaxed);
+#endif
+    fn(li, b >> li);
+    b += u64{1} << li;
   }
-  if (b < b1) fn(0u, b);                       // trailing partial level-0 bucket
 }
 
 /// Folds the decomposition of [b0, b1) into a min/max Range. Returns a

tests/CMakeLists.txt

@@ -22,6 +22,8 @@ else()
     "or mesa-libEGL-devel) or configure with -DPLOT_BUILD_TESTS=OFF.")
 endif()
 
+find_package(Threads REQUIRED)   # std::thread in the push/read race test
+
 add_executable(plot_tests
   main.cpp
   test_transform.cpp
@@ -31,7 +33,10 @@ add_executable(plot_tests
   test_series.cpp
   egl_headless.cpp
 )
-target_link_libraries(plot_tests PRIVATE plot::core ${plot_egl_library})
+target_link_libraries(plot_tests PRIVATE plot::core ${plot_egl_library} Threads::Threads)
+# Same switch the library is built with: the decomposition-cost test reads the
+# counter behind it (see include/plot/data/detail/cover_count.hpp).
+target_compile_definitions(plot_tests PRIVATE PLOT_PYRAMID_COUNT_COVERS)
 plot_apply_warnings(plot_tests)
 
 add_test(NAME plot_tests COMMAND plot_tests)

tests/test_minmax_pyramid.cpp

@@ -1,4 +1,5 @@
 #include "harness.hpp"
+#include <plot/data/detail/cover_count.hpp>
 #include <plot/data/minmax_pyramid.hpp>
 #include <random>
 
@@ -104,3 +105,26 @@ PLOT_TEST(pyramid_clear_and_rebuild) {
   PLOT_CHECK(p.size() == 1);
   PLOT_CHECK_NEAR(p.reduce(0, 1).max, 7.0, 1e-6);
 }
+
+PLOT_TEST(pyramid_cover_cost_is_logarithmic_for_unaligned_starts) {
+  // Answers cannot show this: a decomposition that walks the range one level-0
+  // bucket at a time returns exactly the same min/max as the O(log n) one. So
+  // count the folded aggregates. For this query the bad pass emitted 15624
+  // entries (the level-0 loop consumed the whole remainder); picking a level per
+  // emission leaves 19, at most two per level, and the aligned start stays at 7.
+  MinMaxPyramid p;
+  p.reserve(1'000'000);
+  for (u64 i = 0; i < 1'000'000; ++i) p.append(0.0f);
+  PLOT_CHECK(p.level(0).size() == 1'000'000 / kBucket);   // 15625, no partial bucket
+
+  auto folds_for = [&](u64 begin, u64 end) {
+    plot::data::detail::cover_entry_count().store(0, std::memory_order_relaxed);
+    (void)p.reduce_buckets(begin, end);
+    return plot::data::detail::cover_entry_count().load(std::memory_order_relaxed);
+  };
+  const u64 aligned = folds_for(0, 1'000'000);     // first whole bucket 0
+  const u64 unaligned = folds_for(1, 1'000'000);   // first whole bucket 1: odd
+  PLOT_CHECK(aligned <= 64);
+  PLOT_CHECK(unaligned < 100);
+  PLOT_CHECK(unaligned <= u64{4} * aligned);       // misalignment costs entries, not a scan
+}

tests/test_series.cpp

@@ -4,6 +4,7 @@
 #include <algorithm>
 #include <cmath>
 #include <random>
+#include <thread>
 #include <vector>
 
 PLOT_TEST(array_series_bounds_and_gather) {
@@ -69,10 +70,61 @@ PLOT_TEST(array_series_exact_range_queries_match_bruteforce) {
 
 PLOT_TEST(stream_series_exact_queries_survive_ring_wrap) {
   plot::data::StreamSeries s(1u << 10, 0.0, 1.0);
-  for (int i = 0; i < 3000; ++i) s.push(f32(i % 97));   // wraps the ring twice
+  constexpr u64 kPushes = 3000;
+  constexpr u64 kFirstRetained = kPushes - (1u << 10);  // 1976
+  for (u64 i = 0; i < kPushes; ++i) s.push(f32(i % 97));   // wraps the ring twice
   s.acquire();
   const u64 n = s.size();
+  PLOT_CHECK(n == 1u << 10);
+  PLOT_CHECK_NEAR(s.x0(), 1976.0, 1e-9);
+  PLOT_CHECK(s.x_bounds().min == s.x0());
+  PLOT_CHECK(s.pyramid().size() == n);
+
   const plot::Range r = s.reduce(0, n);                 // retained window, wrapped
   PLOT_CHECK_NEAR(r.max, 96.0, 1e-4);
   PLOT_CHECK_NEAR(r.min, 0.0, 1e-4);
+
+  double sum = 0.0, sumsq = 0.0;
+  for (u64 i = kFirstRetained; i < kPushes; ++i) {
+    const double v = static_cast<double>(i % 97);
+    sum += v;
+    sumsq += v * v;
+  }
+  PLOT_CHECK_NEAR(s.sum(0, n), sum, 1e-2);
+  PLOT_CHECK_NEAR(s.sumsq(0, n), sumsq, 1e-1);
+
+  std::vector<f32> out;
+  s.gather(0, 0, n, out);                               // level 0: raw samples
+  PLOT_CHECK(out.size() == 2 * n);
+  PLOT_CHECK_NEAR(out[0], s.x0(), 1e-6);
+  PLOT_CHECK_NEAR(out[1], f32(kFirstRetained % 97), 1e-6);
+  PLOT_CHECK_NEAR(out[2 * n - 2], s.x0() + f64(n - 1), 1e-6);
+}
+
+PLOT_TEST(stream_series_push_and_read_are_race_free) {
+  // The ring overwrites, so the producer reuses the slot of the snapshot's oldest
+  // retained sample while the consumer may still be reading it. Relaxed atomic
+  // slots make that defined; nothing else in the suite has two threads, so this
+  // is the case ThreadSanitizer actually flags when that is not the case.
+  plot::data::StreamSeries s(1u << 8, 0.0, 1.0);
+  constexpr u64 kPushes = 50'000;
+  std::thread producer([&s] {
+    for (u64 i = 0; i < kPushes; ++i) s.push(f32(i % 971));
+  });
+  u64 reads = 0;
+  while (s.produced() < kPushes) {
+    s.acquire();
+    const u64 n = s.size();
+    if (n == 0) continue;
+    (void)s.reduce(0, n);
+    (void)s.sum(0, n);
+    (void)s.sumsq(0, n);
+    std::vector<f32> out;
+    s.gather(0, 0, n, out);                             // reads the oldest edge
+    PLOT_CHECK(out.size() == 2 * n);
+    ++reads;
+  }
+  producer.join();
+  PLOT_CHECK(s.produced() == kPushes);
+  PLOT_CHECK(reads > 0);
 }