diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7758c16..8ad59c8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,7 +30,7 @@ jobs: run: cmake -S . -B build-bench -DLOGIT_BENCH_ENABLE=ON -DLOGIT_BENCH_WITH_SPDLOG=ON -DCMAKE_BUILD_TYPE=Release -DCMAKE_CXX_STANDARD=${{ matrix.std }} -DLOGIT_WITH_SYSLOG=ON -DLOGIT_WITH_WIN_EVENT_LOG=OFF - name: Build benchmarks # if: ${{ github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/stable') }} - run: cmake --build build-bench --target logit_bench logit_bench_flush_test logit_public_macro_bench logit_public_macro_formatted_bench logit_hotpath_bench logit_hotpath_bench_legacy benchmark_validation_test + run: cmake --build build-bench --target logit_bench logit_bench_flush_test logit_public_macro_bench logit_public_macro_formatted_bench logit_hotpath_bench logit_hotpath_bench_legacy logit_exec_mx_bench logit_exec_mx_bench_concurrent benchmark_validation_test - name: Run spdlog async flush regression run: ./build-bench/logit_bench_flush_test - name: Run public macro benchmark smoke @@ -53,6 +53,12 @@ jobs: run: | ./build-bench/logit_hotpath_bench ./build-bench/logit_hotpath_bench_legacy + - name: Run exec_mx capability A/B smoke + env: + LOGIT_EXEC_MX_BENCH_TOTAL: 2000 + run: | + ./build-bench/logit_exec_mx_bench + ./build-bench/logit_exec_mx_bench_concurrent - name: Run latency benchmarks # if: ${{ github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/stable') }} timeout-minutes: 20 diff --git a/bench/CMakeLists.txt b/bench/CMakeLists.txt index a072878..8863655 100644 --- a/bench/CMakeLists.txt +++ b/bench/CMakeLists.txt @@ -56,6 +56,17 @@ target_compile_definitions(logit_hotpath_bench_legacy PRIVATE LOGIT_BENCH_LEGACY target_link_libraries(logit_hotpath_bench_legacy PRIVATE log-it-cpp::log-it-cpp) set_target_properties(logit_hotpath_bench_legacy PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}) +add_executable(logit_exec_mx_bench logger_exec_mx_bench.cpp) +target_compile_features(logit_exec_mx_bench PRIVATE cxx_std_17) +target_link_libraries(logit_exec_mx_bench PRIVATE log-it-cpp::log-it-cpp) +set_target_properties(logit_exec_mx_bench PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}) + +add_executable(logit_exec_mx_bench_concurrent logger_exec_mx_bench.cpp) +target_compile_features(logit_exec_mx_bench_concurrent PRIVATE cxx_std_17) +target_compile_definitions(logit_exec_mx_bench_concurrent PRIVATE LOGIT_BENCH_CONCURRENT_DISPATCH=1) +target_link_libraries(logit_exec_mx_bench_concurrent PRIVATE log-it-cpp::log-it-cpp) +set_target_properties(logit_exec_mx_bench_concurrent PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}) + add_executable(benchmark_validation_test benchmark_validation_test.cpp) target_compile_features(benchmark_validation_test PRIVATE cxx_std_17) set_target_properties(benchmark_validation_test PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}) diff --git a/bench/logger_exec_mx_bench.cpp b/bench/logger_exec_mx_bench.cpp new file mode 100644 index 0000000..1883476 --- /dev/null +++ b/bench/logger_exec_mx_bench.cpp @@ -0,0 +1,141 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace { + +class CountingLogger final : public logit::ILogger { +public: + void log(const logit::LogRecord&, const std::string&) override { + m_count.fetch_add(1, std::memory_order_relaxed); + } + +#ifdef LOGIT_BENCH_CONCURRENT_DISPATCH + bool supports_concurrent_log() const noexcept override { return true; } +#endif + + std::string get_string_param(const logit::LoggerParam&) const override { return {}; } + std::int64_t get_int_param(const logit::LoggerParam&) const override { return 0; } + double get_float_param(const logit::LoggerParam&) const override { return 0.0; } + void set_log_level(logit::LogLevel level) override { + m_level.store(static_cast(level), std::memory_order_relaxed); + } + logit::LogLevel get_log_level() const override { + return static_cast(m_level.load(std::memory_order_relaxed)); + } + void wait() override {} + + std::size_t count() const { return m_count.load(std::memory_order_relaxed); } + +private: + std::atomic m_count{0}; + std::atomic m_level{static_cast(logit::LogLevel::LOG_LVL_TRACE)}; +}; + +class PassthroughFormatter final : public logit::ILogFormatter { +public: + void set_timestamp_offset(std::int64_t) override {} + std::string format(const logit::LogRecord& record) const override { return record.format; } + +#ifdef LOGIT_BENCH_CONCURRENT_DISPATCH + bool supports_concurrent_format() const noexcept override { return true; } +#endif +}; + +std::size_t env_size(const char* name, std::size_t fallback) { + if (const char* value = std::getenv(name)) { + try { + return static_cast(std::stoull(value)); + } catch (...) { + } + } + return fallback; +} + +long long run_workload( + logit::Logger& logger, + const logit::LogRecord& record, + std::size_t producers, + std::size_t total) { + std::vector per_thread(producers, total / producers); + for (std::size_t i = 0; i < total % producers; ++i) ++per_thread[i]; + + std::mutex mutex; + std::condition_variable condition; + std::size_t ready = 0; + bool start = false; + std::vector threads; + threads.reserve(producers); + + for (std::size_t i = 0; i < producers; ++i) { + threads.emplace_back([&, i]() { + { + std::unique_lock lock(mutex); + ++ready; + condition.notify_all(); + condition.wait(lock, [&]() { return start; }); + } + for (std::size_t n = 0; n < per_thread[i]; ++n) logger.log(record); + }); + } + + std::chrono::steady_clock::time_point begin; + { + std::unique_lock lock(mutex); + condition.wait(lock, [&]() { return ready == producers; }); + begin = std::chrono::steady_clock::now(); + start = true; + condition.notify_all(); + } + + for (auto& thread : threads) thread.join(); + logger.wait(); + return std::chrono::duration_cast( + std::chrono::steady_clock::now() - begin).count(); +} + +} // namespace + +int main() { + const std::size_t total = env_size("LOGIT_EXEC_MX_BENCH_TOTAL", 200000); + auto sink = std::make_unique(); + auto* sink_ptr = sink.get(); + auto& logger = logit::Logger::get_instance(); + logger.add_logger(std::move(sink), std::make_unique()); + + const logit::LogRecord record( + logit::LogLevel::LOG_LVL_INFO, 0, std::string(), -1, + std::string(), std::string("prepared message"), std::string(), -1, false, false); + const std::array producer_counts{{1, 4, 16, 32}}; + + std::cout << "exec-mx mode=" +#ifdef LOGIT_BENCH_CONCURRENT_DISPATCH + << "concurrent"; +#else + << "serialized"; +#endif + std::cout << " total=" << total << '\n'; + + std::size_t expected = 0; + for (std::size_t producers : producer_counts) { + const long long elapsed = run_workload(logger, record, producers, total); + expected += total; + if (sink_ptr->count() != expected) return 1; + const double ns_per_call = static_cast(elapsed) / static_cast(total); + std::cout << "exec-mx producers=" << producers + << " elapsed_ns=" << elapsed + << " ns_per_call=" << ns_per_call << '\n'; + } + return 0; +} diff --git a/docs/adr/0006-concurrent-dispatch-capability.md b/docs/adr/0006-concurrent-dispatch-capability.md new file mode 100644 index 0000000..a26fd8f --- /dev/null +++ b/docs/adr/0006-concurrent-dispatch-capability.md @@ -0,0 +1,41 @@ +# ADR 0006: Explicit capability for concurrent dispatch + +- Status: Accepted +- Date: 2026-09-15 + +## Context + +`Logger::log()` historically serialized formatter and backend calls with one +execution mutex per strategy. This protects mutable formatter state, backend +state, and lifecycle operations, but it also makes independent producer calls +contend on that mutex. A benchmark result from one sink cannot establish a +thread-safety contract for every custom formatter or backend. + +## Decision + +`ILogFormatter::supports_concurrent_format()` and +`ILogger::supports_concurrent_log()` are explicit opt-in capability hooks. Both +must return `true` before `Logger` elides a strategy's `exec_mx` during dispatch. +The default is `false`, preserving serialized behavior for all existing and +custom implementations. + +An opting-in formatter must support concurrent `format()` calls and concurrent +configuration through `set_timestamp_offset()`. An opting-in backend must make +`log()`, `get_log_level()`, `set_log_level()`, `clear_logs()`, `wait()`, and +`shutdown()` safe when called concurrently, and must keep its resources alive +until those calls finish. The backend remains responsible for its own output +ordering and serialization where required. + +The capability is stored when a logger strategy is registered, so the normal +dispatch path does not perform virtual capability checks. The execution mutex +remains in the fallback path and is still used by lifecycle/configuration APIs. +Built-in backends do not opt in until each backend's complete lifecycle and +formatter contract has been audited. + +## Consequences + +Safe custom pairs can avoid the per-strategy execution mutex under producer +contention. Existing behavior and safety guarantees are unchanged by default. +The benchmark and regression test compare serialized and explicitly opted-in +pairs across 1, 4, 16, and 32 producers; their measurements are machine +specific and are not evidence that arbitrary backends may opt in. diff --git a/docs/adr/README.md b/docs/adr/README.md index 71cb586..9e9ac38 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -28,3 +28,4 @@ explains **why the current boundary or trade-off exists**. - [0003 — Immutable logger registry snapshots](0003-immutable-registry-snapshots.md) - [0004 — TimeShield compatibility and dependency reuse](0004-timeshield-compatibility.md) - [0005 — Benchmark evidence and comparison methodology](0005-benchmark-methodology.md) +- [0006 — Explicit capability for concurrent dispatch](0006-concurrent-dispatch-capability.md) diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 79cc468..d2a7224 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -86,6 +86,16 @@ The prepared-message/direct-dispatch pipeline and a true public macro benchmark calls `LOGIT_INFO(...)` are separate scenarios with different work contracts; their results must not be presented as one number. +`logit_exec_mx_bench` and `logit_exec_mx_bench_concurrent` are a guarded +lock-elision experiment. They use the same prepared `LogRecord` and a small +thread-safe counting backend/formatter pair; the first target keeps the default +serialized path, while the second explicitly opts into the concurrency +capabilities. Both report 1, 4, 16, and 32 producer runs. These binaries are +research tools, not a recommendation to opt in arbitrary backends. A backend +or formatter must satisfy the complete lifecycle contract in +[`ADR 0006`](adr/0006-concurrent-dispatch-capability.md) before returning the +capability flag. + `logit_public_macro_bench` and `logit_public_macro_formatted_bench` are focused public-API smoke benchmarks. Both invoke `LOGIT_INFO(...)` from multiple producer threads and therefore include argument-name parsing, `args_array` diff --git a/docs/future-plans.md b/docs/future-plans.md index 0cacffc..da6eac3 100644 --- a/docs/future-plans.md +++ b/docs/future-plans.md @@ -57,9 +57,10 @@ Legend: a versioned fixture contract with compiler/toolchain/commit/queue/flush metadata, and a 1/4/16/32-producer matrix are now covered. Keep publication numbers tied to a fixed machine and toolchain. -- [ ] **Concurrency fast-path research** — only after documenting a formal - thread-safety capability for formatters/backends. Do not remove `exec_mx` - based on benchmark results alone. +- [x] **Concurrency fast-path research** — explicit formatter/backend capability + hooks, serialized fallback, lifecycle regression coverage, and a 1/4/16/32 + producer A/B benchmark are documented in ADR 0006. Built-in backends remain + serialized until their complete contracts are audited. ## Intentionally deferred diff --git a/include/logit_cpp/logit/Logger.hpp b/include/logit_cpp/logit/Logger.hpp index 6d180a6..ee76670 100644 --- a/include/logit_cpp/logit/Logger.hpp +++ b/include/logit_cpp/logit/Logger.hpp @@ -71,6 +71,10 @@ namespace logit { strategy->formatter = std::move(formatter); strategy->single_mode = single_mode; strategy->enabled = true; + strategy->concurrent_dispatch = + strategy->logger && + strategy->logger->supports_concurrent_log() && + (!strategy->formatter || strategy->formatter->supports_concurrent_format()); LoggerWriteLock lock(m_loggers_mx); if (m_shutdown.load(std::memory_order_acquire)) return; @@ -200,26 +204,14 @@ namespace logit { const auto& strategy = (*strategies)[strategy_index]; if (!strategy) return; - std::lock_guard exec_lock(strategy->exec_mx); - if (m_shutdown.load(std::memory_order_acquire)) return; - if (!strategy->enabled.load(std::memory_order_relaxed)) return; - if (!record.raw_mode && - static_cast(record.log_level) < static_cast(strategy->logger->get_log_level())) return; - dispatch_to_strategy(*strategy, record); + dispatch_to_strategy_if_allowed(*strategy, record, true); return; } for (const auto& strategy : *strategies) { if (!strategy) continue; - std::lock_guard exec_lock(strategy->exec_mx); - if (m_shutdown.load(std::memory_order_acquire)) return; - if (strategy->single_mode.load(std::memory_order_relaxed)) continue; - if (!strategy->enabled.load(std::memory_order_relaxed)) continue; - if (!record.raw_mode && - static_cast(record.log_level) < static_cast(strategy->logger->get_log_level())) continue; - - dispatch_to_strategy(*strategy, record); + dispatch_to_strategy_if_allowed(*strategy, record, false); } } @@ -552,9 +544,35 @@ namespace logit { std::unique_ptr formatter; ///< The formatter instance. std::atomic single_mode{false}; ///< Flag indicating if the logger is in single mode. std::atomic enabled{true}; ///< Flag indicating if the logger is enabled. + bool concurrent_dispatch = false; ///< Explicit formatter/backend lock-elision capability. mutable std::mutex exec_mx; ///< Protects formatter+logger invocation. }; + void dispatch_to_strategy_if_allowed( + LoggerStrategy& strategy, + const LogRecord& record, + bool targeted) { + if (strategy.concurrent_dispatch) { + dispatch_to_strategy_if_allowed_unlocked(strategy, record, targeted); + return; + } + + std::lock_guard exec_lock(strategy.exec_mx); + dispatch_to_strategy_if_allowed_unlocked(strategy, record, targeted); + } + + void dispatch_to_strategy_if_allowed_unlocked( + LoggerStrategy& strategy, + const LogRecord& record, + bool targeted) { + if (m_shutdown.load(std::memory_order_acquire)) return; + if (!targeted && strategy.single_mode.load(std::memory_order_relaxed)) return; + if (!strategy.enabled.load(std::memory_order_relaxed)) return; + if (!record.raw_mode && + static_cast(record.log_level) < static_cast(strategy.logger->get_log_level())) return; + dispatch_to_strategy(strategy, record); + } + void dispatch_to_strategy(LoggerStrategy& strategy, const LogRecord& record) { if (record.raw_mode) { strategy.logger->log(record, record.format); diff --git a/include/logit_cpp/logit/formatter/ILogFormatter.hpp b/include/logit_cpp/logit/formatter/ILogFormatter.hpp index 54dae8e..98ceb29 100644 --- a/include/logit_cpp/logit/formatter/ILogFormatter.hpp +++ b/include/logit_cpp/logit/formatter/ILogFormatter.hpp @@ -42,6 +42,16 @@ namespace logit { /// this to enable fast-path handling without extra string copies. /// \return True if the formatter is passthrough, false otherwise. virtual bool is_passthrough() const noexcept { return false; } + + /// \brief Indicates whether formatting may run concurrently with dispatch configuration. + /// + /// Returning true opts this formatter into the `Logger` lock-elision + /// path. Implementations must be safe for concurrent `format()` calls + /// and for concurrent calls to their configuration methods, including + /// `set_timestamp_offset()`. The default keeps the existing serialized + /// behavior for custom formatters. + /// \return True only when concurrent formatting and reconfiguration are safe. + virtual bool supports_concurrent_format() const noexcept { return false; } }; // ILogFormatter }; // namespace logit diff --git a/include/logit_cpp/logit/loggers/ILogger.hpp b/include/logit_cpp/logit/loggers/ILogger.hpp index 8cc9b51..e1ef10e 100644 --- a/include/logit_cpp/logit/loggers/ILogger.hpp +++ b/include/logit_cpp/logit/loggers/ILogger.hpp @@ -57,6 +57,17 @@ namespace logit { /// \param message The formatted log message. virtual void log(const LogRecord& record, const std::string& message) = 0; + /// \brief Indicates whether backend dispatch may run concurrently. + /// + /// Returning true opts this backend into the `Logger` lock-elision + /// path. Implementations must make `log()`, `get_log_level()`, + /// `set_log_level()`, `clear_logs()`, `wait()`, and `shutdown()` safe + /// when called concurrently, and must keep all owned resources alive + /// until those operations complete. The default preserves serialized + /// dispatch for existing and custom backends. + /// \return True only when concurrent dispatch and lifecycle operations are safe. + virtual bool supports_concurrent_log() const noexcept { return false; } + /// \brief Retrieves a string parameter from the logger. /// Derived classes should implement this to return specific string-based parameters. /// \param param The parameter type to retrieve. diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 045abf3..7475f16 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -49,6 +49,7 @@ else() logger_legacy_targeted_path_test.cpp logger_snapshot_read_path_test.cpp logger_clear_api_test.cpp + logger_dispatch_capability_test.cpp memory_logger_backend_test.cpp memory_logger_callback_test.cpp memory_logger_concurrency_test.cpp @@ -161,4 +162,10 @@ else() endif() endif() endforeach() + + add_executable(logger_dispatch_concurrent_test logger_dispatch_capability_test.cpp) + target_compile_features(logger_dispatch_concurrent_test PRIVATE cxx_std_11) + target_compile_definitions(logger_dispatch_concurrent_test PRIVATE LOGIT_TEST_CONCURRENT_CAPABILITY=1) + target_link_libraries(logger_dispatch_concurrent_test PRIVATE log-it-cpp) + add_test(NAME logger_dispatch_concurrent_test COMMAND logger_dispatch_concurrent_test) endif() diff --git a/tests/logger_dispatch_capability_test.cpp b/tests/logger_dispatch_capability_test.cpp new file mode 100644 index 0000000..ebbf097 --- /dev/null +++ b/tests/logger_dispatch_capability_test.cpp @@ -0,0 +1,122 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace { + +class OverlapProbeLogger final : public logit::ILogger { +public: + void log(const logit::LogRecord&, const std::string&) override { + const int active = m_active.fetch_add(1, std::memory_order_acq_rel) + 1; + int observed = m_max_active.load(std::memory_order_relaxed); + while (active > observed && + !m_max_active.compare_exchange_weak( + observed, active, std::memory_order_relaxed, std::memory_order_relaxed)) { + } + // Keep the critical section open long enough for the opt-in test to + // observe overlap; the serialized fallback remains deterministic. + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + m_active.fetch_sub(1, std::memory_order_acq_rel); + m_calls.fetch_add(1, std::memory_order_relaxed); + } + +#ifdef LOGIT_TEST_CONCURRENT_CAPABILITY + bool supports_concurrent_log() const noexcept override { return true; } +#endif + + std::string get_string_param(const logit::LoggerParam&) const override { return {}; } + std::int64_t get_int_param(const logit::LoggerParam&) const override { return 0; } + double get_float_param(const logit::LoggerParam&) const override { return 0.0; } + void set_log_level(logit::LogLevel level) override { + m_level.store(static_cast(level), std::memory_order_relaxed); + } + logit::LogLevel get_log_level() const override { + return static_cast(m_level.load(std::memory_order_relaxed)); + } + void wait() override {} + + int max_active() const { return m_max_active.load(std::memory_order_relaxed); } + int calls() const { return m_calls.load(std::memory_order_relaxed); } + +private: + std::atomic m_active{0}; + std::atomic m_max_active{0}; + std::atomic m_calls{0}; + std::atomic m_level{static_cast(logit::LogLevel::LOG_LVL_TRACE)}; +}; + +class StatelessFormatter final : public logit::ILogFormatter { +public: + void set_timestamp_offset(std::int64_t) override {} + std::string format(const logit::LogRecord& record) const override { return record.format; } + +#ifdef LOGIT_TEST_CONCURRENT_CAPABILITY + bool supports_concurrent_format() const noexcept override { return true; } +#endif +}; + +void run_probe(OverlapProbeLogger& probe, std::size_t workers) { + std::mutex mutex; + std::condition_variable condition; + std::size_t ready = 0; + bool start = false; + std::vector threads; + threads.reserve(workers); + + const logit::LogRecord record( + logit::LogLevel::LOG_LVL_INFO, 0, std::string(), -1, + std::string(), std::string("probe"), std::string(), -1, false, false); + + for (std::size_t i = 0; i < workers; ++i) { + threads.emplace_back([&]() { + { + std::unique_lock lock(mutex); + ++ready; + condition.notify_all(); + condition.wait(lock, [&]() { return start; }); + } + logit::Logger::get_instance().log(record); + }); + } + + { + std::unique_lock lock(mutex); + condition.wait(lock, [&]() { return ready == workers; }); + start = true; + condition.notify_all(); + } + + for (auto& thread : threads) thread.join(); +} + +} // namespace + +int main() { + std::unique_ptr probe(new OverlapProbeLogger()); + auto* probe_ptr = probe.get(); + logit::Logger::get_instance().add_logger( + std::move(probe), std::unique_ptr(new StatelessFormatter())); + + constexpr std::size_t workers = 8; + run_probe(*probe_ptr, workers); + assert(probe_ptr->calls() == static_cast(workers)); + +#ifdef LOGIT_TEST_CONCURRENT_CAPABILITY + // The explicitly opted-in pair must be callable concurrently. + assert(probe_ptr->max_active() >= 2); +#else + // Existing/custom implementations retain the serialized fallback. + assert(probe_ptr->max_active() == 1); +#endif + + return 0; +}