diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a295401..b009f52 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,7 +24,7 @@ jobs: - name: Install run: cmake --install build --prefix install - name: Test - run: ctest --test-dir build + run: ctest --test-dir build --output-on-failure - name: Configure consumer project run: cmake -S tests/install_consumer -B build-consumer -DCMAKE_PREFIX_PATH=${{ github.workspace }}/install -DCMAKE_CXX_STANDARD=${{ matrix.std }} - name: Build consumer project @@ -115,7 +115,7 @@ jobs: - name: Build run: cmake --build build - name: Test - run: ctest --test-dir build + run: ctest --test-dir build --output-on-failure tsan: runs-on: ubuntu-latest @@ -131,7 +131,7 @@ jobs: - name: Build run: cmake --build build - name: Test - run: ctest --test-dir build + run: ctest --test-dir build --output-on-failure vcpkg-install: runs-on: ubuntu-latest diff --git a/AGENTS.md b/AGENTS.md index 96e6744..b745dd0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,6 +17,14 @@ Include a body that describes the change. Keep diffs minimal and focused. Do not refactor or apply style changes beyond the lines you directly touch. +## Repository setup +Before configuring or building the project, initialize the git submodules so +embedded dependencies such as TimeShield are present: + +``` +git submodule update --init --recursive +``` + ## Include Policy - Do not use `../` in `#include` directives. - Within a module (`logit/utils/*`, `logit/formatter/*`, `logit/loggers/*`) only include headers located in the same sub-tree using forward paths (for example `#include "compiler/PatternCompiler.hpp"`). diff --git a/docs/backpressure.md b/docs/backpressure.md new file mode 100644 index 0000000..d6072dd --- /dev/null +++ b/docs/backpressure.md @@ -0,0 +1,28 @@ +# Queue Back-Pressure Controls + +The asynchronous task executor backs every logger and can be tuned to handle +high-load bursts. Use the following helpers from +`` when preparing stress tests or +long-running services: + +- `LOGIT_SET_MAX_QUEUE(size)` sets the maximum number of queued tasks. Use a + small `size` to emulate a constrained environment or `0` to remove the + bound completely. +- `LOGIT_SET_QUEUE_POLICY(mode)` switches the overflow strategy. Available + options are `LOGIT_QUEUE_BLOCK`, `LOGIT_QUEUE_DROP_NEWEST`, and + `LOGIT_QUEUE_DROP_OLDEST`. +- `LOGIT_GET_DROPPED_TASKS()` returns the number of tasks that were discarded + under the current configuration. The value is safe to read concurrently with + producers. +- `LOGIT_RESET_DROPPED_TASKS()` clears the drop counter. Call it between + scenarios so each run measures its own losses. + +The drop counter is maintained inside the executor and is updated every time a +publishing policy decides to discard work. Combining the counter with +`TaskExecutor::wait()` makes it easy to assert the expected throughput for each +policy without inspecting private state. + +The queue limits apply globally to every logger instance. After finishing a +burst test remember to restore the capacity or shut down the logging subsystem +with `LOGIT_WAIT()` and `LOGIT_SHUTDOWN()` to avoid interfering with other +scenarios. diff --git a/include/logit_cpp/logit/detail/TaskExecutor.hpp b/include/logit_cpp/logit/detail/TaskExecutor.hpp index fd53cb4..fe02d47 100644 --- a/include/logit_cpp/logit/detail/TaskExecutor.hpp +++ b/include/logit_cpp/logit/detail/TaskExecutor.hpp @@ -8,49 +8,43 @@ #include #include #if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__) -#include -#include -#include + #include + #include + #include #else -#include -#include -#include -#include -#include + #include + #include + #include + #include + #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 + #ifdef LOGIT_USE_MPSC_RING + #include "MpscRingAny.hpp" + #endif #endif namespace logit { namespace detail { /// \brief Queue overflow handling policy. enum class QueuePolicy { DropNewest, DropOldest, Block }; - -#if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__) - + +# if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__) + /// \class TaskExecutor /// \brief Simplified task executor for single-threaded Emscripten builds. /// \thread_safety Not thread-safe. class TaskExecutor { public: - /// \brief Obtain singleton instance. - /// \return Global executor. static TaskExecutor& get_instance() { static TaskExecutor instance; return instance; } - - /// \brief Enqueue task for later execution. - /// \param task Callable to execute. + void add_task(std::function task) { if (!task) return; bool schedule = false; @@ -93,27 +87,27 @@ namespace logit { namespace detail { emscripten_async_call(&TaskExecutor::drain_thunk, this, 0); } } - - /// \brief Run all queued tasks. + void wait() { drain(); } - - /// \brief Drain queue without scheduling new tasks. void shutdown() { drain(); } - - /// \brief Set maximum queue size. - /// \param size Number of tasks allowed (0 for unlimited). + void set_max_queue_size(std::size_t size) { std::lock_guard lk(m_mutex); m_max_queue_size = size; } - - /// \brief Set overflow handling policy. - /// \param policy Policy to apply. + void set_queue_policy(QueuePolicy policy) { std::lock_guard lk(m_mutex); m_overflow_policy = policy; } - + + std::size_t dropped_tasks() const noexcept { + return m_dropped_tasks.load(std::memory_order_relaxed); + } + void reset_dropped_tasks() noexcept { + m_dropped_tasks.store(0, std::memory_order_relaxed); + } + private: TaskExecutor() : m_max_queue_size(0), @@ -125,18 +119,18 @@ namespace logit { namespace detail { TaskExecutor& operator=(const TaskExecutor&) = delete; TaskExecutor(TaskExecutor&&) = delete; TaskExecutor& operator=(TaskExecutor&&) = delete; - + std::deque> m_tasks; std::mutex m_mutex; std::size_t m_max_queue_size; QueuePolicy m_overflow_policy; std::atomic m_dropped_tasks; bool m_scheduled; - + static void drain_thunk(void* arg) { static_cast(arg)->drain(); } - + void drain() { for (;;) { std::function task; @@ -153,80 +147,26 @@ namespace logit { namespace detail { } } }; - -#else - + +# else // !Emscripten or pthreads + /// \class TaskExecutor /// \brief A thread-safe task executor that processes tasks in a dedicated worker thread. /// \thread_safety Thread-safe. class TaskExecutor { public: - /// \brief Get the singleton instance of the TaskExecutor. - /// \return A reference to the single instance of `TaskExecutor`. + /// Singleton (сохраняем вашу реализацию с new). static TaskExecutor& get_instance() { static TaskExecutor* instance = new TaskExecutor(); return *instance; } - - /// \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 +# ifndef LOGIT_USE_MPSC_RING std::unique_lock lock(m_queue_mutex); - if (m_stop_flag.load(std::memory_order_relaxed)) return; + if (m_stop_flag.load(std::memory_order_acquire)) return; if (m_max_queue_size > 0 && m_tasks_queue.size() >= m_max_queue_size) { switch (m_overflow_policy.load(std::memory_order_relaxed)) { case QueuePolicy::DropNewest: @@ -241,132 +181,213 @@ namespace logit { namespace detail { case QueuePolicy::Block: m_queue_condition.wait(lock, [this]() { return m_tasks_queue.size() < m_max_queue_size || - m_stop_flag.load(std::memory_order_relaxed); + m_stop_flag.load(std::memory_order_acquire); }); - if (m_stop_flag.load(std::memory_order_relaxed)) return; + if (m_stop_flag.load(std::memory_order_acquire)) return; break; } } m_tasks_queue.push_back(std::move(task)); lock.unlock(); m_queue_condition.notify_one(); -#endif - } +# else + // Барьер ресайза: пережидаем «горячее» изменение кольца. + 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); }); + } - /// \brief Waits for all tasks in the queue to be processed. + if (m_stop_flag.load(std::memory_order_acquire)) { + return; + } + + std::function local_task = std::move(task); + + for (;;) { + if (m_stop_flag.load(std::memory_order_acquire)) { + return; + } + + const auto policy = m_overflow_policy.load(std::memory_order_relaxed); + + // Реальное backpressure: учитываем "висящие" задачи. + if (policy == QueuePolicy::Block && + m_max_queue_size > 0 && + m_active_tasks.load(std::memory_order_relaxed) >= m_max_queue_size) + { + std::unique_lock lk(m_cv_mutex); + m_cv.wait_for(lk, std::chrono::microseconds(200)); + continue; + } + + // Пытаемся положить в кольцо. + if (m_mpsc_queue.try_push(local_task)) { + m_cv.notify_one(); // разбудить воркера + return; + } + + // Переполнение — применяем политику. + switch (policy) { + case QueuePolicy::DropNewest: + m_dropped_tasks.fetch_add(1, std::memory_order_relaxed); + return; + + case QueuePolicy::DropOldest: + // Безопасная реализация под MPSC: дропаем входящий. + // Это сохраняет порядок и исключает дедлоки при gate. + m_dropped_tasks.fetch_add(1, std::memory_order_relaxed); + return; + + case QueuePolicy::Block: { + std::unique_lock lk(m_cv_mutex); + m_cv.wait_for(lk, std::chrono::microseconds(200)); + break; + } + } + } +# endif + } + + /// Дождаться опустошения. void wait() { -#ifdef LOGIT_USE_MPSC_RING +# ifndef LOGIT_USE_MPSC_RING std::unique_lock lock(m_queue_mutex); m_queue_condition.wait(lock, [this]() { - return ((queue_empty_() && + return ((m_tasks_queue.empty() && m_active_tasks.load(std::memory_order_relaxed) == 0) || - m_stop_flag.load(std::memory_order_relaxed)); + m_stop_flag.load(std::memory_order_acquire)); }); -#else +# else std::unique_lock lock(m_queue_mutex); m_queue_condition.wait(lock, [this]() { - return ((m_tasks_queue.empty() && + return ((queue_empty_() && m_active_tasks.load(std::memory_order_relaxed) == 0) || - m_stop_flag.load(std::memory_order_relaxed)); + m_stop_flag.load(std::memory_order_acquire)); }); -#endif +# 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 +# ifndef LOGIT_USE_MPSC_RING + std::unique_lock lock(m_queue_mutex); + m_stop_flag.store(true, std::memory_order_release); + lock.unlock(); + m_queue_condition.notify_all(); + if (m_worker_thread.joinable()) { + m_worker_thread.join(); + } +# else { std::lock_guard lock(m_queue_mutex); - m_stop_flag.store(true, std::memory_order_relaxed); + m_stop_flag.store(true, std::memory_order_release); } 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); +# endif + } + + /// Изменить ёмкость очереди. + void set_max_queue_size(std::size_t size) { +# ifdef LOGIT_USE_MPSC_RING + // Сигналим продюсерам, чтобы переждали ресайз (до любых ожиданий/стопов). + m_resizing.store(true, std::memory_order_release); + + // Дождаться опустошения очереди + wait(); + + // Акуратно остановить воркер и дождаться его завершения, чтобы он не трогал m_mpsc_queue, пока мы его меняем. + std::unique_lock lk(m_queue_mutex); m_stop_flag.store(true, std::memory_order_relaxed); - lock.unlock(); + lk.unlock(); + + m_cv.notify_all(); m_queue_condition.notify_all(); if (m_worker_thread.joinable()) { m_worker_thread.join(); } -#endif - } + + // Переинициализировать параметры и само кольцо в единственном потоке. + 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(); + + // Снимаем стоп-флаг, перезапускаем воркер… + m_stop_flag.store(false, std::memory_order_relaxed); + m_worker_thread = std::thread(&TaskExecutor::worker_function, this); - /// \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 + // Открываем барьер для продюсеров. + m_resizing.store(false, std::memory_order_release); + m_resize_cv.notify_all(); +# else std::lock_guard lock(m_queue_mutex); m_max_queue_size = size; -#endif +# endif } - - /// \brief Sets the behavior when the queue is full. - /// \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.store(policy, std::memory_order_relaxed); } - + + std::size_t dropped_tasks() const noexcept { + return m_dropped_tasks.load(std::memory_order_relaxed); + } + void reset_dropped_tasks() noexcept { + m_dropped_tasks.store(0, 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. - 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). - 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 = LOGIT_TASK_EXECUTOR_DEFAULT_RING_CAPACITY; ///< 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. + #ifndef LOGIT_USE_MPSC_RING + std::deque> m_tasks_queue; + mutable std::mutex m_queue_mutex; + std::condition_variable m_queue_condition; + std::thread m_worker_thread; + std::atomic m_stop_flag; + std::size_t m_max_queue_size; + std::atomic m_overflow_policy; + 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; ///< Сон продюсеров/воркера. + + std::atomic m_resizing; ///< true — идёт ресайз кольца. + std::condition_variable m_resize_cv; ///< Продюсеры ждут окончания ресайза. + + std::thread m_worker_thread; + std::atomic m_stop_flag; + std::size_t m_max_queue_size; + std::atomic m_overflow_policy; + std::atomic m_dropped_tasks; + std::atomic m_active_tasks; + + const std::size_t m_default_ring_cap = LOGIT_TASK_EXECUTOR_DEFAULT_RING_CAPACITY; + MpscRingAny> m_mpsc_queue; + #endif + void worker_function() { -#ifndef LOGIT_USE_MPSC_RING + #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.load(std::memory_order_relaxed); + return !m_tasks_queue.empty() || m_stop_flag.load(std::memory_order_acquire); }); - if (m_stop_flag.load(std::memory_order_relaxed) && m_tasks_queue.empty()) { + if (m_stop_flag.load(std::memory_order_acquire) && m_tasks_queue.empty()) { break; } task = std::move(m_tasks_queue.front()); @@ -374,7 +395,9 @@ namespace logit { namespace detail { m_active_tasks.fetch_add(1, std::memory_order_relaxed); lock.unlock(); m_queue_condition.notify_one(); + task(); + lock.lock(); m_active_tasks.fetch_sub(1, std::memory_order_relaxed); if (m_tasks_queue.empty() && m_active_tasks.load(std::memory_order_relaxed) == 0) { @@ -382,109 +405,82 @@ namespace logit { namespace detail { } lock.unlock(); } -#else + #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); + m_cv.notify_one(); // освободили in-flight слот } - -#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)) { + m_queue_condition.notify_all(); // для wait() + m_cv.notify_all(); // разбудить продюсеров Block + if (m_stop_flag.load(std::memory_order_acquire)) { break; } } - + if (!drained_any) { std::unique_lock lk(m_cv_mutex); - if (m_stop_flag.load(std::memory_order_relaxed) && queue_empty_()) { + if (m_stop_flag.load(std::memory_order_acquire) && queue_empty_()) { break; } m_cv.wait_for(lk, std::chrono::milliseconds(1)); } } -#endif + #endif } - -#ifdef LOGIT_USE_MPSC_RING - /// \brief Return true if ring appears empty for current consumer position. + + #ifdef LOGIT_USE_MPSC_RING 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. + #endif + TaskExecutor() -#ifndef LOGIT_USE_MPSC_RING + #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), + #else + : m_resizing(false), + m_worker_thread(), + 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 + #endif { m_worker_thread = std::thread(&TaskExecutor::worker_function, this); } - - /// \brief Destructor that stops the worker thread and cleans up resources. + ~TaskExecutor() { shutdown(); } - - // Delete copy constructor and assignment operators to enforce singleton usage. + TaskExecutor(const TaskExecutor&) = delete; TaskExecutor& operator=(const TaskExecutor&) = delete; TaskExecutor(TaskExecutor&&) = delete; TaskExecutor& operator=(TaskExecutor&&) = delete; }; -#endif // defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__) +#endif // Emscripten split }} // namespace logit::detail #endif // _LOGIT_DETAIL_TASK_EXECUTOR_HPP_INCLUDED + diff --git a/include/logit_cpp/logit/log_macros.hpp b/include/logit_cpp/logit/log_macros.hpp index 9e4e1b2..6d0df35 100644 --- a/include/logit_cpp/logit/log_macros.hpp +++ b/include/logit_cpp/logit/log_macros.hpp @@ -2190,6 +2190,14 @@ static_assert(LOGIT_LEVEL_FATAL == static_cast(logit::LogLevel::LOG_LVL_FAT #define LOGIT_SET_QUEUE_POLICY(mode) \ logit::detail::TaskExecutor::get_instance().set_queue_policy(mode) +/// \brief Returns the number of tasks dropped due to overflow. +#define LOGIT_GET_DROPPED_TASKS() \ + logit::detail::TaskExecutor::get_instance().dropped_tasks() + +/// \brief Resets the dropped-tasks counter to zero. +#define LOGIT_RESET_DROPPED_TASKS() \ + logit::detail::TaskExecutor::get_instance().reset_dropped_tasks() + /// \} /// \brief Macro for waiting for all asynchronous loggers to finish processing. diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 823b2f4..587c8be 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -5,7 +5,7 @@ if(LOGIT_EMSCRIPTEN) add_executable(ems_async_flush ems/async_flush.cpp) target_link_libraries(ems_async_flush PRIVATE log-it-cpp) else() - file(GLOB TEST_SOURCES *.cpp) + file(GLOB TEST_SOURCES CONFIGURE_DEPENDS *.cpp) if(NOT LOGIT_WITH_GZIP) list(REMOVE_ITEM TEST_SOURCES ${CMAKE_CURRENT_LIST_DIR}/file_logger_gzip_compression_test.cpp) list(REMOVE_ITEM TEST_SOURCES ${CMAKE_CURRENT_LIST_DIR}/file_logger_external_cmd_compression_test.cpp) @@ -21,5 +21,16 @@ else() add_executable(${test_name} ${test_src}) target_link_libraries(${test_name} PRIVATE log-it-cpp) add_test(NAME ${test_name} COMMAND ${test_name}) + if(test_name STREQUAL "backpressure_policy_test" OR + test_name STREQUAL "backpressure_ordering_test") + set_tests_properties(${test_name} PROPERTIES LABELS "tsan") + endif() + if(test_name STREQUAL "backpressure_ordering_spsc_test") + if(MSVC) + target_compile_options(${test_name} PRIVATE /ULOGIT_USE_MPSC_RING) + else() + target_compile_options(${test_name} PRIVATE -ULOGIT_USE_MPSC_RING) + endif() + endif() endforeach() endif() diff --git a/tests/backpressure_ordering_spsc_test.cpp b/tests/backpressure_ordering_spsc_test.cpp new file mode 100644 index 0000000..baf65cd --- /dev/null +++ b/tests/backpressure_ordering_spsc_test.cpp @@ -0,0 +1,49 @@ +#include + +#include +#include + +namespace { + +constexpr std::size_t kMessages = 128; +constexpr std::size_t kQueueCapacity = 64; + +} // namespace + +int main() { + auto &executor = logit::detail::TaskExecutor::get_instance(); + executor.wait(); + + LOGIT_SET_QUEUE_POLICY(logit::detail::QueuePolicy::Block); + LOGIT_SET_MAX_QUEUE(kQueueCapacity); + LOGIT_RESET_DROPPED_TASKS(); + + std::vector order; + order.reserve(kMessages); + + for (std::size_t i = 0; i < kMessages; ++i) { + executor.add_task([i, &order]() { + order.push_back(i); + }); + } + + executor.wait(); + + if (order.size() != kMessages) { + return 1; + } + + for (std::size_t index = 0; index < order.size(); ++index) { + if (order[index] != index) { + return 2; + } + } + + if (LOGIT_GET_DROPPED_TASKS() != 0) { + return 3; + } + + LOGIT_RESET_DROPPED_TASKS(); + return 0; +} + diff --git a/tests/backpressure_ordering_test.cpp b/tests/backpressure_ordering_test.cpp new file mode 100644 index 0000000..824b45a --- /dev/null +++ b/tests/backpressure_ordering_test.cpp @@ -0,0 +1,81 @@ +#include + +#include +#include +#include +#include +#include +#include + +namespace { + +constexpr std::size_t kProducers = 4; +constexpr std::size_t kMessagesPerProducer = 64; +constexpr std::size_t kQueueCapacity = 32; + +} // namespace + +int main() { + auto &executor = logit::detail::TaskExecutor::get_instance(); + executor.wait(); + + LOGIT_SET_QUEUE_POLICY(logit::detail::QueuePolicy::Block); + LOGIT_SET_MAX_QUEUE(kQueueCapacity); + LOGIT_RESET_DROPPED_TASKS(); + + std::array, kProducers> sequences; + for (auto &sequence : sequences) { + sequence.clear(); + sequence.reserve(kMessagesPerProducer); + } + + std::array sequence_guards{}; + std::atomic start{false}; + + std::vector producers; + producers.reserve(kProducers); + + for (std::size_t producer_id = 0; producer_id < kProducers; ++producer_id) { + producers.emplace_back([producer_id, &executor, &start, &sequence_guards, &sequences]() { + while (!start.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + + for (std::size_t seq = 0; seq < kMessagesPerProducer; ++seq) { + executor.add_task([producer_id, seq, &sequence_guards, &sequences]() { + std::lock_guard lock(sequence_guards[producer_id]); + sequences[producer_id].push_back(seq); + }); + } + }); + } + + start.store(true, std::memory_order_release); + + for (auto &producer : producers) { + producer.join(); + } + + executor.wait(); + + if (LOGIT_GET_DROPPED_TASKS() != 0) { + return 1; + } + + for (std::size_t producer_id = 0; producer_id < kProducers; ++producer_id) { + const auto &sequence = sequences[producer_id]; + if (sequence.size() != kMessagesPerProducer) { + return 2; + } + + for (std::size_t expected = 0; expected < sequence.size(); ++expected) { + if (sequence[expected] != expected) { + return 3; + } + } + } + + LOGIT_RESET_DROPPED_TASKS(); + return 0; +} + diff --git a/tests/backpressure_policy_test.cpp b/tests/backpressure_policy_test.cpp new file mode 100644 index 0000000..f93ddd3 --- /dev/null +++ b/tests/backpressure_policy_test.cpp @@ -0,0 +1,289 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +constexpr std::size_t kSingleProducerBurst = 32; +constexpr std::size_t kSingleProducerQueueCapacity = 4; +constexpr std::size_t kMultiProducerThreads = 4; +constexpr std::size_t kMessagesPerProducer = 32; +constexpr std::size_t kMultiProducerQueueCapacity = 16; +constexpr auto kSlowTaskDelay = std::chrono::milliseconds{5}; + +struct ScenarioResult { + std::chrono::steady_clock::duration publish_duration{}; + std::chrono::steady_clock::duration enforced_gate_delay{}; + std::size_t processed{}; + std::size_t dropped{}; +}; + +ScenarioResult run_single_producer_scenario( + logit::detail::QueuePolicy policy, + bool hold_consumer, + std::chrono::steady_clock::duration gate_delay = + std::chrono::steady_clock::duration::zero()) { + auto &executor = logit::detail::TaskExecutor::get_instance(); + LOGIT_SET_QUEUE_POLICY(policy); + LOGIT_RESET_DROPPED_TASKS(); + + std::atomic processed{0}; + std::condition_variable gate_cv; + std::mutex gate_mutex; + const bool gating_requested = hold_consumer || + gate_delay > std::chrono::steady_clock::duration::zero(); + bool gate_open = !gating_requested; + + const auto start = std::chrono::steady_clock::now(); + std::thread publisher([&executor, + &processed, + gating_requested, + &gate_cv, + &gate_mutex, + &gate_open]() { + for (std::size_t i = 0; i < kSingleProducerBurst; ++i) { + executor.add_task([&processed, gating_requested, &gate_cv, &gate_mutex, &gate_open]() { + if (gating_requested) { + std::unique_lock lock(gate_mutex); + gate_cv.wait(lock, [&gate_open]() { return gate_open; }); + } + std::this_thread::sleep_for(kSlowTaskDelay); + processed.fetch_add(1, std::memory_order_relaxed); + }); + } + }); + + if (gating_requested) { + if (gate_delay > std::chrono::steady_clock::duration::zero()) { + std::this_thread::sleep_for(gate_delay); + } + { + std::lock_guard lock(gate_mutex); + gate_open = true; + } + gate_cv.notify_all(); + } + + publisher.join(); + const auto publish_duration = std::chrono::steady_clock::now() - start; + + executor.wait(); + + ScenarioResult result{}; + result.publish_duration = publish_duration; + if (gate_delay > std::chrono::steady_clock::duration::zero()) { + result.enforced_gate_delay = gate_delay; + } + result.processed = processed.load(std::memory_order_relaxed); + result.dropped = LOGIT_GET_DROPPED_TASKS(); + return result; +} + +struct MultiProducerResult { + std::size_t processed{}; + std::size_t dropped{}; +}; + +MultiProducerResult run_multi_producer_scenario( + logit::detail::QueuePolicy policy, + bool hold_consumer, + std::chrono::steady_clock::duration gate_delay = + std::chrono::steady_clock::duration::zero()) { + auto &executor = logit::detail::TaskExecutor::get_instance(); + LOGIT_SET_QUEUE_POLICY(policy); + LOGIT_RESET_DROPPED_TASKS(); + + std::atomic processed{0}; + std::atomic start_flag{false}; + std::condition_variable gate_cv; + std::mutex gate_mutex; + const bool gating_requested = hold_consumer || + gate_delay > std::chrono::steady_clock::duration::zero(); + bool gate_open = !gating_requested; + + std::vector producers; + producers.reserve(kMultiProducerThreads); + + for (std::size_t i = 0; i < kMultiProducerThreads; ++i) { + producers.emplace_back([&executor, + &processed, + &start_flag, + gating_requested, + &gate_cv, + &gate_mutex, + &gate_open]() { + while (!start_flag.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + for (std::size_t j = 0; j < kMessagesPerProducer; ++j) { + executor.add_task([&processed, gating_requested, &gate_cv, &gate_mutex, &gate_open]() { + if (gating_requested) { + std::unique_lock lock(gate_mutex); + gate_cv.wait(lock, [&gate_open]() { return gate_open; }); + } + std::this_thread::sleep_for(kSlowTaskDelay); + processed.fetch_add(1, std::memory_order_relaxed); + }); + } + }); + } + + start_flag.store(true, std::memory_order_release); + + for (auto &producer : producers) { + producer.join(); + } + + if (gating_requested) { + if (gate_delay > std::chrono::steady_clock::duration::zero()) { + std::this_thread::sleep_for(gate_delay); + } + { + std::lock_guard lock(gate_mutex); + gate_open = true; + } + gate_cv.notify_all(); + } + + executor.wait(); + + MultiProducerResult result{}; + result.processed = processed.load(std::memory_order_relaxed); + result.dropped = LOGIT_GET_DROPPED_TASKS(); + return result; +} + +} // namespace + +int main() { + auto &executor = logit::detail::TaskExecutor::get_instance(); + executor.wait(); + + LOGIT_SET_MAX_QUEUE(kSingleProducerQueueCapacity); + + const auto deterministic_gate_delay = + kSlowTaskDelay * (kSingleProducerBurst / 2); + const auto block_result = + run_single_producer_scenario(logit::detail::QueuePolicy::Block, + true, + deterministic_gate_delay); + if (block_result.processed != kSingleProducerBurst) { + return 1; + } + if (block_result.dropped != 0) { + return 2; + } + + const auto drop_newest_result = + run_single_producer_scenario(logit::detail::QueuePolicy::DropNewest, + true, + deterministic_gate_delay); + if (drop_newest_result.dropped == 0) { + return 3; + } + if (drop_newest_result.processed + drop_newest_result.dropped != kSingleProducerBurst) { + return 4; + } + const auto single_min_survivors = (std::min)(kSingleProducerQueueCapacity, kSingleProducerBurst); + const auto single_max_survivors = (std::min)(kSingleProducerQueueCapacity + 1, kSingleProducerBurst); + if (drop_newest_result.processed < single_min_survivors) { + return 5; + } + if (drop_newest_result.processed > single_max_survivors) { + return 6; + } + + const auto drop_oldest_result = + run_single_producer_scenario(logit::detail::QueuePolicy::DropOldest, + true, + deterministic_gate_delay); + if (drop_oldest_result.dropped == 0) { + return 7; + } + if (drop_oldest_result.processed + drop_oldest_result.dropped != kSingleProducerBurst) { + return 8; + } + const auto expected_single_oldest_drops = + kSingleProducerBurst - drop_oldest_result.processed; + if (drop_oldest_result.dropped != expected_single_oldest_drops) { + return 9; + } + if (drop_oldest_result.processed < single_min_survivors) { + return 10; + } + if (drop_oldest_result.processed > single_max_survivors) { + return 11; + } + + const auto gate_tolerance = kSlowTaskDelay * 2; + if ((block_result.publish_duration + gate_tolerance) < deterministic_gate_delay) { + return 12; + } + + LOGIT_RESET_DROPPED_TASKS(); + + LOGIT_SET_MAX_QUEUE(kMultiProducerQueueCapacity); + + const auto total_messages = kMultiProducerThreads * kMessagesPerProducer; + + const auto block_multi_result = run_multi_producer_scenario(logit::detail::QueuePolicy::Block, + false); + if (block_multi_result.processed != total_messages) { + return 13; + } + if (block_multi_result.dropped != 0) { + return 14; + } + + const auto drop_newest_multi = run_multi_producer_scenario( + logit::detail::QueuePolicy::DropNewest, + true, + deterministic_gate_delay); + if (drop_newest_multi.dropped == 0) { + return 15; + } + if (drop_newest_multi.processed + drop_newest_multi.dropped != total_messages) { + return 16; + } + const auto multi_min_survivors = (std::min)(kMultiProducerQueueCapacity, total_messages); + const auto multi_max_survivors = (std::min)(kMultiProducerQueueCapacity + 1, total_messages); + if (drop_newest_multi.processed < multi_min_survivors) { + return 17; + } + if (drop_newest_multi.processed > multi_max_survivors) { + return 18; + } + + const auto drop_oldest_multi = run_multi_producer_scenario( + logit::detail::QueuePolicy::DropOldest, + true, + deterministic_gate_delay); + if (drop_oldest_multi.dropped == 0) { + return 19; + } + if (drop_oldest_multi.processed + drop_oldest_multi.dropped != total_messages) { + return 20; + } + const auto expected_multi_oldest_drops = + total_messages - drop_oldest_multi.processed; + if (drop_oldest_multi.dropped != expected_multi_oldest_drops) { + return 21; + } + if (drop_oldest_multi.processed < multi_min_survivors) { + return 22; + } + if (drop_oldest_multi.processed > multi_max_survivors) { + return 23; + } + + LOGIT_RESET_DROPPED_TASKS(); + return 0; +} +