From 63b18cd5e087f10dec6da4347ab1ea1858648484 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 7 Oct 2026 18:13:08 +0300 Subject: [PATCH 1/2] test(bench): compare executor policy bookkeeping Add a bounded Block versus DropNewest TaskExecutor batch probe with an oversized queue and zero-drop validation. Document it as a policy-branch signal rather than an attribution of individual private synchronization operations. --- bench/task_ownership_profile.cpp | 23 ++++++++++++++++++++++- docs/benchmarks.md | 5 ++++- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/bench/task_ownership_profile.cpp b/bench/task_ownership_profile.cpp index 2721976..3611d4d 100644 --- a/bench/task_ownership_profile.cpp +++ b/bench/task_ownership_profile.cpp @@ -110,7 +110,7 @@ double measure_ring_prebuilt_std_function(std::size_t warmup, worker.join(); if (consumed.load(std::memory_order_relaxed) != count) { std::cerr << "prebuilt MPSC ring completion mismatch\n"; - std::exit(6); + std::exit(7); } return static_cast(elapsed); }; @@ -253,6 +253,27 @@ int main() { std::cout << "case=taskexecutor_batch_prebuilt ns_per_call=" << executor_batch_ns << '\n'; + executor.set_queue_policy(logit::QueuePolicy::DropNewest); + executor.reset_dropped_tasks(); + const double executor_drop_newest_ns = measure( + [&executor, &prebuilt_task](std::size_t count) { + for (std::size_t i = 0; i < count; ++i) { + executor.add_task(prebuilt_task); + } + }, + [&executor](std::size_t expected) { + executor.wait(); + if (g_completed.load(std::memory_order_relaxed) != expected || + executor.dropped_tasks() != 0) { + std::cerr << "TaskExecutor DropNewest completion mismatch\n"; + std::exit(6); + } + g_completed.store(0, std::memory_order_relaxed); + }, warmup, total, repeats); + std::cout << "case=taskexecutor_batch_drop_newest_prebuilt ns_per_call=" + << executor_drop_newest_ns << '\n'; + executor.set_queue_policy(logit::QueuePolicy::Block); + const double executor_roundtrip_ns = measure( [&executor, &prebuilt_task](std::size_t count) { for (std::size_t i = 0; i < count; ++i) { diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 3accc55..a1234f9 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -322,7 +322,10 @@ Configure it with `LOGIT_TASK_OWNERSHIP_TOTAL`, `LOGIT_TASK_OWNERSHIP_WARMUP`, and `LOGIT_TASK_OWNERSHIP_REPEATS`. The round-trip case defaults to at most 1,000 samples because each sample waits for worker completion; use `LOGIT_TASK_OWNERSHIP_ROUNDTRIP_TOTAL` to override -that independent sample count. +that independent sample count. The `Block` and `DropNewest` batch rows use the +same oversized queue and verify zero drops; their difference is a controlled +policy-branch signal (not an attribution of one private atomic or notification +operation). The flush regression target uses an intentionally delayed asynchronous sink and asserts that `flush()` does not return before every queued message has reached From 180e8485c886a42f964dc8d4ab3e4166c5815dea Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 7 Oct 2026 21:17:59 +0300 Subject: [PATCH 2/2] test(bench): pair executor policy attribution Interleave Block and DropNewest policy cases to reduce temporal bias and drain between every case. Validate completion and zero drops symmetrically, and document backend-specific policy semantics for the attribution probe. --- bench/task_ownership_profile.cpp | 99 +++++++++++++++++++++----------- docs/benchmarks.md | 13 ++++- 2 files changed, 75 insertions(+), 37 deletions(-) diff --git a/bench/task_ownership_profile.cpp b/bench/task_ownership_profile.cpp index 3611d4d..148afb4 100644 --- a/bench/task_ownership_profile.cpp +++ b/bench/task_ownership_profile.cpp @@ -62,6 +62,67 @@ double measure(Loop loop, Cleanup cleanup, std::size_t warmup, static_cast(total); } +struct PolicyPairResult { + double block_ns = 0.0; + double drop_newest_ns = 0.0; +}; + +PolicyPairResult measure_executor_policy_pair( + logit::detail::TaskExecutor& executor, + const std::function& prebuilt_task, + std::size_t warmup, + std::size_t total, + std::size_t repeats) { + auto run_case = [&](logit::QueuePolicy policy, std::size_t count) { + executor.set_queue_policy(policy); + executor.reset_dropped_tasks(); + g_completed.store(0, std::memory_order_relaxed); + + const auto start = std::chrono::steady_clock::now(); + for (std::size_t i = 0; i < count; ++i) { + executor.add_task(prebuilt_task); + } + const auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start).count(); + + executor.wait(); + if (g_completed.load(std::memory_order_relaxed) != count || + executor.dropped_tasks() != 0) { + std::cerr << "TaskExecutor policy completion mismatch\n"; + std::exit(6); + } + g_completed.store(0, std::memory_order_relaxed); + return static_cast(elapsed); + }; + + run_case(logit::QueuePolicy::Block, warmup); + run_case(logit::QueuePolicy::DropNewest, warmup); + + std::vector block_samples; + std::vector drop_samples; + block_samples.reserve(repeats); + drop_samples.reserve(repeats); + for (std::size_t repeat = 0; repeat < repeats; ++repeat) { + if ((repeat % 2) == 0) { + block_samples.push_back(run_case(logit::QueuePolicy::Block, total)); + drop_samples.push_back(run_case(logit::QueuePolicy::DropNewest, total)); + } else { + drop_samples.push_back(run_case(logit::QueuePolicy::DropNewest, total)); + block_samples.push_back(run_case(logit::QueuePolicy::Block, total)); + } + } + + std::sort(block_samples.begin(), block_samples.end()); + std::sort(drop_samples.begin(), drop_samples.end()); + PolicyPairResult result; + result.block_ns = static_cast(block_samples[block_samples.size() / 2]) / + static_cast(total); + result.drop_newest_ns = + static_cast(drop_samples[drop_samples.size() / 2]) / + static_cast(total); + return result; +} + #if defined(LOGIT_USE_MPSC_RING) double measure_ring_prebuilt_std_function(std::size_t warmup, std::size_t total, @@ -236,42 +297,12 @@ int main() { << function_copy_ns << '\n'; g_completed.store(0, std::memory_order_relaxed); - const double executor_batch_ns = measure( - [&executor, &prebuilt_task](std::size_t count) { - for (std::size_t i = 0; i < count; ++i) { - executor.add_task(prebuilt_task); - } - }, - [&executor](std::size_t expected) { - executor.wait(); - if (g_completed.load(std::memory_order_relaxed) != expected) { - std::cerr << "TaskExecutor batch completion mismatch\n"; - std::exit(3); - } - g_completed.store(0, std::memory_order_relaxed); - }, warmup, total, repeats); + const auto policy_result = measure_executor_policy_pair( + executor, prebuilt_task, warmup, total, repeats); std::cout << "case=taskexecutor_batch_prebuilt ns_per_call=" - << executor_batch_ns << '\n'; - - executor.set_queue_policy(logit::QueuePolicy::DropNewest); - executor.reset_dropped_tasks(); - const double executor_drop_newest_ns = measure( - [&executor, &prebuilt_task](std::size_t count) { - for (std::size_t i = 0; i < count; ++i) { - executor.add_task(prebuilt_task); - } - }, - [&executor](std::size_t expected) { - executor.wait(); - if (g_completed.load(std::memory_order_relaxed) != expected || - executor.dropped_tasks() != 0) { - std::cerr << "TaskExecutor DropNewest completion mismatch\n"; - std::exit(6); - } - g_completed.store(0, std::memory_order_relaxed); - }, warmup, total, repeats); + << policy_result.block_ns << '\n'; std::cout << "case=taskexecutor_batch_drop_newest_prebuilt ns_per_call=" - << executor_drop_newest_ns << '\n'; + << policy_result.drop_newest_ns << '\n'; executor.set_queue_policy(logit::QueuePolicy::Block); const double executor_roundtrip_ns = measure( diff --git a/docs/benchmarks.md b/docs/benchmarks.md index a1234f9..62c9817 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -323,9 +323,16 @@ Configure it with `LOGIT_TASK_OWNERSHIP_TOTAL`, round-trip case defaults to at most 1,000 samples because each sample waits for worker completion; use `LOGIT_TASK_OWNERSHIP_ROUNDTRIP_TOTAL` to override that independent sample count. The `Block` and `DropNewest` batch rows use the -same oversized queue and verify zero drops; their difference is a controlled -policy-branch signal (not an attribution of one private atomic or notification -operation). +same oversized queue and verify zero drops. Cases are paired and interleaved +(with the order reversed on alternate repeats), and each case drains the +executor before the next case. For the `mpsc_ring` backend, `Block` evaluates +the additional `m_active_tasks` backpressure check while `DropNewest` +short-circuits that Block-only check. Their difference is therefore a +controlled policy/backpressure signal, not the exact cost of one private +atomic, branch, or notification operation. For the `mutex_deque` backend, +this workload does not reach the full queue condition, so switching policies +is a control/equivalence case rather than an exercised branch difference. Both +policies use symmetric zero-drop validation. The flush regression target uses an intentionally delayed asynchronous sink and asserts that `flush()` does not return before every queued message has reached