perf(runtime): publish worker counters without atomic RMW

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

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
 

examples/microbench.cpp

@@ -1,6 +1,7 @@
 #include <elio/runtime/scheduler.hpp>
 #include <elio/coro/task.hpp>
 #include <elio/log/macros.hpp>
+#include <atomic>
 #include <iostream>
 #include <chrono>
 #include <vector>
@@ -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<size_t> 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<nanoseconds>(end - start).count();
+
+        std::cout << "Atomic counter fetch_add: "
+                  << (static_cast<double>(ns) / N)
+                  << " ns/update" << std::endl;
+    }
+
+    {
+        size_t local = 0;
+        std::atomic<size_t> 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<nanoseconds>(end - start).count();
+
+        std::cout << "Single-writer counter publish: "
+                  << (static_cast<double>(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();

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;
             }
 

include/elio/runtime/worker_thread.hpp

@@ -49,7 +49,6 @@ class worker_thread {
         , queue_(std::make_unique<chase_lev_deque<void>>())
         , inbox_(std::make_unique<mpsc_queue<void>>())
         , 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<size_t> overflow_size_{0};
     std::atomic<bool> draining_{false};
-    // Hot-write fields (owner thread writes per task) — isolated cache line
-    alignas(64) std::atomic<size_t> 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<size_t> tasks_executed_{0};
+    size_t steals_executed_local_{0};
     std::atomic<size_t> steals_executed_{0};
 
     // Idle flag (owner writes, other threads read) — isolated cache line

tests/unit/test_scheduler.cpp

@@ -37,6 +37,20 @@ task<void> increment_counter_task(std::atomic<int>* counter) {
     co_return;
 }
 
+task<void> yield_for_metric_reads(size_t sample_count,
+                                  std::atomic<size_t>* sample_ready,
+                                  std::atomic<size_t>* sample_acknowledged,
+                                  std::atomic<bool>* 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<void> 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<size_t> sample_ready{0};
+    std::atomic<size_t> sample_acknowledged{0};
+    std::atomic<bool> 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);

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