Fix utils header ODR issues

.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:

.github/workflows/odr_check.yml

@@ -0,0 +1,41 @@
+name: ODR checks
+
+on:
+  push
+
+jobs:
+  odr:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: tests/odr
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: recursive
+
+      - name: Install dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y build-essential
+
+      - name: Build Logit ODR header
+        run: |
+          g++ -std=c++17 -I ../../include -I ../../libs/time-shield-cpp/include -c get_logger_a.cpp
+          g++ -std=c++17 -I ../../include -I ../../libs/time-shield-cpp/include -c get_logger_b.cpp
+          g++ -std=c++17 -I ../../include -I ../../libs/time-shield-cpp/include -c get_logger_main.cpp
+          g++ get_logger_a.o get_logger_b.o get_logger_main.o -o app_logger
+
+      - name: Run ConsoleApp ODR header
+        run: ./app_logger
+
+      - name: Build Task Executor ODR header
+        run: |
+          g++ -std=c++17 -I ../../include -I ../../libs/time-shield-cpp/include -c get_executor_a.cpp
+          g++ -std=c++17 -I ../../include -I ../../libs/time-shield-cpp/include -c get_executor_b.cpp
+          g++ -std=c++17 -I ../../include -I ../../libs/time-shield-cpp/include -c get_executor_main.cpp
+          g++ get_executor_a.o get_executor_b.o get_executor_main.o -o app_executor
+
+      - name: Run Task Executor ODR header
+        run: ./app_executor
+

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

@@ -3,11 +3,15 @@
 #ifdef LOGIT_BENCH_HAVE_SPDLOG
 
 #include <algorithm>
+#include <chrono>
+#include <condition_variable>
 #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 +36,12 @@ 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();
+            m_retired.clear();
+            m_pending_count = 0;
+        }
         if (m_sink == SinkKind::File) {
             std::filesystem::create_directories("bench/results");
             std::lock_guard<std::mutex> lock(m_mutex);
@@ -43,14 +53,21 @@ 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});
+        ++m_pending_count;
+    }
+
     void log(const spdlog::details::log_msg& msg) override {
-        const auto* payload_ptr = reinterpret_cast<const MessagePayload*>(msg.source.funcname);
-        if (!payload_ptr) {
-            return;
+        const char* func = msg.source.funcname;
+        if (msg.payload.size() == 0 || !func || *func == '\0') {
+            return; // Flush/control messages have no payload attached.
         }
+
+        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 {}
@@ -64,15 +81,59 @@ class SpdlogAdapter::MeasuringSink : public spdlog::sinks::sink {
         }
     }
 
+    void complete_pending() {
+        std::vector<Pending> pending;
+        {
+            std::unique_lock<std::mutex> lock(m_pending_mx);
+            m_pending_cv.wait_for(lock, std::chrono::milliseconds(100), [&]{ return m_pending_count == 0; });
+            pending.swap(m_pending);
+            m_pending_count = 0;
+        }
+
+        std::vector<std::unique_ptr<MessagePayload>> retired;
+        retired.reserve(pending.size());
+        for (auto& entry : pending) {
+            if (entry.token.active && m_recorder) {
+                m_recorder->complete(entry.token);
+            }
+            if (entry.payload) {
+                retired.push_back(std::move(entry.payload));
+            }
+        }
+        m_retired.insert(m_retired.end(),
+                         std::make_move_iterator(retired.begin()),
+                         std::make_move_iterator(retired.end()));
+    }
+
 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);
+                --m_pending_count;
+                if (m_pending_count == 0) m_pending_cv.notify_all();
+            }
+        }
+
+        const MessagePayload& msg = owned ? *owned : payload;
+
+        if (token.active && m_recorder) {
+            m_recorder->complete(token);
+        }
+        if (owned) {
+            std::lock_guard<std::mutex> lock(m_pending_mx);
+            m_retired.push_back(std::move(owned));
         }
         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';
             }
         }
     }
@@ -81,6 +142,15 @@ 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::vector<std::unique_ptr<MessagePayload>> m_retired;
+    std::size_t m_pending_count = 0;
+    std::condition_variable m_pending_cv;
+    std::mutex m_pending_mx;
 };
 
 SpdlogAdapter::SpdlogAdapter() = default;
@@ -90,6 +160,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();
@@ -123,18 +197,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;
 };