perf(runtime): sample worker progress for block detection

CHANGELOG.md

@@ -52,6 +52,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Sampled worker block detection**: The autoscaler now detects non-idle
+  workers with no completed-resume progress by sampling their execution
+  counters at `tick_interval`, removing the unconditional clock read and atomic
+  timestamp publication from every coroutine resume. `on_block` remains a
+  best-effort diagnostic whose notification can trail `block_threshold` by a
+  sampling interval. `worker_thread::enable_task_time_tracking()` provides an
+  explicit opt-in to exact timestamps; `last_task_time()` retains the implicit
+  opt-in for backward compatibility (#1019).
 - **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

README.md

@@ -407,6 +407,12 @@ size_t executed = sched.total_tasks_executed();
 sched.shutdown();
 ```
 
+Scheduler execution metrics are published from worker-local counters. Optional
+autoscaler `on_block` detection samples those counters at `tick_interval`, so
+ordinary coroutine resumes do not read the clock. Block notifications are
+best-effort and can arrive up to roughly one sampling interval after
+`block_threshold`.
+
 ### Task Spawning
 
 Elio provides flexible ways to spawn concurrent tasks:

examples/microbench.cpp

@@ -128,7 +128,42 @@ int main() {
                   << " ns/update" << std::endl;
     }
 
-    // 6. Measure atomic fence alone
+    // 6. Compare exact timestamps with the disabled diagnostic fast path
+    {
+        std::atomic<steady_clock::time_point> last_task_time{
+            steady_clock::now()};
+
+        auto start = high_resolution_clock::now();
+        for (int i = 0; i < N; ++i) {
+            last_task_time.store(steady_clock::now(),
+                                 std::memory_order_relaxed);
+        }
+        auto end = high_resolution_clock::now();
+        auto ns = duration_cast<nanoseconds>(end - start).count();
+
+        std::cout << "Exact task timestamp publish: "
+                  << (static_cast<double>(ns) / N)
+                  << " ns/update" << std::endl;
+    }
+
+    {
+        std::atomic<bool> track_task_time{false};
+
+        auto start = high_resolution_clock::now();
+        for (int i = 0; i < N; ++i) {
+            if (track_task_time.load(std::memory_order_relaxed)) {
+                std::atomic_signal_fence(std::memory_order_seq_cst);
+            }
+        }
+        auto end = high_resolution_clock::now();
+        auto ns = duration_cast<nanoseconds>(end - start).count();
+
+        std::cout << "Disabled task timestamp check: "
+                  << (static_cast<double>(ns) / N)
+                  << " ns/update" << std::endl;
+    }
+
+    // 7. Measure atomic fence alone
     {
         auto start = high_resolution_clock::now();
         for (int i = 0; i < N; ++i) {
@@ -140,7 +175,7 @@ int main() {
         std::cout << "Atomic release fence: " << (ns / N) << " ns" << std::endl;
     }
 
-    // 7. Measure eventfd write
+    // 8. Measure eventfd write
     {
         int fd = eventfd(0, EFD_NONBLOCK);
         uint64_t val = 1;
@@ -156,7 +191,7 @@ int main() {
         close(fd);
     }
 
-    // 8. Full spawn path (with running scheduler) - includes alloc + spawn
+    // 9. Full spawn path (with running scheduler) - includes alloc + spawn
     {
         runtime::scheduler sched(4);
         sched.start();
@@ -178,7 +213,7 @@ int main() {
         sched.shutdown();
     }
 
-    // 9. Measure warmed-up worker overhead
+    // 10. Measure warmed-up worker overhead
     {
         runtime::scheduler sched(4);
         sched.start();

include/elio/runtime/autoscaler.hpp

@@ -6,10 +6,85 @@
 #include <thread>
 #include <atomic>
 #include <mutex>
+#include <optional>
 #include <stop_token>
+#include <vector>
 
 namespace elio::runtime {
 
+namespace detail {
+
+class worker_progress_observation {
+public:
+    using clock = std::chrono::steady_clock;
+
+    template<typename Worker>
+    [[nodiscard]] std::optional<clock::duration> sample(
+        Worker* worker, clock::time_point now) noexcept {
+        if (!worker) {
+            reset();
+            return std::nullopt;
+        }
+
+        if constexpr (requires { worker->is_running(); }) {
+            if (!worker->is_running()) {
+                reset();
+                return std::nullopt;
+            }
+        }
+
+        if constexpr (requires { worker->tasks_executed(); }) {
+            const size_t tasks_executed = worker->tasks_executed();
+            const void* identity = static_cast<const void*>(worker);
+            if (worker->is_idle()) {
+                // Keep the latest identity and counter snapshot so the next
+                // active sample can classify the transition without stale data.
+                worker_ = identity;
+                tasks_executed_ = tasks_executed;
+                last_progress_ = now;
+                active_ = false;
+                return std::nullopt;
+            }
+
+            if (worker_ != identity || !active_ ||
+                tasks_executed_ != tasks_executed) {
+                worker_ = identity;
+                tasks_executed_ = tasks_executed;
+                last_progress_ = now;
+                active_ = true;
+                return clock::duration::zero();
+            }
+
+            return now - last_progress_;
+        } else {
+            if (worker->is_idle()) {
+                reset();
+                return std::nullopt;
+            }
+            // Preserve compatibility with custom Scheduler worker types that
+            // implemented the original timestamp-based on_block contract. A
+            // custom last_task_time() may intentionally enable timestamp
+            // collection as part of that pre-existing observation contract.
+            return now - worker->last_task_time();
+        }
+    }
+
+    void reset() noexcept {
+        worker_ = nullptr;
+        tasks_executed_ = 0;
+        last_progress_ = {};
+        active_ = false;
+    }
+
+private:
+    const void* worker_{nullptr};
+    size_t tasks_executed_{0};
+    clock::time_point last_progress_{};
+    bool active_{false};
+};
+
+}  // namespace detail
+
 // Main autoscaler class
 template<typename Scheduler, typename... Triggers>
 class autoscaler_impl {
@@ -26,6 +101,7 @@ class autoscaler_impl {
     void start(Scheduler* sched) {
         bool expected = false;
         if (!running_.compare_exchange_strong(expected, true)) return;
+        block_observations_.clear();
         scheduler_ = sched;
         thread_ = std::jthread([this](std::stop_token st) { run(st); });
     }
@@ -189,17 +265,18 @@ class autoscaler_impl {
         (void)pending;
         (void)last_idle_time;
 
+        // Shrinking discards observations for removed workers. A later worker
+        // at the same index therefore starts with a fresh progress baseline.
+        block_observations_.resize(num_workers);
         for (size_t i = 0; i < num_workers; ++i) {
             auto* worker = scheduler_->get_worker(i);
-            if (worker && !worker->is_idle()) {
-                auto last_time = worker->last_task_time();
-                auto blocked_duration = now - last_time;
-                if (blocked_duration > cfg.block_threshold) {
-                    // Execute block actions
-                    if constexpr (sizeof...(Actions) > 0) {
-                        auto blocked_ms = std::chrono::duration_cast<std::chrono::milliseconds>(blocked_duration);
-                        execute_block_actions<Actions...>(i, blocked_ms);
-                    }
+            auto blocked_duration = block_observations_[i].sample(worker, now);
+            if (blocked_duration && *blocked_duration > cfg.block_threshold) {
+                if constexpr (sizeof...(Actions) > 0) {
+                    auto blocked_ms =
+                        std::chrono::duration_cast<std::chrono::milliseconds>(
+                            *blocked_duration);
+                    execute_block_actions<Actions...>(i, blocked_ms);
                 }
             }
         }
@@ -299,6 +376,7 @@ class autoscaler_impl {
     std::atomic<bool> running_;
     Scheduler* scheduler_;
     std::jthread thread_;
+    std::vector<detail::worker_progress_observation> block_observations_;
 };
 
 // Convenience type alias

include/elio/runtime/scheduler.hpp

@@ -1990,7 +1990,6 @@ inline void worker_thread::run_task(std::coroutine_handle<> handle) noexcept {
     coro::detail::frame_context_scope frame_scope(promise);
     handle.resume();
     record_task_execution();
-    update_last_task_time();
 
     // Note: We do NOT check done() or call destroy() here.
     // If the task completed, its final_suspend will self-destruct (fire-and-forget)

include/elio/runtime/worker_thread.hpp

@@ -258,8 +258,19 @@ class worker_thread {
         return idle_.load(std::memory_order_relaxed);
     }
 
-    /// Get the last time a task was executed
-    [[nodiscard]] std::chrono::steady_clock::time_point last_task_time() const noexcept {
+    /// Enable exact task-time diagnostics for this worker.
+    void enable_task_time_tracking() noexcept {
+        track_task_time_.store(true, std::memory_order_relaxed);
+    }
+
+    /// Get the last recorded execution time.
+    /// This also enables tracking to preserve the original observation API.
+    [[nodiscard]] std::chrono::steady_clock::time_point
+    last_task_time() const noexcept {
+        // Enabling observational instrumentation is logically const. Keep this
+        // compatibility side effect here while exposing the explicit mutator
+        // above for call sites that want their instrumentation intent visible.
+        track_task_time_.store(true, std::memory_order_relaxed);
         return last_task_time_.load(std::memory_order_relaxed);
     }
 
@@ -306,6 +317,9 @@ class worker_thread {
     void record_task_execution() noexcept {
         tasks_executed_.store(++tasks_executed_local_,
                               std::memory_order_relaxed);
+        if (track_task_time_.load(std::memory_order_relaxed)) [[unlikely]] {
+            update_last_task_time();
+        }
     }
 
     void record_successful_steal() noexcept {
@@ -355,8 +369,12 @@ class worker_thread {
     // Idle flag (owner writes, other threads read) — isolated cache line
     alignas(64) std::atomic<bool> idle_{false};
 
-    // Slow-update fields — isolated cache line
-    alignas(64) std::atomic<std::chrono::steady_clock::time_point> last_task_time_{std::chrono::steady_clock::now()};
+    // Diagnostic timestamps are opt-in so normal resumes avoid clock reads.
+    // The mutable flag supports the legacy const observation entry point.
+    alignas(64) mutable std::atomic<bool> track_task_time_{false};
+    // Isolate diagnostic readers from the flag checked on every resume.
+    alignas(64) std::atomic<std::chrono::steady_clock::time_point>
+        last_task_time_{std::chrono::steady_clock::now()};
     bool needs_sync_ = false;          // Whether current task needs memory synchronization
     wait_strategy strategy_;           // Configurable wait strategy
     std::unique_ptr<io::io_context> io_context_;  // Per-worker io_context