From 82bc16ea106faabe83bf43face63d1d20cb6d44b Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Mon, 15 Sep 2025 03:13:36 +0300 Subject: [PATCH 1/3] feat(task-executor): add drop policies and tracking --- include/logit_cpp/logit/LogMacros.hpp | 12 +++-- .../logit_cpp/logit/detail/TaskExecutor.hpp | 53 ++++++++++++++----- 2 files changed, 50 insertions(+), 15 deletions(-) diff --git a/include/logit_cpp/logit/LogMacros.hpp b/include/logit_cpp/logit/LogMacros.hpp index 87a1b28..124e0d1 100644 --- a/include/logit_cpp/logit/LogMacros.hpp +++ b/include/logit_cpp/logit/LogMacros.hpp @@ -1340,14 +1340,20 @@ static_assert(LOGIT_LEVEL_FATAL == static_cast(logit::LogLevel::LOG_LVL_FAT #define LOGIT_SET_MAX_QUEUE(size) \ logit::detail::TaskExecutor::get_instance().set_max_queue_size(size) -/// \brief Queue policy for dropping tasks when the queue is full. -#define LOGIT_QUEUE_DROP logit::detail::QueuePolicy::Drop +/// \brief Queue policy for dropping the newest task when the queue is full. +#define LOGIT_QUEUE_DROP_NEWEST logit::detail::QueuePolicy::DropNewest + +/// \brief Queue policy for dropping the oldest task when the queue is full. +#define LOGIT_QUEUE_DROP_OLDEST logit::detail::QueuePolicy::DropOldest + +/// \brief Backward-compatible alias for dropping the newest task. +#define LOGIT_QUEUE_DROP LOGIT_QUEUE_DROP_NEWEST /// \brief Queue policy for blocking when the queue is full. #define LOGIT_QUEUE_BLOCK logit::detail::QueuePolicy::Block /// \brief Sets the behavior when the queue is full. -/// \param mode LOGIT_QUEUE_DROP or LOGIT_QUEUE_BLOCK. +/// \param mode LOGIT_QUEUE_DROP_NEWEST, LOGIT_QUEUE_DROP_OLDEST or LOGIT_QUEUE_BLOCK. #define LOGIT_SET_QUEUE_POLICY(mode) \ logit::detail::TaskExecutor::get_instance().set_queue_policy(mode) diff --git a/include/logit_cpp/logit/detail/TaskExecutor.hpp b/include/logit_cpp/logit/detail/TaskExecutor.hpp index fc4e396..bd94d6d 100644 --- a/include/logit_cpp/logit/detail/TaskExecutor.hpp +++ b/include/logit_cpp/logit/detail/TaskExecutor.hpp @@ -6,6 +6,7 @@ /// \brief Defines the TaskExecutor class, which manages task execution in a separate thread. #include +#include #if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__) #include #include @@ -20,7 +21,7 @@ namespace logit { namespace detail { /// \brief Queue overflow handling policy. - enum class QueuePolicy { Drop, Block }; + enum class QueuePolicy { DropNewest, DropOldest, Block }; #if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__) @@ -36,6 +37,20 @@ namespace logit { namespace detail { void add_task(std::function task) { if (!task) return; const bool schedule = m_tasks.empty(); + if (m_max_queue_size > 0 && m_tasks.size() >= m_max_queue_size) { + switch (m_overflow_policy) { + case QueuePolicy::DropNewest: + ++m_dropped_tasks; + return; + case QueuePolicy::DropOldest: + m_tasks.pop_front(); + ++m_dropped_tasks; + break; + case QueuePolicy::Block: + drain(); + break; + } + } m_tasks.push_back(std::move(task)); if (schedule) { emscripten_async_call(&TaskExecutor::drain_thunk, this, 0); @@ -45,11 +60,11 @@ namespace logit { namespace detail { void wait() { drain(); } void shutdown() { drain(); } - void set_max_queue_size(std::size_t) {} - void set_queue_policy(QueuePolicy) {} + void set_max_queue_size(std::size_t size) { m_max_queue_size = size; } + void set_queue_policy(QueuePolicy policy) { m_overflow_policy = policy; } private: - TaskExecutor() = default; + TaskExecutor() : m_max_queue_size(0), m_overflow_policy(QueuePolicy::Block), m_dropped_tasks(0) {} ~TaskExecutor() = default; TaskExecutor(const TaskExecutor&) = delete; TaskExecutor& operator=(const TaskExecutor&) = delete; @@ -57,6 +72,9 @@ namespace logit { namespace detail { TaskExecutor& operator=(TaskExecutor&&) = delete; std::deque> m_tasks; + std::size_t m_max_queue_size; + QueuePolicy m_overflow_policy; + std::atomic m_dropped_tasks; static void drain_thunk(void* arg) { static_cast(arg)->drain(); @@ -93,13 +111,21 @@ namespace logit { namespace detail { std::unique_lock lock(m_queue_mutex); if (m_stop_flag) return; if (m_max_queue_size > 0 && m_tasks_queue.size() >= m_max_queue_size) { - if (m_overflow_policy == QueuePolicy::Drop) { - return; + switch (m_overflow_policy) { + case QueuePolicy::DropNewest: + ++m_dropped_tasks; + return; + case QueuePolicy::DropOldest: + m_tasks_queue.pop(); + ++m_dropped_tasks; + break; + case QueuePolicy::Block: + m_queue_condition.wait(lock, [this]() { + return m_tasks_queue.size() < m_max_queue_size || m_stop_flag; + }); + if (m_stop_flag) return; + break; } - m_queue_condition.wait(lock, [this]() { - return m_tasks_queue.size() < m_max_queue_size || m_stop_flag; - }); - if (m_stop_flag) return; } m_tasks_queue.push(std::move(task)); lock.unlock(); @@ -138,7 +164,9 @@ namespace logit { namespace detail { } /// \brief Sets the behavior when the queue is full. - /// \param policy QueuePolicy::Drop to discard tasks or QueuePolicy::Block to wait. + /// \param policy QueuePolicy::DropNewest to discard the incoming task, + /// QueuePolicy::DropOldest to discard the oldest task, + /// or QueuePolicy::Block to wait. void set_queue_policy(QueuePolicy policy) { std::lock_guard lock(m_queue_mutex); m_overflow_policy = policy; @@ -152,6 +180,7 @@ namespace logit { namespace detail { bool m_stop_flag; ///< Flag indicating if the worker thread should stop. std::size_t m_max_queue_size; ///< Maximum number of tasks in the queue (0 for unlimited). QueuePolicy m_overflow_policy; ///< Policy for handling queue overflow. + std::atomic m_dropped_tasks; ///< Number of discarded tasks due to overflow. /// \brief The worker thread function that processes tasks from the queue. void worker_function() { @@ -173,7 +202,7 @@ namespace logit { namespace detail { } /// \brief Private constructor to enforce the singleton pattern. - TaskExecutor() : m_stop_flag(false), m_max_queue_size(0), m_overflow_policy(QueuePolicy::Block) { + TaskExecutor() : m_stop_flag(false), m_max_queue_size(0), m_overflow_policy(QueuePolicy::Block), m_dropped_tasks(0) { m_worker_thread = std::thread(&TaskExecutor::worker_function, this); } From ceb4a8d3dabe7d1d3dceaf7ea41c5732bfa46078 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Mon, 15 Sep 2025 03:24:19 +0300 Subject: [PATCH 2/3] refactor(task-executor): use deque for overflow --- include/logit_cpp/logit/detail/TaskExecutor.hpp | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/include/logit_cpp/logit/detail/TaskExecutor.hpp b/include/logit_cpp/logit/detail/TaskExecutor.hpp index bd94d6d..030fa30 100644 --- a/include/logit_cpp/logit/detail/TaskExecutor.hpp +++ b/include/logit_cpp/logit/detail/TaskExecutor.hpp @@ -12,7 +12,7 @@ #include #else #include -#include +#include #include #include #include @@ -116,7 +116,7 @@ namespace logit { namespace detail { ++m_dropped_tasks; return; case QueuePolicy::DropOldest: - m_tasks_queue.pop(); + m_tasks_queue.pop_front(); ++m_dropped_tasks; break; case QueuePolicy::Block: @@ -127,7 +127,7 @@ namespace logit { namespace detail { break; } } - m_tasks_queue.push(std::move(task)); + m_tasks_queue.push_back(std::move(task)); lock.unlock(); m_queue_condition.notify_one(); } @@ -173,7 +173,7 @@ namespace logit { namespace detail { } private: - std::queue> m_tasks_queue; ///< Queue holding tasks to be executed. + std::deque> m_tasks_queue; ///< Queue holding tasks to be executed. mutable std::mutex m_queue_mutex; ///< Mutex to protect access to the task queue. std::condition_variable m_queue_condition; ///< Condition variable to signal task availability. std::thread m_worker_thread; ///< Worker thread for executing tasks. @@ -194,7 +194,7 @@ namespace logit { namespace detail { break; } task = std::move(m_tasks_queue.front()); - m_tasks_queue.pop(); + m_tasks_queue.pop_front(); lock.unlock(); m_queue_condition.notify_one(); task(); From 7498179a4d7806ce1548666e22b7f1614729a3b4 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Mon, 15 Sep 2025 03:24:24 +0300 Subject: [PATCH 3/3] fix(task-executor): wait for running tasks --- .../logit_cpp/logit/detail/TaskExecutor.hpp | 24 +++++++++++-------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/include/logit_cpp/logit/detail/TaskExecutor.hpp b/include/logit_cpp/logit/detail/TaskExecutor.hpp index 030fa30..670e4cf 100644 --- a/include/logit_cpp/logit/detail/TaskExecutor.hpp +++ b/include/logit_cpp/logit/detail/TaskExecutor.hpp @@ -36,7 +36,6 @@ namespace logit { namespace detail { void add_task(std::function task) { if (!task) return; - const bool schedule = m_tasks.empty(); if (m_max_queue_size > 0 && m_tasks.size() >= m_max_queue_size) { switch (m_overflow_policy) { case QueuePolicy::DropNewest: @@ -51,6 +50,7 @@ namespace logit { namespace detail { break; } } + const bool schedule = m_tasks.empty(); m_tasks.push_back(std::move(task)); if (schedule) { emscripten_async_call(&TaskExecutor::drain_thunk, this, 0); @@ -134,14 +134,10 @@ namespace logit { namespace detail { /// \brief Waits for all tasks in the queue to be processed. void wait() { - m_queue_condition.notify_one(); - for (;;) { - std::unique_lock lock(m_queue_mutex); - if (m_tasks_queue.empty() || m_stop_flag) break; - lock.unlock(); - std::this_thread::yield(); - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - } + std::unique_lock lock(m_queue_mutex); + m_queue_condition.wait(lock, [this]() { + return (m_tasks_queue.empty() && m_active_tasks.load() == 0) || m_stop_flag; + }); } /// \brief Shuts down the TaskExecutor by stopping the worker thread. @@ -181,6 +177,7 @@ namespace logit { namespace detail { std::size_t m_max_queue_size; ///< Maximum number of tasks in the queue (0 for unlimited). QueuePolicy m_overflow_policy; ///< Policy for handling queue overflow. std::atomic m_dropped_tasks; ///< Number of discarded tasks due to overflow. + std::atomic m_active_tasks; ///< Number of tasks currently running. /// \brief The worker thread function that processes tasks from the queue. void worker_function() { @@ -195,14 +192,21 @@ namespace logit { namespace detail { } task = std::move(m_tasks_queue.front()); m_tasks_queue.pop_front(); + ++m_active_tasks; lock.unlock(); m_queue_condition.notify_one(); task(); + lock.lock(); + --m_active_tasks; + if (m_tasks_queue.empty() && m_active_tasks == 0) { + m_queue_condition.notify_all(); + } + lock.unlock(); } } /// \brief Private constructor to enforce the singleton pattern. - TaskExecutor() : m_stop_flag(false), m_max_queue_size(0), m_overflow_policy(QueuePolicy::Block), m_dropped_tasks(0) { + TaskExecutor() : m_stop_flag(false), m_max_queue_size(0), m_overflow_policy(QueuePolicy::Block), m_dropped_tasks(0), m_active_tasks(0) { m_worker_thread = std::thread(&TaskExecutor::worker_function, this); }