perf(runtime): coalesce external submission wakes

CHANGELOG.md

@@ -52,6 +52,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Coalesced external-submission wakes**: Each worker now issues at most one
+  `eventfd` notification for a burst of cross-thread task submissions. Before
+  blocking, the worker clears the pending-wake claim and rechecks both external
+  queues, preserving the unconditional-wake correctness guarantee without one
+  syscall per submitted task. Interrupted eventfd writes are retried before the
+  wake remains claimed (#1026).
 - **Owner-local continuation routing**: `scheduler::try_schedule()` now keeps an
   already-suspended coroutine on the current worker when that worker satisfies
   its affinity and active-I/O ownership constraints. Initial task admission

README.md

@@ -301,6 +301,8 @@ The scheduler manages a pool of worker threads, each with a local task queue. Ke
 - **Work stealing**: Idle threads steal tasks from busy threads
 - **Owner-local continuations**: eligible same-worker resumptions bypass the
   external inbox and wake path while remaining stealable
+- **Coalesced external wakes**: bursts targeting one worker share an outstanding
+  eventfd notification without sampling an idle flag or risking a lost wake
 - **Per-worker I/O context**: Pending operations stay pinned to the worker/backend generation that accepted them
 - **Dynamic sizing**: Adjust thread count at runtime
 - **Load balancing**: Automatic task distribution
@@ -519,7 +521,9 @@ Elio achieves competitive performance through careful optimization:
 
 ### Key Optimizations
 
-- **Unconditional Wake**: Cross-thread submissions always wake the target worker; eventfd deduplication prevents lost wakes
+- **Coalesced Submission Wake**: The first cross-thread submission claims an
+  eventfd wake; later submissions share it until the worker clears the claim
+  and rechecks its queues before blocking
 - **io_uring Batch Submit**: Automatic batching of I/O operations
 - **Queue-based Scheduling Fast Path**: bounded MPSC inbox (~5 ns) +
   Chase-Lev deque (~13 ns), with a locked overflow queue for rare sustained
@@ -547,7 +551,8 @@ Elio achieves competitive performance through careful optimization:
 cd build
 cmake --build . --target quick_benchmark microbench scheduler_service_benchmark io_benchmark benchmark scalability_test
 
-# Quick benchmark (spawn, context switch, yield, scheduler reschedule)
+# Quick benchmark (spawn, context switch, yield, scheduler reschedule,
+# external submission burst)
 ./examples/quick_benchmark
 
 # Microbenchmarks (individual operations)

examples/quick_benchmark.cpp

@@ -86,6 +86,15 @@ coro::task<void> measure_scheduler_reschedules(
     co_return;
 }
 
+coro::task<void> hold_worker_for_external_burst(
+    std::atomic<bool>* started, std::atomic<bool>* release) {
+    started->store(true, std::memory_order_release);
+    while (!release->load(std::memory_order_acquire)) {
+        std::this_thread::yield();
+    }
+    co_return;
+}
+
 // Time-based spawn overhead benchmark
 void benchmark_spawn_overhead() {
     const int batch_size = 10000;
@@ -271,6 +280,46 @@ void benchmark_scheduler_reschedule() {
               << ", max=" << stats.max << ")" << std::endl;
 }
 
+// Measure producer-side enqueue cost when many submissions target one busy
+// worker and therefore share one outstanding wake notification.
+void benchmark_external_submission_burst() {
+    constexpr int submissions_per_burst = 4000;
+    std::vector<double> samples;
+
+    const auto bench_start = steady_clock::now();
+    while (duration_cast<seconds>(steady_clock::now() - bench_start) <
+           MIN_BENCH_DURATION) {
+        runtime::scheduler sched(1);
+        sched.start();
+
+        std::atomic<bool> holder_started{false};
+        std::atomic<bool> release_holder{false};
+        sched.go_to(0, hold_worker_for_external_burst,
+                    &holder_started, &release_holder);
+        while (!holder_started.load(std::memory_order_acquire)) {
+            std::this_thread::yield();
+        }
+
+        const auto start = steady_clock::now();
+        for (int i = 0; i < submissions_per_burst; ++i) {
+            sched.spawn_to(0, std::noop_coroutine());
+        }
+        const auto elapsed = duration_cast<nanoseconds>(
+            steady_clock::now() - start).count();
+        samples.push_back(
+            static_cast<double>(elapsed) / submissions_per_burst);
+
+        release_holder.store(true, std::memory_order_release);
+        sched.shutdown();
+    }
+
+    const auto stats = bench_stats::compute(samples);
+    std::cout << "External submission burst: " << std::fixed
+              << std::setprecision(2) << stats.avg
+              << " ns/submit (min=" << stats.min
+              << ", max=" << stats.max << ")" << std::endl;
+}
+
 int main() {
     log::logger::instance().set_level(log::level::error);
 
@@ -281,6 +330,7 @@ int main() {
     benchmark_context_switch();
     benchmark_yield();
     benchmark_scheduler_reschedule();
+    benchmark_external_submission_burst();
 
     std::cout << "=== Done ===" << std::endl;
 

include/elio/io/epoll_backend.hpp

@@ -621,7 +621,10 @@ class epoll_backend : public io_backend {
 
     void notify() noexcept override {
         uint64_t val = 1;
-        ssize_t ret = ::write(wake_fd_, &val, sizeof(val));
+        ssize_t ret;
+        do {
+            ret = ::write(wake_fd_, &val, sizeof(val));
+        } while (ret < 0 && errno == EINTR);
         if (ret < 0 && errno != EAGAIN) {
             int error = errno;
             detail::run_noexcept([&]() {

include/elio/io/io_uring_backend.hpp

@@ -728,7 +728,10 @@ class io_uring_backend final : public io_backend {
 
     void notify() noexcept override {
         uint64_t val = 1;
-        ssize_t ret = ::write(wake_fd_, &val, sizeof(val));
+        ssize_t ret;
+        do {
+            ret = ::write(wake_fd_, &val, sizeof(val));
+        } while (ret < 0 && errno == EINTR);
         if (ret < 0 && errno != EAGAIN) {
             int error = errno;
             detail::run_noexcept([&]() {

include/elio/runtime/scheduler.hpp

@@ -1883,6 +1883,13 @@ inline void worker_thread::run() {
                 }
                 break;
             }
+            reset_submission_wake_before_poll();
+            if (has_external_submission()) {
+                continue;
+            }
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+            record_blocking_poll_for_test();
+#endif
             io_context_->poll(std::chrono::milliseconds(50));
             continue;
         }
@@ -2139,11 +2146,20 @@ inline void worker_thread::poll_io_when_idle() {
     // Mark as idle before any blocking so diagnostics can observe wait state.
     idle_.store(true, std::memory_order_release);
 
+    // A submit that observed the previous outstanding wake may have skipped
+    // its eventfd write. Clear that claim before the queue recheck: work
+    // published earlier is found below, while a later producer claims the
+    // clear state and wakes the poll.
+    reset_submission_wake_before_poll();
+    if (has_external_submission()) {
+        idle_.store(false, std::memory_order_relaxed);
+        return;
+    }
+
     // Optional spinning phase (if configured via wait_strategy)
     if (strategy_.spin_iterations > 0) {
         for (size_t i = 0; i < strategy_.spin_iterations; ++i) {
-            if (inbox_->size_approx() > 0 ||
-                overflow_size_.load(std::memory_order_acquire) > 0) {
+            if (has_external_submission()) {
                 idle_.store(false, std::memory_order_relaxed);
                 return;
             }
@@ -2157,6 +2173,9 @@ inline void worker_thread::poll_io_when_idle() {
 
     // Single unified wait: blocks on I/O backend (epoll/io_uring)
     // Both I/O completions AND task wake-ups (via eventfd) will unblock this
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+    record_blocking_poll_for_test();
+#endif
     io_context_->poll(std::chrono::milliseconds(idle_timeout_ms));
 
     // Clear idle flag after waking up

include/elio/runtime/worker_thread.hpp

@@ -25,6 +25,8 @@ inline std::atomic<bool> overflow_transfer_paused_for_test{false};
 inline std::atomic<bool> pause_queue_snapshot_for_test{false};
 inline std::atomic<bool> queue_snapshot_paused_for_test{false};
 inline std::atomic<bool> queue_transfer_waiting_for_test{false};
+inline std::atomic<bool> pause_before_submission_wake_reset_for_test{false};
+inline std::atomic<bool> submission_wake_reset_paused_for_test{false};
 inline std::atomic<io::io_context::backend_type> worker_io_backend_for_test{
     io::io_context::backend_type::auto_detect};
 }  // namespace detail
@@ -100,12 +102,7 @@ class worker_thread {
             if (!inbox_->push(handle.address())) {
                 return push_result::full;
             }
-            // Always wake on cross-thread submit. The eventfd dedupes via its
-            // counter — calling wake() on a busy worker just bumps the counter
-            // and is consumed in the next poll cycle. The previous "lazy wake"
-            // load on idle_ raced with the worker's relaxed idle_=false store,
-            // missing wakes and producing 10 ms tail latency on weak hardware.
-            wake();
+            wake_for_submission();
             return push_result::accepted;
         };
 
@@ -155,7 +152,7 @@ class worker_thread {
                 overflow_.push_back(handle.address());
                 overflow_size_.fetch_add(1, std::memory_order_release);
             }
-            wake();
+            wake_for_submission();
         } catch (...) {
             // The handle was not published if overflow allocation failed.
             return false;
@@ -175,7 +172,7 @@ class worker_thread {
             overflow_.push_back(handle.address());
             overflow_size_.fetch_add(1, std::memory_order_release);
         }
-        wake();
+        wake_for_submission();
         return true;
     }
 #endif
@@ -200,6 +197,20 @@ class worker_thread {
         return steals_executed_.load(std::memory_order_relaxed);
     }
 
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+    [[nodiscard]] size_t submission_wake_calls_for_test() const noexcept {
+        return submission_wake_calls_for_test_.load(std::memory_order_relaxed);
+    }
+
+    [[nodiscard]] size_t blocking_poll_calls_for_test() const noexcept {
+        return blocking_poll_calls_for_test_.load(std::memory_order_relaxed);
+    }
+
+    void record_blocking_poll_for_test() noexcept {
+        blocking_poll_calls_for_test_.fetch_add(1, std::memory_order_relaxed);
+    }
+#endif
+
     [[nodiscard]] size_t queue_size() const noexcept {
         std::unique_lock<std::mutex> transfer_lock(transfer_mutex_,
                                                    std::try_to_lock);
@@ -329,6 +340,47 @@ class worker_thread {
 
     void request_stop() noexcept;
 
+    void wake_for_submission() noexcept {
+        // Every producer performs a release RMW, including producers that
+        // share an existing wake. The owner's acquire exchange before poll
+        // therefore observes the whole release sequence before rechecking the
+        // MPSC inbox; a failed CAS would not publish skipped submissions.
+        if (submission_wake_pending_.exchange(
+                true, std::memory_order_acq_rel)) {
+            return;
+        }
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        submission_wake_calls_for_test_.fetch_add(
+            1, std::memory_order_relaxed);
+#endif
+        wake();
+    }
+
+    void reset_submission_wake_before_poll() noexcept {
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        if (detail::pause_before_submission_wake_reset_for_test.load(
+                std::memory_order_acquire)) {
+            detail::submission_wake_reset_paused_for_test.store(
+                true, std::memory_order_release);
+            detail::submission_wake_reset_paused_for_test.notify_all();
+            while (detail::pause_before_submission_wake_reset_for_test.load(
+                    std::memory_order_acquire)) {
+                detail::pause_before_submission_wake_reset_for_test.wait(
+                    true, std::memory_order_acquire);
+            }
+            detail::submission_wake_reset_paused_for_test.store(
+                false, std::memory_order_release);
+        }
+#endif
+        (void)submission_wake_pending_.exchange(
+            false, std::memory_order_acq_rel);
+    }
+
+    [[nodiscard]] bool has_external_submission() const noexcept {
+        return !inbox_->empty() ||
+            overflow_size_.load(std::memory_order_acquire) > 0;
+    }
+
     void leave_draining_mode() noexcept {
         draining_.store(false, std::memory_order_release);
     }
@@ -366,6 +418,16 @@ class worker_thread {
     size_t steals_executed_local_{0};
     std::atomic<size_t> steals_executed_{0};
 
+    // The first external submit in a busy/blocked interval owns the eventfd
+    // wake. The owner clears this before polling and then rechecks the queues,
+    // closing the clear-versus-block lost-wake window without sampling idle_.
+    alignas(64) std::atomic<bool> submission_wake_pending_{false};
+
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+    std::atomic<size_t> submission_wake_calls_for_test_{0};
+    std::atomic<size_t> blocking_poll_calls_for_test_{0};
+#endif
+
     // Idle flag (owner writes, other threads read) — isolated cache line
     alignas(64) std::atomic<bool> idle_{false};