From 9a086b7d69b70cf3a77142dcdc9d61315b11c3a6 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Sun, 27 Sep 2026 21:51:26 +0200 Subject: [PATCH] test(data): make the push/read race test deterministic The completion check was scheduling-dependent and the test could lose its TSan overlap window. The producer now signals done with a release store, the consumer reads in a do/while keyed on that flag so at least one read is guaranteed by construction, and a started/reading handshake makes the producer's remaining pushes provably overlap the reader's loop. Bounded to 4096 pushes on the 256-slot ring, and the assertions moved after join() so a failure reports instead of terminating. With the pre-fix ring restored, four TSan runs still flag push vs copy_from, so the case keeps its teeth. --- tests/test_series.cpp | 63 ++++++++++++++++++++++++++++++++----------- 1 file changed, 47 insertions(+), 16 deletions(-) diff --git a/tests/test_series.cpp b/tests/test_series.cpp index abfc519..e3ccf3f 100644 --- a/tests/test_series.cpp +++ b/tests/test_series.cpp @@ -2,6 +2,7 @@ #include #include #include +#include #include #include #include @@ -104,27 +105,57 @@ PLOT_TEST(stream_series_exact_queries_survive_ring_wrap) { 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)); + // slots make that defined; nothing else in the suite has two threads, so this is + // the case ThreadSanitizer actually flags when the slots are plain. + // + // The two flags make the overlap structural, not scheduling luck: the reader + // waits until the producer is already pushing, acknowledges, and only then does + // the producer push the rest -- so the reader's window spans pushes that are in + // flight by construction. The reader then loops until the producer reports done, + // which also makes "at least one read" true by construction instead of by + // winning a race with a fast producer. + plot::data::StreamSeries s(1u << 8, 0.0, 1.0); // one read spans reused slots + constexpr u64 kStride = 512; // pushes before the reader starts + constexpr u64 kPushes = 4'096; // total: a few thousand only + std::atomic started{false}; + std::atomic reading{false}; + std::atomic done{false}; + + std::thread producer([&] { + for (u64 i = 0; i < kStride; ++i) s.push(f32(i % 971)); + started.store(true, std::memory_order_release); + while (!reading.load(std::memory_order_acquire)) { + } + for (u64 i = kStride; i < kPushes; ++i) s.push(f32(i % 971)); + done.store(true, std::memory_order_release); }); + + while (!started.load(std::memory_order_acquire)) { + } + reading.store(true, std::memory_order_release); + u64 reads = 0; - while (s.produced() < kPushes) { + bool sizes_ok = true; + bool finished = false; + do { + finished = done.load(std::memory_order_acquire); 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); + if (n > 0) { + (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 + if (out.size() != 2 * n) sizes_ok = false; + } ++reads; - } + } while (!finished); + producer.join(); + // Checked only after the join: a throw while the producer is joinable would + // std::terminate instead of reporting the failure. PLOT_CHECK(s.produced() == kPushes); - PLOT_CHECK(reads > 0); + PLOT_CHECK(reads >= 1); + PLOT_CHECK(sizes_ok); }