feat(data): SPSC ring buffer, incremental min/max LOD pyramid, series types

src/data/array_series.cpp

@@ -0,0 +1,68 @@
+#include <plot/data/array_series.hpp>
+
+#include <algorithm>
+#include <cstddef>
+#include <utility>
+
+namespace plot::data {
+
+ArraySeries::ArraySeries(std::vector<f32> y, f64 x0, f64 dx) { reset(std::move(y), x0, dx); }
+
+void ArraySeries::reset(std::vector<f32> y, f64 x0, f64 dx) {
+  y_ = std::move(y);
+  x0_ = x0;
+  dx_ = dx;
+  commit();
+}
+
+void ArraySeries::commit() {
+  const u64 n = static_cast<u64>(y_.size());
+  pyramid_.clear();
+  pyramid_.reserve(n);
+
+  f32 lo = 0.0f, hi = 0.0f;
+  bool first = true;
+  for (const f32 v : y_) {
+    pyramid_.append(v);
+    if (first) {
+      lo = v;
+      hi = v;
+      first = false;
+    } else {
+      lo = std::min(lo, v);
+      hi = std::max(hi, v);
+    }
+  }
+  y_bounds_ = first ? Range{0.0, 0.0} : Range{f64(lo), f64(hi)};
+  x_bounds_ = n == 0 ? Range{x0_, x0_} : Range{x0_, x0_ + f64(n - 1) * dx_};
+}
+
+u64 ArraySeries::size() const { return static_cast<u64>(y_.size()); }
+
+f64 ArraySeries::x0() const { return x0_; }
+
+f64 ArraySeries::dx() const { return dx_; }
+
+Range ArraySeries::x_bounds() const { return x_bounds_; }
+
+Range ArraySeries::y_bounds() const { return y_bounds_; }
+
+const MinMaxPyramid& ArraySeries::pyramid() const { return pyramid_; }
+
+void ArraySeries::gather(u32 level, u64 begin, u64 end, std::vector<f32>& out_xy) const {
+  gather_impl(level, begin, end, out_xy);
+}
+
+void ArraySeries::copy_raw(u64 begin, u64 end, f32* out) const {
+  const u64 want = end > begin ? end - begin : 0;
+  if (want == 0) return;
+  const u64 n = static_cast<u64>(y_.size());
+  const u64 b = std::min(begin, n);
+  const u64 e = std::min(std::max(end, b), n);
+  const auto first = y_.begin();
+  std::copy(first + static_cast<std::ptrdiff_t>(b), first + static_cast<std::ptrdiff_t>(e), out);
+  std::fill(out + static_cast<std::ptrdiff_t>(e - b),
+            out + static_cast<std::ptrdiff_t>(want), 0.0f);
+}
+
+}  // namespace plot::data

src/data/minmax_pyramid.cpp

@@ -0,0 +1,240 @@
+#include <plot/data/minmax_pyramid.hpp>
+
+#include <algorithm>
+#include <cmath>
+
+namespace plot::data {
+namespace {
+
+/// Samples covered by one level-l bucket.
+constexpr u64 level_span(u32 l) { return u64{kBucket} << l; }
+
+/// Level-l buckets needed to hold n samples (ceiling division).
+constexpr u64 buckets_for(u64 n, u32 l) {
+  const u64 span = level_span(l);
+  return n / span + (n % span == 0 ? u64{0} : u64{1});
+}
+
+/// First level-0 bucket boundary at or after v.
+constexpr u64 ceil_to_bucket(u64 v) {
+  return (v / kBucket + (v % kBucket == 0 ? u64{0} : u64{1})) * kBucket;
+}
+
+/// Canonical, coarsest-first decomposition of the level-0 bucket span [b0, b1)
+/// into pyramid entries, calling fn(level, index) for each.
+///
+/// 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
+/// decomposition has O(log) entries, so every query below is O(log n).
+template <typename Fn>
+void cover_buckets(const MinMaxPyramid& p, u64 b0, u64 b1, Fn&& fn) {
+  const u64 complete = p.size() / kBucket;
+  u64 b = b0;
+  for (u32 li = p.level_count(); li-- > 0 && b < b1;) {
+    const u64 lb = u64{1} << li;               // level-0 buckets per level-li bucket
+    const u64 avail = p.level(li).size();
+    while (b < b1 && (b % lb) == 0 && b + lb <= b1 && b + lb <= complete &&
+           (b / lb) < avail) {
+      fn(li, b / lb);
+      b += lb;
+    }
+  }
+  if (b < b1) fn(0u, b);                       // trailing partial level-0 bucket
+}
+
+/// Folds the decomposition of [b0, b1) into a min/max Range. Returns a
+/// zero-span Range when there is nothing to fold.
+Range fold_range(const MinMaxPyramid& p, u64 b0, u64 b1) {
+  Range r{};
+  bool first = true;
+  cover_buckets(p, b0, b1, [&](u32 l, u64 i) {
+    const MinMax m = p.level(l)[static_cast<usize>(i)];
+    if (first) {
+      r.min = f64(m.min);
+      r.max = f64(m.max);
+      first = false;
+    } else {
+      r.min = std::min(r.min, f64(m.min));
+      r.max = std::max(r.max, f64(m.max));
+    }
+  });
+  return first ? Range{0.0, 0.0} : r;
+}
+
+}  // namespace
+
+void MinMaxPyramid::add_level() {
+  levels_.emplace_back();
+  sums_.emplace_back();
+  sumsqs_.emplace_back();
+}
+
+void MinMaxPyramid::reserve(u64 n) {
+  if (n > reserved_n_) reserved_n_ = n;
+  const u64 want = std::max(std::max(n, size_), u64{2});
+
+  // Materialise every level n samples can need -- buckets *and* prefix sums --
+  // and size each one up front, so a run of appends allocates nothing at all.
+  u32 max_levels = 1;
+  for (u64 span = level_span(0); span <= want && max_levels < 63; span <<= 1) ++max_levels;
+  while (levels_.size() < max_levels) add_level();
+  for (u32 l = 0; l < levels_.size(); ++l) {
+    const usize want_l = static_cast<usize>(buckets_for(want, l));
+    levels_[l].reserve(want_l);
+    sums_[l].reserve(want_l);
+    sumsqs_[l].reserve(want_l);
+  }
+}
+
+void MinMaxPyramid::append(f32 v) {
+  const u64 prev = size_;
+  const f64 vd = f64(v);
+  // sum_[l].back() is the sum over every sample appended so far, and the newest
+  // sample always lands in the last bucket of every level, so the last prefix
+  // entry of every level is exactly the running total.
+  const f64 total_before = sums_[0].empty() ? 0.0 : sums_[0].back();
+  const f64 total = total_before + vd;
+  const f64 sq_before = sumsqs_[0].empty() ? 0.0 : sumsqs_[0].back();
+  const f64 total_sq = sq_before + vd * vd;
+
+  const u64 idx = prev / kBucket;
+  if (idx == levels_[0].size()) {
+    levels_[0].push_back(MinMax{v, v});
+    sums_[0].push_back(total);
+    sumsqs_[0].push_back(total_sq);
+  } else {
+    MinMax& cur = levels_[0][static_cast<usize>(idx)];
+    cur.min = std::min(cur.min, v);
+    cur.max = std::max(cur.max, v);
+    sums_[0][static_cast<usize>(idx)] = total;
+    sumsqs_[0][static_cast<usize>(idx)] = total_sq;
+  }
+
+  for (u32 l = 1; l < 63; ++l) {
+    // A level only carries data once the level below has more than one bucket: a
+    // one-child parent would just duplicate it.
+    if (levels_[l - 1].size() < 2) break;
+    if (l == live_levels_) {
+      ++live_levels_;
+      if (l == levels_.size()) {   // no reserve() sized this level up front
+        add_level();
+        if (reserved_n_ != 0) {
+          const usize want = static_cast<usize>(buckets_for(reserved_n_, l));
+          levels_[l].reserve(want);
+          sums_[l].reserve(want);
+          sumsqs_[l].reserve(want);
+        }
+      }
+    }
+    const u64 ci = prev / level_span(l - 1);   // current bucket at level l-1
+    const u64 parent = ci / 2;                 // its parent at level l
+    MinMax m;
+    if ((ci & 1u) != 0) {
+      const MinMax& a = levels_[l - 1][static_cast<usize>(ci - 1)];
+      const MinMax& b = levels_[l - 1][static_cast<usize>(ci)];
+      m.min = std::min(a.min, b.min);
+      m.max = std::max(a.max, b.max);
+    } else {
+      m = levels_[l - 1][static_cast<usize>(ci)];
+    }
+    if (parent == levels_[l].size()) {
+      levels_[l].push_back(m);
+      sums_[l].push_back(total);
+      sumsqs_[l].push_back(total_sq);
+    } else {
+      levels_[l][static_cast<usize>(parent)] = m;
+      sums_[l][static_cast<usize>(parent)] = total;
+      sumsqs_[l][static_cast<usize>(parent)] = total_sq;
+    }
+  }
+  size_ = prev + 1;
+}
+
+void MinMaxPyramid::clear() {
+  size_ = 0;
+  live_levels_ = 1;
+  for (auto& lv : levels_) lv.clear();
+  for (auto& s : sums_) s.clear();
+  for (auto& s : sumsqs_) s.clear();
+}
+
+std::span<const MinMax> MinMaxPyramid::level(u32 l) const {
+  if (l >= level_count()) return {};
+  return levels_[l];
+}
+
+Range MinMaxPyramid::reduce_buckets(u64 begin, u64 end) const {
+  const u64 e = std::min(end, size_);
+  const u64 b = std::min(begin, e);
+  const u64 b0 = ceil_to_bucket(b) / kBucket;   // first whole bucket inside
+  const u64 b1 = e / kBucket;                   // one past the last whole bucket
+  if (b1 <= b0) return Range{0.0, 0.0};
+  return fold_range(*this, b0, b1);
+}
+
+f64 MinMaxPyramid::sum_buckets(u64 begin, u64 end) const {
+  const u64 e = std::min(end, size_);
+  const u64 b = std::min(begin, e);
+  const u64 b0 = ceil_to_bucket(b) / kBucket;
+  const u64 b1 = e / kBucket;
+  if (b1 <= b0) return 0.0;
+  f64 total = 0.0;
+  cover_buckets(*this, b0, b1, [&](u32 l, u64 i) {
+    const f64 prev = i == 0 ? 0.0 : sums_[l][static_cast<usize>(i - 1)];
+    total += sums_[l][static_cast<usize>(i)] - prev;
+  });
+  return total;
+}
+
+f64 MinMaxPyramid::sumsq_buckets(u64 begin, u64 end) const {
+  const u64 e = std::min(end, size_);
+  const u64 b = std::min(begin, e);
+  const u64 b0 = ceil_to_bucket(b) / kBucket;
+  const u64 b1 = e / kBucket;
+  if (b1 <= b0) return 0.0;
+  f64 total = 0.0;
+  cover_buckets(*this, b0, b1, [&](u32 l, u64 i) {
+    const f64 prev = i == 0 ? 0.0 : sumsqs_[l][static_cast<usize>(i - 1)];
+    total += sumsqs_[l][static_cast<usize>(i)] - prev;
+  });
+  return total;
+}
+
+Range MinMaxPyramid::reduce(u64 begin, u64 end) const {
+  const u64 e = std::min(end, size_);
+  const u64 b = std::min(begin, e);
+  if (e <= b) return Range{0.0, 0.0};
+  const u64 b0 = b / kBucket;                   // first bucket holding a sample
+  const u64 b1 = (e - 1) / kBucket + 1;         // one past the last one
+  return fold_range(*this, b0, b1);
+}
+
+u32 MinMaxPyramid::level_for(u64 n, f64 samples_per_pixel) const {
+  // The ratio test also rejects NaN, and the clamp keeps a huge (or infinite)
+  // samples-per-pixel from making the floor-to-u32 conversion ill-formed.
+  const f64 ratio = samples_per_pixel / f64(kBucket);
+  if (n == 0 || !(ratio > 1.0)) return 0;
+  const f64 lg = std::floor(std::log2(ratio));
+  u32 l = lg >= f64(62) ? 62u : static_cast<u32>(lg);
+  while (l > 0) {
+    const bool exists = l < level_count() && level(l).size() >= 2;
+    if (exists && buckets_for(n, l) >= 2) break;
+    --l;
+  }
+  return l;
+}
+
+u64 MinMaxPyramid::snap(f64 x, f64 x0, f64 dx) const {
+  if (dx <= 0.0 || size_ == 0) return 0;
+  const f64 i = (x - x0) / dx;
+  // Bound the round before it: llround() is undefined for values a long long
+  // cannot hold, and this also maps NaN to index 0.
+  if (!(i > 0.0)) return 0;
+  if (i >= f64(size_)) return size_ - 1;
+  const u64 idx = static_cast<u64>(std::llround(i));
+  return idx >= size_ ? size_ - 1 : idx;
+}
+
+}  // namespace plot::data

src/data/series.cpp

@@ -0,0 +1,153 @@
+#include <plot/data/series.hpp>
+
+#include <algorithm>
+#include <cstddef>
+
+namespace plot::data {
+namespace {
+
+/// Raw samples read per copy_raw() call while gathering level 0. Bounded so the
+/// gather path never allocates, and large enough to amortise the call.
+constexpr u64 kGatherChunk = 256;
+
+}  // namespace
+
+Series::QueryParts Series::split_query(u64 begin, u64 end) {
+  QueryParts p;
+  // First level-0 bucket boundary at or after begin, and the last at or before
+  // end: [interior_begin, interior_end) is exactly the part the pyramid can
+  // answer with whole buckets.
+  const u64 up = (begin / kBucket + (begin % kBucket == 0 ? u64{0} : u64{1})) * kBucket;
+  const u64 down = (end / kBucket) * kBucket;
+  if (up < down) {
+    p.interior_begin = up;
+    p.interior_end = down;
+  }
+  // The two edge spans never overlap and never exceed one bucket each: when the
+  // interior is empty they tile [begin, end) together, otherwise they are the
+  // head and tail buckets half inside the range.
+  p.left_begin = begin;
+  p.left_end = std::min(end, up);
+  p.right_begin = std::max(down, p.left_end);
+  p.right_end = end;
+  return p;
+}
+
+u64 Series::copy_edges(const QueryParts& p, f32* out) const {
+  u64 n = 0;
+  if (p.left_begin < p.left_end) {
+    copy_raw(p.left_begin, p.left_end, out);
+    n = p.left_end - p.left_begin;
+  }
+  if (p.right_begin < p.right_end) {
+    copy_raw(p.right_begin, p.right_end, out + static_cast<std::ptrdiff_t>(n));
+    n += p.right_end - p.right_begin;
+  }
+  return n;
+}
+
+Range Series::reduce(u64 begin, u64 end) const {
+  const u64 e = std::min(end, size());
+  if (begin >= e) return Range{0.0, 0.0};
+  const QueryParts p = split_query(begin, e);
+
+  Range r{};
+  bool first = true;
+  if (p.interior_begin < p.interior_end) {
+    r = pyramid().reduce_buckets(p.interior_begin, p.interior_end);
+    first = false;
+  }
+
+  // At most 2*kBucket - 1 edge samples, so the buffer is enough without
+  // allocating: each edge span is under one bucket, and when the interior is
+  // empty the two spans tile a range shorter than two buckets.
+  f32 edge[2 * kBucket];
+  const u64 n = copy_edges(p, edge);
+  for (u64 i = 0; i < n; ++i) {
+    const f64 v = f64(edge[i]);
+    if (first) {
+      r.min = v;
+      r.max = v;
+      first = false;
+    } else {
+      r.min = std::min(r.min, v);
+      r.max = std::max(r.max, v);
+    }
+  }
+  return first ? Range{0.0, 0.0} : r;
+}
+
+f64 Series::sum(u64 begin, u64 end) const {
+  const u64 e = std::min(end, size());
+  if (begin >= e) return 0.0;
+  const QueryParts p = split_query(begin, e);
+
+  f64 total = 0.0;
+  if (p.interior_begin < p.interior_end) {
+    total = pyramid().sum_buckets(p.interior_begin, p.interior_end);
+  }
+  f32 edge[2 * kBucket];
+  const u64 n = copy_edges(p, edge);
+  for (u64 i = 0; i < n; ++i) total += f64(edge[i]);
+  return total;
+}
+
+f64 Series::sumsq(u64 begin, u64 end) const {
+  const u64 e = std::min(end, size());
+  if (begin >= e) return 0.0;
+  const QueryParts p = split_query(begin, e);
+
+  f64 total = 0.0;
+  if (p.interior_begin < p.interior_end) {
+    total = pyramid().sumsq_buckets(p.interior_begin, p.interior_end);
+  }
+  f32 edge[2 * kBucket];
+  const u64 n = copy_edges(p, edge);
+  for (u64 i = 0; i < n; ++i) {
+    const f64 v = f64(edge[i]);
+    total += v * v;
+  }
+  return total;
+}
+
+void Series::gather_impl(u32 level, u64 begin, u64 end, std::vector<f32>& out_xy) const {
+  const u64 n = size();
+  const u64 b = std::min(begin, n);
+  const u64 e = std::min(std::max(end, b), n);
+  if (b >= e) return;
+
+  const f64 x_0 = x0();
+  const f64 step = dx();
+  if (level == 0) {
+    f32 raw[kGatherChunk];
+    for (u64 i = b; i < e;) {
+      const u64 take = std::min<u64>(kGatherChunk, e - i);
+      copy_raw(i, i + take, raw);
+      for (u64 k = 0; k < take; ++k) {
+        out_xy.push_back(f32(x_0 + f64(i + k) * step));
+        out_xy.push_back(raw[k]);
+      }
+      i += take;
+    }
+    return;
+  }
+
+  // Level >= 1: one envelope pair per bucket, x at the bucket's first sample
+  // (min) and at its last sample (max), which keeps x monotonic while covering
+  // the bucket's span.
+  const u64 span = u64{kBucket} << level;
+  const std::span<const MinMax> buckets = pyramid().level(level);
+  const u64 first = b / span;
+  const u64 last = std::min<u64>((e - 1) / span + 1, static_cast<u64>(buckets.size()));
+  for (u64 bi = first; bi < last; ++bi) {
+    const MinMax m = buckets[static_cast<usize>(bi)];
+    const u64 first_sample = bi * span;
+    const u64 last_sample = std::min(first_sample + span, n) - 1;
+    out_xy.push_back(f32(x_0 + f64(first_sample) * step));
+    out_xy.push_back(m.min);
+    out_xy.push_back(f32(x_0 + f64(last_sample) * step));
+    out_xy.push_back(m.max);
+  }
+}
+
+}  // namespace plot::data

src/data/stream_series.cpp

@@ -0,0 +1,115 @@
+#include <plot/data/stream_series.hpp>
+
+#include <algorithm>
+#include <cstddef>
+
+namespace plot::data {
+namespace {
+
+/// Samples copied per ring read on the ingest path, so rebuilding a saturated
+/// window costs no allocation.
+constexpr u64 kReadChunk = 256;
+
+}  // namespace
+
+StreamSeries::StreamSeries(u64 capacity, f64 x0, f64 dx)
+    : ring_(capacity), x0_(x0), dx_(dx) {}
+
+void StreamSeries::push(f32 y) { ring_.push(y); }
+
+void StreamSeries::push(const f32* ys, u64 n) { ring_.push(ys, n); }
+
+u64 StreamSeries::produced() const { return ring_.produced(); }
+
+void StreamSeries::acquire() {
+  // One atomic read: every absolute index below is relative to this single count,
+  // so the producer may keep pushing without the snapshot shifting under us.
+  const u64 p = ring_.produced();
+  const u64 cap = ring_.capacity();
+  const u64 base = p > cap ? p - cap : 0;   // oldest index the ring still holds
+  const u64 n = p - base;                   // retained, published sample count
+
+  if (base != base_) {
+    // The window slid: samples the pyramid covered are gone, and because bucket
+    // boundaries are absolute, extending it incrementally would mis-align every
+    // bucket. Rebuild from the retained samples instead.
+    pyramid_.clear();
+    pyramid_.reserve(n);
+    f32 lo = 0.0f, hi = 0.0f;
+    bool first = true;
+    f32 chunk[kReadChunk];
+    for (u64 i = 0; i < n;) {
+      const u64 take = std::min<u64>(kReadChunk, n - i);
+      ring_.copy_from(base + i, take, chunk);
+      for (u64 k = 0; k < take; ++k) {
+        const f32 v = chunk[k];
+        pyramid_.append(v);
+        if (first) {
+          lo = v;
+          hi = v;
+          first = false;
+        } else {
+          lo = std::min(lo, v);
+          hi = std::max(hi, v);
+        }
+      }
+      i += take;
+    }
+    y_bounds_ = first ? Range{0.0, 0.0} : Range{f64(lo), f64(hi)};
+    base_ = base;
+  } else {
+    // Contiguous: only the samples produced since the last acquire() are new.
+    if (n == 0) y_bounds_ = Range{0.0, 0.0};
+    f32 chunk[kReadChunk];
+    for (u64 i = end_ - base_; i < n;) {
+      const u64 take = std::min<u64>(kReadChunk, n - i);
+      ring_.copy_from(base + i, take, chunk);
+      for (u64 k = 0; k < take; ++k) {
+        const f32 v = chunk[k];
+        pyramid_.append(v);
+        if (i + k == 0) {
+          y_bounds_ = Range{f64(v), f64(v)};
+        } else {
+          y_bounds_.min = std::min(y_bounds_.min, f64(v));
+          y_bounds_.max = std::max(y_bounds_.max, f64(v));
+        }
+      }
+      i += take;
+    }
+  }
+  end_ = p;
+}
+
+u64 StreamSeries::size() const { return end_ - base_; }
+
+f64 StreamSeries::x0() const { return x0_ + f64(base_) * dx_; }
+
+f64 StreamSeries::dx() const { return dx_; }
+
+Range StreamSeries::x_bounds() const {
+  const f64 n = f64(size());
+  return n == 0.0 ? Range{x0(), x0()} : Range{x0(), x0() + (n - 1.0) * dx_};
+}
+
+Range StreamSeries::y_bounds() const { return y_bounds_; }
+
+const MinMaxPyramid& StreamSeries::pyramid() const { return pyramid_; }
+
+void StreamSeries::gather(u32 level, u64 begin, u64 end, std::vector<f32>& out_xy) const {
+  gather_impl(level, begin, end, out_xy);
+}
+
+void StreamSeries::copy_raw(u64 begin, u64 end, f32* out) const {
+  const u64 want = end > begin ? end - begin : 0;
+  if (want == 0) return;
+  const u64 n = size();
+  const u64 b = std::min(begin, n);
+  const u64 e = std::min(std::max(end, b), n);
+  // Retained index b is absolute base_ + b: fixed for the whole copy, so a
+  // pushing producer cannot shift the window midway.
+  ring_.copy_from(base_ + b, e - b, out);
+  std::fill(out + static_cast<std::ptrdiff_t>(e - b),
+            out + static_cast<std::ptrdiff_t>(want), 0.0f);
+}
+
+}  // namespace plot::data