Fix spdlog async bench crash and MPSC dequeue reliability

bench/adapters/SpdlogAdapter.cpp

@@ -44,10 +44,12 @@ class SpdlogAdapter::MeasuringSink : public spdlog::sinks::sink {
     }
 
     void log(const spdlog::details::log_msg& msg) override {
-        const auto* payload_ptr = reinterpret_cast<const MessagePayload*>(msg.source.funcname);
-        if (!payload_ptr) {
-            return;
+        const char* func = msg.source.funcname;
+        if (msg.payload.size() == 0 || !func || *func == '\0') {
+            return; // Flush/control messages have no payload attached.
         }
+
+        const auto* payload_ptr = reinterpret_cast<const MessagePayload*>(func);
         auto* payload = const_cast<MessagePayload*>(payload_ptr);
         consume(*payload);
         delete payload;

bench/logit_bench.cpp

@@ -119,6 +119,7 @@ std::chrono::nanoseconds run_workload(
     // Barrier to start together.
     std::mutex start_mx;
     std::condition_variable start_cv;
+    std::condition_variable ready_cv;
     bool start_flag = false;
     std::size_t ready = 0;
 
@@ -132,7 +133,7 @@ std::chrono::nanoseconds run_workload(
             {
                 std::unique_lock<std::mutex> lk(start_mx);
                 ++ready;
-                if (ready == scenario.producers) start_cv.notify_one();
+                if (ready == scenario.producers) ready_cv.notify_one();
                 start_cv.wait(lk, [&]{ return start_flag; });
             }
             for (std::size_t n = 0; n < per_thread[i]; ++n) {
@@ -150,7 +151,7 @@ std::chrono::nanoseconds run_workload(
     std::chrono::steady_clock::time_point t0;
     {
         std::unique_lock<std::mutex> lk(start_mx);
-        start_cv.wait(lk, [&]{ return ready == scenario.producers; });
+        ready_cv.wait(lk, [&]{ return ready == scenario.producers; });
         if (measure_duration) t0 = std::chrono::steady_clock::now();
         start_flag = true;
         start_cv.notify_all();
@@ -225,6 +226,10 @@ ScenarioResult execute_scenario(
         log_info(oss.str());
     }
 
+    // Ensure async pipelines (e.g., spdlog thread pool) are fully drained before
+    // destroying the recorder referenced by sinks.
+    adapter.flush();
+
     const auto sum = recorder.finalize();
 
     double thr = 0.0;

docs/TaskExecutor.md

@@ -191,7 +191,13 @@ const auto lost = LOGIT_GET_DROPPED_TASKS();
   `set_queue_policy()`.
 * The hot-resize barrier uses `m_resizing` and `m_resize_cv` so producers never
   touch a ring buffer that is being rebuilt. This eliminates the data races that
-  TSAN previously reported on `try_pop()` vs. buffer assignment.
+  TSAN previously reported on `try_pop()` vs. buffer assignment. The barrier
+  only drops once the worker thread fully stops and the queue drains; if a sink
+  blocks the worker or `QueuePolicy::Block` keeps `m_active_tasks` above the
+  limit for more than one second, `set_max_queue_size()` abandons the hot
+  resize, clears `m_resizing`, and leaves the existing ring untouched so
+  producers cannot wait indefinitely. Non-MPSC builds perform the resize as an
+  atomic update of `m_max_queue_size`, so they are not subject to this stall.
 * Non-MPSC builds rely solely on mutexes and had no known data races.
 * The Emscripten path is single-threaded and should not be used concurrently.
 

include/logit_cpp/logit/detail/MpscRingAny.hpp

@@ -112,32 +112,35 @@ namespace logit { namespace detail {
         /// \brief Try to dequeue value into out. Non-blocking.
         /// \return true on success; false if queue is empty.
         bool try_pop(T& out) noexcept {
-            std::size_t pos = m_dequeue_pos.load(std::memory_order_relaxed);
-            Cell& c = m_cells[pos % m_cap];
-            std::size_t seq = c.m_seq.load(std::memory_order_acquire);
-
-            // When ready, seq == pos + 1
-            std::intptr_t diff =
-                static_cast<std::intptr_t>(seq) - static_cast<std::intptr_t>(pos + 1);
-
-            if (diff == 0) {
-                if (!m_dequeue_pos.compare_exchange_strong(
-                        pos, pos + 1,
-                        std::memory_order_relaxed,
-                        std::memory_order_relaxed)) {
-                    return false; // Single consumer: should be rare.
+            for (;;) {
+                std::size_t pos = m_dequeue_pos.load(std::memory_order_relaxed);
+                Cell& c = m_cells[pos % m_cap];
+                std::size_t seq = c.m_seq.load(std::memory_order_acquire);
+
+                // When ready, seq == pos + 1
+                std::intptr_t diff =
+                    static_cast<std::intptr_t>(seq) - static_cast<std::intptr_t>(pos + 1);
+
+                if (diff == 0) {
+                    if (!m_dequeue_pos.compare_exchange_weak(
+                            pos, pos + 1,
+                            std::memory_order_relaxed,
+                            std::memory_order_relaxed)) {
+                        // Spurious failure: retry until we own the slot.
+                        continue;
+                    }
+
+                    T* p = reinterpret_cast<T*>(&c.m_storage);
+                    out = std::move(*p);
+                    p->~T();
+
+                    // Mark cell free for next cycle.
+                    c.m_seq.store(pos + m_cap, std::memory_order_release);
+                    return true;
                 }
-
-                T* p = reinterpret_cast<T*>(&c.m_storage);
-                out = std::move(*p);
-                p->~T();
-
-                // Mark cell free for next cycle.
-                c.m_seq.store(pos + m_cap, std::memory_order_release);
-                return true;
+
+                return false; // Empty or not yet published.
             }
-
-            return false; // Empty or not yet published.
         }
 
         /// \brief Lightweight emptiness check for current consumer position.

include/logit_cpp/logit/detail/TaskExecutor.hpp

@@ -12,12 +12,12 @@
   #include <deque>
   #include <mutex>
   #include <emscripten/emscripten.h>
-#else
-  #include <thread>
-  #include <deque>
-  #include <mutex>
-  #include <condition_variable>
-  #include <chrono>
+    #else
+        #include <thread>
+        #include <deque>
+        #include <mutex>
+        #include <condition_variable>
+        #include <chrono>
 #endif
 
 // Enable lock-free MPSC ring integration (non-Emscripten) by defining:
@@ -323,8 +323,18 @@ namespace logit { namespace detail {
             // Tell producers to pause before any wait()/stop conditions run.
             m_resizing.store(true, std::memory_order_release);
 
-            // Drain the queue completely.
-            wait();
+            // Drain the queue completely, but do not wait forever if the worker
+            // is stalled (e.g., blocked sink or backpressure keeping
+            // m_active_tasks > 0). If we fail to drain before the deadline,
+            // abort the resize and re-open the barrier so producers can
+            // continue.
+            const auto deadline = std::chrono::steady_clock::now() +
+                                  std::chrono::seconds(1);
+            if (!wait_until_idle_(deadline)) {
+                m_resizing.store(false, std::memory_order_release);
+                m_resize_cv.notify_all();
+                return;
+            }
 
             // Stop the worker so it cannot touch m_mpsc_queue during the resize.
             std::unique_lock<std::mutex> lk(m_queue_mutex);
@@ -474,6 +484,15 @@ namespace logit { namespace detail {
         bool queue_empty_() const noexcept {
             return m_mpsc_queue.empty();
         }
+
+        bool wait_until_idle_(std::chrono::steady_clock::time_point deadline) {
+            std::unique_lock<std::mutex> lock(m_queue_mutex);
+            return m_queue_condition.wait_until(lock, deadline, [this]() {
+                return ((queue_empty_() &&
+                         m_active_tasks.load(std::memory_order_relaxed) == 0) ||
+                        m_stop_flag.load(std::memory_order_acquire));
+            });
+        }
     #endif
 
         TaskExecutor()