GCC Code Coverage Report

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