perf(data): absolute-space pyramid so acquire() is O(produced), not O(window)

Frame cost was O(capacity x levels) once the ring saturated: acquire() saw the
window base move (it always moves once the ring wraps), cleared the pyramid and
re-appended all 1,048,576 retained samples, and append() walked every live level
with no early stop -- 13,647,936 bucket merges and 73.8 ms per frame at a
1M-sample window, scaling with the window, which is what capped the 1 MS/s demo
at 32-58 ms/frame while it was still filling.

Index the pyramid in absolute bucket space instead. Level l bucket k covers the
samples [k * (kBucket << l), (k + 1) * (kBucket << l)), so a bucket's span never
depends on where the window starts and the window can slide over the aggregate
instead of forcing it to be rebuilt. Each level keeps its complete buckets in a
deque plus the absolute base of its first entry; append() folds a sample into the
single in-progress level-0 bucket and, only when that bucket completes, cascades
upward, stopping at the first level that gains no new entry; a new drop_before()
slides the window by discarding leading buckets and remembers where the window
begins. acquire() is now: drop to the new base, then append exactly what was
produced -- 256 merges and 0.034 ms per frame for 8192 new samples, independent
of the window, and memory bounded by the window (32,767 entries after 50M samples
through a 1M ring, the same as after 10M). A producer that laps the consumer by
more than capacity() between two acquires restarts the window rather than
splicing a gap.

Interface: `level(l) -> std::span<const MinMax>` cannot survive a ring and is
replaced by level_base/level_size/bucket in absolute bucket coordinates, plus
window_base/window_end/drop_before. The *_buckets queries stay in window
coordinates (index 0 = oldest retained sample), so Series keeps its window
semantics for callers; its exactness split now aligns in absolute space
(window_base() + the ceil/floor) because the bucket grid is absolute. The sums
move from per-level prefix sums to per-bucket ones -- the only representation a
sliding front can keep exact -- and StreamSeries::y_bounds() becomes
reduce(0, size()), exact in O(log n + kBucket) per frame instead of a scan of the
window. The pyramid owns the published window now, so there is no second copy of
base/end to keep in step, and gather() emits only buckets that are both complete
and wholly retained.

Tests: test-only merge counters (mirroring PLOT_PYRAMID_COUNT_COVERS, absent from
every non-test build) pin an acquire proportional to production and independent
of capacity, an append cost amortised over 16 live levels, and bounded memory;
brute-force checks pin absolute bucket aggregates across slides, sliding-window
reduce/sum/sumsq over unaligned ranges, and the level-1 gather mapping. The demo
now draws 946 samples/px at 1.03 ms/frame (967 fps) where the saturated window
cost 54.96 ms/frame (18 fps).

tests/CMakeLists.txt

@@ -40,9 +40,10 @@ add_executable(plot_tests
   egl_headless.cpp
 )
 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)
+# Same switches the library is built with: the cost tests read the counters
+# behind them (see include/plot/data/detail/{cover,merge}_count.hpp).
+target_compile_definitions(plot_tests PRIVATE PLOT_PYRAMID_COUNT_COVERS
+                                             PLOT_PYRAMID_COUNT_MERGES)
 plot_apply_warnings(plot_tests)
 
 add_test(NAME plot_tests COMMAND plot_tests)

tests/test_minmax_pyramid.cpp

@@ -1,12 +1,36 @@
 #include "harness.hpp"
 #include <plot/data/detail/cover_count.hpp>
+#include <plot/data/detail/merge_count.hpp>
 #include <plot/data/minmax_pyramid.hpp>
+#include <algorithm>
 #include <random>
 
 using plot::data::kBucket;
+using plot::data::MinMax;
 using plot::data::MinMaxPyramid;
 using plot::Range;
 
+namespace {
+
+/// Brute-force aggregate of the absolute span of level-`l` bucket `b`, from the
+/// same value function the pyramid was fed. This is the oracle the absolute
+/// bucket indexing is judged against: a bucket that has slid inside the window
+/// must still aggregate exactly its own absolute samples.
+template <typename ValueFn>
+MinMax brute_bucket(u32 l, u64 b, const ValueFn& value) {
+  const u64 span = u64{kBucket} << l;
+  const u64 first = b * span;
+  MinMax m{value(first), value(first)};
+  for (u64 i = first + 1; i < first + span; ++i) {
+    const f32 v = value(i);
+    m.min = std::min(m.min, v);
+    m.max = std::max(m.max, v);
+  }
+  return m;
+}
+
+}  // namespace
+
 PLOT_TEST(pyramid_bucket_extremes_preserved) {
   MinMaxPyramid p;
   p.reserve(1000);
@@ -107,7 +131,8 @@ PLOT_TEST(pyramid_clear_and_rebuild) {
   // bucket and ask the bucket-grained query for its extrema.
   for (u32 i = 1; i < kBucket; ++i) p.append(1.0f);
   PLOT_CHECK(p.size() == kBucket);
-  PLOT_CHECK(p.level(0).size() == 1);
+  PLOT_CHECK(p.level_base(0) == 0);
+  PLOT_CHECK(p.level_size(0) == 1);
   PLOT_CHECK_NEAR(p.reduce_buckets(0, kBucket).max, 7.0, 1e-6);
   PLOT_CHECK_NEAR(p.reduce_buckets(0, kBucket).min, 1.0, 1e-6);
 }
@@ -121,7 +146,7 @@ PLOT_TEST(pyramid_cover_cost_is_logarithmic_for_unaligned_starts) {
   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
+  PLOT_CHECK(p.level_size(0) == 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);
@@ -134,3 +159,131 @@ PLOT_TEST(pyramid_cover_cost_is_logarithmic_for_unaligned_starts) {
   PLOT_CHECK(unaligned < 100);
   PLOT_CHECK(unaligned <= u64{4} * aligned);       // misalignment costs entries, not a scan
 }
+
+PLOT_TEST(pyramid_drop_before_slides_the_window) {
+  MinMaxPyramid p;
+  for (u64 i = 0; i < 4096; ++i) p.append(f32(i % 13));
+  PLOT_CHECK(p.window_base() == 0);
+  PLOT_CHECK(p.window_end() == 4096);
+  PLOT_CHECK(p.size() == 4096);
+
+  p.drop_before(1000);
+  PLOT_CHECK(p.window_base() == 1000);
+  PLOT_CHECK(p.size() == 3096);
+  // The window's first *whole* bucket starts at 1024: the bucket the drop cut
+  // into is not representable as a whole bucket and is never queried.
+  PLOT_CHECK(p.level_base(0) == 1024 / kBucket);
+  PLOT_CHECK(p.level_base(0) + p.level_size(0) == 4096 / kBucket);
+
+  f32 lo = 1.0f, hi = -1.0f;
+  for (u64 i = 1024; i < 4096; ++i) {
+    lo = std::min(lo, f32(i % 13));
+    hi = std::max(hi, f32(i % 13));
+  }
+  const Range r = p.reduce_buckets(0, p.size());
+  PLOT_CHECK_NEAR(r.min, lo, 1e-6);
+  PLOT_CHECK_NEAR(r.max, hi, 1e-6);
+  // Window [1000, 1024) holds no whole bucket at all: an empty answer, not one
+  // that quietly reaches into the dropped samples.
+  PLOT_CHECK(p.reduce_buckets(0, 24).span() == 0.0);
+  PLOT_CHECK(p.sum_buckets(0, 24) == 0.0);
+  PLOT_CHECK(p.sumsq_buckets(0, 24) == 0.0);
+
+  p.drop_before(1000);                    // at the base: no-op
+  PLOT_CHECK(p.window_base() == 1000);
+  p.drop_before(4096);                    // at the end: the window is empty
+  PLOT_CHECK(p.size() == 0);
+  p.append(5.0f);                         // and appending starts a fresh window
+  PLOT_CHECK(p.size() == 1);
+  PLOT_CHECK(p.window_base() == 4096);
+}
+
+PLOT_TEST(pyramid_bucket_aggregates_stay_absolute_across_slides) {
+  // The invariant the sliding window rests on: a bucket that has slid inside the
+  // window is never recomputed, so it must already hold the aggregate of its own
+  // absolute span -- at every level, after any number of drops.
+  MinMaxPyramid p;
+  const auto value = [](u64 i) { return f32((i * 37u) % 101u) - 50.0f; };
+  constexpr u64 kTotal = 20'000;
+  constexpr u64 kWindow = 4'000;
+  for (u64 i = 0; i < kTotal; ++i) {
+    p.append(value(i));
+    if ((i + 1) % 997 == 0 && i + 1 > kWindow) p.drop_before(i + 1 - kWindow);
+  }
+  // The last drop is the last multiple of 997 at or below kTotal.
+  PLOT_CHECK(p.window_base() == (kTotal / 997) * 997 - kWindow);
+  for (u32 l = 0; l < p.level_count(); ++l) {
+    PLOT_CHECK(p.level_size(l) > 0);
+    const u64 span = u64{kBucket} << l;
+    for (u64 b = p.level_base(l); b < p.level_base(l) + p.level_size(l); ++b) {
+      const MinMax want = brute_bucket(l, b, value);
+      const MinMax got = p.bucket(l, b);
+      PLOT_CHECK_NEAR(got.min, want.min, 1e-6);
+      PLOT_CHECK_NEAR(got.max, want.max, 1e-6);
+      PLOT_CHECK(b * span >= p.window_base());          // wholly inside the window
+      PLOT_CHECK((b + 1) * span <= p.window_end());     // and complete
+    }
+  }
+}
+
+PLOT_TEST(pyramid_sliding_window_bucket_queries_match_bruteforce) {
+  MinMaxPyramid p;
+  const auto value = [](u64 i) { return f32(i % 89) * 0.25f - 5.0f; };
+  constexpr u64 kTotal = 12'000;
+  constexpr u64 kWindow = 3'000;
+  for (u64 i = 0; i < kTotal; ++i) {
+    p.append(value(i));
+    if ((i + 1) % 613 != 0) continue;
+    const u64 abs0 = i + 1 > kWindow ? i + 1 - kWindow : 0;
+    p.drop_before(abs0);
+    const u64 n = p.size();
+    // Window coordinates: window index j is absolute abs0 + j.
+    const u64 first = abs0 / kBucket + (abs0 % kBucket == 0 ? u64{0} : u64{1});
+    const u64 last = (abs0 + n) / kBucket;
+    if (last <= first) continue;
+    f32 lo = 1e30f, hi = -1e30f;
+    f64 sum = 0.0, sq = 0.0;
+    for (u64 a = first * kBucket; a < last * kBucket; ++a) {
+      const f32 v = value(a);
+      lo = std::min(lo, v);
+      hi = std::max(hi, v);
+      sum += f64(v);
+      sq += f64(v) * f64(v);
+    }
+    const u64 wb = first * kBucket - abs0;
+    const u64 we = last * kBucket - abs0;
+    const Range r = p.reduce_buckets(wb, we);
+    PLOT_CHECK_NEAR(r.min, lo, 1e-6);
+    PLOT_CHECK_NEAR(r.max, hi, 1e-6);
+    PLOT_CHECK_NEAR(p.sum_buckets(wb, we), sum, 1e-3);
+    PLOT_CHECK_NEAR(p.sumsq_buckets(wb, we), sq, 1e-1);
+  }
+}
+
+PLOT_TEST(pyramid_append_cost_is_amortised_constant) {
+  // A sample must not walk the live levels: level l gains an entry once per
+  // kBucket << l samples, so N appends perform ~N/32 merges however many levels
+  // the pyramid holds. Walking every level per append costs ~N * levels.
+  MinMaxPyramid p;
+  p.reserve(1u << 20);
+  constexpr u64 kAppends = 1u << 20;
+  for (u64 i = 0; i < kAppends; ++i) p.append(0.5f);
+  PLOT_CHECK(p.level_count() >= 10);   // 2^20 samples: 16384 level-0 buckets
+
+  plot::data::detail::merge_count().store(0, std::memory_order_relaxed);
+  for (u64 i = 0; i < kAppends; ++i) p.append(0.5f);
+  const u64 merges = plot::data::detail::merge_count().load(std::memory_order_relaxed);
+  // ~N/64 + N/128 + ... = ~N/32; the bound leaves 2x room, and is four orders of
+  // magnitude below the N * live_levels an unbounded walk would perform.
+  PLOT_CHECK(merges <= kAppends / 16);
+
+  // Sliding must not add per-sample work either: the appends below run against a
+  // window whose base moves with every sample.
+  plot::data::detail::merge_count().store(0, std::memory_order_relaxed);
+  for (u64 i = 0; i < kAppends; ++i) {
+    p.drop_before(i + 1);
+    p.append(0.5f);
+  }
+  const u64 slid = plot::data::detail::merge_count().load(std::memory_order_relaxed);
+  PLOT_CHECK(slid <= kAppends / 16);
+}

tests/test_series.cpp

@@ -1,13 +1,18 @@
 #include "harness.hpp"
 #include <plot/data/array_series.hpp>
+#include <plot/data/detail/merge_count.hpp>
 #include <plot/data/stream_series.hpp>
 #include <algorithm>
+#include <array>
 #include <atomic>
+#include <chrono>
 #include <cmath>
 #include <random>
 #include <thread>
 #include <vector>
 
+using plot::data::kBucket;
+
 PLOT_TEST(array_series_bounds_and_gather) {
   std::vector<f32> y{1, 5, -2, 3};
   plot::data::ArraySeries s(std::move(y), /*x0=*/10.0, /*dx=*/0.5);
@@ -159,3 +164,184 @@ PLOT_TEST(stream_series_push_and_read_are_race_free) {
   PLOT_CHECK(reads >= 1);
   PLOT_CHECK(sizes_ok);
 }
+
+namespace {
+
+/// Buckets the pyramid holds across every level: its footprint, not its window.
+u64 pyramid_entries(const plot::data::MinMaxPyramid& p) {
+  u64 total = 0;
+  for (u32 l = 0; l < p.level_count(); ++l) total += p.level_size(l);
+  return total;
+}
+
+}  // namespace
+
+PLOT_TEST(stream_series_sliding_window_queries_match_bruteforce) {
+  // After the base has moved many times, an exact query must still match a scan
+  // of exactly the retained samples. The pyramid's buckets live in absolute
+  // sample space while the window's base is arbitrary, and that meeting point is
+  // where an off-by-one hides -- so test the sliding case, not just the static
+  // ArraySeries one.
+  constexpr u64 kCapacity = 1u << 14;
+  constexpr u64 kPushes = 60'000;
+  const auto value = [](u64 i) { return f32((i * 29u) % 173u) - 86.0f; };
+  plot::data::StreamSeries s(kCapacity, 0.0, 1e-6);
+  for (u64 i = 0; i < kPushes; i += 337) {
+    const u64 n = std::min<u64>(337, kPushes - i);
+    for (u64 k = 0; k < n; ++k) s.push(value(i + k));
+    s.acquire();   // the window's base moves on every one of these
+  }
+
+  const u64 n = s.size();
+  const u64 base = s.produced() - n;
+  PLOT_CHECK(n == kCapacity);
+  PLOT_CHECK(base > 2 * kCapacity);             // the window really did slide
+  PLOT_CHECK(s.pyramid().size() == n);          // pyramid() indexes the window
+  PLOT_CHECK(s.pyramid().window_base() == base);
+  PLOT_CHECK(s.pyramid().window_end() == s.produced());
+  PLOT_CHECK_NEAR(s.x0(), f64(base) * 1e-6, 1e-12);
+  PLOT_CHECK(s.x_bounds().min == s.x0());
+  PLOT_CHECK_NEAR(s.x_bounds().max, f64(base + n - 1) * 1e-6, 1e-9);
+
+  // Brute force in window coordinates over the retained samples only.
+  const auto brute = [&](u64 a, u64 b) {
+    f64 lo = f64(value(base + a)), hi = lo;
+    f64 sum = 0.0, sq = 0.0;
+    for (u64 i = a; i < b; ++i) {
+      const f64 v = f64(value(base + i));
+      lo = std::min(lo, v);
+      hi = std::max(hi, v);
+      sum += v;
+      sq += v * v;
+    }
+    return std::array<f64, 4>{lo, hi, sum, sq};
+  };
+
+  const u64 starts[] = {0, 1, 63, 64, 65, 1000, 7919, n - 200};
+  for (const u64 a : starts) {
+    for (u64 b = a + 1; b <= n; b += 1301) {
+      const std::array<f64, 4> want = brute(a, b);
+      const plot::Range r = s.reduce(a, b);
+      PLOT_CHECK_NEAR(r.min, want[0], 1e-4);
+      PLOT_CHECK_NEAR(r.max, want[1], 1e-4);
+      PLOT_CHECK_NEAR(s.sum(a, b), want[2], 1e-2);
+      PLOT_CHECK_NEAR(s.sumsq(a, b), want[3], 1e-1);
+    }
+  }
+  const std::array<f64, 4> whole = brute(0, n);
+  PLOT_CHECK_NEAR(s.y_bounds().min, whole[0], 1e-6);
+  PLOT_CHECK_NEAR(s.y_bounds().max, whole[1], 1e-6);
+
+  // The level accessors are in absolute bucket coordinates: a stored bucket must
+  // aggregate its own absolute span, the sliding window included.
+  const u64 span = u64{kBucket} << 1;
+  const u64 lbase = s.pyramid().level_base(1);
+  const u64 lsize = s.pyramid().level_size(1);
+  PLOT_CHECK(lsize >= 2);
+  for (u64 b = lbase; b < lbase + std::min<u64>(lsize, 4); ++b) {
+    const std::array<f64, 4> want = brute(b * span - base, (b + 1) * span - base);
+    const plot::data::MinMax m = s.pyramid().bucket(1, b);
+    PLOT_CHECK_NEAR(m.min, want[0], 1e-6);
+    PLOT_CHECK_NEAR(m.max, want[1], 1e-6);
+  }
+
+  // level-1 gather: one envelope pair per retained bucket the range overlaps, x
+  // at the bucket's own first and last sample. This is the mapping the renderer
+  // draws from.
+  std::vector<f32> out;
+  s.gather(1, 0, n, out);
+  const u64 first = std::max(base / span, lbase);
+  const u64 last = std::min((base + n - 1) / span + 1, lbase + lsize);
+  PLOT_CHECK(out.size() == 4 * (last - first));
+  for (u64 bi = first; bi < last; ++bi) {
+    const std::array<f64, 4> want = brute(bi * span - base, (bi + 1) * span - base);
+    const plot::usize k = static_cast<plot::usize>(bi - first);
+    const u64 first_sample = bi * span - base;
+    const u64 last_sample = std::min((bi + 1) * span, base + n) - 1 - base;
+    // x is in data coordinates: x0() is the retained window's first sample, so
+    // window index j plots at (base + j) * dx.
+    PLOT_CHECK_NEAR(out[4 * k], f32(f64(base + first_sample) * 1e-6), 1e-9);
+    PLOT_CHECK_NEAR(out[4 * k + 1], want[0], 1e-6);
+    PLOT_CHECK_NEAR(out[4 * k + 2], f32(f64(base + last_sample) * 1e-6), 1e-9);
+    PLOT_CHECK_NEAR(out[4 * k + 3], want[1], 1e-6);
+  }
+}
+
+PLOT_TEST(stream_series_steady_state_acquire_is_proportional_to_produced) {
+  // The defect this pins: a saturated ring whose base moved every frame used to
+  // rebuild the whole window, costing O(capacity * live_levels) merges per
+  // acquire. It must cost O(samples produced since the last acquire) instead, so
+  // the same production through a 16x larger window must cost the same.
+  const auto merges_for = [](u64 capacity) {
+    plot::data::StreamSeries s(capacity, 0.0, 1.0);
+    constexpr u64 kSaturate = 1u << 20;
+    for (u64 i = 0; i < kSaturate; ++i) s.push(f32(i % 977));
+    s.acquire();
+    PLOT_CHECK(s.size() == capacity);
+
+    constexpr u64 kFrames = 64;
+    constexpr u64 kFrame = 1u << 13;   // 8192 samples: 1 MS/s at ~122 fps
+    plot::data::detail::merge_count().store(0, std::memory_order_relaxed);
+    for (u64 f = 0; f < kFrames; ++f) {
+      for (u64 k = 0; k < kFrame; ++k) s.push(f32((kSaturate + f * kFrame + k) % 977));
+      s.acquire();
+    }
+    return plot::data::detail::merge_count().load(std::memory_order_relaxed);
+  };
+
+  constexpr u64 kProduced = 64 * (1u << 13);
+  const u64 small = merges_for(1u << 16);   // 64k-sample window
+  const u64 big = merges_for(1u << 20);     // 1M-sample window: 16x to rebuild
+  // ~kProduced/32 in practice; rebuilding the 1M window alone would merge
+  // kCapacity/64 * live_levels ~ 245'000 buckets per acquire.
+  PLOT_CHECK(small <= kProduced / 16);
+  PLOT_CHECK(big <= kProduced / 16);
+  PLOT_CHECK(big <= 2 * small + kBucket);
+}
+
+PLOT_TEST(stream_series_steady_state_acquire_is_fast) {
+  // Wall-clock backstop for the merge count above: rebuilding the 1M-sample
+  // window measured ~44 ms, so 128 such frames would need ~5.6 s. The sliding
+  // path does the same work in a few milliseconds; the bound is two orders of
+  // magnitude clear of both, so it is not a thin timing margin.
+  plot::data::StreamSeries s(1u << 20, 0.0, 1.0);
+  constexpr u64 kSaturate = 1u << 20;
+  for (u64 i = 0; i < kSaturate; ++i) s.push(f32(i % 977));
+  s.acquire();
+
+  constexpr u64 kFrames = 128;
+  constexpr u64 kFrame = 1u << 13;
+  const auto t0 = std::chrono::steady_clock::now();
+  for (u64 f = 0; f < kFrames; ++f) {
+    for (u64 k = 0; k < kFrame; ++k) s.push(f32((kSaturate + f * kFrame + k) % 977));
+    s.acquire();
+  }
+  const f64 ms =
+      std::chrono::duration<f64>(std::chrono::steady_clock::now() - t0).count() * 1.0e3;
+  PLOT_CHECK(s.size() == 1u << 20);
+  PLOT_CHECK(ms < 250.0);
+}
+
+PLOT_TEST(stream_series_pyramid_memory_is_bounded_by_the_window) {
+  // 50M samples through a 1M-sample ring: the pyramid's footprint must track the
+  // retained window (n/64 level-0 buckets plus the levels over them), not the
+  // samples ever pushed.
+  constexpr u64 kCapacity = 1u << 20;
+  constexpr u64 kTotal = 50u * (1u << 20);
+  constexpr u64 kChunk = 1u << 19;
+  plot::data::StreamSeries s(kCapacity, 0.0, 1.0);
+  u64 entries_after_10m = 0;
+  for (u64 p = 0; p < kTotal; p += kChunk) {
+    for (u64 i = 0; i < kChunk; ++i) s.push(f32((p + i) % 251));
+    s.acquire();
+    if (p + kChunk == 10u * (1u << 20)) entries_after_10m = pyramid_entries(s.pyramid());
+  }
+  const u64 entries = pyramid_entries(s.pyramid());
+  const u64 bound = 2 * (kCapacity / kBucket) + kBucket;
+  PLOT_CHECK(s.produced() == kTotal);
+  PLOT_CHECK(s.size() == kCapacity);
+  PLOT_CHECK(entries <= bound);
+  // 5x the samples pushed, same footprint: levels deeper than one chunk can
+  // differ by a bucket each, hence the slack.
+  PLOT_CHECK(entries <= entries_after_10m + kBucket);
+}