diff --git a/README.md b/README.md index 00091be..f3a6867 100644 --- a/README.md +++ b/README.md @@ -113,11 +113,7 @@ default** and covers anything the hooked path doesn't. From an in-depth review pass. Each item is fixed test-first (a failing test, then the fix) and lands as its own commit; "verify" items are confirmed real before any change, and dropped if not. -Cross-process / ABI: -- **`log_ring` torn-text window** — verify the MPSC overwrite race; fix or bound it. - Test coverage: -- **`log_ring`** — threaded push/drain + wrap-skip generation test (currently zero coverage). - **Seqlock reader paths** — torn-read retry, odd-sequence skip, attempt-exhaustion → false. - **Version/magic mismatch rejection** — negative test that the hook refuses a bad version. - **Dedicated hook tests** — `focus_spoof`, `vk_hook` (present), `d3d9_hook`. diff --git a/common/include/coop/log_ring.hpp b/common/include/coop/log_ring.hpp index ffda442..f975134 100644 --- a/common/include/coop/log_ring.hpp +++ b/common/include/coop/log_ring.hpp @@ -92,12 +92,19 @@ inline void log_ring_push(LogRing& r, std::uint32_t pid, std::uint32_t level, st { const std::uint64_t idx = r.write_index.fetch_add(1, std::memory_order_acq_rel); LogRecord& rec = log_ring_records(&r)[idx % r.capacity]; + // Seqlock write. Mark the slot in-progress (seq 0) and fence BEFORE touching the record, so a + // consumer still reading the slot's previous occupant sees seq change and bails instead of reading + // half-overwritten text; publish the new generation only after the text is fully written. Without + // this the consumer's single seq check passed before the read, so an overwrite mid-read tore it. + rec.seq.store(0, std::memory_order_relaxed); + std::atomic_thread_fence(std::memory_order_release); rec.pid = pid; rec.level = level; rec.millis = millis; std::strncpy(rec.text, text, kLogMsgLen - 1); rec.text[kLogMsgLen - 1] = '\0'; - rec.seq.store(idx + 1, std::memory_order_release); // publish: record is ready + std::atomic_thread_fence(std::memory_order_release); + rec.seq.store(idx + 1, std::memory_order_relaxed); // publish: generation idx is ready } // Consumer (host): emit each new record since `cursor` (advanced in place). Skips @@ -116,16 +123,28 @@ inline void log_ring_drain(LogRing& r, std::uint64_t& cursor, F&& emit) for (; i < w; ++i) { LogRecord& rec = recs[i % r.capacity]; - const std::uint64_t s = rec.seq.load(std::memory_order_acquire); - if (s == i + 1) + const std::uint64_t s1 = rec.seq.load(std::memory_order_acquire); + if (s1 <= i) { - emit(rec); // ready + break; // generation i not written yet (in-flight, or being overwritten); retry next call } - else if (s <= i) + if (s1 != i + 1) { - break; // slot not written for this generation yet (in-flight); retry later + continue; // s1 > i+1: overwritten by a later generation before we got here; lost, skip } - // s > i + 1: overwritten before we read it; skip (lost) + // Seqlock read: copy the record out, then re-check seq. A producer overwriting this slot stores + // seq 0 before it writes and the new generation after, so any change means our copy may be torn. + LogRecord snap{}; + snap.pid = rec.pid; + snap.level = rec.level; + snap.millis = rec.millis; + std::memcpy(snap.text, rec.text, kLogMsgLen); + std::atomic_thread_fence(std::memory_order_acquire); + if (rec.seq.load(std::memory_order_relaxed) == i + 1) + { + emit(snap); // consistent snapshot + } + // else: overwritten while we copied -> skip (lost) } cursor = i; } diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 33665e9..439a96b 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -76,6 +76,12 @@ add_executable(mkb_ring_test mkb_ring_test.cpp) target_link_libraries(mkb_ring_test PRIVATE coop_common) add_test(NAME mkb_ring_test COMMAND mkb_ring_test) +# Unit test for the lossy MPSC log ring: deterministic wrap-drop + a threaded torn-read guard +# (many producers + a slow consumer on a small ring) that the seqlock copy+recheck must keep clean. +add_executable(log_ring_test log_ring_test.cpp) +target_link_libraries(log_ring_test PRIVATE coop_common) +add_test(NAME log_ring_test COMMAND log_ring_test) + # Unit test for the audio mixer math (decode/sum/soft-clip/encode). Header-only. add_executable(audio_mix_test audio_mix_test.cpp) target_include_directories(audio_mix_test PRIVATE ${CMAKE_SOURCE_DIR}/host/src) @@ -321,6 +327,7 @@ coop_output_subdir(tests detour_gate_test hook_install_test mkb_ring_test + log_ring_test mkb_map_test audio_mix_test tone_analysis_test diff --git a/tests/log_ring_test.cpp b/tests/log_ring_test.cpp new file mode 100644 index 0000000..022cc0d --- /dev/null +++ b/tests/log_ring_test.cpp @@ -0,0 +1,172 @@ +// 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; +}