perf: benchmark hot path and reduce logger dispatch overhead

.github/workflows/ci.yml

@@ -27,10 +27,12 @@ jobs:
         run: ctest --test-dir build --output-on-failure
       - name: Configure benchmarks
         # if: ${{ github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/stable') }}
-        run: cmake -S . -B build-bench -DLOGIT_BENCH_ENABLE=ON -DLOGIT_BENCH_WITH_SPDLOG=ON -DCMAKE_CXX_STANDARD=${{ matrix.std }} -DLOGIT_WITH_SYSLOG=ON -DLOGIT_WITH_WIN_EVENT_LOG=OFF
+        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
+        run: cmake --build build-bench --target logit_bench logit_bench_flush_test
+      - name: Run spdlog async flush regression
+        run: ./build-bench/logit_bench_flush_test
       - name: Run latency benchmarks
         # if: ${{ github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/stable') }}
         timeout-minutes: 20

bench/CMakeLists.txt

@@ -31,9 +31,27 @@ if(LOGIT_BENCH_WITH_SPDLOG)
         include(FetchContent)
         FetchContent_Declare(spdlog
             GIT_REPOSITORY https://github.com/gabime/spdlog.git
-            GIT_TAG v1.12.0
+            GIT_TAG v1.17.0
         )
         FetchContent_MakeAvailable(spdlog)
     endif()
     target_link_libraries(logit_bench PRIVATE spdlog::spdlog)
+
+    add_executable(logit_bench_flush_test
+        spdlog_flush_test.cpp
+        adapters/SpdlogAdapter.cpp
+    )
+    target_compile_features(logit_bench_flush_test PRIVATE cxx_std_17)
+    target_include_directories(logit_bench_flush_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
+    target_compile_definitions(logit_bench_flush_test PRIVATE LOGIT_BENCH_HAVE_SPDLOG=1)
+    target_link_libraries(logit_bench_flush_test PRIVATE spdlog::spdlog)
+    set_target_properties(logit_bench_flush_test PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}
+    )
+    foreach(config IN ITEMS DEBUG RELEASE RELWITHDEBINFO MINSIZEREL)
+        set_target_properties(logit_bench_flush_test PROPERTIES
+            RUNTIME_OUTPUT_DIRECTORY_${config} ${CMAKE_BINARY_DIR}
+        )
+    endforeach()
+    add_test(NAME logit_bench_flush_test COMMAND logit_bench_flush_test)
 endif()

bench/LatencyRecorder.hpp

@@ -1,5 +1,9 @@
 #pragma once
 
+#if defined(_WIN32) && !defined(NOMINMAX)
+#define NOMINMAX
+#endif
+
 #include <algorithm>
 #include <atomic>
 #include <chrono>
@@ -11,6 +15,10 @@
 #include <vector>
 #include <cmath>
 
+#ifdef max
+#undef max
+#endif
+
 namespace logit_bench {
 
     /**

bench/Scenario.hpp

@@ -24,6 +24,7 @@ struct Scenario {
     std::size_t producers      = 1;
     std::size_t message_bytes  = 0;
     std::size_t total_messages = 0;
+    std::size_t queue_capacity = 0;
 };
 
 } // namespace logit_bench

bench/adapters/LogItAdapter.cpp

@@ -105,7 +105,8 @@ namespace logit_bench {
         };
 
         void consume(int slot_line, std::string_view text) {
-            // slot-only completion
+            // Record sink-entry latency; file I/O happens below and is not
+            // part of this completion marker.
             if (slot_line >= 0 && m_recorder) {
                 m_recorder->complete_slot(static_cast<std::uint64_t>(slot_line));
             }

bench/adapters/SpdlogAdapter.cpp

@@ -9,6 +9,7 @@
 #include <limits>
 #include <memory>
 #include <mutex>
+#include <condition_variable>
 #include <string>
 #include <string_view>
 
@@ -43,7 +44,9 @@ namespace logit_bench {
         }
 
         void log(const spdlog::details::log_msg& msg) override {
-            // slot is stored in msg.source.line
+            // Record sink-entry latency; file I/O happens below and is not
+            // part of this completion marker. The slot is stored in
+            // msg.source.line.
             const int line = msg.source.line;
             if (line >= 0 && m_recorder) {
                 m_recorder->complete_slot(static_cast<std::uint64_t>(line));
@@ -63,17 +66,37 @@ namespace logit_bench {
 
         void flush() override {
             std::lock_guard<std::mutex> lock(m_mutex);
+            flush_file_locked();
+            ++m_flush_generation;
+            m_flush_cv.notify_all();
+        }
+
+        std::uint64_t flush_generation() const {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            return m_flush_generation;
+        }
+
+        void wait_for_flush(std::uint64_t generation) const {
+            std::unique_lock<std::mutex> lock(m_mutex);
+            m_flush_cv.wait(lock, [&]() {
+                return m_flush_generation > generation;
+            });
+        }
+
+    private:
+        void flush_file_locked() {
             if (m_file.is_open()) {
                 m_file.flush();
             }
         }
-
-    private:
+
         SinkKind m_sink = SinkKind::Null;
         std::shared_ptr<LatencyRecorder> m_recorder;
-    
+
         std::ofstream m_file;
-        std::mutex m_mutex;
+        mutable std::mutex m_mutex;
+        mutable std::condition_variable m_flush_cv;
+        std::uint64_t m_flush_generation = 0;
     };
 
     SpdlogAdapter::SpdlogAdapter() = default;
@@ -104,8 +127,9 @@ namespace logit_bench {
 
         std::string logger_name = m_async ? "logit_bench_async" : "logit_bench_sync";
         if (m_async) {
-            const std::size_t queue_size =
-                std::max<std::size_t>(kDefaultQueue, scenario.total_messages * 2);
+            const std::size_t queue_size = scenario.queue_capacity > 0
+                ? scenario.queue_capacity
+                : kDefaultQueue;
 
             spdlog::init_thread_pool(queue_size, 1);
 
@@ -146,10 +170,15 @@ namespace logit_bench {
 
     void SpdlogAdapter::flush() {
         if (m_logger) {
-            m_logger->flush();
-        }
-        if (m_sink) {
-            m_sink->flush();
+            if (m_async && m_sink) {
+                const auto generation = m_sink->flush_generation();
+                m_logger->flush();
+                // async_logger::flush() enqueues a marker. The sink-side
+                // generation is advanced only when the worker executes it.
+                m_sink->wait_for_flush(generation);
+            } else {
+                m_logger->flush();
+            }
         }
     }
 

bench/logit_bench.cpp

@@ -1,4 +1,5 @@
 #include <array>
+#include <algorithm>
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
@@ -206,6 +207,11 @@ std::chrono::nanoseconds run_workload(
     adapter.flush();
     touch_watchdog();
 
+    if (record_latency && recorder.completed() != recorder.recorded()) {
+        throw std::runtime_error(
+            "adapter.flush() returned before all recorded messages reached the sink");
+    }
+
     if (!measure_duration) return std::chrono::nanoseconds(0);
     auto t1 = std::chrono::steady_clock::now();
     return std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0);
@@ -299,22 +305,42 @@ void append_csv(
 {
     namespace fs = std::filesystem;
     const fs::path csv_path{"bench/results/latency.csv"};
+    const std::string expected_header =
+        "lib,async,sink,producers,msg_bytes,total,queue_capacity,"
+        "p50_ns,p99_ns,p999_ns,throughput";
     fs::create_directories(csv_path.parent_path());
 
     const bool write_header = !fs::exists(csv_path) || fs::file_size(csv_path) == 0;
 
+    if (!write_header) {
+        std::ifstream in(csv_path);
+        std::string header;
+        if (!in || !std::getline(in, header)) {
+            throw std::runtime_error("Failed to read latency.csv schema header");
+        }
+        if (!header.empty() && header.back() == '\r') {
+            header.pop_back();
+        }
+        if (header != expected_header) {
+            throw std::runtime_error(
+                "Unsupported bench/results/latency.csv schema; rename or remove "
+                "the existing file before running this benchmark");
+        }
+    }
+
     std::ofstream out(csv_path, std::ios::app);
     if (!out) throw std::runtime_error("Failed to open latency.csv for writing");
 
     if (write_header) {
-        out << "lib,async,sink,producers,msg_bytes,total,p50_ns,p99_ns,p999_ns,throughput\n";
+        out << expected_header << '\n';
     }
     out << library << ','
         << (scenario.async ? 1 : 0) << ','
         << sink_name(scenario.sink) << ','
         << scenario.producers << ','
         << scenario.message_bytes << ','
         << scenario.total_messages << ','
+        << scenario.queue_capacity << ','
         << summary.p50_ns << ','
         << summary.p99_ns << ','
         << summary.p999_ns << ','
@@ -333,6 +359,7 @@ void print_summary(
         << " producers=" << scenario.producers
         << " bytes=" << scenario.message_bytes
         << " total=" << scenario.total_messages
+        << " queue=" << scenario.queue_capacity
         << " p50=" << result.summary.p50_ns
         << "ns p99=" << result.summary.p99_ns
         << "ns p999=" << result.summary.p999_ns
@@ -361,17 +388,26 @@ int main() {
         // Matrix
         const std::array<bool, 2> async_modes{false, true};
         const std::array<SinkKind, 2> sinks{SinkKind::Null, SinkKind::File};
-        const std::array<std::size_t, 3> producer_counts{1, 4, 16};
+        const std::array<std::size_t, 4> producer_counts{1, 4, 16, 32};
         const std::array<std::size_t, 3> message_sizes{40, 200, 1024};
 
         // Totals (can be overridden by env):
         const std::size_t total_messages  = get_env_size_t("LOGIT_BENCH_TOTAL", 200000);
         const std::size_t warmup_messages = get_env_size_t("LOGIT_BENCH_WARMUP", 4096);
         const std::size_t timeout_seconds = get_env_size_t("LOGIT_BENCH_TIMEOUT_SEC", 1200);
+        const std::size_t queue_capacity = get_env_size_t(
+            "LOGIT_BENCH_QUEUE_CAPACITY",
+            std::max<std::size_t>(8192, total_messages * 2));
+        if (queue_capacity == 0) {
+            throw std::invalid_argument(
+                "LOGIT_BENCH_QUEUE_CAPACITY must be greater than zero for "
+                "a comparative benchmark");
+        }
 
         const BenchFilter filter = load_filter();
 
-        LOGIT_SET_MAX_QUEUE(total_messages);
+        LOGIT_SET_MAX_QUEUE(queue_capacity);
+        LOGIT_SET_QUEUE_POLICY(LOGIT_QUEUE_BLOCK);
 
         if (timeout_seconds > 0) {
             watchdog = std::thread([timeout_seconds, &watchdog_done, &watchdog_progress]() {
@@ -405,6 +441,12 @@ int main() {
                             scenario.producers      = producers;
                             scenario.message_bytes  = msg_bytes;
                             scenario.total_messages = total_messages;
+                            scenario.queue_capacity = queue_capacity;
+
+                            // Keep the global LogIt executor on the same
+                            // bounded/blocking contract as the spdlog adapter.
+                            LOGIT_SET_MAX_QUEUE(scenario.queue_capacity);
+                            LOGIT_SET_QUEUE_POLICY(LOGIT_QUEUE_BLOCK);
 
                             {
                                 std::ostringstream oss;

bench/spdlog_flush_test.cpp

@@ -0,0 +1,31 @@
+#include <cstddef>
+#include <memory>
+#include <string>
+
+#include "LatencyRecorder.hpp"
+#include "Scenario.hpp"
+#include "adapters/SpdlogAdapter.hpp"
+
+int main() {
+    logit_bench::Scenario scenario;
+    scenario.async = true;
+    scenario.sink = logit_bench::SinkKind::Null;
+    scenario.producers = 1;
+    scenario.message_bytes = 1;
+    scenario.total_messages = 64;
+    scenario.queue_capacity = 8;
+
+    logit_bench::SpdlogAdapter adapter;
+    auto recorder = std::make_shared<logit_bench::LatencyRecorder>(
+        scenario.total_messages);
+    adapter.set_recorder_handle(recorder);
+    adapter.prepare(scenario, *recorder);
+
+    for (std::size_t i = 0; i < scenario.total_messages; ++i) {
+        const auto token = recorder->begin(true);
+        adapter.log(token, std::string_view("x", 1));
+    }
+
+    adapter.flush();
+    return recorder->completed() == scenario.total_messages ? 0 : 1;
+}