From 47c246fb32988fb365f4b065d4a9816b5c0416d4 Mon Sep 17 00:00:00 2001
From: Aster Seker <git.electroyar@gmail.com>
Date: Thu, 4 Dec 2025 00:29:43 +0300
Subject: [PATCH] fix(bench): prevent start barrier deadlock

---
 bench/logit_bench.cpp                         |  5 +--
 docs/TaskExecutor.md                          |  8 ++++-
 .../logit_cpp/logit/detail/TaskExecutor.hpp   | 35 ++++++++++++++-----
 3 files changed, 37 insertions(+), 11 deletions(-)

diff --git a/bench/logit_bench.cpp b/bench/logit_bench.cpp
index 98b2974..adf84dc 100644
--- a/bench/logit_bench.cpp
+++ b/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();
diff --git a/docs/TaskExecutor.md b/docs/TaskExecutor.md
index f93392f..154c3bb 100644
--- a/docs/TaskExecutor.md
+++ b/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.
 
diff --git a/include/logit_cpp/logit/detail/TaskExecutor.hpp b/include/logit_cpp/logit/detail/TaskExecutor.hpp
index 3af0e6e..da813c0 100644
--- a/include/logit_cpp/logit/detail/TaskExecutor.hpp
+++ b/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()