bench: add contract-matched async benchmark

.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 logit_exec_mx_bench logit_exec_mx_bench_concurrent benchmark_validation_test
+        run: cmake --build build-bench --target logit_bench logit_bench_async_contract logit_bench_async_payload_contract_test 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
@@ -47,6 +47,8 @@ jobs:
         run: ./build-bench/logit_public_macro_formatted_bench
       - name: Run benchmark validation tests
         run: ./build-bench/benchmark_validation_test
+      - name: Run async payload contract regression
+        run: ./build-bench/logit_bench_async_payload_contract_test
       - name: Run logger hot-path A/B smoke
         env:
           LOGIT_HOTPATH_BENCH_TOTAL: 20000
@@ -67,6 +69,17 @@ jobs:
           LOGIT_BENCH_TOTAL: 20000
           LOGIT_BENCH_WARMUP: 2000
         run: ./build-bench/logit_bench
+      - name: Run contract-matched async benchmark smoke
+        env:
+          LOGIT_BENCH_FILTER_ASYNC: "1"
+          LOGIT_BENCH_FILTER_SINK: "null"
+          LOGIT_BENCH_FILTER_PRODUCERS: "4"
+          LOGIT_BENCH_FILTER_BYTES: "200"
+          LOGIT_BENCH_TOTAL: 2000
+          LOGIT_BENCH_WARMUP: 200
+          LOGIT_BENCH_TIMEOUT_SEC: 120
+          LOGIT_BENCH_OUTPUT: bench/results/latency-async-contract.csv
+        run: ./build-bench/logit_bench_async_contract
       - 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

README-RU.md

@@ -939,6 +939,14 @@ LogIt++ включает библиотеку *fmt* для форматиров
 
 Запустите `./build/bench/logit_bench`, чтобы получить полный набор измерений (sync/async × null/file × количество продюсеров × размер сообщений). Результаты дописываются в `bench/results/latency.csv` по одной строке на каждую библиотеку/комбинацию. При необходимости сократите нагрузку с помощью переменных окружения `LOGIT_BENCH_TOTAL` и `LOGIT_BENCH_WARMUP`.
 
+Для сравнения async/null с одинаковой стоимостью передачи payload со spdlog
+соберите и запустите `logit_bench_async_contract`. В этом режиме полное
+сообщение проходит через очередь LogIt++ и результат записывается в отдельный
+CSV `bench/results/latency-async-contract.csv` (или в путь из
+`LOGIT_BENCH_OUTPUT`). Не смешивайте эти строки с обычным async/null режимом,
+где LogIt++ передаёт только marker. Сам target запускает только сценарий
+`async=1`, `sink=null`.
+
 ### Что на самом деле измеряет бенчмарк
 
 Полная методика, ограничения сравнения и правила интерпретации находятся в

README.md

@@ -1167,6 +1167,13 @@ Run `./build/bench/logit_bench` to record the full matrix (sync/async × null/fi
 are appended to `bench/results/latency.csv` with one row per library/combination. Override the workload via `LOGIT_BENCH_TOTAL`
 and `LOGIT_BENCH_WARMUP` environment variables if you need a lighter run.
 
+For a payload-matched async/null comparison with spdlog, build and run
+`logit_bench_async_contract`; it carries the full message through the LogIt++
+worker queue and writes by default to `bench/results/latency-async-contract.csv`
+(or to the path selected by `LOGIT_BENCH_OUTPUT`). Do not combine those rows
+with the default marker-only async/null workload. The target itself runs only
+the `async=1`, `sink=null` scenario.
+
 ### What this benchmark measures
 
 See the canonical [benchmark guide](docs/benchmarks.md) for the measurement

bench/BenchmarkMetadata.hpp

@@ -27,6 +27,7 @@ struct BenchmarkMetadata {
     std::string queue_policy;
     std::string latency_completion;
     std::string flush_barrier;
+    std::string workload_contract;
 };
 
 inline std::string benchmark_env(const char* name, const char* fallback) {
@@ -138,7 +139,8 @@ inline BenchmarkMetadata make_benchmark_metadata(
         std::string queue_capacity,
         std::string queue_policy,
         std::string latency_completion,
-        std::string flush_barrier) {
+        std::string flush_barrier,
+        std::string workload_contract) {
     const std::string compiler = benchmark_compiler();
     const std::string compiler_version = benchmark_compiler_version();
     std::string toolchain = compiler + "-" + compiler_version;
@@ -157,7 +159,8 @@ inline BenchmarkMetadata make_benchmark_metadata(
         std::move(queue_capacity),
         std::move(queue_policy),
         std::move(latency_completion),
-        std::move(flush_barrier)};
+        std::move(flush_barrier),
+        std::move(workload_contract)};
 }
 
 inline bool benchmark_value_unknown(const std::string& value) {
@@ -187,7 +190,8 @@ inline void validate_comparable_metadata(const BenchmarkMetadata& metadata,
         {"queue_capacity", &metadata.queue_capacity, true},
         {"queue_policy", &metadata.queue_policy, true},
         {"latency_completion", &metadata.latency_completion, false},
-        {"flush_barrier", &metadata.flush_barrier, false}};
+        {"flush_barrier", &metadata.flush_barrier, false},
+        {"workload_contract", &metadata.workload_contract, false}};
 
     for (const auto& value : values) {
         if (benchmark_value_unknown(*value.value) ||
@@ -210,7 +214,7 @@ inline void print_benchmark_metadata(std::ostream& out,
                                      const BenchmarkMetadata& metadata,
                                      std::size_t total_messages,
                                      std::size_t warmup_messages) {
-    out << "benchmark-fixture version=1"
+    out << "benchmark-fixture version=2"
         << " source_commit=" << metadata.source_commit
         << " compiler=" << metadata.compiler
         << " compiler_version=" << metadata.compiler_version
@@ -225,6 +229,7 @@ inline void print_benchmark_metadata(std::ostream& out,
         << " queue_policy=" << metadata.queue_policy
         << " latency_completion=" << metadata.latency_completion
         << " flush_barrier=" << metadata.flush_barrier
+        << " workload_contract=" << metadata.workload_contract
         << " total=" << total_messages
         << " warmup=" << warmup_messages << '\n';
 }

bench/BenchmarkValidation.hpp

@@ -15,7 +15,7 @@ inline void validate_queue_capacity(std::size_t capacity) {
 
 inline const char* latency_csv_header() {
     return "lib,async,sink,producers,msg_bytes,total,queue_capacity,"
-           "p50_ns,p99_ns,p999_ns,throughput";
+           "workload_contract,p50_ns,p99_ns,p999_ns,throughput";
 }
 
 inline void validate_latency_csv_header(std::string header) {
@@ -24,7 +24,7 @@ inline void validate_latency_csv_header(std::string header) {
     }
     if (header != latency_csv_header()) {
         throw std::runtime_error(
-            "Unsupported bench/results/latency.csv schema; rename or remove "
+            "Unsupported benchmark CSV schema; rename or remove "
             "the existing file before running this benchmark");
     }
 }

bench/CMakeLists.txt

@@ -26,6 +26,23 @@ endforeach()
 
 target_link_libraries(logit_bench PRIVATE log-it-cpp::log-it-cpp)
 
+add_executable(logit_bench_async_contract ${LOGIT_BENCH_SOURCES})
+target_include_directories(logit_bench_async_contract PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
+target_compile_definitions(logit_bench_async_contract PRIVATE
+    LOGIT_BENCH_BUILD_TYPE="$<CONFIG>"
+    LOGIT_BENCH_CONTRACT_MATCHED_ASYNC=1
+)
+target_compile_features(logit_bench_async_contract PRIVATE cxx_std_17)
+set_target_properties(logit_bench_async_contract PROPERTIES
+    RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}
+)
+foreach(config IN ITEMS DEBUG RELEASE RELWITHDEBINFO MINSIZEREL)
+    set_target_properties(logit_bench_async_contract PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY_${config} ${CMAKE_BINARY_DIR}
+    )
+endforeach()
+target_link_libraries(logit_bench_async_contract PRIVATE log-it-cpp::log-it-cpp)
+
 add_executable(logit_public_macro_bench public_macro_bench.cpp)
 target_compile_features(logit_public_macro_bench PRIVATE cxx_std_17)
 target_compile_definitions(logit_public_macro_bench PRIVATE LOGIT_BENCH_BUILD_TYPE=\"$<CONFIG>\")
@@ -72,8 +89,27 @@ target_compile_features(benchmark_validation_test PRIVATE cxx_std_17)
 set_target_properties(benchmark_validation_test PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
 add_test(NAME benchmark_validation_test COMMAND benchmark_validation_test)
 
+add_executable(logit_bench_async_payload_contract_test
+    async_payload_contract_test.cpp
+    adapters/LogItAdapter.cpp
+)
+target_include_directories(logit_bench_async_payload_contract_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
+target_compile_features(logit_bench_async_payload_contract_test PRIVATE cxx_std_17)
+target_link_libraries(logit_bench_async_payload_contract_test PRIVATE log-it-cpp::log-it-cpp)
+set_target_properties(logit_bench_async_payload_contract_test PROPERTIES
+    RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}
+)
+foreach(config IN ITEMS DEBUG RELEASE RELWITHDEBINFO MINSIZEREL)
+    set_target_properties(logit_bench_async_payload_contract_test PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY_${config} ${CMAKE_BINARY_DIR}
+    )
+endforeach()
+add_test(NAME logit_bench_async_payload_contract_test
+         COMMAND logit_bench_async_payload_contract_test)
+
 if(LOGIT_BENCH_WITH_SPDLOG)
     target_compile_definitions(logit_bench PRIVATE LOGIT_BENCH_HAVE_SPDLOG=1)
+    target_compile_definitions(logit_bench_async_contract PRIVATE LOGIT_BENCH_HAVE_SPDLOG=1)
     if(NOT TARGET spdlog::spdlog)
         include(FetchContent)
         FetchContent_Declare(spdlog
@@ -83,6 +119,7 @@ if(LOGIT_BENCH_WITH_SPDLOG)
         FetchContent_MakeAvailable(spdlog)
     endif()
     target_link_libraries(logit_bench PRIVATE spdlog::spdlog)
+    target_link_libraries(logit_bench_async_contract PRIVATE spdlog::spdlog)
 
     add_executable(logit_bench_flush_test
         spdlog_flush_test.cpp

bench/Scenario.hpp

@@ -1,7 +1,9 @@
 #pragma once
 
 #include <cstddef>
+#include <functional>
 #include <string>
+#include <string_view>
 
 namespace logit_bench {
 
@@ -10,6 +12,11 @@ enum class SinkKind {
     File,
 };
 
+enum class AsyncPayloadMode {
+    MarkerOnly,
+    FullMessage,
+};
+
 inline std::string sink_name(SinkKind sink) {
     switch (sink) {
         case SinkKind::Null: return "null";
@@ -21,6 +28,8 @@ inline std::string sink_name(SinkKind sink) {
 struct Scenario {
     bool        async          = false;
     SinkKind    sink           = SinkKind::Null;
+    AsyncPayloadMode async_payload = AsyncPayloadMode::MarkerOnly;
+    std::function<void(std::string_view)> async_payload_observer;
     std::size_t producers      = 1;
     std::size_t message_bytes  = 0;
     std::size_t total_messages = 0;

bench/adapters/LogItAdapter.cpp

@@ -38,6 +38,8 @@ namespace logit_bench {
         void configure(const Scenario& scenario, LatencyRecorder& recorder) {
             m_async = scenario.async;
             m_sink = scenario.sink;
+            m_async_payload = scenario.async_payload;
+            m_async_payload_observer = scenario.async_payload_observer;
             m_recorder = &recorder;
 
             if (m_sink == SinkKind::File) {
@@ -60,6 +62,17 @@ namespace logit_bench {
             }
 
             if (m_sink == SinkKind::Null) {
+                if (m_async_payload == AsyncPayloadMode::FullMessage) {
+                    AsyncPayload payload;
+                    payload.slot_line = slot_line;
+                    payload.text = message;
+                    logit::detail::TaskExecutor::get_instance().add_task(
+                        [this, payload = std::move(payload)]() mutable {
+                            consume(payload.slot_line, payload.text);
+                        });
+                    return;
+                }
+
                 logit::detail::TaskExecutor::get_instance().add_task([this, slot_line]() {
                     consume(slot_line, std::string_view{});
                 });
@@ -121,6 +134,10 @@ namespace logit_bench {
             if (slot_line >= 0 && m_recorder) {
                 m_recorder->complete_slot(static_cast<std::uint64_t>(slot_line));
             }
+
+            if (m_async_payload_observer) {
+                m_async_payload_observer(text);
+            }
 
             if (m_sink == SinkKind::File) {
                 std::lock_guard<std::mutex> lock(m_file_mutex);
@@ -137,6 +154,8 @@ namespace logit_bench {
 
         bool m_async = false;
         SinkKind m_sink = SinkKind::Null;
+        AsyncPayloadMode m_async_payload = AsyncPayloadMode::MarkerOnly;
+        std::function<void(std::string_view)> m_async_payload_observer;
         LatencyRecorder* m_recorder = nullptr;
 
         std::ofstream m_file;

bench/async_payload_contract_test.cpp

@@ -0,0 +1,33 @@
+#include "adapters/LogItAdapter.hpp"
+
+#include <cstddef>
+#include <mutex>
+#include <string>
+#include <string_view>
+
+int main() {
+    using namespace logit_bench;
+
+    const std::string expected(200, 'P');
+    std::mutex observer_mutex;
+    std::string observed;
+
+    Scenario scenario;
+    scenario.async = true;
+    scenario.sink = SinkKind::Null;
+    scenario.async_payload = AsyncPayloadMode::FullMessage;
+    scenario.async_payload_observer = [&](std::string_view payload) {
+        std::lock_guard<std::mutex> lock(observer_mutex);
+        observed.assign(payload.data(), payload.size());
+    };
+
+    LatencyRecorder recorder(1);
+    LogItAdapter adapter;
+    adapter.prepare(scenario, recorder);
+    adapter.log(recorder.begin(true), expected);
+    adapter.flush();
+    recorder.wait_for_all();
+
+    std::lock_guard<std::mutex> lock(observer_mutex);
+    return observed == expected ? 0 : 1;
+}

bench/benchmark_validation_test.cpp

@@ -25,11 +25,22 @@ int main() {
     }
     if (!rejected_legacy_schema) return 2;
 
+    bool rejected_pre_contract_schema = false;
+    try {
+        validate_latency_csv_header(
+            "lib,async,sink,producers,msg_bytes,total,queue_capacity,"
+            "p50_ns,p99_ns,p999_ns,throughput");
+    } catch (const std::runtime_error&) {
+        rejected_pre_contract_schema = true;
+    }
+    if (!rejected_pre_contract_schema) return 3;
+
     validate_latency_csv_header(std::string(latency_csv_header()) + "\r");
     validate_latency_csv_header(latency_csv_header());
 
     const auto comparable = make_benchmark_metadata(
-        "8192", "block", "sink-entry", "all-prior-work-drained");
+        "8192", "block", "sink-entry", "all-prior-work-drained",
+        "test/workload");
     auto complete = comparable;
     complete.source_commit = "test-commit";
     complete.compiler = "test-compiler";
@@ -51,7 +62,7 @@ int main() {
     } catch (const std::runtime_error&) {
         rejected_unknown_metadata = true;
     }
-    if (!rejected_unknown_metadata) return 3;
+    if (!rejected_unknown_metadata) return 4;
 
     return 0;
 }