From d9909f2115e63c306a001f1ef3cea8d4351a7079 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Sun, 27 Sep 2026 21:42:53 +0200 Subject: [PATCH] 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 | 5 +++ include/plot/data/detail/cover_count.hpp | 29 +++++++++++++ include/plot/data/ring_buffer.hpp | 44 +++++++++++-------- include/plot/data/stream_series.hpp | 26 ++++++++---- src/data/minmax_pyramid.cpp | 39 ++++++++++++----- tests/CMakeLists.txt | 7 ++- tests/test_minmax_pyramid.cpp | 24 +++++++++++ tests/test_series.cpp | 54 +++++++++++++++++++++++- 8 files changed, 190 insertions(+), 38 deletions(-) create mode 100644 include/plot/data/detail/cover_count.hpp diff --git a/CMakeLists.txt b/CMakeLists.txt index e848e45..b42a364 100644 --- a/CMakeLists.txt +++ b/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() diff --git a/include/plot/data/detail/cover_count.hpp b/include/plot/data/detail/cover_count.hpp new file mode 100644 index 0000000..3c0604e --- /dev/null +++ b/include/plot/data/detail/cover_count.hpp @@ -0,0 +1,29 @@ +#pragma once + +#ifdef PLOT_PYRAMID_COUNT_COVERS + +#include + +#include + +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& cover_entry_count() { + static std::atomic count{0}; + return count; +} + +} // namespace plot::data::detail + +#endif // PLOT_PYRAMID_COUNT_COVERS diff --git a/include/plot/data/ring_buffer.hpp b/include/plot/data/ring_buffer.hpp index 9e0ecf1..a8b77db 100644 --- a/include/plot/data/ring_buffer.hpp +++ b/include/plot/data/ring_buffer.hpp @@ -1,8 +1,6 @@ #pragma once -#include #include #include -#include #include #include @@ -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 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)), @@ -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(i & mask_)] = v; + buf_[static_cast(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(). - const T& operator[](u64 i) const noexcept { - return buf_[static_cast((base() + i) & mask_)]; + /// < 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`, splitting the copy at the ring's wrap point. The caller must hold - /// a snapshot that retains those indices; this never consults the live count, so - /// a pushing producer cannot shift the read. + /// 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; - const u64 s = absolute & mask_; - const u64 run = std::min(n, cap_ - s); - const auto first = buf_.begin() + static_cast(s); - std::copy(first, first + static_cast(run), out); - if (run < n) { - std::copy(buf_.begin(), buf_.begin() + static_cast(n - run), - out + static_cast(run)); + // 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); } } @@ -94,7 +104,7 @@ class RingBuffer { private: u64 cap_; u64 mask_; - std::vector buf_; + std::vector> buf_; std::atomic produced_{0}; }; diff --git a/include/plot/data/stream_series.hpp b/include/plot/data/stream_series.hpp index 3922df1..817ee8d 100644 --- a/include/plot/data/stream_series.hpp +++ b/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 -/// at the window's oldest edge. That read is not ordered against the write, so a -/// race detector flags it: the value is still a single f32 (no tearing on -/// 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 -/// pyramid itself never mixes windows, because acquire() derives the whole window -/// from one count. Ingest that is race-free even at the edge 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. +/// producer can overwrite the slots a consumer is reading even while it holds a +/// snapshot. Those slots are relaxed atomics (see RingBuffer), so this is not a +/// data race and no value tears; what it is not is a frozen window. +/// +/// The bound is the sample count, not the geometry. A snapshot count fixes *which +/// absolute indices* a read walks, not that the slots still hold those samples: +/// every push after the snapshot overwrites exactly one slot, the one belonging to +/// the sample capacity() positions behind it. So if the producer pushes k samples +/// 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); diff --git a/src/data/minmax_pyramid.cpp b/src/data/minmax_pyramid.cpp index 20f0e73..0e9d6ce 100644 --- a/src/data/minmax_pyramid.cpp +++ b/src/data/minmax_pyramid.cpp @@ -1,3 +1,4 @@ +#include #include #include @@ -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 -/// decomposition has O(log) entries, so every query below is O(log n). +/// single trailing partial level-0 bucket, which is emitted at level 0. +/// +/// 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 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;) { - 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; + while (b < b1) { + u32 li = top; + for (;;) { + const u64 lb = u64{1} << li; + if ((b % lb) == 0 && b + lb <= limit && + (b / lb) < static_cast(p.level(li).size())) { + 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 diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 4a0309f..1217f6e 100644 --- a/tests/CMakeLists.txt +++ b/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) diff --git a/tests/test_minmax_pyramid.cpp b/tests/test_minmax_pyramid.cpp index 432a5d5..605081e 100644 --- a/tests/test_minmax_pyramid.cpp +++ b/tests/test_minmax_pyramid.cpp @@ -1,4 +1,5 @@ #include "harness.hpp" +#include #include #include @@ -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 +} diff --git a/tests/test_series.cpp b/tests/test_series.cpp index 93a5b0d..1b15a29 100644 --- a/tests/test_series.cpp +++ b/tests/test_series.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include 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(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 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 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); }