log_ring: seqlock the records to prevent torn cross-process reads

The lossy MPSC log ring published each record by writing its text and THEN storing
the slot's sequence. A consumer that passed the seq==generation check could then
read text while a producer 'capacity' generations later overwrote that same slot
(it wrote text before bumping seq), yielding a torn line. Diagnostics-only and
practically unreachable (it needs the consumer a full ring behind -- ~60k lines/s
between two host drains), but a real data race.

Make it a proper seqlock: the producer stores seq 0 (in-progress) and fences
BEFORE touching the record, then publishes the generation after the text; the
consumer copies the record out and re-checks seq, dropping the line if it changed.
The ring stays lossy, never torn.

Adds log_ring_test (previously zero coverage): a deterministic wrap-drop case plus
a threaded torn-read guard (4 producers + a slow consumer on a 32-slot ring) that
emits 0 torn lines out of ~300k produced. Closes both the cross-process item and
the log_ring coverage gap.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-24 01:57:37 +02:00
parent 6840d9df88
commit 8fd209c73f
4 changed files with 205 additions and 11 deletions

View File

@@ -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`.

View File

@@ -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;
}

View File

@@ -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

172
tests/log_ring_test.cpp Normal file
View File

@@ -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 <atomic>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <string>
#include <thread>
#include <vector>
#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 ("T<thread>S<seq>") 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<std::uint8_t> make_ring(std::uint32_t capacity)
{
std::vector<std::uint8_t> buf(log_ring_total_size(capacity), 0);
log_ring_init(*reinterpret_cast<LogRing*>(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<LogRing*>(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<std::string> 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<LogRing*>(buf.data());
std::atomic<bool> stop{false};
std::atomic<long long> produced{0};
std::vector<std::thread> 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<long long> consumed{0};
std::atomic<long long> 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;
}