From c72a3e12c2ea161a159b52780385d78e35c5e871 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 4 Dec 2025 00:45:06 +0300 Subject: [PATCH] fix(concurrency): harden mpsc pop and bench flushing --- bench/adapters/SpdlogAdapter.cpp | 8 +-- bench/logit_bench.cpp | 9 +++- docs/TaskExecutor.md | 8 ++- .../logit_cpp/logit/detail/MpscRingAny.hpp | 51 ++++++++++--------- .../logit_cpp/logit/detail/TaskExecutor.hpp | 35 ++++++++++--- 5 files changed, 73 insertions(+), 38 deletions(-) diff --git a/bench/adapters/SpdlogAdapter.cpp b/bench/adapters/SpdlogAdapter.cpp index 5cb22c1..30b1a0a 100644 --- a/bench/adapters/SpdlogAdapter.cpp +++ b/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(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(func); auto* payload = const_cast(payload_ptr); consume(*payload); delete payload; diff --git a/bench/logit_bench.cpp b/bench/logit_bench.cpp index 98b2974..1a03395 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 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 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; 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/MpscRingAny.hpp b/include/logit_cpp/logit/detail/MpscRingAny.hpp index a4c206a..1a747d7 100644 --- a/include/logit_cpp/logit/detail/MpscRingAny.hpp +++ b/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(seq) - static_cast(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(seq) - static_cast(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(&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(&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. 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 #include #include -#else - #include - #include - #include - #include - #include + #else + #include + #include + #include + #include + #include #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 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 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()