// Unit test for the lossy MPSC log ring (common/include/coop/log_ring.hpp). Two parts: // 1. Deterministic single-threaded wrap: pushing more than `capacity` lines drops the oldest and // drains the rest in order. // 2. Threaded torn-read guard: many producer threads hammer a small ring while one consumer drains // a small ring (so slots wrap constantly). Each line's text is a unique token repeated across the // whole buffer; a torn read (a slot overwritten mid-copy) would mix two tokens, which the consumer // detects. The ring is lossy by design, so we assert lines are never TORN -- not that none are // lost. Without the seqlock copy+recheck (and the producer's in-progress marker), this trips. #include #include #include #include #include #include #include #include "coop/log_ring.hpp" using namespace coop; namespace { int g_failures = 0; void check(bool ok, const char* what) { std::printf("%s %s\n", ok ? " ok:" : "FAIL:", what); if (!ok) { ++g_failures; } } // A line whose whole text is one token ("TS") repeated, space-separated. A torn read // (two generations mixed in one slot) yields tokens that aren't all equal. std::string make_line(unsigned thread, unsigned long long seq) { char token[32]; std::snprintf(token, sizeof(token), "T%02uS%010llu", thread, seq); std::string s; while (s.size() + std::strlen(token) + 1 < kLogMsgLen - 1) { s += token; s += ' '; } return s; } // True if every space-separated token in `text` is identical (i.e. the line wasn't torn). bool line_consistent(const char* text) { std::string first; std::string cur; for (const char* p = text;; ++p) { if (*p == ' ' || *p == '\0') { if (!cur.empty()) { if (first.empty()) { first = cur; } else if (cur != first) { return false; } cur.clear(); } if (*p == '\0') { break; } } else { cur.push_back(*p); } } return !first.empty(); } std::vector make_ring(std::uint32_t capacity) { std::vector buf(log_ring_total_size(capacity), 0); log_ring_init(*reinterpret_cast(buf.data()), capacity); return buf; } } // namespace int main() { // 1. Deterministic wrap: capacity 8, push 11 -> the oldest 3 drop, drain 3..10 in order. { auto buf = make_ring(8); auto& ring = *reinterpret_cast(buf.data()); for (unsigned long long n = 0; n < 11; ++n) { const std::string line = make_line(0, n); log_ring_push(ring, 1234, LogLevel_Info, n, line.c_str()); } std::vector got; std::uint64_t cursor = 0; log_ring_drain(ring, cursor, [&](const LogRecord& rec) { got.push_back(rec.text); }); check(got.size() == 8, "wrap: drains exactly capacity lines after overflow"); bool ordered = true; for (std::size_t k = 0; k < got.size(); ++k) { ordered = ordered && got[k] == make_line(0, 3 + k); // oldest 3 (0,1,2) dropped } check(ordered, "wrap: keeps the newest `capacity` lines, in order, oldest dropped"); } // 2. Threaded torn-read guard: small ring, several producers, one slow-ish consumer. { constexpr std::uint32_t kCap = 32; auto buf = make_ring(kCap); auto& ring = *reinterpret_cast(buf.data()); std::atomic stop{false}; std::atomic produced{0}; std::vector producers; for (unsigned t = 0; t < 4; ++t) { producers.emplace_back([&, t] { unsigned long long n = 0; while (!stop.load(std::memory_order_relaxed)) { const std::string line = make_line(t, n++); log_ring_push(ring, t, LogLevel_Info, n, line.c_str()); produced.fetch_add(1, std::memory_order_relaxed); } }); } std::atomic consumed{0}; std::atomic torn{0}; std::thread consumer([&] { std::uint64_t cursor = 0; auto drain = [&] { log_ring_drain(ring, cursor, [&](const LogRecord& rec) { consumed.fetch_add(1, std::memory_order_relaxed); if (!line_consistent(rec.text)) { torn.fetch_add(1, std::memory_order_relaxed); } }); }; while (!stop.load(std::memory_order_relaxed)) { drain(); std::this_thread::sleep_for(std::chrono::microseconds(50)); // fall behind so slots wrap } drain(); }); std::this_thread::sleep_for(std::chrono::milliseconds(1500)); stop.store(true, std::memory_order_relaxed); for (auto& p : producers) { p.join(); } consumer.join(); std::printf(" produced=%lld consumed=%lld torn=%lld\n", produced.load(), consumed.load(), torn.load()); check(produced.load() > 100000, "threaded: producers ran a real workload"); check(consumed.load() > 0, "threaded: consumer received lines (ring wrapped, most are lost -- ok)"); check(torn.load() == 0, "threaded: no torn line ever emitted (seqlock copy+recheck holds)"); } std::printf(g_failures == 0 ? "PASS log_ring_test\n" : "FAILED log_ring_test (%d)\n", g_failures); return g_failures == 0 ? 0 : 1; }