From bfe6dcf41cd9bfea8fa0fd96c0b5c487eadda639 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 18 Sep 2025 05:07:16 +0300 Subject: [PATCH] docs(task-executor): document queue variants and hot resize Describe TaskExecutor implementations, update README and changelog, and add Doxygen notes. --- CHANGELOG.md | 16 ++ README.md | 12 + docs/TaskExecutor.md | 208 ++++++++++++++++++ .../logit_cpp/logit/detail/TaskExecutor.hpp | 141 +++++++----- 4 files changed, 321 insertions(+), 56 deletions(-) create mode 100644 CHANGELOG.md create mode 100644 docs/TaskExecutor.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..3962422 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,16 @@ +# Changelog + +## Unreleased + +### Added +- Optional lock-free MPSC ring queue for `TaskExecutor` (enable via + `LOGIT_USE_MPSC_RING`) bringing low-overhead multi-producer support compared + to previous mutex-only releases. +- Hot queue resizing for the MPSC build guarded by `m_resizing` and + `m_resize_cv`, allowing capacity changes without dropping accepted tasks. + +### Changed +- `QueuePolicy::DropOldest` now drops the incoming task when + `LOGIT_USE_MPSC_RING` is defined. This preserves FIFO execution of accepted + work, avoids producer/consumer contention, and keeps the implementation + TSAN-friendly. diff --git a/README.md b/README.md index a898a23..d6b054b 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,18 @@ LOGIT_SYSERR_ERROR("Deleting temp directory failed"); --- +## Backpressure and hot resize + +The asynchronous `TaskExecutor` supports both a mutex-protected deque and an +optional lock-free MPSC ring (enable via `LOGIT_USE_MPSC_RING`). Queue overflow +policies (`Block`, `DropNewest`, `DropOldest`) behave consistently across both +implementations, with the MPSC build intentionally dropping the *incoming* task +for `DropOldest` to keep accepted work ordered. The ring build also allows +"hot" queue resizes where producers briefly wait while the worker rebuilds the +ring buffer without losing in-flight tasks. See +[`docs/TaskExecutor.md`](docs/TaskExecutor.md) for a full breakdown and tuning +tips. + ## Features - **Flexible Log Formatting**: diff --git a/docs/TaskExecutor.md b/docs/TaskExecutor.md new file mode 100644 index 0000000..fc4e9f4 --- /dev/null +++ b/docs/TaskExecutor.md @@ -0,0 +1,208 @@ +# TaskExecutor Implementation Notes + +The asynchronous task executor powers every non-blocking logger. It accepts +work from multiple producer threads and drains it on a dedicated worker. This +document describes how the executor behaves across build configurations, +provides guidance on tuning the backpressure policies, and explains the +lifetime guarantees that logger integrations rely on. + +## 1. Implementation variants + +### Default deque worker (without `LOGIT_USE_MPSC_RING`) + +* Structure: one worker thread (`m_worker_thread`) consumes a `std::deque` + protected by `m_queue_mutex`. +* Synchronisation: producers and the worker coordinate through + `m_queue_condition` and the `m_stop_flag` atomic. +* Backpressure policies are implemented literally on the protected deque. +* Intended for environments where a simple mutex-protected queue is sufficient + or where the lock-free ring cannot be used. + +### Lock-free MPSC ring (`LOGIT_USE_MPSC_RING`) + +* Structure: producers push tasks into `m_mpsc_queue`, a lock-free + `MpscRingAny>` with a single consumer thread. +* Synchronisation primitives: + * `m_cv` + `m_cv_mutex` coordinate sleepers for both the worker and producers + that wait for capacity during `QueuePolicy::Block`. + * `m_queue_condition` wakes `wait()` callers once the queue drains. + * `m_active_tasks` tracks in-flight work so that `Block` limits concurrent + execution and `wait()` can determine quiescence. + * `m_stop_flag` terminates the worker and stops accepting new tasks. +* Enables very low producer overhead while maintaining FIFO ordering on the + consumer side. + +### Emscripten builds without pthreads + +* Structure: single-threaded `std::deque` guarded by `m_mutex`. +* No dedicated worker thread is created. Instead, tasks are drained via + `emscripten_async_call` scheduled from the main loop. +* Not thread-safe — intended for WebAssembly builds where pthreads are not + available. + +## 2. Backpressure semantics + +`QueuePolicy` controls what happens when the queue reaches `max_queue_size` +(`0` means "unbounded"). + +* `Block` + * Uses `m_active_tasks` to count in-flight work. If the counter reaches the + limit, producers wait. The non-MPSC build waits on + `m_queue_condition`. The MPSC build parks on `m_cv` with short sleeps while + the worker drains tasks. This policy avoids loss at the expense of + producer-side backpressure. +* `DropNewest` + * Non-MPSC: the incoming task is discarded when the deque is full. + * MPSC: identical semantics — the incoming task is dropped and + `m_dropped_tasks` is incremented. +* `DropOldest` + * Non-MPSC: the oldest dequeued element is removed, then the incoming task is + enqueued, providing literal "drop the oldest" behaviour. + * MPSC: **drop-incoming semantics**. The executor rejects the incoming task + instead of racing to remove an old element. This preserves the order of + tasks already accepted by the consumer, avoids lock-step coordination + between multiple producers and the worker, and keeps the implementation + TSAN-clean. `m_dropped_tasks` still counts these rejections. + +The drop counter is observable via `TaskExecutor::dropped_tasks()` and exposed to +end users through `LOGIT_GET_DROPPED_TASKS()`. + +## 3. Hot queue resize (`LOGIT_USE_MPSC_RING`) + +`set_max_queue_size()` performs a "hot" resize without tearing down the +application. + +1. `m_resizing` is set to `true` with release semantics. +2. `wait()` drains the queue and ensures `m_active_tasks == 0`. +3. The worker is stopped by setting `m_stop_flag`, notifying sleepers, and + joining the thread so it no longer touches `m_mpsc_queue`. +4. In a single thread the ring is rebuilt with the new capacity. The resize + keeps `m_dropped_tasks` intact but resets `m_active_tasks` to 0 because the + queue is empty. +5. The worker thread is restarted and the stop flag cleared. +6. `m_resizing` flips back to `false` and `m_resize_cv.notify_all()` wakes + producers that parked at the start of `add_task()`. + +While the resize is in progress, producers briefly wait on `m_resize_cv`. No +accepted tasks are lost, and the consumer thread never observes partially +initialised ring buffers. + +## 4. Ordering and completion guarantees + +* Exactly one consumer thread executes tasks, so work is processed in the order + accepted by the consumer. +* When the ring build is enabled, `DropNewest` and `DropOldest` both drop the + incoming task; accepted tasks keep their order. +* `wait()` returns once the queue is empty and `m_active_tasks == 0`, or when a + shutdown is requested. +* `shutdown()` blocks until the worker thread terminates. It is safe to call + multiple times. + +## 5. Singleton and lifetime management + +`TaskExecutor::get_instance()` intentionally stores the singleton inside a +`static TaskExecutor* instance = new TaskExecutor();`. This lets the executor +outlive static destructors inside logger components. Applications may call +`shutdown()` explicitly (for example during test teardown), but the singleton +remains valid until the process terminates. + +## 6. Emscripten (no pthreads) + +When targeting Emscripten without pthread support: + +* The executor remains single-threaded and therefore not thread-safe. +* `Block` is approximated by invoking `drain()` from the producer path until the + deque has room. `DropNewest`/`DropOldest` mirror the deque operations exactly. +* Tasks are executed by `emscripten_async_call`, which schedules a drain on the + browser event loop. This keeps logging compatible with the cooperative + execution model used in WebAssembly UI scenarios. +* Typical use cases: browser-hosted tools or demos that need asynchronous-style + logging without pulling in pthread support. + +## 7. API surface and macros + +Public methods exposed by `TaskExecutor`: + +* `set_max_queue_size(std::size_t size)` — change the queue capacity (`0` + disables the limit). Trigger a hot resize on MPSC builds. +* `set_queue_policy(QueuePolicy policy)` — change overflow behaviour. +* `add_task(std::function fn)` — enqueue work for the background worker. +* `wait()` — block until the queue drains or stop is requested. +* `shutdown()` — stop the worker thread and release resources. +* `dropped_tasks()` and `reset_dropped_tasks()` — inspect or reset the overflow + counter. + +Macros in `` map directly onto these calls: + +* `LOGIT_SET_MAX_QUEUE(size)` → `set_max_queue_size(size)` +* `LOGIT_SET_QUEUE_POLICY(mode)` → `set_queue_policy(mode)` +* `LOGIT_QUEUE_BLOCK`, `LOGIT_QUEUE_DROP_NEWEST`, `LOGIT_QUEUE_DROP_OLDEST` + select the enum value. +* `LOGIT_GET_DROPPED_TASKS()` and `LOGIT_RESET_DROPPED_TASKS()` forward to the + counter helpers. + +### Examples + +Basic setup using macros: + +```cpp +#include + +int main() { + LOGIT_ADD_CONSOLE_DEFAULT(); + LOGIT_SET_QUEUE_POLICY(LOGIT_QUEUE_BLOCK); + + LOGIT_INFO("async logging is live"); + LOGIT_WAIT(); +} +``` + +Hot resize while the system is running (only with `LOGIT_USE_MPSC_RING`): + +```cpp +auto& executor = logit::detail::TaskExecutor::get_instance(); +LOGIT_SET_QUEUE_POLICY(LOGIT_QUEUE_BLOCK); + +// Later, increase the capacity without losing accepted tasks. +LOGIT_SET_MAX_QUEUE(1024); // producers briefly wait for resize to finish +``` + +Inspecting drops under `DropNewest`: + +```cpp +LOGIT_SET_QUEUE_POLICY(LOGIT_QUEUE_DROP_NEWEST); +LOGIT_SET_MAX_QUEUE(16); +LOGIT_RESET_DROPPED_TASKS(); + +for (int i = 0; i < 1000; ++i) { + LOGIT_INFO("burst", i); +} + +LOGIT_WAIT(); +const auto lost = LOGIT_GET_DROPPED_TASKS(); +``` + +## 8. Thread-safety and TSAN considerations + +* All public methods on non-Emscripten builds are thread-safe. Producers may + call `add_task()` concurrently with `set_max_queue_size()` and + `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. +* 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. + +## 9. Performance and tuning + +* `QueuePolicy::Block` limits the number of in-flight tasks tracked by + `m_active_tasks`. Use it to introduce producer-side backpressure when the + downstream sinks are expensive. +* The worker drains up to 2048 tasks per iteration when the ring is enabled. + Increase this "budget" in `TaskExecutor::worker_function()` if your workload + generates extremely large bursts and the worker sleeps too often. Reducing it + can lower per-iteration latency for latency-sensitive applications. +* Adjust `LOGIT_TASK_EXECUTOR_DEFAULT_RING_CAPACITY` at compile time to select a + different default capacity when `LOGIT_USE_MPSC_RING` is active. +* Monitor `dropped_tasks()` during load testing to verify that the chosen policy + matches the application's tolerance for loss. diff --git a/include/logit_cpp/logit/detail/TaskExecutor.hpp b/include/logit_cpp/logit/detail/TaskExecutor.hpp index fe02d47..a033358 100644 --- a/include/logit_cpp/logit/detail/TaskExecutor.hpp +++ b/include/logit_cpp/logit/detail/TaskExecutor.hpp @@ -3,7 +3,8 @@ #define _LOGIT_DETAIL_TASK_EXECUTOR_HPP_INCLUDED /// \file TaskExecutor.hpp -/// \brief Defines the TaskExecutor class, which manages task execution in a separate thread. +/// \brief Task executor used by asynchronous loggers. +/// \details Detailed design notes are available in docs/TaskExecutor.md. #include #include @@ -30,21 +31,31 @@ namespace logit { namespace detail { - /// \brief Queue overflow handling policy. - enum class QueuePolicy { DropNewest, DropOldest, Block }; + /// \brief Queue overflow handling policy used by TaskExecutor. + enum class QueuePolicy { + DropNewest, ///< Reject the incoming task when the queue is full. + DropOldest, ///< Drop the oldest queued task (or the incoming one in MPSC builds). + Block ///< Producers wait until capacity is available. + }; # if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__) /// \class TaskExecutor /// \brief Simplified task executor for single-threaded Emscripten builds. + /// \details The Emscripten variant keeps behaviour compatible with the + /// browser event loop. See docs/TaskExecutor.md for the high level design. /// \thread_safety Not thread-safe. class TaskExecutor { public: + /// \brief Returns the singleton executor instance. static TaskExecutor& get_instance() { static TaskExecutor instance; return instance; } - + + /// \brief Enqueue a task to be executed on the async drain. + /// \note Backpressure policies mirror the deque implementation described + /// in docs/TaskExecutor.md. void add_task(std::function task) { if (!task) return; bool schedule = false; @@ -87,23 +98,29 @@ namespace logit { namespace detail { emscripten_async_call(&TaskExecutor::drain_thunk, this, 0); } } - + + /// \brief Drain the queue synchronously. void wait() { drain(); } + /// \brief Shut down the executor by draining all queued work. void shutdown() { drain(); } - + + /// \brief Change the maximum queue size (`0` disables the limit). void set_max_queue_size(std::size_t size) { std::lock_guard lk(m_mutex); m_max_queue_size = size; } - + + /// \brief Update the queue overflow policy. void set_queue_policy(QueuePolicy policy) { std::lock_guard lk(m_mutex); m_overflow_policy = policy; } - + + /// \brief Return the number of tasks dropped by the overflow policy. std::size_t dropped_tasks() const noexcept { return m_dropped_tasks.load(std::memory_order_relaxed); } + /// \brief Reset the drop counter to zero. void reset_dropped_tasks() noexcept { m_dropped_tasks.store(0, std::memory_order_relaxed); } @@ -151,17 +168,25 @@ namespace logit { namespace detail { # else // !Emscripten or pthreads /// \class TaskExecutor - /// \brief A thread-safe task executor that processes tasks in a dedicated worker thread. + /// \brief Thread-safe task executor backed by a dedicated worker thread. + /// \details The full design, including backpressure semantics and hot + /// resizing, is described in docs/TaskExecutor.md. /// \thread_safety Thread-safe. class TaskExecutor { public: - /// Singleton (сохраняем вашу реализацию с new). + /// \brief Returns the global executor instance. + /// \note The pointer-based singleton intentionally outlives static + /// logger destructors so logging remains available during process + /// shutdown. static TaskExecutor& get_instance() { static TaskExecutor* instance = new TaskExecutor(); return *instance; } - - /// Добавить задачу. + + /// \brief Enqueue a task for asynchronous execution. + /// \note `QueuePolicy::DropOldest` drops the incoming task when + /// `LOGIT_USE_MPSC_RING` is defined. See docs/TaskExecutor.md for the + /// rationale. void add_task(std::function task) { if (!task) return; # ifndef LOGIT_USE_MPSC_RING @@ -191,7 +216,7 @@ namespace logit { namespace detail { lock.unlock(); m_queue_condition.notify_one(); # else - // Барьер ресайза: пережидаем «горячее» изменение кольца. + // Hot-resize barrier: wait until the ring rebuild is finished. if (m_resizing.load(std::memory_order_acquire)) { std::unique_lock lk(m_cv_mutex); m_resize_cv.wait(lk, [this]{ return !m_resizing.load(std::memory_order_acquire); }); @@ -210,7 +235,7 @@ namespace logit { namespace detail { const auto policy = m_overflow_policy.load(std::memory_order_relaxed); - // Реальное backpressure: учитываем "висящие" задачи. + // Backpressure relies on the count of in-flight tasks. if (policy == QueuePolicy::Block && m_max_queue_size > 0 && m_active_tasks.load(std::memory_order_relaxed) >= m_max_queue_size) @@ -220,21 +245,21 @@ namespace logit { namespace detail { continue; } - // Пытаемся положить в кольцо. + // Try to push into the ring buffer. if (m_mpsc_queue.try_push(local_task)) { - m_cv.notify_one(); // разбудить воркера + m_cv.notify_one(); // wake the worker return; } - - // Переполнение — применяем политику. + + // Apply the configured overflow policy when the ring is full. switch (policy) { case QueuePolicy::DropNewest: m_dropped_tasks.fetch_add(1, std::memory_order_relaxed); return; - + case QueuePolicy::DropOldest: - // Безопасная реализация под MPSC: дропаем входящий. - // Это сохраняет порядок и исключает дедлоки при gate. + // Safe MPSC behaviour: drop the incoming task. + // Preserves ordering and avoids producer/consumer deadlocks. m_dropped_tasks.fetch_add(1, std::memory_order_relaxed); return; @@ -248,7 +273,7 @@ namespace logit { namespace detail { # endif } - /// Дождаться опустошения. + /// \brief Block until the queue is empty or a shutdown is requested. void wait() { # ifndef LOGIT_USE_MPSC_RING std::unique_lock lock(m_queue_mutex); @@ -267,7 +292,7 @@ namespace logit { namespace detail { # endif } - /// Остановить воркер. + /// \brief Stop the worker thread and drain outstanding tasks. void shutdown() { # ifndef LOGIT_USE_MPSC_RING std::unique_lock lock(m_queue_mutex); @@ -290,42 +315,44 @@ namespace logit { namespace detail { # endif } - /// Изменить ёмкость очереди. + /// \brief Update the maximum queue size (`0` disables the limit). + /// \details On MPSC builds this performs the "hot" resize described in + /// docs/TaskExecutor.md. void set_max_queue_size(std::size_t size) { # ifdef LOGIT_USE_MPSC_RING - // Сигналим продюсерам, чтобы переждали ресайз (до любых ожиданий/стопов). + // Tell producers to pause before any wait()/stop conditions run. m_resizing.store(true, std::memory_order_release); - // Дождаться опустошения очереди + // Drain the queue completely. wait(); - - // Акуратно остановить воркер и дождаться его завершения, чтобы он не трогал m_mpsc_queue, пока мы его меняем. + + // Stop the worker so it cannot touch m_mpsc_queue during the resize. std::unique_lock lk(m_queue_mutex); m_stop_flag.store(true, std::memory_order_relaxed); - lk.unlock(); - + lk.unlock(); + m_cv.notify_all(); m_queue_condition.notify_all(); if (m_worker_thread.joinable()) { m_worker_thread.join(); } - - // Переинициализировать параметры и само кольцо в единственном потоке. - lk.lock(); - m_max_queue_size = size; - const std::size_t cap = - (m_max_queue_size == 0 ? m_default_ring_cap : m_max_queue_size); - m_mpsc_queue = MpscRingAny>(cap); - // обнулить счётчики (не обязательно, но логично при "чистой" очереди). - m_active_tasks.store(0, std::memory_order_relaxed); - // m_dropped_tasks оставляем как есть — тесты его сами сбрасывают макросом. - lk.unlock(); - - // Снимаем стоп-флаг, перезапускаем воркер… + + // Reinitialise the parameters and the ring on a single thread. + lk.lock(); + m_max_queue_size = size; + const std::size_t cap = + (m_max_queue_size == 0 ? m_default_ring_cap : m_max_queue_size); + m_mpsc_queue = MpscRingAny>(cap); + // Reset counters (except drops) because the queue is empty. + m_active_tasks.store(0, std::memory_order_relaxed); + // Keep m_dropped_tasks untouched — tests manage it via macros. + lk.unlock(); + + // Clear the stop flag and restart the worker thread. m_stop_flag.store(false, std::memory_order_relaxed); m_worker_thread = std::thread(&TaskExecutor::worker_function, this); - // Открываем барьер для продюсеров. + // Re-open the barrier so producers can continue. m_resizing.store(false, std::memory_order_release); m_resize_cv.notify_all(); # else @@ -334,15 +361,17 @@ namespace logit { namespace detail { # endif } - /// Политика переполнения. + /// \brief Change the overflow policy for newly submitted tasks. void set_queue_policy(QueuePolicy policy) { std::lock_guard lock(m_queue_mutex); m_overflow_policy.store(policy, std::memory_order_relaxed); } - + + /// \brief Return the number of tasks dropped by the overflow policy. std::size_t dropped_tasks() const noexcept { return m_dropped_tasks.load(std::memory_order_relaxed); } + /// \brief Reset the overflow counter to zero. void reset_dropped_tasks() noexcept { m_dropped_tasks.store(0, std::memory_order_relaxed); } @@ -359,14 +388,14 @@ namespace logit { namespace detail { std::atomic m_dropped_tasks; std::atomic m_active_tasks; #else - mutable std::mutex m_queue_mutex; ///< Для wait()/смены политики. - std::condition_variable m_queue_condition; ///< Будим wait() на полном drain. - - std::condition_variable m_cv; ///< Будим воркер / продюсеров. - std::mutex m_cv_mutex; ///< Сон продюсеров/воркера. + mutable std::mutex m_queue_mutex; ///< Guards wait() and policy changes. + std::condition_variable m_queue_condition; ///< Notifies wait() once the queue drains. + + std::condition_variable m_cv; ///< Wakes the worker or producers. + std::mutex m_cv_mutex; ///< Protects producer/worker sleeps. - std::atomic m_resizing; ///< true — идёт ресайз кольца. - std::condition_variable m_resize_cv; ///< Продюсеры ждут окончания ресайза. + std::atomic m_resizing; ///< true while a hot resize is in flight. + std::condition_variable m_resize_cv; ///< Producers wait here during a resize. std::thread m_worker_thread; std::atomic m_stop_flag; @@ -418,13 +447,13 @@ namespace logit { namespace detail { task(); m_active_tasks.fetch_sub(1, std::memory_order_relaxed); - m_cv.notify_one(); // освободили in-flight слот + m_cv.notify_one(); // freed an in-flight slot } if (queue_empty_() && m_active_tasks.load(std::memory_order_relaxed) == 0) { std::unique_lock lock(m_queue_mutex); - m_queue_condition.notify_all(); // для wait() - m_cv.notify_all(); // разбудить продюсеров Block + m_queue_condition.notify_all(); // notify wait() + m_cv.notify_all(); // wake producers blocked on Block if (m_stop_flag.load(std::memory_order_acquire)) { break; }