GCC Code Coverage Report

Directory: ./
Coverage: low: ≥ 0% medium: ≥ 75.0% high: ≥ 90.0%
Coverage Exec / Excl / Total
Lines: 91.6% 489 / 0 / 534
Functions: 100.0% 59 / 0 / 59
Branches: 74.9% 257 / 0 / 343
src/internal/async_logger.cpp
Line Branch Exec Source
1 #include "internal/async_logger.hpp"
2
3 #include "DetourModKit/diagnostics.hpp"
4
5 #include "internal/async_logger_queue.hpp"
6 #include "internal/lifecycle_context.hpp"
7 #include "platform.hpp"
8 #include "internal/win_file_stream.hpp"
9
10 #include <algorithm>
11 #include <condition_variable>
12 #include <cstdint>
13 #include <cstring>
14 #include <iomanip>
15 #include <new>
16 #include <span>
17 #include <system_error>
18 #include <thread>
19 #include <type_traits>
20
21 namespace DetourModKit
22 {
23 using detail::acquire_module_ref;
24 using detail::LogMessage;
25 using detail::release_module_ref;
26
27 // The string pool, per-message record, and MPMC queue are implementation-only types that live in
28 // DetourModKit::detail (see internal/async_logger_queue.hpp). Their out-of-line definitions are grouped in the
29 // namespace block below; the AsyncLogger pimpl definitions that follow reach them through the using-declarations
30 // above.
31 namespace detail
32 {
33 #if defined(DMK_ENABLE_TEST_SEAMS)
34 // Lets tests drive the writer-detach-and-leak branch without entering the real loader lock.
35 bool (*g_async_logger_loader_lock_override)() noexcept = nullptr;
36
37 // AsyncLogger instances currently alive. The retention root is a self-reference, so "was this writer retained
38 // or destroyed?" is otherwise unobservable from outside: a retained writer is reachable only from itself.
39 std::atomic<std::size_t> g_async_logger_live_count_for_test{0};
40 #endif
41
42 398 bool async_logger_must_not_block() noexcept
43 {
44 #if defined(DMK_ENABLE_TEST_SEAMS)
45 398 return !blocking_teardown_permitted(g_async_logger_loader_lock_override);
46 #else
47 return !blocking_teardown_permitted();
48 #endif
49 }
50
51 #if defined(DMK_ENABLE_TEST_SEAMS)
52 // Test-only gates for deterministic shutdown interleavings.
53 std::atomic<std::atomic<bool> *> g_async_logger_writer_gate{nullptr};
54 std::atomic<std::atomic<bool> *> g_async_logger_producer_gate{nullptr};
55 std::atomic<bool> g_async_logger_producer_waiting{false};
56 std::atomic<bool> g_async_logger_flush_waiting{false};
57 // Pauses a producer AFTER it increments m_pending_messages but BEFORE it publishes the queue slot, so a test
58 // can hold the writer in the in-flight window (pending != 0, queue empty) and prove it parks rather than
59 // busy-spinning.
60 std::atomic<std::atomic<bool> *> g_async_logger_prepush_gate{nullptr};
61 std::atomic<bool> g_async_logger_prepush_waiting{false};
62 // Counts each entry into the writer's idle park branch, so a test can prove a producer preempted mid-publish
63 // does not turn the idle path into an unbounded busy loop.
64 std::atomic<std::atomic<size_t> *> g_async_logger_idle_park_counter{nullptr};
65 // Counts each record drained through the writer's one-record progress floor. The floor keeps the writer live
66 // when a batch reservation cannot be secured, so it is reached only on that path: a test proves both that a
67 // refused reservation really did route through it, and that an ordinary configuration never does.
68 std::atomic<std::atomic<size_t> *> g_async_logger_batch_floor_counter{nullptr};
69 // Makes flush_with_timeout hold m_flush_mutex and spin until the gate clears, so a test can prove a
70 // Drop-policy producer completes its enqueue without acquiring that control-plane mutex.
71 std::atomic<std::atomic<bool> *> g_async_logger_flush_mutex_gate{nullptr};
72 // Counts the final-producer wake-event signal in finish_producer, so a test can prove the parked stopping
73 // writer was released by that signal rather than by its flush-interval timeout.
74 std::atomic<std::atomic<size_t> *> g_async_logger_finish_wake_counter{nullptr};
75 // Counts entry into the Block overflow branch after admission. This gives shutdown tests an exact handshake
76 // for the retry phase instead of a timing guess.
77 std::atomic<std::atomic<size_t> *> g_async_logger_block_entry_counter{nullptr};
78 // Pauses a Block producer at the retry boundary. The deadline proof releases this gate after Stopping is
79 // visible, and production establishes the deadline immediately after the release.
80 std::atomic<std::atomic<bool> *> g_async_logger_block_entry_gate{nullptr};
81 // Publishes the Block retry-loop start time from the producer thread for a scheduling-independent deadline
82 // assertion. Nanoseconds use steady_clock's epoch only as an opaque common origin.
83 std::atomic<std::atomic<std::int64_t> *> g_async_logger_block_start_ns{nullptr};
84 // Runs after a batch has been formatted but before its final flush so a test can invalidate the sink at the
85 // exact durability boundary.
86 void (*g_async_logger_before_flush_probe)(WinFileStream &) noexcept = nullptr;
87 #endif
88
89 20 StringPool::StringPool() noexcept
90 {
91 20 std::lock_guard<std::mutex> lock(m_pool_mutex);
92 20 grow_pool_locked();
93 20 }
94
95 25 void StringPool::grow_pool_locked() noexcept
96 {
97 25 Block *existing = m_head.load(std::memory_order_relaxed);
98 25 size_t count = 0;
99
2/2
✓ Branch 7 → 4 taken 11 times.
✓ Branch 7 → 8 taken 25 times.
 36 for (Block *b = existing; b; b = b->next)
100 {
101
1/2
✗ Branch 4 → 5 not taken.
✓ Branch 4 → 6 taken 11 times.
 11 if (++count >= MEMORY_POOL_BLOCK_COUNT)
102 {
103 ✗ return;
104 }
105 }
106
107 // Block is over-aligned via its alignas(64) data member, so it must be allocated through the aligned
108 // operator new; the plain overload is not required to honour an alignment stricter than
109 // __STDCPP_DEFAULT_NEW_ALIGNMENT__ (typically 16 on x64), which would be alignment UB. The allocation is
110 // also nothrow: this runs underneath the noexcept logging path, so on out-of-memory it must leave the pool
111 // unchanged and let the caller fall back to a nothrow heap string (or drop the message) rather than let
112 // std::bad_alloc escape and terminate.
113 25 void *raw = ::operator new(sizeof(Block), std::align_val_t{alignof(Block)}, std::nothrow);
114
1/2
✗ Branch 9 → 10 not taken.
✓ Branch 9 → 11 taken 25 times.
 25 if (!raw)
115 {
116 ✗ return;
117 }
118
1/2
✗ Branch 13 → 14 not taken.
✓ Branch 13 → 15 taken 25 times.
 25 Block *new_block = new (raw) Block();
119
120 25 new_block->next = existing;
121 25 new_block->free_list = nullptr;
122
123 25 PoolSlot *slots = reinterpret_cast<PoolSlot *>(new_block->data);
124 // Slot construction must not throw, otherwise a partially built block could leak with no unwinding under
125 // this noexcept function. std::string's default constructor is noexcept, so the loop below is provably
126 // no-throw.
127 static_assert(
128 std::is_nothrow_default_constructible_v<PoolSlot>,
129 "PoolSlot must be nothrow-default-constructible so grow_pool_locked stays no-throw"
130 );
131
2/2
✓ Branch 24 → 16 taken 400 times.
✓ Branch 24 → 25 taken 25 times.
 425 for (size_t i = 0; i < POOL_SLOTS_PER_BLOCK; ++i)
132 {
133
1/2
✗ Branch 18 → 19 not taken.
✓ Branch 18 → 20 taken 400 times.
 400 new (&slots[i]) PoolSlot();
134
2/2
✓ Branch 20 → 21 taken 375 times.
✓ Branch 20 → 22 taken 25 times.
 400 slots[i].next_free = (i + 1 < POOL_SLOTS_PER_BLOCK) ? &slots[i + 1] : nullptr;
135 }
136 25 new_block->free_list = &slots[0];
137
138 25 m_head.store(new_block, std::memory_order_release);
139 }
140
141 31 StringPool &StringPool::instance() noexcept
142 {
143 // Constructed once into function-local static storage and never destroyed. A Meyers singleton would be
144 // destroyed at static teardown and race late LogMessage destructors that call into deallocate()
145 // (use-after-free under DLL unload and loader-lock teardown). A heap-allocated singleton (`*new
146 // StringPool()`) would instead require a throwing operator new whose std::bad_alloc would escape this
147 // noexcept accessor and terminate the host. Placement-new into static storage avoids both: the object lives
148 // for the whole process, its destructor never runs, and construction performs no throwing allocation
149 // because grow_pool_locked() is nothrow. The bounded block leak (at most MEMORY_POOL_BLOCK_COUNT blocks,
150 // each a POOL_SLOTS_PER_BLOCK * sizeof(PoolSlot)-byte slot array) is released by the OS at process exit.
151 alignas(StringPool) static unsigned char storage[sizeof(StringPool)];
152
4/6
✓ Branch 2 → 3 taken 20 times.
✓ Branch 2 → 10 taken 11 times.
✓ Branch 4 → 5 taken 20 times.
✗ Branch 4 → 10 not taken.
✗ Branch 8 → 9 not taken.
✓ Branch 8 → 10 taken 20 times.
 31 static StringPool *const pool = ::new (static_cast<void *>(storage)) StringPool();
153 31 return *pool;
154 }
155
156 513 StringPool::PoolSlot *StringPool::claim_free_slot() noexcept
157 {
158
2/2
✓ Branch 6 → 3 taken 519 times.
✓ Branch 6 → 7 taken 5 times.
 524 for (Block *b = m_head.load(std::memory_order_relaxed); b; b = b->next)
159 {
160
2/2
✓ Branch 3 → 4 taken 508 times.
✓ Branch 3 → 5 taken 11 times.
 519 if (b->free_list)
161 {
162 508 PoolSlot *slot = b->free_list;
163 508 b->free_list = slot->next_free;
164 508 return slot;
165 }
166 }
167 5 return nullptr;
168 }
169
170 513 std::string *StringPool::allocate(size_t size) noexcept
171 {
172
2/2
✓ Branch 2 → 3 taken 6 times.
✓ Branch 2 → 12 taken 507 times.
 513 if (size > MAX_POOLED_STRING_SIZE)
173 {
174
5/6
✓ Branch 4 → 5 taken 4 times.
✓ Branch 4 → 7 taken 2 times.
✓ Branch 8 → 9 taken 4 times.
✓ Branch 8 → 11 taken 2 times.
✗ Branch 9 → 10 not taken.
✓ Branch 9 → 11 taken 4 times.
 6 return new (std::nothrow) std::string();
175 }
176
177 507 std::lock_guard<std::mutex> lock(m_pool_mutex);
178
179 508 PoolSlot *slot = claim_free_slot();
180
2/2
✓ Branch 14 → 15 taken 5 times.
✓ Branch 14 → 17 taken 503 times.
 508 if (!slot)
181 {
182 5 grow_pool_locked();
183 5 slot = claim_free_slot();
184 }
185
186
1/2
✓ Branch 17 → 18 taken 508 times.
✗ Branch 17 → 20 not taken.
 508 if (slot)
187 {
188 508 slot->str.clear();
189 508 return &slot->str;
190 }
191
192 ✗ return new (std::nothrow) std::string();
193 508 }
194
195 510 void StringPool::deallocate(std::string *ptr) noexcept
196 {
197
1/2
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 4 taken 510 times.
 510 if (!ptr)
198 508 return;
199
200 510 std::lock_guard<std::mutex> lock(m_pool_mutex);
201
202
2/2
✓ Branch 12 → 6 taken 688 times.
✓ Branch 12 → 13 taken 4 times.
 692 for (Block *b = m_head.load(std::memory_order_relaxed); b; b = b->next)
203 {
204 688 const auto *block_begin = reinterpret_cast<const char *>(b->data);
205 688 const auto *block_end = block_begin + POOL_SLOTS_PER_BLOCK * sizeof(PoolSlot);
206 688 const auto *raw_ptr = reinterpret_cast<const char *>(ptr);
207
208
3/4
✓ Branch 6 → 7 taken 508 times.
✓ Branch 6 → 11 taken 180 times.
✓ Branch 7 → 8 taken 508 times.
✗ Branch 7 → 11 not taken.
 688 if (raw_ptr >= block_begin && raw_ptr < block_end)
209 {
210 508 auto offset = static_cast<size_t>(raw_ptr - block_begin);
211 508 PoolSlot *slot = reinterpret_cast<PoolSlot *>(b->data) + (offset / sizeof(PoolSlot));
212 508 slot->str.clear();
213 508 return_slot_locked(slot, b);
214 508 return;
215 }
216 }
217
218 // Not a pool allocation: heap fallback. The delete is performed under m_pool_mutex to serialize with
219 // concurrent deallocate() calls that walk the block list above. Without the lock, a concurrent deallocate
220 // could see a partially updated free list. The lock does not prevent double-free of heap pointers (those
221 // are not tracked); callers must ensure each pointer is deallocated exactly once. The cost is a single
222 // free() call (or no-op for SSO-sized strings).
223
1/2
✓ Branch 13 → 14 taken 4 times.
✗ Branch 13 → 16 not taken.
 4 delete ptr;
224
2/2
✓ Branch 18 → 19 taken 4 times.
✓ Branch 18 → 21 taken 508 times.
 512 }
225
226 508 void StringPool::return_slot_locked(PoolSlot *slot, Block *block) noexcept
227 {
228 508 slot->next_free = block->free_list;
229 508 block->free_list = slot;
230 508 }
231
232 // buffer is intentionally left uninitialized on this hot path (see async_logger_queue.hpp): only [0, length)
233 // is written before any read, so zero-filling it every message would be wasted work.
234 // NOLINTNEXTLINE(cppcoreguidelines-pro-type-member-init)
235 7157 LogMessage::LogMessage(LogLevel lvl, std::string_view msg) noexcept
236 7157 : level(lvl), timestamp(std::chrono::system_clock::now())
237 {
238 7158 const size_t msg_size = std::min(msg.size(), MAX_VALID_LENGTH);
239
240
2/2
✓ Branch 5 → 6 taken 7068 times.
✓ Branch 5 → 10 taken 13 times.
 7081 if (msg_size <= MAX_INLINE_SIZE)
241 {
242 7068 std::memcpy(buffer.data(), msg.data(), msg_size);
243 7046 length = msg_size;
244 }
245 else
246 {
247 13 overflow = StringPool::instance().allocate(msg_size);
248
2/2
✓ Branch 12 → 13 taken 11 times.
✓ Branch 12 → 17 taken 2 times.
 13 if (overflow)
249 {
250 try
251 {
252
2/4
✓ Branch 13 → 14 taken 11 times.
✗ Branch 13 → 19 not taken.
✓ Branch 14 → 15 taken 11 times.
✗ Branch 14 → 19 not taken.
 11 overflow->assign(msg.substr(0, msg_size));
253 11 length = overflow->size();
254 }
255 ✗ catch (...)
256 {
257 ✗ StringPool::instance().deallocate(overflow);
258 ✗ overflow = nullptr;
259 ✗ length = 0;
260 // The assign threw under OOM: mark the record failed so the producer drops and counts it
261 // rather than enqueuing a zero-length husk.
262 ✗ failed = true;
263 ✗ }
264 }
265 else
266 {
267 // Overflow allocation failed (OOM): mark the record failed (not merely empty) so it is dropped
268 // and counted, never enqueued as an empty timestamped line.
269 2 length = 0;
270 2 failed = true;
271 }
272 }
273 7059 }
274
275 2810753 LogMessage::~LogMessage() noexcept
276 {
277 2810753 reset();
278 2810751 }
279
280 // Move transfers ownership of the overflow pointer without touching the StringPool. Exactly one LogMessage
281 // owns the pointer at any time, and only reset() (called by the eventual owner's destructor) returns it.
282 // NOLINTNEXTLINE(cppcoreguidelines-pro-type-member-init) buffer is filled by the length-guarded memcpy below
283 3333 LogMessage::LogMessage(LogMessage &&other) noexcept
284 3333 : level(other.level), timestamp(other.timestamp), length(other.length), overflow(other.overflow),
285 3333 failed(other.failed)
286 {
287
4/4
✓ Branch 2 → 3 taken 3331 times.
✓ Branch 2 → 9 taken 2 times.
✓ Branch 3 → 4 taken 3328 times.
✓ Branch 3 → 9 taken 3 times.
 3333 if (length > 0 && !overflow)
288 {
289 9984 std::memcpy(buffer.data(), other.buffer.data(), length);
290 }
291 3333 other.overflow = nullptr;
292 3333 other.length = 0;
293 3333 other.failed = false;
294 3333 }
295
296 11673 LogMessage &LogMessage::operator=(LogMessage &&other) noexcept
297 {
298
1/2
✓ Branch 2 → 3 taken 11673 times.
✗ Branch 2 → 12 not taken.
 11673 if (this != &other)
299 {
300 11673 reset();
301 11666 level = other.level;
302 11666 timestamp = other.timestamp;
303 11666 length = other.length;
304 11666 overflow = other.overflow;
305 11666 failed = other.failed;
306
3/4
✓ Branch 4 → 5 taken 11666 times.
✗ Branch 4 → 11 not taken.
✓ Branch 5 → 6 taken 11663 times.
✓ Branch 5 → 11 taken 3 times.
 11666 if (length > 0 && !overflow)
307 {
308 34989 std::memcpy(buffer.data(), other.buffer.data(), length);
309 }
310 11666 other.overflow = nullptr;
311 11666 other.length = 0;
312 11666 other.failed = false;
313 }
314 11666 return *this;
315 }
316
317 3651 std::string_view LogMessage::message() const noexcept
318 {
319
2/2
✓ Branch 2 → 3 taken 9 times.
✓ Branch 2 → 4 taken 3642 times.
 3651 if (overflow)
320 {
321 9 return *overflow;
322 }
323 3642 return std::string_view(buffer.data(), length);
324 }
325
326 5037 bool LogMessage::is_valid() const noexcept
327 {
328
2/2
✓ Branch 2 → 3 taken 2 times.
✓ Branch 2 → 4 taken 5035 times.
 5037 if (failed)
329 {
330 2 return false;
331 }
332
2/2
✓ Branch 4 → 5 taken 8 times.
✓ Branch 4 → 7 taken 5027 times.
 5035 if (overflow)
333 {
334 8 return length == overflow->size();
335 }
336 5027 return length <= MAX_INLINE_SIZE;
337 }
338
339 2822346 void LogMessage::reset() noexcept
340 {
341
2/2
✓ Branch 2 → 3 taken 11 times.
✓ Branch 2 → 6 taken 2822335 times.
 2822346 if (overflow)
342 {
343 11 StringPool::instance().deallocate(overflow);
344 11 overflow = nullptr;
345 }
346 2822346 length = 0;
347 2822346 failed = false;
348 2822346 }
349
350 428 size_t DynamicMPMCQueue::validated_capacity(size_t capacity)
351 {
352
4/4
✓ Branch 2 → 3 taken 422 times.
✓ Branch 2 → 4 taken 6 times.
✓ Branch 3 → 4 taken 3 times.
✓ Branch 3 → 7 taken 419 times.
 428 if ((capacity & (capacity - 1)) != 0 || capacity < 2)
353 {
354
1/2
✓ Branch 5 → 6 taken 9 times.
✗ Branch 5 → 9 not taken.
 9 throw std::invalid_argument("DynamicMPMCQueue capacity must be a power of 2 and at least 2");
355 }
356 419 return capacity;
357 }
358
359 428 DynamicMPMCQueue::DynamicMPMCQueue(size_t capacity)
360 428 : m_capacity(validated_capacity(capacity)), m_mask(m_capacity - 1),
361 419 m_buffer(std::make_unique<Slot[]>(m_capacity))
362 {
363
2/2
✓ Branch 17 → 7 taken 2847604 times.
✓ Branch 17 → 18 taken 419 times.
 2848023 for (size_t i = 0; i < m_capacity; ++i)
364 {
365 2847604 m_buffer[i].sequence.store(i, std::memory_order_relaxed);
366 }
367 419 }
368
369 11275 bool DynamicMPMCQueue::try_push(LogMessage &item) noexcept
370 {
371 22595 size_t pos = m_enqueue_pos.load(std::memory_order_relaxed);
372
373 for (;;)
374 {
375 12001 Slot &slot = m_buffer[pos & m_mask];
376 12049 size_t seq = slot.sequence.load(std::memory_order_acquire);
377 12066 intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
378
379
2/2
✓ Branch 18 → 19 taken 6461 times.
✓ Branch 18 → 40 taken 5605 times.
 12066 if (diff == 0)
380 {
381
2/2
✓ Branch 27 → 28 taken 5840 times.
✓ Branch 27 → 50 taken 660 times.
 12961 if (m_enqueue_pos.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed))
382 {
383 5840 slot.data = std::move(item);
384 5840 slot.sequence.store(pos + 1, std::memory_order_release);
385 5839 return true;
386 }
387 }
388
2/2
✓ Branch 40 → 41 taken 5590 times.
✓ Branch 40 → 42 taken 15 times.
 5605 else if (diff < 0)
389 {
390 5590 return false;
391 }
392 else
393 {
394 36 pos = m_enqueue_pos.load(std::memory_order_relaxed);
395 }
396 681 }
397 }
398
399 7441 bool DynamicMPMCQueue::try_pop(LogMessage &item) noexcept
400 {
401 14882 size_t pos = m_dequeue_pos.load(std::memory_order_relaxed);
402
403 for (;;)
404 {
405 9227 Slot &slot = m_buffer[pos & m_mask];
406 8968 size_t seq = slot.sequence.load(std::memory_order_acquire);
407 8910 intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
408
409
2/2
✓ Branch 18 → 19 taken 7176 times.
✓ Branch 18 → 40 taken 1734 times.
 8910 if (diff == 0)
410 {
411
2/2
✓ Branch 27 → 28 taken 5835 times.
✓ Branch 27 → 50 taken 1539 times.
 14550 if (m_dequeue_pos.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed))
412 {
413 11670 item = std::move(slot.data);
414 5835 slot.sequence.store(pos + m_capacity, std::memory_order_release);
415 5835 return true;
416 }
417 }
418
2/2
✓ Branch 40 → 41 taken 1605 times.
✓ Branch 40 → 42 taken 129 times.
 1734 else if (diff < 0)
419 {
420 1605 return false;
421 }
422 else
423 {
424 376 pos = m_dequeue_pos.load(std::memory_order_relaxed);
425 }
426 1786 }
427 }
428
429 1764 size_t DynamicMPMCQueue::try_pop_batch(std::vector<LogMessage> &items, size_t max_count) noexcept
430 {
431
2/2
✓ Branch 2 → 3 taken 1 time.
✓ Branch 2 → 4 taken 1763 times.
 1764 if (max_count == 0)
432 {
433 1 return 0;
434 }
435
436 // Reserve headroom so the push_back loop below never reallocates. This runs on the writer's noexcept
437 // frames, so a throwing allocator here must not escape: on bad_alloc the reserve is skipped and the pop
438 // is capped to whatever spare capacity the vector already holds. Failing closed to a smaller batch is
439 // correct behaviour for an out-of-memory host; terminating it is not.
440 try
441 {
442
2/2
✓ Branch 5 → 6 taken 1562 times.
✓ Branch 5 → 24 taken 201 times.
 1763 items.reserve(items.size() + max_count);
443 }
444 201 catch (...)
445 {
446 201 }
447
448 // Never pop more than fits in the reserved capacity. Within capacity, push_back performs no allocation and
449 // the LogMessage move constructor is noexcept, so the loop cannot throw even if the reserve above failed.
450 1763 const size_t headroom = items.capacity() - items.size();
451 1763 const size_t budget = std::min(max_count, headroom);
452
453 1763 size_t count = 0;
454 1763 LogMessage msg;
455
456
6/6
✓ Branch 15 → 16 taken 4621 times.
✓ Branch 15 → 19 taken 467 times.
✓ Branch 17 → 18 taken 3325 times.
✓ Branch 17 → 19 taken 1296 times.
✓ Branch 20 → 11 taken 3325 times.
✓ Branch 20 → 21 taken 1763 times.
 5088 while (count < budget && try_pop(msg))
457 {
458 3325 items.push_back(std::move(msg));
459 3325 ++count;
460 }
461
462 1763 return count;
463 1763 }
464
465 1720 size_t DynamicMPMCQueue::size() const noexcept
466 {
467 1720 size_t enq = m_enqueue_pos.load(std::memory_order_relaxed);
468 1720 size_t deq = m_dequeue_pos.load(std::memory_order_relaxed);
469
1/2
✓ Branch 16 → 17 taken 1720 times.
✗ Branch 16 → 18 not taken.
 1720 return (enq >= deq) ? (enq - deq) : 0;
470 }
471
472 1696 bool DynamicMPMCQueue::empty() const noexcept
473 {
474 1696 return size() == 0;
475 }
476 } // namespace detail
477
478 // The AsyncLogger pimpl: every member and method that touches the queue, string pool, writer thread, or flush
479 // synchronization lives here, so the header (internal/async_logger.hpp) names none of it. AsyncLogger forwards
480 // each public call to the matching Impl method.
481 struct AsyncLogger::Impl
482 {
483 Impl(
484 const AsyncLoggerConfig &config,
485 std::shared_ptr<detail::WinFileStream> file_stream,
486 std::shared_ptr<std::mutex> log_mutex
487 );
488 ~Impl() noexcept;
489
490 Impl(const Impl &) = delete;
491 Impl &operator=(const Impl &) = delete;
492 Impl(Impl &&) = delete;
493 Impl &operator=(Impl &&) = delete;
494
495 [[nodiscard]] bool
496 enqueue(LogLevel level, std::string_view message, std::atomic<std::size_t> *retired_drop_counter) noexcept;
497 [[nodiscard]] bool flush_with_timeout(std::chrono::milliseconds timeout) noexcept;
498 void flush() noexcept;
499 void shutdown() noexcept;
500 [[nodiscard]] bool is_running() const noexcept;
501 #if defined(DMK_ENABLE_TEST_SEAMS)
502 [[nodiscard]] bool is_writer_waiting() const noexcept;
503 #endif
504 [[nodiscard]] size_t queue_size() const noexcept;
505 [[nodiscard]] size_t dropped_count() const noexcept;
506 void reset_dropped_count() noexcept;
507 void set_timestamp_format(std::string timestamp_format) noexcept;
508 void set_file_stream(std::shared_ptr<detail::WinFileStream> file_stream) noexcept;
509 // True once shutdown() detached the writer under the loader lock instead of joining it. The public
510 // destructor reads this to decide whether the Impl (and the queue / wake event / flush channel / file stream
511 // the detached writer still touches) may be destroyed or must be leaked in place.
512 [[nodiscard]] bool writer_was_detached() const noexcept;
513
514 void writer_thread_func() noexcept;
515 void finish_producer() noexcept;
516 [[nodiscard]] size_t write_batch(std::span<const detail::LogMessage> messages) noexcept;
517 bool handle_overflow(detail::LogMessage &&message) noexcept;
518 // Wakes a parked writer after a successful push. SetEvent acquires no DMK control-plane mutex, so a
519 // callback-safe Drop-policy producer cannot stall behind a flusher or the writer here. The producer publishes
520 // the queue slot before it reads m_writer_waiting; the writer publishes m_writer_waiting before it rechecks for
521 // work and parks. Those seq_cst operations close the lost-wakeup window, and the auto-reset event retains one
522 // signal when a producer wins the race immediately before the wait.
523 void notify_writer() noexcept;
524
525 detail::DynamicMPMCQueue m_queue;
526 AsyncLoggerConfig m_config;
527
528 std::shared_ptr<detail::WinFileStream> m_file_stream;
529 std::shared_ptr<std::mutex> m_log_mutex;
530
531 std::jthread m_writer_thread;
532 // Counted reference on the module the writer thread's code lives in, taken before the thread is created.
533 // shutdown() releases it after a clean join, or leaks it on the loader-lock detach path so the writer's code
534 // stays mapped. void* keeps the pimpl header-light; it holds an HMODULE in the implementation. See
535 // detail::acquire_module_ref.
536 void *m_writer_self_ref{nullptr};
537 // Single-owner shutdown state. Async: producers enqueue and the writer drains. Stopping: a stop was
538 // requested, so the retained writer finishes the drain while new callback-safe producers drop and count.
539 // Stopped: the writer has drained, flushed the sink, and acknowledged the drain. The writer owns the
540 // Stopping->Stopped transition at its own tail (the loader-lock abandon path returns and leaves everything to
541 // it); shutdown() only requests the stop.
542 enum class State : std::uint8_t
543 {
544 Async,
545 Stopping,
546 Stopped
547 };
548 std::atomic<State> m_state{State::Stopped};
549 // Producers register before observing m_state. This prevents the writer from exiting while an admitted
550 // producer has not published its queue slot yet.
551 std::atomic<size_t> m_active_producers{0};
552 // Latched by shutdown() when it detaches the writer on the loader-lock path. Read by ~AsyncLogger to keep the
553 // Impl alive past the detached writer. A latched flag (rather than re-querying is_loader_lock_held() in the
554 // destructor) avoids a TOCTOU where the loader-lock state differs between the detach decision and the free.
555 std::atomic<bool> m_writer_detached{false};
556
557 // Flush-drain acknowledgement channel, used only by control-plane flushers (flush_with_timeout) and the
558 // signalers that release them (the writer after a batch drains the pending count to zero, and finish_producer
559 // when the last admitted producer leaves during shutdown). The producer wake path does NOT touch this mutex;
560 // it uses m_wake_event, so a callback-safe producer never contends a control-plane lock.
561 std::mutex m_flush_mutex;
562 std::condition_variable m_flush_cv;
563
564 // Auto-reset event the writer parks on when idle. A producer SetEvents it (mutex-free) to wake the writer;
565 // shutdown() SetEvents it so a parked writer observes the stop. Held as a Win32 HANDLE created before the
566 // writer thread starts. Closed by ~Impl on the normal join path; on the loader-lock detach path the whole
567 // Impl (and this handle) is leaked in place so the still-running detached writer keeps a live event.
568 HANDLE m_wake_event{nullptr};
569
570 // Set true by the writer immediately before it parks on m_wake_event and cleared when it wakes. Producers read
571 // it (seq_cst) after a successful queue push; the seq_cst order makes a racing push either visible to the
572 // writer's pre-park work recheck or visible here as a parked-writer wake through SetEvent.
573 std::atomic<bool> m_writer_waiting{false};
574
575 std::atomic<size_t> m_pending_messages{0};
576 std::atomic<size_t> m_dropped_messages{0};
577 };
578
579 408 AsyncLogger::Impl::Impl(
580 const AsyncLoggerConfig &config,
581 std::shared_ptr<detail::WinFileStream> file_stream,
582 std::shared_ptr<std::mutex> log_mutex
583 408 )
584
1/2
✓ Branch 3 → 4 taken 405 times.
✗ Branch 3 → 89 not taken.
 813 : m_queue(config.queue_capacity), m_config(config), m_file_stream(std::move(file_stream)),
585 810 m_log_mutex(std::move(log_mutex))
586 {
587
2/2
✓ Branch 20 → 21 taken 54 times.
✓ Branch 20 → 22 taken 351 times.
 405 if (m_config.timestamp_format.empty())
588 {
589
1/2
✓ Branch 21 → 22 taken 54 times.
✗ Branch 21 → 77 not taken.
 54 m_config.timestamp_format = DEFAULT_ASYNC_TIMESTAMP_FORMAT;
590 }
591
1/2
✗ Branch 23 → 24 not taken.
✓ Branch 23 → 27 taken 405 times.
 405 if (!m_config.validate())
592 {
593 ✗ throw std::invalid_argument("Invalid AsyncLoggerConfig");
594 }
595 // Bound the drain request to what the queue can ever hold. A larger batch_size buys nothing (the queue cannot
596 // produce more records than its capacity) and costs liveness: the writer reserves batch_size headroom before
597 // popping, and a request past the vector's max_size throws length_error on every single attempt, which is a
598 // permanent zero-progress pop rather than a transient one. Clamping on the owned copy leaves the caller's
599 // configuration untouched, and rejecting instead would turn a harmless oversize into a construction failure on
600 // a path that reaches Session bootstrap.
601 405 m_config.batch_size = std::min(m_config.batch_size, m_config.queue_capacity);
602
603
2/2
✓ Branch 29 → 30 taken 1 time.
✓ Branch 29 → 33 taken 404 times.
 405 if (!m_file_stream)
604 {
605
1/2
✓ Branch 31 → 32 taken 1 time.
✗ Branch 31 → 58 not taken.
 1 throw std::invalid_argument("file_stream cannot be null");
606 }
607
608
2/2
✓ Branch 34 → 35 taken 1 time.
✓ Branch 34 → 38 taken 403 times.
 404 if (!m_log_mutex)
609 {
610
1/2
✓ Branch 36 → 37 taken 1 time.
✗ Branch 36 → 60 not taken.
 1 throw std::invalid_argument("log_mutex cannot be null");
611 }
612
613 // Hold a counted reference on this module before creating the writer thread. Once std::jthread returns, the
614 // writer may already be executing this TU's code, so the keepalive has to predate the thread start. shutdown()
615 // releases it after a clean join or leaks it on the loader-lock detach path.
616 403 const HMODULE writer_self_ref = acquire_module_ref(diagnostics::ModulePinReason::AsyncLogger);
617
1/2
✗ Branch 39 → 40 not taken.
✓ Branch 39 → 44 taken 403 times.
 403 if (writer_self_ref == nullptr)
618 {
619 throw std::system_error(
620 ✗ static_cast<int>(GetLastError()),
621 std::system_category(),
622 "AsyncLogger: acquire_module_ref failed"
623 ✗ );
624 }
625
626 // Auto-reset (bManualReset FALSE), initially non-signaled. Created before the writer thread starts so the
627 // thread can park on it immediately; a producer or shutdown() signals it to wake the parked writer.
628
1/2
✓ Branch 44 → 45 taken 403 times.
✗ Branch 44 → 77 not taken.
 403 m_wake_event = ::CreateEventW(nullptr, FALSE, FALSE, nullptr);
629
1/2
✗ Branch 45 → 46 not taken.
✓ Branch 45 → 51 taken 403 times.
 403 if (m_wake_event == nullptr)
630 {
631 ✗ release_module_ref(writer_self_ref, diagnostics::ModulePinReason::AsyncLogger);
632 throw std::system_error(
633 ✗ static_cast<int>(GetLastError()),
634 std::system_category(),
635 "AsyncLogger: CreateEventW failed"
636 ✗ );
637 }
638
639 403 m_state.store(State::Async, std::memory_order_release);
640 try
641 {
642
1/2
✓ Branch 52 → 53 taken 403 times.
✗ Branch 52 → 66 not taken.
 403 m_writer_thread = std::jthread(&AsyncLogger::Impl::writer_thread_func, this);
643 }
644 ✗ catch (...)
645 {
646 ✗ m_state.store(State::Stopped, std::memory_order_release);
647 ✗ ::CloseHandle(m_wake_event);
648 ✗ m_wake_event = nullptr;
649 ✗ release_module_ref(writer_self_ref, diagnostics::ModulePinReason::AsyncLogger);
650 ✗ throw;
651 ✗ }
652 403 m_writer_self_ref = writer_self_ref;
653 417 }
654
655 370 AsyncLogger::Impl::~Impl() noexcept
656 {
657 370 shutdown();
658 // Reached only off the loader-lock detach path: shutdown() joined the writer, so the wake event is no longer
659 // waited on and is closed here. On the detach path the public ~AsyncLogger leaks this Impl in place (its
660 // release()), so this destructor never runs and the handle stays live for the still-running detached writer.
661
1/2
✓ Branch 3 → 4 taken 370 times.
✗ Branch 3 → 6 not taken.
 370 if (m_wake_event != nullptr)
662 {
663 370 ::CloseHandle(m_wake_event);
664 370 m_wake_event = nullptr;
665 }
666 370 }
667
668 5121 bool AsyncLogger::Impl::enqueue(
669 LogLevel level,
670 std::string_view message,
671 std::atomic<std::size_t> *retired_drop_counter
672 ) noexcept
673 {
674 // The seq_cst registration and state check form an admission handshake with shutdown's state transition. A
675 // producer that observes Async remains visible until it publishes or drops; one that arrives after Stopping
676 // cannot publish.
677 5121 m_active_producers.fetch_add(1, std::memory_order_seq_cst);
678
2/2
✓ Branch 5 → 6 taken 5 times.
✓ Branch 5 → 13 taken 5126 times.
 5121 if (m_state.load(std::memory_order_seq_cst) != State::Async)
679 {
680
2/2
✓ Branch 6 → 7 taken 4 times.
✓ Branch 6 → 8 taken 1 time.
 5 std::atomic<std::size_t> &drop_counter =
681 retired_drop_counter != nullptr ? *retired_drop_counter : m_dropped_messages;
682 5 drop_counter.fetch_add(1, std::memory_order_relaxed);
683 5 finish_producer();
684 5 return false;
685 }
686
687 #if defined(DMK_ENABLE_TEST_SEAMS)
688
2/2
✓ Branch 14 → 15 taken 4 times.
✓ Branch 14 → 21 taken 5095 times.
 5126 if (auto *gate = detail::g_async_logger_producer_gate.load(std::memory_order_acquire))
689 {
690 4 detail::g_async_logger_producer_waiting.store(true, std::memory_order_release);
691
2/2
✓ Branch 19 → 17 taken 522875 times.
✓ Branch 19 → 20 taken 4 times.
 522879 while (gate->load(std::memory_order_acquire))
692 {
693 522875 std::this_thread::yield();
694 }
695 4 detail::g_async_logger_producer_waiting.store(false, std::memory_order_release);
696 }
697 #endif
698
699 5099 LogMessage msg(level, message);
700
2/2
✓ Branch 23 → 24 taken 2 times.
✓ Branch 23 → 28 taken 5190 times.
 5035 if (!msg.is_valid())
701 {
702 // An over-long message whose overflow allocation failed is an invalid, zero-length husk. Drop and count
703 // it rather than enqueue an empty timestamped line that misrepresents the lost content.
704 2 m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
705 2 finish_producer();
706 2 return false;
707 }
708
709 // Increment before push so flush cannot observe zero while a message is already in the queue but not yet
710 // counted.
711 5190 m_pending_messages.fetch_add(1, std::memory_order_seq_cst);
712 #if defined(DMK_ENABLE_TEST_SEAMS)
713 // Test-only pause in the in-flight window: pending is counted but the slot is not published yet.
714
2/2
✓ Branch 31 → 32 taken 1 time.
✓ Branch 31 → 38 taken 5105 times.
 5190 if (auto *gate = detail::g_async_logger_prepush_gate.load(std::memory_order_acquire))
715 {
716 1 detail::g_async_logger_prepush_waiting.store(true, std::memory_order_release);
717
2/2
✓ Branch 36 → 34 taken 482185 times.
✓ Branch 36 → 37 taken 1 time.
 482186 while (gate->load(std::memory_order_acquire))
718 {
719 482185 std::this_thread::yield();
720 }
721 1 detail::g_async_logger_prepush_waiting.store(false, std::memory_order_release);
722 }
723 #endif
724
2/2
✓ Branch 39 → 40 taken 3492 times.
✓ Branch 39 → 43 taken 1718 times.
 5106 if (m_queue.try_push(msg))
725 {
726 3492 notify_writer();
727 3500 finish_producer();
728 3496 return true;
729 }
730 // Push failed: undo the pre-increment before entering overflow handling
731 1718 m_pending_messages.fetch_sub(1, std::memory_order_seq_cst);
732 1718 const bool handled = handle_overflow(std::move(msg));
733 1717 finish_producer();
734 1721 return handled;
735 5219 }
736
737 153 bool AsyncLogger::Impl::flush_with_timeout(std::chrono::milliseconds timeout) noexcept
738 {
739 // Succeeds only on a genuine drain acknowledgement (the pending count reaching zero), never merely because
740 // the writer stopped. A stopped writer with an undrained queue must report a failed flush, not a false
741 // success. The predicate is checked before the wait, so an already-drained logger returns at once.
742 153 std::unique_lock<std::mutex> lock(m_flush_mutex);
743 #if defined(DMK_ENABLE_TEST_SEAMS)
744 153 detail::g_async_logger_flush_waiting.store(true, std::memory_order_release);
745 // Hold m_flush_mutex (the control-plane mutex) while a fixture proves a Drop-policy producer's enqueue
746 // completes without acquiring it. wait_for below only runs once the gate clears and the lock is released.
747
2/2
✓ Branch 5 → 6 taken 1 time.
✓ Branch 5 → 10 taken 152 times.
 153 if (auto *gate = detail::g_async_logger_flush_mutex_gate.load(std::memory_order_acquire))
748 {
749
2/2
✓ Branch 9 → 7 taken 653 times.
✓ Branch 9 → 10 taken 1 time.
 654 while (gate->load(std::memory_order_acquire))
750 {
751 653 std::this_thread::yield();
752 }
753 }
754 #endif
755 153 const bool flushed = m_flush_cv.wait_for(
756 lock,
757 timeout,
758 348 [this]() noexcept
759 {
760
2/2
✓ Branch 9 → 10 taken 153 times.
✓ Branch 9 → 19 taken 195 times.
 849 return m_pending_messages.load(std::memory_order_acquire) == 0 &&
761
1/2
✓ Branch 17 → 18 taken 153 times.
✗ Branch 17 → 19 not taken.
 654 m_active_producers.load(std::memory_order_seq_cst) == 0;
762 }
763 );
764 #if defined(DMK_ENABLE_TEST_SEAMS)
765 153 detail::g_async_logger_flush_waiting.store(false, std::memory_order_release);
766 #endif
767 306 return flushed;
768 153 }
769
770 14 void AsyncLogger::Impl::flush() noexcept
771 {
772 14 (void)flush_with_timeout(DEFAULT_FLUSH_TIMEOUT);
773 14 }
774
775 1135 void AsyncLogger::Impl::shutdown() noexcept
776 {
777 1135 State expected = State::Async;
778
2/2
✓ Branch 3 → 4 taken 737 times.
✓ Branch 3 → 5 taken 398 times.
 1135 if (!m_state.compare_exchange_strong(expected, State::Stopping, std::memory_order_seq_cst))
779 {
780 // Already Stopping or Stopped: shutdown ran once and owns the teardown.
781 765 return;
782 }
783
784 // Wake a parked writer so it observes Stopping and drains to exit. SetEvent takes no lock, which is required
785 // on the loader-lock abandon path, where waiting for a writer-owned mutex could deadlock process teardown. A
786 // flusher blocked on m_flush_cv is released separately, by the writer's post-drain notify at its tail.
787 398 ::SetEvent(m_wake_event);
788
789
1/2
✓ Branch 7 → 8 taken 398 times.
✗ Branch 7 → 17 not taken.
 398 if (m_writer_thread.joinable())
790 {
791
2/2
✓ Branch 9 → 10 taken 28 times.
✓ Branch 9 → 14 taken 370 times.
 398 if (detail::async_logger_must_not_block())
792 {
793 // Loader-lock abandon: publish the stop and return immediately. The retained writer alone finishes
794 // the drain, owns final sink access, resets the pending counter, and performs the Stopping->Stopped
795 // transition at its tail. Draining or zeroing the counter here would make this thread a second
796 // consumer of the same queue and sink and could underflow the counter against the writer's own
797 // decrement. Detach the writer and leak its module reference (taken before thread creation) so its
798 // code stays mapped, and latch the detach so ~AsyncLogger leaks this Impl in place instead of freeing
799 // the queue / wake event / flush channel / file stream the writer still reads.
800 try
801 {
802
1/2
✓ Branch 10 → 11 taken 28 times.
✗ Branch 10 → 20 not taken.
 28 m_writer_thread.detach();
803 }
804 ✗ catch (...)
805 {
806 // The Impl is abandoned below, so a still-joinable jthread remains live storage rather than
807 // reaching its joining destructor under the loader lock.
808 ✗ }
809 28 m_writer_detached.store(true, std::memory_order_release);
810 28 DetourModKit::diagnostics::record_intentional_leak(
811 DetourModKit::diagnostics::LeakSubsystem::AsyncLogger
812 );
813 28 return;
814 }
815
816 try
817 {
818
1/2
✓ Branch 14 → 15 taken 370 times.
✗ Branch 14 → 23 not taken.
 370 m_writer_thread.join();
819 }
820 ✗ catch (...)
821 {
822 try
823 {
824 ✗ if (m_writer_thread.joinable())
825 {
826 ✗ m_writer_thread.detach();
827 }
828 }
829 ✗ catch (...)
830 {
831 // The leaked Impl keeps a still-joinable thread object alive.
832 ✗ }
833 ✗ m_writer_detached.store(true, std::memory_order_release);
834 ✗ DetourModKit::diagnostics::record_intentional_leak(
835 DetourModKit::diagnostics::LeakSubsystem::AsyncLogger
836 );
837 ✗ return;
838 ✗ }
839 // Joined off the loader lock: the writer's code is done, so drop the reference taken before thread
840 // creation. Another reference on the module still exists (the caller running this teardown), so this is
841 // never terminal.
842 370 release_module_ref(static_cast<HMODULE>(m_writer_self_ref), diagnostics::ModulePinReason::AsyncLogger);
843 370 m_writer_self_ref = nullptr;
844 }
845
846 // The writer waits for every admitted producer and owns the complete drain before join returns.
847 }
848
849 688 bool AsyncLogger::Impl::is_running() const noexcept
850 {
851 688 return m_state.load(std::memory_order_acquire) == State::Async;
852 }
853
854 #if defined(DMK_ENABLE_TEST_SEAMS)
855 1492 bool AsyncLogger::Impl::is_writer_waiting() const noexcept
856 {
857 1492 return m_writer_waiting.load(std::memory_order_acquire);
858 }
859 #endif
860
861 15 size_t AsyncLogger::Impl::queue_size() const noexcept
862 {
863 15 return m_queue.size();
864 }
865
866 367 size_t AsyncLogger::Impl::dropped_count() const noexcept
867 {
868 734 return m_dropped_messages.load(std::memory_order_relaxed);
869 }
870
871 1 void AsyncLogger::Impl::reset_dropped_count() noexcept
872 {
873 1 m_dropped_messages.store(0, std::memory_order_release);
874 1 }
875
876 2 void AsyncLogger::Impl::set_timestamp_format(std::string timestamp_format) noexcept
877 {
878 // No lock taken here: the caller holds the shared log mutex (m_log_mutex), which is the same mutex the writer
879 // thread takes before it reads m_config.timestamp_format in write_batch / enqueue / handle_overflow. Because
880 // the caller holds it, the writer cannot be mid-read, so the assignment is race-free; taking the mutex here
881 // would self-deadlock the reconfigure path that already owns it. std::string move-assignment is noexcept, so
882 // the by-value parameter (copied in the caller's throwing context) makes this frame genuinely no-throw.
883 2 m_config.timestamp_format = std::move(timestamp_format);
884 2 }
885
886 2 void AsyncLogger::Impl::set_file_stream(std::shared_ptr<detail::WinFileStream> file_stream) noexcept
887 {
888 // Same locking rule as set_timestamp_format: the caller holds m_log_mutex, which every reader of m_file_stream
889 // takes, so the writer thread cannot be mid-write. Logger::reconfigure_locked opens the replacement sink before
890 // it retires the old one, so the writer never observes a closed stream between the two.
891 2 m_file_stream = std::move(file_stream);
892 2 }
893
894 715 bool AsyncLogger::Impl::writer_was_detached() const noexcept
895 {
896 715 return m_writer_detached.load(std::memory_order_acquire);
897 }
898
899 5221 void AsyncLogger::Impl::finish_producer() noexcept
900 {
901 5221 const size_t previous = m_active_producers.fetch_sub(1, std::memory_order_seq_cst);
902 // Signal flush waiters only while stopping, where the last admitted producer draining out is the transition
903 // that lets a shutdown-time flush observe m_active_producers == 0 and complete. During Async this path stays
904 // off the flush condition variable so a producer never contends the control-plane mutex on a hot enqueue; an
905 // Async flush waiter is instead released by the writer's post-drain notify or, failing that, by its own wait
906 // timeout, so its return value stays a genuine drain acknowledgement in every case.
907
6/6
✓ Branch 4 → 5 taken 4455 times.
✓ Branch 4 → 8 taken 766 times.
✓ Branch 6 → 7 taken 9 times.
✓ Branch 6 → 8 taken 4446 times.
✓ Branch 9 → 10 taken 9 times.
✓ Branch 9 → 19 taken 5212 times.
 5221 if (previous == 1 && m_state.load(std::memory_order_seq_cst) != State::Async)
908 {
909 9 m_flush_cv.notify_all();
910 // The stopping writer's loop stays alive on m_active_producers != 0, so the transition to zero is also
911 // its exit condition. A producer that leaves WITHOUT publishing a slot (a dropped over-inline message)
912 // never reaches notify_writer, and a writer already parked on the wake event otherwise inherits the
913 // full flush interval before it rechecks. The same seq_cst Dekker pairing as the push path applies: the
914 // writer publishes m_writer_waiting before it loads m_active_producers, so either it sees this exit and
915 // skips the wait, or this load sees the parked flag and signals.
916
2/2
✓ Branch 12 → 13 taken 1 time.
✓ Branch 12 → 19 taken 8 times.
 9 if (m_writer_waiting.load(std::memory_order_seq_cst))
917 {
918 #if defined(DMK_ENABLE_TEST_SEAMS)
919
1/2
✓ Branch 14 → 15 taken 1 time.
✗ Branch 14 → 18 not taken.
 1 if (auto *counter = detail::g_async_logger_finish_wake_counter.load(std::memory_order_acquire))
920 {
921 1 counter->fetch_add(1, std::memory_order_relaxed);
922 }
923 #endif
924 1 ::SetEvent(m_wake_event);
925 }
926 }
927 5221 }
928
929 3812 void AsyncLogger::Impl::notify_writer() noexcept
930 {
931 // The caller has already published the queue slot. In the seq_cst order either the writer's pre-park work
932 // recheck observes that push, or this load observes the writer's parked flag and signals the wake event.
933 // SetEvent takes no control-plane mutex, so a callback-safe Drop-policy producer cannot stall behind a
934 // flusher or the writer. When the writer is actively draining the flag is false and this is a single atomic
935 // load with no syscall, so the streaming hot path stays syscall-free.
936
2/2
✓ Branch 3 → 4 taken 461 times.
✓ Branch 3 → 5 taken 3351 times.
 3812 if (m_writer_waiting.load(std::memory_order_seq_cst))
937 {
938 461 ::SetEvent(m_wake_event);
939 }
940 3812 }
941
942 403 void AsyncLogger::Impl::writer_thread_func() noexcept
943 {
944 // Per-idle-cycle cap on the cooperative yields the writer spins through when the pending count
945 // shows an in-flight push whose queue slot has not landed yet. Small and fixed so a producer
946 // preempted mid-publish cannot turn the idle path into a tight hot loop.
947 403 constexpr size_t INFLIGHT_SPIN_LIMIT = 8;
948
949 // No pre-reserve here: this frame is noexcept, so a throwing reserve would std::terminate on host OOM.
950 // try_pop_batch owns the (fail-closed) reservation and only pops within the capacity it can secure. After the
951 // first pop the batch retains its capacity across the clear()s below, so the steady-state reserve is a no-op.
952 403 std::vector<LogMessage> batch;
953
954 403 auto last_flush = std::chrono::steady_clock::now();
955
956 2155 while (m_state.load(std::memory_order_seq_cst) == State::Async ||
957
8/8
✓ Branch 115 → 116 taken 829 times.
✓ Branch 115 → 126 taken 1326 times.
✓ Branch 123 → 124 taken 825 times.
✓ Branch 123 → 126 taken 4 times.
✓ Branch 125 → 126 taken 427 times.
✓ Branch 125 → 127 taken 398 times.
✓ Branch 128 → 4 taken 1757 times.
✓ Branch 128 → 129 taken 398 times.
 2984 m_active_producers.load(std::memory_order_seq_cst) != 0 || !m_queue.empty())
958 {
959 #if defined(DMK_ENABLE_TEST_SEAMS)
960 // Test-only pause point: hold the writer off the queue while a fixture inspects the pre-drain state.
961 1757 for (auto *gate = detail::g_async_logger_writer_gate.load(std::memory_order_acquire);
962
6/6
✓ Branch 8 → 9 taken 75 times.
✓ Branch 8 → 12 taken 1711 times.
✓ Branch 10 → 11 taken 29 times.
✓ Branch 10 → 12 taken 46 times.
✓ Branch 13 → 5 taken 29 times.
✓ Branch 13 → 14 taken 1757 times.
 1786 gate && gate->load(std::memory_order_acquire);
963 29 gate = detail::g_async_logger_writer_gate.load(std::memory_order_acquire))
964 {
965 29 std::this_thread::sleep_for(std::chrono::milliseconds(1));
966 }
967 #endif
968 1757 batch.clear();
969 1757 (void)m_queue.try_pop_batch(batch, m_config.batch_size);
970
971 1757 size_t drained = batch.size();
972 1757 size_t failed = 0;
973
2/2
✓ Branch 17 → 18 taken 896 times.
✓ Branch 17 → 21 taken 861 times.
 1757 if (drained != 0)
974 {
975 896 failed = write_batch(batch);
976 }
977
2/2
✓ Branch 22 → 23 taken 200 times.
✓ Branch 22 → 35 taken 661 times.
 861 else if (!m_queue.empty())
978 {
979 // The batch reserve could not secure headroom, so the pop returned nothing while records are queued.
980 // Without a floor the loop condition stays true, the idle gates below never run, and the writer spins
981 // at 100% CPU forever, taking shutdown's join() with it. Drain one record through stack storage
982 // instead: try_pop moves into an existing object and allocates nothing, so forward progress no longer
983 // depends on the allocator recovering. The accounting below is shared with the batch arm, so a
984 // one-record cycle updates exactly what a batch cycle does.
985 200 LogMessage one;
986
1/2
✓ Branch 25 → 26 taken 200 times.
✗ Branch 25 → 33 not taken.
 200 if (m_queue.try_pop(one))
987 {
988 200 failed = write_batch(std::span<const LogMessage>(&one, 1));
989 200 drained = 1;
990 #if defined(DMK_ENABLE_TEST_SEAMS)
991
1/2
✓ Branch 29 → 30 taken 200 times.
✗ Branch 29 → 33 not taken.
 200 if (auto *counter = detail::g_async_logger_batch_floor_counter.load(std::memory_order_acquire))
992 {
993 200 counter->fetch_add(1, std::memory_order_relaxed);
994 }
995 #endif
996 }
997 200 }
998
999
2/2
✓ Branch 35 → 36 taken 1096 times.
✓ Branch 35 → 46 taken 661 times.
 1757 if (drained != 0)
1000 {
1001 // Publish sink loss before acknowledging the drain. A flush waiter that observes pending == 0 must
1002 // also observe the complete drop count for every record the writer just consumed.
1003
2/2
✓ Branch 36 → 37 taken 8 times.
✓ Branch 36 → 40 taken 1088 times.
 1096 if (failed != 0)
1004 {
1005 8 m_dropped_messages.fetch_add(failed, std::memory_order_relaxed);
1006 }
1007 {
1008 1096 std::lock_guard<std::mutex> flock(m_flush_mutex);
1009 1096 m_pending_messages.fetch_sub(drained, std::memory_order_acq_rel);
1010 1096 }
1011 1096 m_flush_cv.notify_all();
1012 1096 last_flush = std::chrono::steady_clock::now();
1013 }
1014 else
1015 {
1016 // Nothing was drainable this cycle. A producer bumps m_pending_messages before it publishes its queue
1017 // slot, so a non-zero pending count here means a push is in flight. That holds whether or not the
1018 // queue reports itself empty, because a claimed-but-unpublished slot counts toward its size. The idle
1019 // gates are therefore conditioned on this branch (no progress) rather than on emptiness, which a
1020 // claimed but not yet readable slot does not satisfy while offering nothing to drain. A small fixed
1021 // yield budget handles the common few-instruction window. A producer preempted beyond it sends the
1022 // writer to the event wait below instead of restarting an unbounded yield loop.
1023 773 for (size_t spin = 0;
1024
6/6
✓ Branch 49 → 50 taken 759 times.
✓ Branch 49 → 61 taken 14 times.
✓ Branch 51 → 52 taken 756 times.
✓ Branch 51 → 61 taken 3 times.
✓ Branch 62 → 47 taken 112 times.
✓ Branch 62 → 63 taken 661 times.
 1529 spin < INFLIGHT_SPIN_LIMIT && m_state.load(std::memory_order_acquire) == State::Async &&
1025
2/2
✓ Branch 59 → 60 taken 112 times.
✓ Branch 59 → 61 taken 644 times.
 1512 m_pending_messages.load(std::memory_order_seq_cst) != 0;
1026 ++spin)
1027 {
1028 112 std::this_thread::yield();
1029 }
1030
1031 661 auto now = std::chrono::steady_clock::now();
1032
2/2
✓ Branch 67 → 68 taken 22 times.
✓ Branch 67 → 77 taken 639 times.
 661 if (now - last_flush >= m_config.flush_interval)
1033 {
1034 22 std::lock_guard<std::mutex> lock(*m_log_mutex);
1035
1/2
✓ Branch 72 → 73 taken 22 times.
✗ Branch 72 → 75 not taken.
 22 if (m_file_stream->is_open())
1036 {
1037 22 m_file_stream->flush();
1038 }
1039 22 last_flush = now;
1040 22 }
1041
1042 // Park on the wake event. Publish the parked flag first (seq_cst) so a producer that publishes a slot
1043 // after this point observes it and SetEvents the event; the recheck below closes the lost-wakeup
1044 // window. Parking on the event instead of returning immediately from a pending-count predicate bounds
1045 // the idle work: the writer sleeps until the producer signals or a bounded recheck expires.
1046 661 m_writer_waiting.store(true, std::memory_order_seq_cst);
1047 661 const bool has_pending = m_pending_messages.load(std::memory_order_seq_cst) != 0;
1048
2/2
✓ Branch 86 → 87 taken 3 times.
✓ Branch 86 → 95 taken 658 times.
 664 const bool keep_running = m_state.load(std::memory_order_seq_cst) == State::Async ||
1049
1/2
✓ Branch 94 → 95 taken 3 times.
✗ Branch 94 → 96 not taken.
 6 m_active_producers.load(std::memory_order_seq_cst) != 0;
1050
1/4
✗ Branch 97 → 98 not taken.
✓ Branch 97 → 99 taken 661 times.
✗ Branch 98 → 99 not taken.
✗ Branch 98 → 111 not taken.
 661 if (keep_running || has_pending)
1051 {
1052 #if defined(DMK_ENABLE_TEST_SEAMS)
1053
2/2
✓ Branch 100 → 101 taken 13 times.
✓ Branch 100 → 104 taken 648 times.
 661 if (auto *counter = detail::g_async_logger_idle_park_counter.load(std::memory_order_acquire))
1054 {
1055 13 counter->fetch_add(1, std::memory_order_relaxed);
1056 }
1057 #endif
1058 // A push in flight (has_pending, nothing drainable) parks for a bounded 1 ms recheck instead of
1059 // the full interval, so the rare lost-wakeup case (a producer that read m_writer_waiting ==
1060 // false just before the store above, then published without signalling) self-heals in a
1061 // millisecond without burning a core. A genuinely idle writer sleeps the whole interval and the
1062 // next producer's SetEvent wakes it.
1063 661 const auto interval = m_config.flush_interval.count();
1064
3/4
✓ Branch 105 → 106 taken 17 times.
✓ Branch 105 → 107 taken 644 times.
✓ Branch 107 → 108 taken 644 times.
✗ Branch 107 → 109 not taken.
 661 const DWORD wait_ms = has_pending ? 1u
1065 : static_cast<DWORD>(
1066 interval < 1 ? 1
1067 644 : interval > 0x7FFFFFFF ? 0x7FFFFFFF
1068 : interval
1069 );
1070 661 ::WaitForSingleObject(m_wake_event, wait_ms);
1071 }
1072 656 m_writer_waiting.store(false, std::memory_order_seq_cst);
1073 }
1074 }
1075
1076 {
1077 398 std::lock_guard<std::mutex> lock(*m_log_mutex);
1078
2/2
✓ Branch 133 → 134 taken 395 times.
✓ Branch 133 → 136 taken 3 times.
 398 if (m_file_stream->is_open())
1079 {
1080 395 m_file_stream->flush();
1081 }
1082 398 }
1083
1084 {
1085 398 std::lock_guard<std::mutex> lock(m_flush_mutex);
1086 398 m_pending_messages.store(0, std::memory_order_release);
1087 398 m_flush_cv.notify_all();
1088 398 }
1089
1090 // The writer is the single owner of the Stopping->Stopped transition: it publishes Stopped only after the
1091 // queue is fully drained, the sink flushed, and the pending counter zeroed above. On the loader-lock abandon
1092 // path shutdown() returned early and left this final acknowledgement to the retained writer.
1093 398 m_state.store(State::Stopped, std::memory_order_release);
1094 398 }
1095
1096 1096 size_t AsyncLogger::Impl::write_batch(std::span<const LogMessage> messages) noexcept
1097 {
1098 1096 std::lock_guard<std::mutex> lock(*m_log_mutex);
1099
1100
5/6
✓ Branch 6 → 7 taken 1089 times.
✓ Branch 6 → 10 taken 7 times.
✗ Branch 9 → 10 not taken.
✓ Branch 9 → 11 taken 1089 times.
✓ Branch 12 → 13 taken 7 times.
✓ Branch 12 → 14 taken 1089 times.
 1096 if (!m_file_stream->is_open() || !m_file_stream->good())
1101 {
1102 7 return messages.size();
1103 }
1104
1105 try
1106 {
1107 // Cache the localtime result across consecutive messages that share the same second to avoid repeated CRT
1108 // lock acquisition inside localtime_s.
1109 1089 std::time_t cached_second{-1};
1110 1089 std::tm cached_tm{};
1111
1112
2/2
✓ Branch 62 → 16 taken 3486 times.
✓ Branch 62 → 63 taken 1089 times.
 5664 for (const auto &msg : messages)
1113 {
1114 3486 const auto time_t = std::chrono::system_clock::to_time_t(msg.timestamp);
1115
1116
2/2
✓ Branch 23 → 24 taken 1089 times.
✓ Branch 23 → 25 taken 2397 times.
 3486 if (time_t != cached_second)
1117 {
1118 1089 cached_second = time_t;
1119 #if defined(_WIN32) || defined(_MSC_VER)
1120
1/2
✓ Branch 24 → 25 taken 1089 times.
✗ Branch 24 → 80 not taken.
 1089 localtime_s(&cached_tm, &time_t);
1121 #else
1122 localtime_r(&time_t, &cached_tm);
1123 #endif
1124 }
1125
1126 const auto ms =
1127
2/4
✓ Branch 26 → 27 taken 3486 times.
✗ Branch 26 → 77 not taken.
✓ Branch 27 → 28 taken 3486 times.
✗ Branch 27 → 77 not taken.
 3486 std::chrono::duration_cast<std::chrono::milliseconds>(msg.timestamp.time_since_epoch()) % 1000;
1128
1129 3486 *m_file_stream << "[" << std::put_time(&cached_tm, m_config.timestamp_format.c_str()) << "."
1130
5/10
✓ Branch 29 → 30 taken 3486 times.
✗ Branch 29 → 80 not taken.
✓ Branch 32 → 33 taken 3486 times.
✗ Branch 32 → 80 not taken.
✓ Branch 33 → 34 taken 3486 times.
✗ Branch 33 → 80 not taken.
✓ Branch 35 → 36 taken 3486 times.
✗ Branch 35 → 80 not taken.
✓ Branch 39 → 40 taken 3486 times.
✗ Branch 39 → 80 not taken.
 3486 << std::setfill('0') << std::setw(3) << ms.count() << std::setfill(' ') << "] "
1131
4/8
✓ Branch 41 → 42 taken 3486 times.
✗ Branch 41 → 80 not taken.
✓ Branch 42 → 43 taken 3486 times.
✗ Branch 42 → 80 not taken.
✓ Branch 43 → 44 taken 3486 times.
✗ Branch 43 → 80 not taken.
✓ Branch 46 → 47 taken 3486 times.
✗ Branch 46 → 80 not taken.
 3486 << "[" << std::setw(7) << std::left << to_string(msg.level) << "] :: " << msg.message()
1132
4/8
✓ Branch 48 → 49 taken 3486 times.
✗ Branch 48 → 80 not taken.
✓ Branch 49 → 50 taken 3486 times.
✗ Branch 49 → 80 not taken.
✓ Branch 51 → 52 taken 3486 times.
✗ Branch 51 → 80 not taken.
✓ Branch 52 → 53 taken 3486 times.
✗ Branch 52 → 80 not taken.
 3486 << '\n';
1133 }
1134
1135 #if defined(DMK_ENABLE_TEST_SEAMS)
1136
2/2
✓ Branch 63 → 64 taken 1 time.
✓ Branch 63 → 66 taken 1088 times.
 1089 if (auto *probe = detail::g_async_logger_before_flush_probe)
1137 {
1138 1 probe(*m_file_stream);
1139 }
1140 #endif
1141
1/2
✓ Branch 67 → 68 taken 1089 times.
✗ Branch 67 → 82 not taken.
 1089 m_file_stream->flush();
1142 }
1143 ✗ catch (...)
1144 {
1145 ✗ return messages.size();
1146 ✗ }
1147
1148 // An insertion or final flush failure leaves the stream unhealthy. Record the whole dequeued batch as
1149 // unconfirmed because the buffer does not expose which complete record boundaries reached the file.
1150
2/2
✓ Branch 70 → 71 taken 1088 times.
✓ Branch 70 → 72 taken 1 time.
 1089 return m_file_stream->good() ? 0 : messages.size();
1151 1096 }
1152
1153 1712 bool AsyncLogger::Impl::handle_overflow(LogMessage &&message) noexcept
1154 {
1155
4/5
✓ Branch 2 → 3 taken 1273 times.
✓ Branch 2 → 6 taken 306 times.
✓ Branch 2 → 23 taken 6 times.
✓ Branch 2 → 77 taken 128 times.
✗ Branch 2 → 136 not taken.
 1712 switch (m_config.overflow_policy)
1156 {
1157 1273 case OverflowPolicy::DropNewest:
1158 1273 m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
1159 1273 return false;
1160
1161 306 case OverflowPolicy::DropOldest:
1162 {
1163 306 LogMessage oldest;
1164
1/2
✓ Branch 8 → 9 taken 306 times.
✗ Branch 8 → 18 not taken.
 306 if (m_queue.try_pop(oldest))
1165 {
1166 // Count the evicted oldest message as dropped
1167 306 m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
1168
1/2
✓ Branch 12 → 13 taken 306 times.
✗ Branch 12 → 15 not taken.
 306 if (m_queue.try_push(message))
1169 {
1170 // Net effect on m_pending_messages: pop(-1) + push(+1) = 0
1171 306 notify_writer();
1172 306 return true;
1173 }
1174 // Pop succeeded but push failed: net -1
1175 ✗ m_pending_messages.fetch_sub(1, std::memory_order_seq_cst);
1176 }
1177 // Count the new message as dropped (separate from the evicted oldest above). m_dropped_messages counts
1178 // individual lost messages, not overflow events.
1179 ✗ m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
1180 ✗ return false;
1181 306 }
1182
1183 6 case OverflowPolicy::Block:
1184 {
1185 #if defined(DMK_ENABLE_TEST_SEAMS)
1186
2/2
✓ Branch 24 → 25 taken 2 times.
✓ Branch 24 → 28 taken 4 times.
 6 if (auto *counter = detail::g_async_logger_block_entry_counter.load(std::memory_order_acquire))
1187 {
1188 2 counter->fetch_add(1, std::memory_order_relaxed);
1189 }
1190
2/2
✓ Branch 29 → 30 taken 2 times.
✓ Branch 29 → 34 taken 4 times.
 6 if (auto *gate = detail::g_async_logger_block_entry_gate.load(std::memory_order_acquire))
1191 {
1192
2/2
✓ Branch 33 → 31 taken 22 times.
✓ Branch 33 → 34 taken 2 times.
 24 while (gate->load(std::memory_order_acquire))
1193 {
1194 22 std::this_thread::yield();
1195 }
1196 }
1197 #endif
1198 6 const auto deadline = std::chrono::steady_clock::now() + m_config.block_timeout_ms;
1199 #if defined(DMK_ENABLE_TEST_SEAMS)
1200
2/2
✓ Branch 37 → 38 taken 1 time.
✓ Branch 37 → 51 taken 5 times.
 6 if (auto *start_ns = detail::g_async_logger_block_start_ns.load(std::memory_order_acquire))
1201 {
1202 1 const auto now_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
1203 1 std::chrono::steady_clock::now().time_since_epoch()
1204 );
1205 1 start_ns->store(now_ns.count(), std::memory_order_release);
1206 }
1207 #endif
1208 6 size_t spin_count = 0;
1209
1210 // Pre-increment so flush sees the in-flight message throughout the retry loop
1211 6 m_pending_messages.fetch_add(1, std::memory_order_seq_cst);
1212
1213
2/2
✓ Branch 70 → 54 taken 3627 times.
✓ Branch 70 → 71 taken 1 time.
 3628 while (std::chrono::steady_clock::now() < deadline)
1214 {
1215
2/2
✓ Branch 55 → 56 taken 5 times.
✓ Branch 55 → 58 taken 3622 times.
 3627 if (m_queue.try_push(message))
1216 {
1217 5 notify_writer();
1218 5 return true;
1219 }
1220
1221
2/2
✓ Branch 58 → 59 taken 165 times.
✓ Branch 58 → 60 taken 3457 times.
 3622 if (spin_count < m_config.spin_backoff_iterations)
1222 {
1223 165 ++spin_count;
1224 }
1225
2/2
✓ Branch 60 → 61 taken 3428 times.
✓ Branch 60 → 63 taken 29 times.
 3457 else if (spin_count < m_config.block_max_spin_iterations)
1226 {
1227 3428 std::this_thread::yield();
1228 3428 ++spin_count;
1229 }
1230 else
1231 {
1232 29 std::this_thread::sleep_for(std::chrono::milliseconds(1));
1233 }
1234 }
1235 // Timed out: undo the pre-increment
1236 1 m_pending_messages.fetch_sub(1, std::memory_order_seq_cst);
1237 1 m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
1238 1 return false;
1239 }
1240
1241 128 case OverflowPolicy::SyncFallback:
1242 {
1243 128 std::lock_guard<std::mutex> lock(*m_log_mutex);
1244
5/6
✓ Branch 81 → 82 taken 127 times.
✓ Branch 81 → 85 taken 1 time.
✗ Branch 84 → 85 not taken.
✓ Branch 84 → 86 taken 127 times.
✓ Branch 87 → 88 taken 1 time.
✓ Branch 87 → 91 taken 127 times.
 128 if (!m_file_stream->is_open() || !m_file_stream->good())
1245 {
1246 // The synchronous fallback could not reach a healthy sink, so the message is lost. Count it as a
1247 // drop rather than reporting a silent failure.
1248 1 m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
1249 1 return false;
1250 }
1251
1252 254 const auto time_t = std::chrono::system_clock::to_time_t(message.timestamp);
1253 127 std::tm tm_buf{};
1254
1255 #if defined(_WIN32) || defined(_MSC_VER)
1256 127 localtime_s(&tm_buf, &time_t);
1257 #else
1258 localtime_r(&time_t, &tm_buf);
1259 #endif
1260
1261 const auto ms =
1262 127 std::chrono::duration_cast<std::chrono::milliseconds>(message.timestamp.time_since_epoch()) % 1000;
1263 127 *m_file_stream << "[" << std::put_time(&tm_buf, m_config.timestamp_format.c_str()) << "."
1264 127 << std::setfill('0') << std::setw(3) << ms.count() << std::setfill(' ') << "] "
1265 127 << "[" << std::setw(7) << std::left << to_string(message.level)
1266 127 << "] :: " << message.message() << '\n';
1267 127 m_file_stream->flush();
1268
1269
1/2
✗ Branch 129 → 130 not taken.
✓ Branch 129 → 133 taken 127 times.
 127 if (m_file_stream->fail())
1270 {
1271 // The synchronous write left the stream failed, so the message did not durably reach the sink.
1272 ✗ m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
1273 ✗ return false;
1274 }
1275 127 return true;
1276 128 }
1277
1278 ✗ default:
1279 ✗ m_dropped_messages.fetch_add(1, std::memory_order_relaxed);
1280 ✗ return false;
1281 }
1282 }
1283
1284 // AsyncLogger is a thin facade: construction builds the Impl (which validates the config and starts the writer
1285 // thread), and every public method forwards to it. The out-of-line destructor sees the complete Impl so the
1286 // unique_ptr can delete it (Impl::~Impl drains and joins the writer).
1287 408 AsyncLogger::AsyncLogger(
1288 const AsyncLoggerConfig &config,
1289 std::shared_ptr<detail::WinFileStream> file_stream,
1290 std::shared_ptr<std::mutex> log_mutex
1291 408 )
1292 816 : m_impl(std::make_unique<Impl>(config, std::move(file_stream), std::move(log_mutex)))
1293 {
1294 #if defined(DMK_ENABLE_TEST_SEAMS)
1295 detail::g_async_logger_live_count_for_test.fetch_add(1, std::memory_order_relaxed);
1296 #endif
1297 403 }
1298
1299 372 AsyncLogger::~AsyncLogger() noexcept
1300 {
1301 #if defined(DMK_ENABLE_TEST_SEAMS)
1302 detail::g_async_logger_live_count_for_test.fetch_sub(1, std::memory_order_relaxed);
1303 #endif
1304
1/2
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 372 times.
 372 if (!m_impl)
1305 {
1306 ✗ return;
1307 }
1308
1309 // Drive the writer to a stop. Under the loader lock shutdown() cannot join, so it detaches the writer (which
1310 // keeps reading m_queue / m_flush_cv / m_file_stream until it observes the stop) and latches m_writer_detached.
1311 372 m_impl->shutdown();
1312
1313
2/2
✓ Branch 11 → 12 taken 2 times.
✓ Branch 11 → 14 taken 370 times.
 372 if (m_impl->writer_was_detached())
1314 {
1315 // The writer is still running against this Impl's members, so ~Impl must NOT run: destroying the condition
1316 // variable while the detached writer is parked on it (or the queue it is draining) is undefined behaviour.
1317 // Abandon the already-heap-allocated Impl in place: release() relinquishes the unique_ptr without
1318 // freeing, so the members outlive the writer with zero further allocation.
1319 // The detached writer's own counted module reference keeps the code pages it executes mapped. The
1320 // intentional-leak event was already recorded inside shutdown()'s detach branch, so it is not recorded a
1321 // second time here.
1322 2 (void)m_impl.release();
1323 2 return;
1324 }
1325
1326 // Off the loader lock the writer was joined by shutdown(), so the unique_ptr destroys the Impl normally. ~Impl
1327 // calls shutdown() again, but the Async->Stopping state CAS makes that an idempotent no-op before the members
1328 // are freed.
1329
4/4
✓ Branch 16 → 17 taken 370 times.
✓ Branch 16 → 18 taken 2 times.
✓ Branch 20 → 21 taken 370 times.
✓ Branch 20 → 22 taken 2 times.
 374 }
1330
1331 350 void AsyncLogger::arm_retention_root(const std::shared_ptr<AsyncLogger> &self) noexcept
1332 {
1333 350 m_retention_root = self;
1334 350 }
1335
1336 319 void AsyncLogger::release_retention_root() noexcept
1337 {
1338 319 m_retention_root.reset();
1339 319 }
1340
1341 2 void AsyncLogger::set_timestamp_format(std::string timestamp_format) noexcept
1342 {
1343 4 m_impl->set_timestamp_format(std::move(timestamp_format));
1344 2 }
1345
1346 2 void AsyncLogger::set_file_stream(std::shared_ptr<detail::WinFileStream> file_stream) noexcept
1347 {
1348 4 m_impl->set_file_stream(std::move(file_stream));
1349 2 }
1350
1351 3596 bool AsyncLogger::enqueue(LogLevel level, std::string_view message) noexcept
1352 {
1353 3596 return m_impl->enqueue(level, message, nullptr);
1354 }
1355
1356 1592 bool AsyncLogger::enqueue_from_facade(
1357 LogLevel level,
1358 std::string_view message,
1359 std::atomic<std::size_t> &facade_drops
1360 ) noexcept
1361 {
1362 1592 return m_impl->enqueue(level, message, &facade_drops);
1363 }
1364
1365 139 bool AsyncLogger::flush_with_timeout(std::chrono::milliseconds timeout) noexcept
1366 {
1367 139 return m_impl->flush_with_timeout(timeout);
1368 }
1369
1370 14 void AsyncLogger::flush() noexcept
1371 {
1372 14 m_impl->flush();
1373 14 }
1374
1375 393 void AsyncLogger::shutdown() noexcept
1376 {
1377 393 m_impl->shutdown();
1378 393 }
1379
1380 688 bool AsyncLogger::is_running() const noexcept
1381 {
1382 688 return m_impl->is_running();
1383 }
1384
1385 #if defined(DMK_ENABLE_TEST_SEAMS)
1386 1492 bool AsyncLogger::is_writer_waiting() const noexcept
1387 {
1388 1492 return m_impl->is_writer_waiting();
1389 }
1390 #endif
1391
1392 15 size_t AsyncLogger::queue_size() const noexcept
1393 {
1394 15 return m_impl->queue_size();
1395 }
1396
1397 367 size_t AsyncLogger::dropped_count() const noexcept
1398 {
1399 367 return m_impl->dropped_count();
1400 }
1401
1402 1 void AsyncLogger::reset_dropped_count() noexcept
1403 {
1404 1 m_impl->reset_dropped_count();
1405 1 }
1406
1407 343 bool AsyncLogger::writer_was_detached() const noexcept
1408 {
1409 343 return m_impl->writer_was_detached();
1410 }
1411
1412 } // namespace DetourModKit
1413