diff --git a/include/plot/data/minmax_pyramid.hpp b/include/plot/data/minmax_pyramid.hpp index a25c6c9..974f839 100644 --- a/include/plot/data/minmax_pyramid.hpp +++ b/include/plot/data/minmax_pyramid.hpp @@ -76,6 +76,19 @@ class MinMaxPyramid { /// whose samples were all non-finite is stored as an empty aggregate /// (count 0, zero sums) that every reader treats as absent. void append(f32 v); + /// Folds a backfilled sample into every aggregate that covers absolute index + /// `absolute_sample`. + /// + /// **Precondition: the sample at that index was appended non-finite.** That is what makes + /// this purely additive -- min/max widen, sum/sumsq accumulate and count rises by one, at + /// every covering level -- so no bucket is ever recomputed and the cost is O(levels) + /// regardless of how wide the repaired span is. Amending a *finite* sample would need the + /// old value removed from sum/sumsq and could not restore min/max at all, which is why + /// this is a one-way NaN-to-value operation and not a general write. + /// + /// An index outside `[window_base(), window_end())` is a no-op: the window slides on the + /// consumer thread, so a caller cannot prove the slot is still held before asking. + void amend(u64 absolute_sample, f32 v); /// Slides the retained window's leading edge to absolute sample index /// `absolute_sample`, dropping every aggregate that covered a sample before it. /// Monotonic, and a no-op only when `absolute_sample <= window_base()`: an empty diff --git a/src/data/minmax_pyramid.cpp b/src/data/minmax_pyramid.cpp index f7c0dd3..b5280c6 100644 --- a/src/data/minmax_pyramid.cpp +++ b/src/data/minmax_pyramid.cpp @@ -179,6 +179,43 @@ void MinMaxPyramid::append(f32 v) { } } +void MinMaxPyramid::amend(u64 absolute_sample, f32 v) { + if (!std::isfinite(v)) return; + if (absolute_sample < base_ || absolute_sample >= count_) return; + + const f64 vd = f64(v); + const auto fold = [&](Entry& e) { + // An empty bucket's `mm` is meaningless (see the Entry note), so the first finite + // sample *sets* the extremes instead of widening from a zero that was never a sample. + if (e.count == 0) { + e.mm.min = v; + e.mm.max = v; + } else { + e.mm.min = std::min(e.mm.min, v); + e.mm.max = std::max(e.mm.max, v); + } + e.sum += vd; + e.sumsq += vd * vd; + ++e.count; + }; + + // The in-progress level-0 bucket, when the index lies inside it. It is not an entry yet, + // so the loop below cannot reach it; leaving it out would lose the amendment for a hole + // whose tail is still filling, which is exactly the hole a prompt backfill arrives into. + if (count_ % kBucket != 0 && absolute_sample >= (count_ / kBucket) * kBucket) { + fold(pending_); + } + + // Every level independently: min/max/sum/sumsq/count are each associative, so folding the + // value into a parent gives the same aggregate as re-merging its children would. + for (u32 l = 0; l < static_cast(levels_.size()); ++l) { + const u64 i = absolute_sample / level_span(l); + if (i < levels_[l].base) continue; + if (i - levels_[l].base >= static_cast(levels_[l].entries.size())) continue; + fold(levels_[l].entries[static_cast(i - levels_[l].base)]); + } +} + void MinMaxPyramid::drop_before(u64 absolute_sample) { if (absolute_sample <= base_) return; if (absolute_sample > count_) { diff --git a/tests/test_minmax_pyramid.cpp b/tests/test_minmax_pyramid.cpp index 20c3dbf..124023f 100644 --- a/tests/test_minmax_pyramid.cpp +++ b/tests/test_minmax_pyramid.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include using plot::data::kBucket; @@ -342,3 +343,69 @@ PLOT_TEST(pyramid_level_cap_keeps_the_span_shift_defined) { p.append(1.0f); PLOT_CHECK(p.level_count() >= 1); } + +PLOT_TEST(amend_folds_a_backfilled_sample_into_every_covering_level) { + // A hole is appended as NaN -- which the pyramid already stores as an empty aggregate -- + // and amended later when the packet that owned it finally arrives. The amendment has to + // reach every level that covers the index, because the renderer reads whichever level the + // zoom picked: repairing level 0 alone would make a trace whole when zoomed in and still + // broken when zoomed out. + MinMaxPyramid p; + p.reserve(8 * kBucket); + for (u64 i = 0; i < 8 * kBucket; ++i) { + // One NaN, at an index comfortably inside the first bucket of every level. + p.append(i == 5 ? std::numeric_limits::quiet_NaN() : 1.0f); + } + // Before: every covering bucket aggregates kBucket-1 ones and knows nothing of index 5. + PLOT_CHECK(p.bucket_count(0, 0) == kBucket - 1); + PLOT_CHECK_NEAR(p.bucket(0, 0).max, 1.0); + + p.amend(5, 7.0f); + + // Level 0 gained the sample. + PLOT_CHECK(p.bucket_count(0, 0) == kBucket); + PLOT_CHECK_NEAR(p.bucket(0, 0).max, 7.0); + PLOT_CHECK_NEAR(p.bucket_sum(0, 0), f64(kBucket - 1) + 7.0); + PLOT_CHECK_NEAR(p.bucket_sumsq(0, 0), f64(kBucket - 1) + 49.0); + // And so did every coarser level that covers it. + for (u32 l = 1; l < p.level_count(); ++l) { + if (p.level_size(l) == 0) continue; + const u64 span = u64{kBucket} << l; + const u64 b = 5 / span; + if (b < p.level_base(l) || b - p.level_base(l) >= p.level_size(l)) continue; + PLOT_CHECK_NEAR(p.bucket(l, b).max, 7.0); + PLOT_CHECK(p.bucket_count(l, b) == span); + } +} + +PLOT_TEST(amend_makes_an_all_nan_bucket_present_again) { + // A bucket whose every sample was non-finite is stored with count 0, which every reader + // treats as absent. One amendment has to make it present -- min and max become the value + // rather than widening from a meaningless zero, which is the case a plain fold gets wrong. + MinMaxPyramid p; + p.reserve(4 * kBucket); + const f32 nan = std::numeric_limits::quiet_NaN(); + for (u64 i = 0; i < 2 * kBucket; ++i) p.append(i < kBucket ? nan : 1.0f); + PLOT_CHECK(p.bucket_count(0, 0) == 0); + + p.amend(3, -2.0f); + + PLOT_CHECK(p.bucket_count(0, 0) == 1); + PLOT_CHECK_NEAR(p.bucket(0, 0).min, -2.0); + PLOT_CHECK_NEAR(p.bucket(0, 0).max, -2.0); +} + +PLOT_TEST(amend_outside_the_window_is_a_no_op) { + // An index the window has already dropped, or one not yet appended, has nowhere to go. + // Silently ignoring it is what lets the caller amend without first proving the ring still + // holds the slot -- a race it cannot win, because the window slides on the other thread. + MinMaxPyramid p; + p.reserve(4 * kBucket); + for (u64 i = 0; i < 4 * kBucket; ++i) p.append(1.0f); + p.drop_before(2 * kBucket); + const u64 before = p.bucket_count(0, 2); + p.amend(5, 99.0f); // dropped + p.amend(100 * kBucket, 99.0f); // never appended + PLOT_CHECK(p.bucket_count(0, 2) == before); + PLOT_CHECK_NEAR(p.bucket(0, 2).max, 1.0); +}