From f6ac6d2eccd88a151515e7032ec72148b2cf2768 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Wed, 12 Aug 2026 16:27:50 +0800 Subject: [PATCH] perf(runtime): publish worker counters without atomic RMW --- CHANGELOG.md | 4 ++ README.md | 2 + examples/microbench.cpp | 41 ++++++++++++++++-- include/elio/runtime/scheduler.hpp | 12 +++--- include/elio/runtime/worker_thread.hpp | 18 ++++++-- tests/unit/test_scheduler.cpp | 57 ++++++++++++++++++++++++++ wiki/Performance-Tuning.md | 3 ++ 7 files changed, 124 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 871c81d4..5cf8b574 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **Single-writer worker metrics**: Coroutine-resume and successful-steal + counters now use owner-local increments with relaxed atomic snapshot stores, + preserving exact monotonic metric reads without atomic read-modify-write + instructions on worker hot paths (#1017). - **Quieter io_uring initialization**: per-context backend initialization and selection diagnostics now use DEBUG instead of the default INFO level (#1011). diff --git a/README.md b/README.md index 4207c2e4..5f2cfa37 100644 --- a/README.md +++ b/README.md @@ -516,6 +516,8 @@ Elio achieves competitive performance through careful optimization: - **Queue-based Scheduling Fast Path**: bounded MPSC inbox (~5 ns) + Chase-Lev deque (~13 ns), with a locked overflow queue for rare sustained bursts +- **Single-writer Metric Publication**: worker execution and steal counters use + owner-local increments with relaxed atomic snapshots for external readers ### Scalability diff --git a/examples/microbench.cpp b/examples/microbench.cpp index 63ee5777..2cec836e 100644 --- a/examples/microbench.cpp +++ b/examples/microbench.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include #include @@ -95,7 +96,39 @@ int main() { while (queue.pop()) {} } - // 5. Measure atomic fence alone + // 5. Compare atomic RMW with single-writer snapshot publication + { + std::atomic published{0}; + + auto start = high_resolution_clock::now(); + for (int i = 0; i < N; ++i) { + published.fetch_add(1, std::memory_order_relaxed); + } + auto end = high_resolution_clock::now(); + auto ns = duration_cast(end - start).count(); + + std::cout << "Atomic counter fetch_add: " + << (static_cast(ns) / N) + << " ns/update" << std::endl; + } + + { + size_t local = 0; + std::atomic published{0}; + + auto start = high_resolution_clock::now(); + for (int i = 0; i < N; ++i) { + published.store(++local, std::memory_order_relaxed); + } + auto end = high_resolution_clock::now(); + auto ns = duration_cast(end - start).count(); + + std::cout << "Single-writer counter publish: " + << (static_cast(ns) / N) + << " ns/update" << std::endl; + } + + // 6. Measure atomic fence alone { auto start = high_resolution_clock::now(); for (int i = 0; i < N; ++i) { @@ -107,7 +140,7 @@ int main() { std::cout << "Atomic release fence: " << (ns / N) << " ns" << std::endl; } - // 6. Measure eventfd write + // 7. Measure eventfd write { int fd = eventfd(0, EFD_NONBLOCK); uint64_t val = 1; @@ -123,7 +156,7 @@ int main() { close(fd); } - // 7. Full spawn path (with running scheduler) - includes alloc + spawn + // 8. Full spawn path (with running scheduler) - includes alloc + spawn { runtime::scheduler sched(4); sched.start(); @@ -145,7 +178,7 @@ int main() { sched.shutdown(); } - // 8. Measure warmed-up worker overhead + // 9. Measure warmed-up worker overhead { runtime::scheduler sched(4); sched.start(); diff --git a/include/elio/runtime/scheduler.hpp b/include/elio/runtime/scheduler.hpp index 77ca2556..66d5f6ae 100644 --- a/include/elio/runtime/scheduler.hpp +++ b/include/elio/runtime/scheduler.hpp @@ -1989,7 +1989,7 @@ inline void worker_thread::run_task(std::coroutine_handle<> handle) noexcept { auto* promise = coro::get_promise_base(handle.address()); coro::detail::frame_context_scope frame_scope(promise); handle.resume(); - tasks_executed_.fetch_add(1, std::memory_order_relaxed); + record_task_execution(); update_last_task_time(); // Note: We do NOT check done() or call destroy() here. @@ -2029,7 +2029,7 @@ inline std::coroutine_handle<> worker_thread::try_steal() noexcept { if (promise && promise->has_active_io_pin()) { if (promise->is_io_pin_owner( worker_id_, io_context_->generation())) { - steals_executed_.fetch_add(1, std::memory_order_relaxed); + record_successful_steal(); return handle; } @@ -2051,7 +2051,7 @@ inline std::coroutine_handle<> worker_thread::try_steal() noexcept { if (promise && promise->is_worker_local()) { if (affinity == worker_id_) { - steals_executed_.fetch_add(1, std::memory_order_relaxed); + record_successful_steal(); return handle; } if (affinity < num_workers) { @@ -2072,7 +2072,7 @@ inline std::coroutine_handle<> worker_thread::try_steal() noexcept { } if (affinity == coro::NO_AFFINITY || affinity == worker_id_) { - steals_executed_.fetch_add(1, std::memory_order_relaxed); + record_successful_steal(); return handle; } @@ -2084,14 +2084,14 @@ inline std::coroutine_handle<> worker_thread::try_steal() noexcept { if (promise) { promise->clear_affinity(); } - steals_executed_.fetch_add(1, std::memory_order_relaxed); + record_successful_steal(); return handle; } } else { if (promise) { promise->clear_affinity(); } - steals_executed_.fetch_add(1, std::memory_order_relaxed); + record_successful_steal(); return handle; } diff --git a/include/elio/runtime/worker_thread.hpp b/include/elio/runtime/worker_thread.hpp index 45081712..746d14f4 100644 --- a/include/elio/runtime/worker_thread.hpp +++ b/include/elio/runtime/worker_thread.hpp @@ -49,7 +49,6 @@ class worker_thread { , queue_(std::make_unique>()) , inbox_(std::make_unique>()) , running_(false) - , tasks_executed_(0) , strategy_(strategy) #ifdef ELIO_RUNTIME_TEST_HOOKS , io_context_(io::io_context::make_worker_owned( @@ -304,6 +303,16 @@ class worker_thread { } private: + void record_task_execution() noexcept { + tasks_executed_.store(++tasks_executed_local_, + std::memory_order_relaxed); + } + + void record_successful_steal() noexcept { + steals_executed_.store(++steals_executed_local_, + std::memory_order_relaxed); + } + void request_stop() noexcept; void leave_draining_mode() noexcept { @@ -336,8 +345,11 @@ class worker_thread { // overcount work, but it must never let idle detection miss accepted work. std::atomic overflow_size_{0}; std::atomic draining_{false}; - // Hot-write fields (owner thread writes per task) — isolated cache line - alignas(64) std::atomic tasks_executed_; + // Only the owner mutates the local counters. Relaxed atomic stores publish + // exact monotonic snapshots without a read-modify-write instruction. + alignas(64) size_t tasks_executed_local_{0}; + std::atomic tasks_executed_{0}; + size_t steals_executed_local_{0}; std::atomic steals_executed_{0}; // Idle flag (owner writes, other threads read) — isolated cache line diff --git a/tests/unit/test_scheduler.cpp b/tests/unit/test_scheduler.cpp index b2870b0c..02b36782 100644 --- a/tests/unit/test_scheduler.cpp +++ b/tests/unit/test_scheduler.cpp @@ -37,6 +37,20 @@ task increment_counter_task(std::atomic* counter) { co_return; } +task yield_for_metric_reads(size_t sample_count, + std::atomic* sample_ready, + std::atomic* sample_acknowledged, + std::atomic* completed) { + for (size_t sample = 1; sample <= sample_count; ++sample) { + co_await yield(); + sample_ready->store(sample, std::memory_order_release); + while (sample_acknowledged->load(std::memory_order_acquire) < sample) { + co_await yield(); + } + } + completed->store(true, std::memory_order_release); +} + task empty_task() { co_return; } @@ -714,6 +728,49 @@ TEST_CASE("Scheduler statistics", "[scheduler]") { sched.shutdown(); } +TEST_CASE("Scheduler task counter remains monotonic during concurrent reads", + "[scheduler][metrics]") { + constexpr size_t sample_count = 128; + scheduler sched(1); + sched.start(); + + std::atomic sample_ready{0}; + std::atomic sample_acknowledged{0}; + std::atomic completed{false}; + const auto initial = sched.worker_tasks_executed(0); + sched.go_to(0, yield_for_metric_reads, sample_count, &sample_ready, + &sample_acknowledged, &completed); + + const auto deadline = std::chrono::steady_clock::now() + scaled_sec(5); + size_t previous = initial; + bool monotonic = true; + size_t active_samples = 0; + for (size_t sample = 1; sample <= sample_count; ++sample) { + while (sample_ready.load(std::memory_order_acquire) < sample && + std::chrono::steady_clock::now() < deadline) { + std::this_thread::yield(); + } + REQUIRE(sample_ready.load(std::memory_order_acquire) >= sample); + + const auto current = sched.worker_tasks_executed(0); + monotonic = monotonic && current >= previous; + previous = current; + active_samples += + !completed.load(std::memory_order_acquire) ? 1U : 0U; + sample_acknowledged.store(sample, std::memory_order_release); + } + + while (!completed.load(std::memory_order_acquire) && + std::chrono::steady_clock::now() < deadline) { + std::this_thread::yield(); + } + REQUIRE(completed.load(std::memory_order_acquire)); + REQUIRE(sched.shutdown(scaled_sec(5))); + CHECK(monotonic); + CHECK(active_samples == sample_count); + CHECK(sched.worker_tasks_executed(0) >= initial + sample_count); +} + TEST_CASE("Scheduler thread-local current", "[scheduler]") { scheduler sched(2); REQUIRE(scheduler::current() == nullptr); diff --git a/wiki/Performance-Tuning.md b/wiki/Performance-Tuning.md index 579464bf..9ec8d145 100644 --- a/wiki/Performance-Tuning.md +++ b/wiki/Performance-Tuning.md @@ -477,6 +477,9 @@ size_t threads = sched.num_threads(); // Current thread count ``` These are lightweight atomic reads suitable for periodic monitoring in production. Combine with `set_thread_count` to implement your own adaptive scaling. +Workers publish execution and successful-steal counts from single-writer local +counters, so metric collection does not require atomic read-modify-write +instructions on coroutine resume or steal paths. ### Logging Overhead