diff --git a/CMakeLists.txt b/CMakeLists.txt index 7533b11..8ba99e4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -14,6 +14,8 @@ if(CMAKE_SYSTEM_NAME STREQUAL "Emscripten") option(LOGIT_EMSCRIPTEN "Build for Emscripten" ON) endif() option(LOGIT_FORCE_ASYNC_OFF "Force disable async logging" OFF) +option(LOGIT_USE_MPSC_RING "Enable lock-free TaskExecutor queue" ON) +option(LOGIT_ENABLE_DROP_OLDEST_SLOWPATH "Enable TaskExecutor DropOldest slow-path" ON) if(NOT DEFINED CMAKE_CXX_STANDARD) set(CMAKE_CXX_STANDARD 11) @@ -42,6 +44,14 @@ target_include_directories(log-it-cpp INTERFACE target_link_libraries(log-it-cpp INTERFACE time_shield::time_shield) +if(LOGIT_USE_MPSC_RING) + target_compile_definitions(log-it-cpp INTERFACE LOGIT_USE_MPSC_RING=1) +endif() + +if(LOGIT_ENABLE_DROP_OLDEST_SLOWPATH) + target_compile_definitions(log-it-cpp INTERFACE LOGIT_ENABLE_DROP_OLDEST_SLOWPATH=1) +endif() + if(LOGIT_EMSCRIPTEN) set(LOGIT_WITH_SYSLOG OFF CACHE BOOL "" FORCE) set(LOGIT_WITH_WIN_EVENT_LOG OFF CACHE BOOL "" FORCE) diff --git a/include/logit_cpp/logit/detail/MpscRingAny.hpp b/include/logit_cpp/logit/detail/MpscRingAny.hpp index fba3fc6..a4c206a 100644 --- a/include/logit_cpp/logit/detail/MpscRingAny.hpp +++ b/include/logit_cpp/logit/detail/MpscRingAny.hpp @@ -37,14 +37,39 @@ namespace logit { namespace detail { /// \brief Destroy remaining elements, if any. ~MpscRingAny() { - T tmp; - while (try_pop(tmp)) { - // Element destroyed via move-from tmp - } + destroy_all_(); } - + MpscRingAny(const MpscRingAny&) = delete; MpscRingAny& operator=(const MpscRingAny&) = delete; + + /// \brief Move construct ring, adopting producer/consumer indices. + MpscRingAny(MpscRingAny&& other) noexcept + : m_cap(other.m_cap), + m_cells(std::move(other.m_cells)), + m_enqueue_pos(other.m_enqueue_pos.load(std::memory_order_relaxed)), + m_dequeue_pos(other.m_dequeue_pos.load(std::memory_order_relaxed)) { + other.m_cap = 0; + other.m_enqueue_pos.store(0, std::memory_order_relaxed); + other.m_dequeue_pos.store(0, std::memory_order_relaxed); + } + + /// \brief Move assign ring, draining existing payload. + MpscRingAny& operator=(MpscRingAny&& other) noexcept { + if (this != &other) { + destroy_all_(); + m_cap = other.m_cap; + m_cells = std::move(other.m_cells); + m_enqueue_pos.store(other.m_enqueue_pos.load(std::memory_order_relaxed), + std::memory_order_relaxed); + m_dequeue_pos.store(other.m_dequeue_pos.load(std::memory_order_relaxed), + std::memory_order_relaxed); + other.m_cap = 0; + other.m_enqueue_pos.store(0, std::memory_order_relaxed); + other.m_dequeue_pos.store(0, std::memory_order_relaxed); + } + return *this; + } /// \brief Capacity of the ring. std::size_t capacity() const noexcept { return m_cap; } @@ -117,6 +142,9 @@ namespace logit { namespace detail { /// \brief Lightweight emptiness check for current consumer position. bool empty() const noexcept { + if (!m_cells || m_cap == 0) { + return true; + } std::size_t pos = m_dequeue_pos.load(std::memory_order_acquire); const Cell& c = m_cells[pos % m_cap]; std::size_t seq = c.m_seq.load(std::memory_order_acquire); @@ -124,8 +152,18 @@ namespace logit { namespace detail { static_cast(seq) - static_cast(pos + 1); return diff != 0; } - + private: + void destroy_all_() noexcept { + if (!m_cells || m_cap == 0) { + return; + } + T tmp; + while (try_pop(tmp)) { + // Element destroyed via move-from tmp + } + } + std::size_t m_cap; std::unique_ptr m_cells; alignas(64) std::atomic m_enqueue_pos; diff --git a/include/logit_cpp/logit/detail/TaskExecutor.hpp b/include/logit_cpp/logit/detail/TaskExecutor.hpp index 6656ef7..5090f1b 100644 --- a/include/logit_cpp/logit/detail/TaskExecutor.hpp +++ b/include/logit_cpp/logit/detail/TaskExecutor.hpp @@ -19,6 +19,17 @@ #include #endif +// Enable lock-free MPSC ring integration (non-Emscripten) by defining: +// #define LOGIT_USE_MPSC_RING +// Optional: enable rare DropOldest slow-path coordination: +// #define LOGIT_ENABLE_DROP_OLDEST_SLOWPATH + +#if !defined(__EMSCRIPTEN__) || defined(__EMSCRIPTEN_PTHREADS__) +# ifdef LOGIT_USE_MPSC_RING +# include "MpscRingAny.hpp" +# endif +#endif + namespace logit { namespace detail { /// \brief Queue overflow handling policy. @@ -53,8 +64,10 @@ namespace logit { namespace detail { ++m_dropped_tasks; return; case QueuePolicy::DropOldest: - m_tasks.pop_front(); - ++m_dropped_tasks; + if (!m_tasks.empty()) { + m_tasks.pop_front(); + ++m_dropped_tasks; + } break; case QueuePolicy::Block: break; // handled after unlocking @@ -146,9 +159,6 @@ namespace logit { namespace detail { /// \class TaskExecutor /// \brief A thread-safe task executor that processes tasks in a dedicated worker thread. /// \thread_safety Thread-safe. - /// - /// This class provides a mechanism for queuing tasks (functions or lambdas) and executing them - /// asynchronously in a background thread. It follows the singleton design pattern. class TaskExecutor { public: /// \brief Get the singleton instance of the TaskExecutor. @@ -161,55 +171,145 @@ namespace logit { namespace detail { /// \brief Adds a task to the queue in a thread-safe manner. /// \param task A function or lambda with no arguments to be executed asynchronously. void add_task(std::function task) { + if (!task) return; +#ifdef LOGIT_USE_MPSC_RING + if (m_stop_flag.load(std::memory_order_acquire)) { + return; + } + + std::function local_task = std::move(task); + + if (m_mpsc_queue.try_push(local_task)) { + m_cv.notify_one(); + return; + } + + switch (m_overflow_policy.load(std::memory_order_relaxed)) { + case QueuePolicy::DropNewest: + ++m_dropped_tasks; + return; + case QueuePolicy::Block: { + for (int i = 0; i < 2000; ++i) { + if (m_mpsc_queue.try_push(local_task)) { + m_cv.notify_one(); + return; + } + } + std::unique_lock lk(m_cv_mutex); + m_cv.wait_for(lk, std::chrono::microseconds(50)); + if (m_mpsc_queue.try_push(local_task)) { + m_cv.notify_one(); + return; + } + ++m_dropped_tasks; + return; + } + case QueuePolicy::DropOldest: +#ifdef LOGIT_ENABLE_DROP_OLDEST_SLOWPATH + { + std::unique_lock lk(m_drop_mutex); + const std::size_t target = ++m_drop_requested; + m_cv.notify_one(); + m_drop_cv.wait_for(lk, std::chrono::milliseconds(2), + [this, target] { return m_drop_done >= target; }); + if (m_mpsc_queue.try_push(local_task)) { + m_cv.notify_one(); + return; + } + ++m_dropped_tasks; + return; + } +#else + ++m_dropped_tasks; + return; +#endif + } +#else std::unique_lock lock(m_queue_mutex); - if (m_stop_flag) return; + if (m_stop_flag.load(std::memory_order_relaxed)) return; if (m_max_queue_size > 0 && m_tasks_queue.size() >= m_max_queue_size) { - switch (m_overflow_policy) { + switch (m_overflow_policy.load(std::memory_order_relaxed)) { case QueuePolicy::DropNewest: ++m_dropped_tasks; return; case QueuePolicy::DropOldest: - m_tasks_queue.pop_front(); - ++m_dropped_tasks; + if (!m_tasks_queue.empty()) { + m_tasks_queue.pop_front(); + ++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; + return m_tasks_queue.size() < m_max_queue_size || + m_stop_flag.load(std::memory_order_relaxed); }); - if (m_stop_flag) return; + if (m_stop_flag.load(std::memory_order_relaxed)) return; break; } } m_tasks_queue.push_back(std::move(task)); lock.unlock(); m_queue_condition.notify_one(); +#endif } /// \brief Waits for all tasks in the queue to be processed. void wait() { +#ifdef LOGIT_USE_MPSC_RING 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; + return ((queue_empty_() && + m_active_tasks.load(std::memory_order_relaxed) == 0) || + m_stop_flag.load(std::memory_order_relaxed)); }); +#else + std::unique_lock lock(m_queue_mutex); + m_queue_condition.wait(lock, [this]() { + return ((m_tasks_queue.empty() && + m_active_tasks.load(std::memory_order_relaxed) == 0) || + m_stop_flag.load(std::memory_order_relaxed)); + }); +#endif } /// \brief Shuts down the TaskExecutor by stopping the worker thread. /// \details This method signals the worker thread to stop and then joins it. void shutdown() { +#ifdef LOGIT_USE_MPSC_RING + { + std::lock_guard lock(m_queue_mutex); + m_stop_flag.store(true, std::memory_order_relaxed); + } + m_cv.notify_all(); + m_queue_condition.notify_all(); + if (m_worker_thread.joinable()) { + m_worker_thread.join(); + } +#else std::unique_lock lock(m_queue_mutex); - m_stop_flag = true; + m_stop_flag.store(true, std::memory_order_relaxed); lock.unlock(); m_queue_condition.notify_all(); if (m_worker_thread.joinable()) { m_worker_thread.join(); } +#endif } /// \brief Sets the maximum size of the task queue. /// \param size Maximum number of tasks in the queue (0 for unlimited). void set_max_queue_size(std::size_t size) { +#ifdef LOGIT_USE_MPSC_RING + std::lock_guard lk(m_queue_mutex); + m_max_queue_size = size; + if (queue_empty_() && m_active_tasks.load(std::memory_order_relaxed) == 0) { + std::size_t cap = (m_max_queue_size == 0 ? m_default_ring_cap : m_max_queue_size); + m_mpsc_queue = MpscRingAny>(cap); + } +#else std::lock_guard lock(m_queue_mutex); m_max_queue_size = size; +#endif } /// \brief Sets the behavior when the queue is full. @@ -218,48 +318,156 @@ namespace logit { namespace detail { /// or QueuePolicy::Block to wait. void set_queue_policy(QueuePolicy policy) { std::lock_guard lock(m_queue_mutex); - m_overflow_policy = policy; + m_overflow_policy.store(policy, std::memory_order_relaxed); } private: +#ifndef LOGIT_USE_MPSC_RING 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. - bool m_stop_flag; ///< Flag indicating if the worker thread should stop. + std::atomic 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_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. +#else + mutable std::mutex m_queue_mutex; ///< Used only for wait()/policy changes. + std::condition_variable m_queue_condition; ///< Notifies waiters on full drain. + + std::condition_variable m_cv; ///< Wake-up for worker on push. + std::mutex m_cv_mutex; ///< Sleep mutex for worker waits. + + std::thread m_worker_thread; ///< Worker thread for executing tasks. + std::atomic m_stop_flag; ///< Flag indicating if the worker thread should stop. + std::size_t m_max_queue_size; ///< Maximum number of tasks requested by user. + std::atomic 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. + const std::size_t m_default_ring_cap = 1024; ///< Default capacity when unlimited requested. + MpscRingAny> m_mpsc_queue; ///< Lock-free bounded MPSC ring. + +#ifdef LOGIT_ENABLE_DROP_OLDEST_SLOWPATH + std::mutex m_drop_mutex; ///< Coordinates DropOldest slow-path. + std::condition_variable m_drop_cv; ///< Producer waits for confirmation. + std::size_t m_drop_requested; ///< Number of requested drops. + std::size_t m_drop_done; ///< Number of drops completed. +#endif +#endif + /// \brief The worker thread function that processes tasks from the queue. void worker_function() { +#ifndef LOGIT_USE_MPSC_RING for (;;) { std::function task; std::unique_lock lock(m_queue_mutex); m_queue_condition.wait(lock, [this]() { - return !m_tasks_queue.empty() || m_stop_flag; + return !m_tasks_queue.empty() || m_stop_flag.load(std::memory_order_relaxed); }); - if (m_stop_flag && m_tasks_queue.empty()) { + if (m_stop_flag.load(std::memory_order_relaxed) && m_tasks_queue.empty()) { break; } task = std::move(m_tasks_queue.front()); m_tasks_queue.pop_front(); - ++m_active_tasks; + m_active_tasks.fetch_add(1, std::memory_order_relaxed); lock.unlock(); m_queue_condition.notify_one(); task(); lock.lock(); - --m_active_tasks; - if (m_tasks_queue.empty() && m_active_tasks == 0) { + m_active_tasks.fetch_sub(1, std::memory_order_relaxed); + if (m_tasks_queue.empty() && m_active_tasks.load(std::memory_order_relaxed) == 0) { m_queue_condition.notify_all(); } lock.unlock(); } +#else + for (;;) { + bool drained_any = false; + std::function task; + + int budget = 2048; + while (budget-- && m_mpsc_queue.try_pop(task)) { + drained_any = true; + m_active_tasks.fetch_add(1, std::memory_order_relaxed); + task(); + m_active_tasks.fetch_sub(1, std::memory_order_relaxed); + } + +#ifdef LOGIT_ENABLE_DROP_OLDEST_SLOWPATH + handle_drop_requests_(); +#endif + + if (queue_empty_() && m_active_tasks.load(std::memory_order_relaxed) == 0) { + std::unique_lock lock(m_queue_mutex); + m_queue_condition.notify_all(); + if (m_stop_flag.load(std::memory_order_relaxed)) { + break; + } + } + + if (!drained_any) { + std::unique_lock lk(m_cv_mutex); + if (m_stop_flag.load(std::memory_order_relaxed) && queue_empty_()) { + break; + } + m_cv.wait_for(lk, std::chrono::milliseconds(1)); + } + } +#endif + } + +#ifdef LOGIT_USE_MPSC_RING + /// \brief Return true if ring appears empty for current consumer position. + bool queue_empty_() const noexcept { + return m_mpsc_queue.empty(); + } + +#ifdef LOGIT_ENABLE_DROP_OLDEST_SLOWPATH + /// \brief Perform requested drops of oldest items (rare path). + void handle_drop_requests_() { + { + std::lock_guard g(m_drop_mutex); + if (m_drop_done >= m_drop_requested) { + return; + } + } + std::unique_lock lk(m_drop_mutex); + while (m_drop_done < m_drop_requested) { + std::function dummy; + if (m_mpsc_queue.try_pop(dummy)) { + ++m_drop_done; + } else { + break; + } + } + m_drop_cv.notify_all(); } +#endif +#endif /// \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), m_active_tasks(0) { + TaskExecutor() +#ifndef LOGIT_USE_MPSC_RING + : m_stop_flag(false), + m_max_queue_size(0), + m_overflow_policy(QueuePolicy::Block), + m_dropped_tasks(0), + m_active_tasks(0) +#else + : m_stop_flag(false), + m_max_queue_size(0), + m_overflow_policy(QueuePolicy::Block), + m_dropped_tasks(0), + m_active_tasks(0), + m_mpsc_queue(m_default_ring_cap) +#ifdef LOGIT_ENABLE_DROP_OLDEST_SLOWPATH + , m_drop_requested(0), + m_drop_done(0) +#endif +#endif + { m_worker_thread = std::thread(&TaskExecutor::worker_function, this); }