Guard spdlog async bench recorder lifetime and add ASAN check

.github/workflows/ci.yml

@@ -147,6 +147,37 @@ jobs:
       - name: Test
         run: ctest --test-dir build --output-on-failure
 
+  bench-asan:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: true
+      - run: git submodule update --init --recursive
+      - name: Configure benchmarks (asan)
+        run: |
+          cmake -S . -B build-bench-asan \
+            -DLOGIT_BENCH_ENABLE=ON \
+            -DLOGIT_BENCH_WITH_SPDLOG=ON \
+            -DCMAKE_BUILD_TYPE=RelWithDebInfo \
+            -DCMAKE_CXX_STANDARD=17 \
+            -DCMAKE_CXX_FLAGS='-fsanitize=address,undefined -fno-omit-frame-pointer -g' \
+            -DCMAKE_EXE_LINKER_FLAGS='-fsanitize=address,undefined'
+      - name: Build benchmarks (asan)
+        run: cmake --build build-bench-asan --target logit_bench
+      - name: Run spdlog async null bench (asan)
+        timeout-minutes: 10
+        env:
+          LOGIT_BENCH_FILTER_LIB: spdlog
+          LOGIT_BENCH_FILTER_ASYNC: "1"
+          LOGIT_BENCH_FILTER_SINK: null
+          LOGIT_BENCH_FILTER_PRODUCERS: "4"
+          LOGIT_BENCH_FILTER_BYTES: "40"
+          LOGIT_BENCH_TOTAL: 200
+          LOGIT_BENCH_WARMUP: 20
+          LOGIT_BENCH_TIMEOUT_SEC: 120
+        run: ./build-bench-asan/logit_bench
+
   vcpkg-install:
     runs-on: ubuntu-latest
     env:

bench/LatencyRecorder.hpp

@@ -4,7 +4,9 @@
 #include <atomic>
 #include <chrono>
 #include <cstdint>
+#include <condition_variable>
 #include <limits>
+#include <mutex>
 #include <stdexcept>
 #include <vector>
 #include <cmath>
@@ -36,7 +38,8 @@ class LatencyRecorder {
     explicit LatencyRecorder(std::size_t total)
         : m_values(total),
           m_expected(total),
-          m_next_slot(0) {}
+          m_next_slot(0),
+          m_completed(0) {}
 
     /**
      * Reserve a slot (if record==true) and capture t0 using steady_clock.
@@ -61,14 +64,24 @@ class LatencyRecorder {
         if (!token.active) return;
         const auto t1_ns = now();
         m_values[token.slot] = t1_ns - token.t0_ns; // distinct slots -> no data race
+        const auto done = m_completed.fetch_add(1, std::memory_order_acq_rel) + 1;
+        if (done == m_expected) {
+            std::lock_guard<std::mutex> lk(m_wait_mx);
+            m_wait_cv.notify_all();
+        }
     }
 
     std::size_t recorded() const {
         return m_next_slot.load(std::memory_order_relaxed);
     }
 
+    void wait_for_all() const {
+        std::unique_lock<std::mutex> lk(m_wait_mx);
+        m_wait_cv.wait(lk, [&]{ return m_completed.load(std::memory_order_acquire) >= m_expected; });
+    }
+
     Summary finalize() const {
-        if (recorded() != m_expected) {
+        if (recorded() != m_expected || m_completed.load(std::memory_order_acquire) != m_expected) {
             throw std::runtime_error("Incomplete latency capture");
         }
         std::vector<std::uint64_t> sorted = m_values;
@@ -102,6 +115,9 @@ class LatencyRecorder {
     std::vector<std::uint64_t> m_values;   // preallocated; no reallocation
     const std::size_t          m_expected; // total messages to record
     std::atomic<std::size_t>   m_next_slot;
+    std::atomic<std::size_t>   m_completed;
+    mutable std::condition_variable m_wait_cv;
+    mutable std::mutex m_wait_mx;
 };
 
 } // namespace logit_bench

bench/adapters/ILoggerAdapter.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <memory>
 #include <string_view>
 
 #include "../LatencyRecorder.hpp"
@@ -18,6 +19,8 @@ class ILoggerAdapter {
     virtual void log(const LatencyRecorder::Token& token, std::string_view message) = 0;
 
     virtual void flush() = 0;
+
+    virtual void set_recorder_handle(std::shared_ptr<LatencyRecorder>) {}
 };
 
 } // namespace logit_bench

bench/adapters/SpdlogAdapter.cpp

@@ -5,9 +5,11 @@
 #include <algorithm>
 #include <filesystem>
 #include <fstream>
+#include <memory>
 #include <mutex>
 #include <string>
 #include <string_view>
+#include <vector>
 
 #include <spdlog/async.h>
 #include <spdlog/async_logger.h>
@@ -32,6 +34,10 @@ class SpdlogAdapter::MeasuringSink : public spdlog::sinks::sink {
     void configure(const Scenario& scenario, LatencyRecorder& recorder) {
         m_sink = scenario.sink;
         m_recorder = &recorder;
+        {
+            std::lock_guard<std::mutex> lock(m_pending_mx);
+            m_pending.clear();
+        }
         if (m_sink == SinkKind::File) {
             std::filesystem::create_directories("bench/results");
             std::lock_guard<std::mutex> lock(m_mutex);
@@ -43,6 +49,11 @@ class SpdlogAdapter::MeasuringSink : public spdlog::sinks::sink {
         }
     }
 
+    void track_token(const LatencyRecorder::Token& token, std::unique_ptr<MessagePayload> payload) {
+        std::lock_guard<std::mutex> lock(m_pending_mx);
+        m_pending.push_back(Pending{std::move(payload), token});
+    }
+
     void log(const spdlog::details::log_msg& msg) override {
         const char* func = msg.source.funcname;
         if (msg.payload.size() == 0 || !func || *func == '\0') {
@@ -51,8 +62,7 @@ class SpdlogAdapter::MeasuringSink : public spdlog::sinks::sink {
 
         const auto* payload_ptr = reinterpret_cast<const MessagePayload*>(func);
         auto* payload = const_cast<MessagePayload*>(payload_ptr);
-        consume(*payload);
-        delete payload;
+        consume(*payload, payload);
     }
 
     void set_pattern(const std::string&) override {}
@@ -66,15 +76,42 @@ class SpdlogAdapter::MeasuringSink : public spdlog::sinks::sink {
         }
     }
 
+    void complete_pending() {
+        std::vector<Pending> pending;
+        {
+            std::lock_guard<std::mutex> lock(m_pending_mx);
+            pending.swap(m_pending);
+        }
+        for (const auto& entry : pending) {
+            if (entry.token.active && m_recorder) {
+                m_recorder->complete(entry.token);
+            }
+        }
+    }
+
 private:
-    void consume(const MessagePayload& payload) {
-        if (payload.token.active && m_recorder) {
-            m_recorder->complete(payload.token);
+    void consume(const MessagePayload& payload, MessagePayload* payload_ptr) {
+        LatencyRecorder::Token token = payload.token;
+        std::unique_ptr<MessagePayload> owned;
+        {
+            std::lock_guard<std::mutex> lock(m_pending_mx);
+            auto it = std::find_if(m_pending.begin(), m_pending.end(), [&](const Pending& p){ return p.payload.get() == payload_ptr; });
+            if (it != m_pending.end()) {
+                token = it->token;
+                owned = std::move(it->payload);
+                m_pending.erase(it);
+            }
+        }
+
+        const MessagePayload& msg = owned ? *owned : payload;
+
+        if (token.active && m_recorder) {
+            m_recorder->complete(token);
         }
         if (m_sink == SinkKind::File) {
             std::lock_guard<std::mutex> lock(m_mutex);
             if (m_file.is_open()) {
-                m_file << payload.text << '\n';
+                m_file << msg.text << '\n';
             }
         }
     }
@@ -83,6 +120,12 @@ class SpdlogAdapter::MeasuringSink : public spdlog::sinks::sink {
     LatencyRecorder* m_recorder = nullptr;
     std::ofstream m_file;
     std::mutex m_mutex;
+    struct Pending {
+        std::unique_ptr<MessagePayload> payload;
+        LatencyRecorder::Token token;
+    };
+    std::vector<Pending> m_pending;
+    std::mutex m_pending_mx;
 };
 
 SpdlogAdapter::SpdlogAdapter() = default;
@@ -92,6 +135,10 @@ SpdlogAdapter::~SpdlogAdapter() {
     spdlog::shutdown();
 }
 
+void SpdlogAdapter::set_recorder_handle(std::shared_ptr<LatencyRecorder> recorder) {
+    m_recorder_handle = std::move(recorder);
+}
+
 void SpdlogAdapter::prepare(const Scenario& scenario, LatencyRecorder& recorder) {
     m_logger.reset();
     m_sink.reset();
@@ -125,18 +172,23 @@ void SpdlogAdapter::log(const LatencyRecorder::Token& token, std::string_view me
     if (!m_logger) {
         return;
     }
-    auto* payload = new MessagePayload();
+    auto payload = std::make_unique<MessagePayload>();
     payload->token = token;
     payload->text.assign(message.data(), message.size());
-    spdlog::source_loc loc{nullptr, 0, reinterpret_cast<const char*>(payload)};
-    m_logger->log(loc, spdlog::level::info, spdlog::string_view_t(payload->text));
+    MessagePayload* payload_ptr = payload.get();
+    if (m_sink) {
+        m_sink->track_token(token, std::move(payload));
+    }
+    spdlog::source_loc loc{nullptr, 0, reinterpret_cast<const char*>(payload_ptr)};
+    m_logger->log(loc, spdlog::level::info, spdlog::string_view_t(payload_ptr->text));
 }
 
 void SpdlogAdapter::flush() {
     if (m_logger) {
         m_logger->flush();
     }
     if (m_sink) {
+        m_sink->complete_pending();
         m_sink->flush();
     }
 }

bench/adapters/SpdlogAdapter.hpp

@@ -24,11 +24,14 @@ class SpdlogAdapter : public ILoggerAdapter {
 
     void flush() override;
 
+    void set_recorder_handle(std::shared_ptr<LatencyRecorder> recorder) override;
+
 private:
     class MeasuringSink;
 
     std::shared_ptr<spdlog::logger> m_logger;
     std::shared_ptr<MeasuringSink> m_sink;
+    std::shared_ptr<LatencyRecorder> m_recorder_handle;
     bool m_async = false;
 };
 

bench/logit_bench.cpp

@@ -9,6 +9,7 @@
 #include <iomanip>
 #include <iostream>
 #include <ctime>
+#include <optional>
 #include <memory>
 #include <mutex>
 #include <stdexcept>
@@ -48,6 +49,50 @@ std::size_t get_env_size_t(const char* name, std::size_t def) {
     return def;
 }
 
+struct BenchFilter {
+    std::optional<std::string> library;
+    std::optional<bool> async;
+    std::optional<SinkKind> sink;
+    std::optional<std::size_t> producers;
+    std::optional<std::size_t> bytes;
+
+    bool matches(const std::string& lib,
+                 bool async_mode,
+                 SinkKind sink_kind,
+                 std::size_t producer_count,
+                 std::size_t msg_bytes) const {
+        if (library && *library != lib) return false;
+        if (async && *async != async_mode) return false;
+        if (sink && *sink != sink_kind) return false;
+        if (producers && *producers != producer_count) return false;
+        if (bytes && *bytes != msg_bytes) return false;
+        return true;
+    }
+};
+
+BenchFilter load_filter() {
+    BenchFilter filter;
+    if (const char* v = std::getenv("LOGIT_BENCH_FILTER_LIB")) {
+        filter.library = std::string(v);
+    }
+    if (const char* v = std::getenv("LOGIT_BENCH_FILTER_ASYNC")) {
+        filter.async = std::string(v) == "1";
+    }
+    if (const char* v = std::getenv("LOGIT_BENCH_FILTER_SINK")) {
+        std::string s(v);
+        std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c){ return static_cast<char>(std::tolower(c)); });
+        if (s == "null") filter.sink = SinkKind::Null;
+        if (s == "file") filter.sink = SinkKind::File;
+    }
+    if (const char* v = std::getenv("LOGIT_BENCH_FILTER_PRODUCERS")) {
+        filter.producers = get_env_size_t("LOGIT_BENCH_FILTER_PRODUCERS", 0);
+    }
+    if (const char* v = std::getenv("LOGIT_BENCH_FILTER_BYTES")) {
+        filter.bytes = get_env_size_t("LOGIT_BENCH_FILTER_BYTES", 0);
+    }
+    return filter;
+}
+
 std::uint64_t steady_now_ns() {
     const auto now_tp = std::chrono::steady_clock::now().time_since_epoch();
     return std::chrono::duration_cast<std::chrono::nanoseconds>(now_tp).count();
@@ -177,10 +222,12 @@ ScenarioResult execute_scenario(
         const Scenario& scenario,
         std::size_t warmup_messages)
 {
-    LatencyRecorder recorder(scenario.total_messages);
+    auto recorder = std::make_shared<LatencyRecorder>(scenario.total_messages);
+
+    adapter.set_recorder_handle(recorder);
 
     // Adapter should keep a pointer/ref to recorder and call complete(token) from its sink.
-    adapter.prepare(scenario, recorder);
+    adapter.prepare(scenario, *recorder);
 
     // Warm-up (no recording, no duration).
     {
@@ -193,7 +240,7 @@ ScenarioResult execute_scenario(
             << " total=" << warmup_messages;
         log_info(oss.str());
     }
-    run_workload(adapter, recorder, scenario, warmup_messages, false, false);
+    run_workload(adapter, *recorder, scenario, warmup_messages, false, false);
     {
         std::ostringstream oss;
         oss << "Warm-up completed lib=" << adapter.library_name()
@@ -215,7 +262,7 @@ ScenarioResult execute_scenario(
             << " total=" << scenario.total_messages;
         log_info(oss.str());
     }
-    const auto dur = run_workload(adapter, recorder, scenario, scenario.total_messages, true, true);
+    const auto dur = run_workload(adapter, *recorder, scenario, scenario.total_messages, true, true);
     {
         std::ostringstream oss;
         oss << "Measure completed lib=" << adapter.library_name()
@@ -230,13 +277,18 @@ ScenarioResult execute_scenario(
     // destroying the recorder referenced by sinks.
     adapter.flush();
 
-    const auto sum = recorder.finalize();
+    recorder->wait_for_all();
+    const auto sum = recorder->finalize();
+
 
     double thr = 0.0;
     if (dur.count() > 0) {
         const double sec = static_cast<double>(dur.count()) / 1'000'000'000.0;
         thr = static_cast<double>(scenario.total_messages) / sec;
     }
+
+    adapter.set_recorder_handle(nullptr);
+
     return ScenarioResult{sum, thr, dur};
 }
 
@@ -318,6 +370,8 @@ int main() {
         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 BenchFilter filter = load_filter();
+
         LOGIT_SET_MAX_QUEUE(total_messages);
 
         if (timeout_seconds > 0) {
@@ -343,6 +397,9 @@ int main() {
                 for (auto sink : sinks) {
                     for (std::size_t producers : producer_counts) {
                         for (std::size_t msg_bytes : message_sizes) {
+                            if (!filter.matches(adapter->library_name(), async_mode, sink, producers, msg_bytes)) {
+                                continue;
+                            }
                             Scenario scenario;
                             scenario.async          = async_mode;
                             scenario.sink           = sink;