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.
This commit is contained in:
1 parent
8f438b049b
commit
9a086b7d69
1 file changed
+42
-11
+42
-11
@@ -2,6 +2,7 @@
|
|||||||
#include <plot/data/array_series.hpp>
|
#include <plot/data/array_series.hpp>
|
||||||
#include <plot/data/stream_series.hpp>
|
#include <plot/data/stream_series.hpp>
|
||||||
#include <algorithm>
|
#include <algorithm>
|
||||||
|
#include <atomic>
|
||||||
#include <cmath>
|
#include <cmath>
|
||||||
#include <random>
|
#include <random>
|
||||||
#include <thread>
|
#include <thread>
|
||||||
@@ -104,27 +105,57 @@ PLOT_TEST(stream_series_exact_queries_survive_ring_wrap) {
|
|||||||
PLOT_TEST(stream_series_push_and_read_are_race_free) {
|
PLOT_TEST(stream_series_push_and_read_are_race_free) {
|
||||||
// The ring overwrites, so the producer reuses the slot of the snapshot's oldest
|
// 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
|
// 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
|
// slots make that defined; nothing else in the suite has two threads, so this is
|
||||||
// is the case ThreadSanitizer actually flags when that is not the case.
|
// the case ThreadSanitizer actually flags when the slots are plain.
|
||||||
plot::data::StreamSeries s(1u << 8, 0.0, 1.0);
|
//
|
||||||
constexpr u64 kPushes = 50'000;
|
// The two flags make the overlap structural, not scheduling luck: the reader
|
||||||
std::thread producer([&s] {
|
// waits until the producer is already pushing, acknowledges, and only then does
|
||||||
for (u64 i = 0; i < kPushes; ++i) s.push(f32(i % 971));
|
// 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<bool> started{false};
|
||||||
|
std::atomic<bool> reading{false};
|
||||||
|
std::atomic<bool> 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;
|
u64 reads = 0;
|
||||||
while (s.produced() < kPushes) {
|
bool sizes_ok = true;
|
||||||
|
bool finished = false;
|
||||||
|
do {
|
||||||
|
finished = done.load(std::memory_order_acquire);
|
||||||
s.acquire();
|
s.acquire();
|
||||||
const u64 n = s.size();
|
const u64 n = s.size();
|
||||||
if (n == 0) continue;
|
if (n > 0) {
|
||||||
(void)s.reduce(0, n);
|
(void)s.reduce(0, n);
|
||||||
(void)s.sum(0, n);
|
(void)s.sum(0, n);
|
||||||
(void)s.sumsq(0, n);
|
(void)s.sumsq(0, n);
|
||||||
std::vector<f32> out;
|
std::vector<f32> out;
|
||||||
s.gather(0, 0, n, out); // reads the oldest edge
|
s.gather(0, 0, n, out); // reads the oldest edge
|
||||||
PLOT_CHECK(out.size() == 2 * n);
|
if (out.size() != 2 * n) sizes_ok = false;
|
||||||
|
}
|
||||||
++reads;
|
++reads;
|
||||||
}
|
} while (!finished);
|
||||||
|
|
||||||
producer.join();
|
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(s.produced() == kPushes);
|
||||||
PLOT_CHECK(reads > 0);
|
PLOT_CHECK(reads >= 1);
|
||||||
|
PLOT_CHECK(sizes_ok);
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user