diff --git a/CHANGELOG.md b/CHANGELOG.md index 3928fc06..b00eba25 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 7ae92d03..b433f4f4 100644 --- a/README.md +++ b/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) diff --git a/examples/quick_benchmark.cpp b/examples/quick_benchmark.cpp index e7d06f3b..cd8efa80 100644 --- a/examples/quick_benchmark.cpp +++ b/examples/quick_benchmark.cpp @@ -86,6 +86,15 @@ coro::task measure_scheduler_reschedules( co_return; } +coro::task hold_worker_for_external_burst( + std::atomic* started, std::atomic* 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 samples; + + const auto bench_start = steady_clock::now(); + while (duration_cast(steady_clock::now() - bench_start) < + MIN_BENCH_DURATION) { + runtime::scheduler sched(1); + sched.start(); + + std::atomic holder_started{false}; + std::atomic 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( + steady_clock::now() - start).count(); + samples.push_back( + static_cast(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; diff --git a/include/elio/io/epoll_backend.hpp b/include/elio/io/epoll_backend.hpp index 96433d80..7fc53168 100644 --- a/include/elio/io/epoll_backend.hpp +++ b/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([&]() { diff --git a/include/elio/io/io_uring_backend.hpp b/include/elio/io/io_uring_backend.hpp index 9a073647..9e854697 100644 --- a/include/elio/io/io_uring_backend.hpp +++ b/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([&]() { diff --git a/include/elio/runtime/scheduler.hpp b/include/elio/runtime/scheduler.hpp index f4ede646..413ddb24 100644 --- a/include/elio/runtime/scheduler.hpp +++ b/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 diff --git a/include/elio/runtime/worker_thread.hpp b/include/elio/runtime/worker_thread.hpp index cacee3e3..fc5e87ae 100644 --- a/include/elio/runtime/worker_thread.hpp +++ b/include/elio/runtime/worker_thread.hpp @@ -25,6 +25,8 @@ inline std::atomic overflow_transfer_paused_for_test{false}; inline std::atomic pause_queue_snapshot_for_test{false}; inline std::atomic queue_snapshot_paused_for_test{false}; inline std::atomic queue_transfer_waiting_for_test{false}; +inline std::atomic pause_before_submission_wake_reset_for_test{false}; +inline std::atomic submission_wake_reset_paused_for_test{false}; inline std::atomic 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 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 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 submission_wake_pending_{false}; + +#ifdef ELIO_RUNTIME_TEST_HOOKS + std::atomic submission_wake_calls_for_test_{0}; + std::atomic blocking_poll_calls_for_test_{0}; +#endif + // Idle flag (owner writes, other threads read) — isolated cache line alignas(64) std::atomic idle_{false}; diff --git a/tests/unit/test_scheduler.cpp b/tests/unit/test_scheduler.cpp index c698e15d..59cfbfa4 100644 --- a/tests/unit/test_scheduler.cpp +++ b/tests/unit/test_scheduler.cpp @@ -27,6 +27,25 @@ using namespace elio::test; // Standalone task functions to avoid lambda capture lifetime issues namespace { +class worker_io_backend_guard { +public: + explicit worker_io_backend_guard( + elio::io::io_context::backend_type backend) + : previous_(elio::runtime::detail::worker_io_backend_for_test.exchange( + backend, std::memory_order_acq_rel)) {} + + worker_io_backend_guard(const worker_io_backend_guard&) = delete; + worker_io_backend_guard& operator=(const worker_io_backend_guard&) = delete; + + ~worker_io_backend_guard() { + elio::runtime::detail::worker_io_backend_for_test.store( + previous_, std::memory_order_release); + } + +private: + elio::io::io_context::backend_type previous_; +}; + task set_executed_task(std::atomic* executed) { executed->store(true); co_return; @@ -37,6 +56,159 @@ task increment_counter_task(std::atomic* counter) { co_return; } +task hold_worker_for_submission_burst( + std::atomic* started, std::atomic* release) { + started->store(true, std::memory_order_release); + while (!release->load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + co_return; +} + +task arm_submission_wake_reset_pause(std::atomic* armed) { + elio::runtime::detail::pause_before_submission_wake_reset_for_test.store( + true, std::memory_order_release); + armed->store(true, std::memory_order_release); + co_return; +} + +task record_blocking_poll_count( + std::atomic* poll_calls_seen, + std::atomic* completed) { + auto* worker = worker_thread::current(); + poll_calls_seen->store(worker ? worker->blocking_poll_calls_for_test() : 0, + std::memory_order_relaxed); + completed->store(true, std::memory_order_release); + co_return; +} + +template +bool wait_for_scheduler_condition(Predicate&& predicate); + +void check_submission_wake_burst( + elio::io::io_context::backend_type backend) { + worker_io_backend_guard backend_guard(backend); + + scheduler sched(1); + sched.start(); + auto* worker = sched.get_worker(0); + REQUIRE(worker != nullptr); + const auto wake_calls_before = worker->submission_wake_calls_for_test(); + + std::atomic holder_started{false}; + std::atomic release_holder{false}; + std::atomic completed{0}; + sched.go_to(0, hold_worker_for_submission_burst, + &holder_started, &release_holder); + + const bool started = wait_for_scheduler_condition([&] { + return holder_started.load(std::memory_order_acquire); + }); + const auto wake_calls_before_burst = + worker->submission_wake_calls_for_test(); + + constexpr int burst_size = 128; + if (started) { + for (int i = 0; i < burst_size; ++i) { + sched.go_to(0, increment_counter_task, &completed); + } + } + const auto wake_calls_after_burst = + worker->submission_wake_calls_for_test(); + release_holder.store(true, std::memory_order_release); + + const bool burst_completed = wait_for_scheduler_condition([&] { + return completed.load(std::memory_order_acquire) == burst_size; + }); + const bool stopped = sched.shutdown(scaled_sec(5)); + + REQUIRE(started); + REQUIRE(burst_completed); + REQUIRE(stopped); + CHECK(wake_calls_before_burst == wake_calls_before + 1); + // The worker may clear the holder's claim immediately before resuming it, + // so the burst can require one fresh wake, but never one wake per task. + CHECK(wake_calls_after_burst >= wake_calls_before_burst); + CHECK(wake_calls_after_burst <= wake_calls_before_burst + 1); +} + +void check_skipped_submission_rechecked_before_blocking( + elio::io::io_context::backend_type backend) { + worker_io_backend_guard backend_guard(backend); + elio::runtime::detail::pause_before_submission_wake_reset_for_test.store( + false, std::memory_order_release); + elio::runtime::detail::submission_wake_reset_paused_for_test.store( + false, std::memory_order_release); + + scheduler sched(1); + sched.start(); + auto* worker = sched.get_worker(0); + REQUIRE(worker != nullptr); + + // Wait until the worker has cleared its claim and entered a blocking poll. + // The holder's wake then remains outstanding while it runs, so both the + // pause task and marker deterministically share that claim. + std::atomic holder_started{false}; + std::atomic release_holder{false}; + std::atomic pause_armed{false}; + std::atomic marker_completed{false}; + std::atomic marker_poll_calls{0}; + const auto poll_calls_before_holder = + worker->blocking_poll_calls_for_test(); + const bool worker_blocking = wait_for_scheduler_condition([&] { + return worker->blocking_poll_calls_for_test() > + poll_calls_before_holder; + }); + const auto wake_calls_before = worker->submission_wake_calls_for_test(); + if (worker_blocking) { + sched.go_to(0, hold_worker_for_submission_burst, + &holder_started, &release_holder); + } + const bool holder_running = wait_for_scheduler_condition([&] { + return holder_started.load(std::memory_order_acquire); + }); + if (holder_running) { + sched.go_to(0, arm_submission_wake_reset_pause, &pause_armed); + } + const auto wake_calls_after_arm = + worker->submission_wake_calls_for_test(); + release_holder.store(true, std::memory_order_release); + + const bool reset_paused = wait_for_scheduler_condition([&] { + return pause_armed.load(std::memory_order_acquire) && + elio::runtime::detail::submission_wake_reset_paused_for_test.load( + std::memory_order_acquire); + }); + const auto poll_calls_before_release = + worker->blocking_poll_calls_for_test(); + if (reset_paused) { + sched.go_to(0, record_blocking_poll_count, + &marker_poll_calls, &marker_completed); + } + const auto wake_calls_before_release = + worker->submission_wake_calls_for_test(); + + elio::runtime::detail::pause_before_submission_wake_reset_for_test.store( + false, std::memory_order_release); + elio::runtime::detail::pause_before_submission_wake_reset_for_test + .notify_all(); + + const bool marker_ran = wait_for_scheduler_condition([&] { + return marker_completed.load(std::memory_order_acquire); + }); + const bool stopped = sched.shutdown(scaled_sec(5)); + + REQUIRE(worker_blocking); + REQUIRE(holder_running); + REQUIRE(reset_paused); + REQUIRE(marker_ran); + REQUIRE(stopped); + CHECK(wake_calls_after_arm == wake_calls_before + 1); + CHECK(wake_calls_before_release == wake_calls_before + 1); + CHECK(marker_poll_calls.load(std::memory_order_relaxed) == + poll_calls_before_release); +} + task yield_for_metric_reads(size_t sample_count, std::atomic* sample_ready, std::atomic* sample_acknowledged, @@ -618,6 +790,40 @@ TEST_CASE("Scheduler keeps an eligible continuation on the owner-local deque", CHECK(fallbacks_after == fallbacks_before); } +TEST_CASE("Scheduler coalesces external submission wakes while a worker is busy", + "[scheduler][performance][wake][io]") { + SECTION("epoll") { + check_submission_wake_burst( + elio::io::io_context::backend_type::epoll); + } + + SECTION("io_uring") { + if (!elio::io::io_uring_backend::is_available()) { + SUCCEED("io_uring is not available at runtime"); + return; + } + check_submission_wake_burst( + elio::io::io_context::backend_type::io_uring); + } +} + +TEST_CASE("Scheduler rechecks skipped submissions before blocking", + "[scheduler][performance][wake][race][regression]") { + SECTION("epoll") { + check_skipped_submission_rechecked_before_blocking( + elio::io::io_context::backend_type::epoll); + } + + SECTION("io_uring") { + if (!elio::io::io_uring_backend::is_available()) { + SUCCEED("io_uring is not available at runtime"); + return; + } + check_skipped_submission_rechecked_before_blocking( + elio::io::io_context::backend_type::io_uring); + } +} + TEST_CASE("go_to affinity reaches the returned task across suspension", "[scheduler][task][io][affinity][regression]") { check_spawn_affinity_case( diff --git a/wiki/API-Reference.md b/wiki/API-Reference.md index e2639c78..544692b9 100644 --- a/wiki/API-Reference.md +++ b/wiki/API-Reference.md @@ -901,7 +901,7 @@ Individual worker that executes tasks. Workers use a unified idle mechanism wher ```cpp class worker_thread { public: - // Schedule a task to this worker (thread-safe, wakes worker if sleeping). + // Schedule a task to this worker (thread-safe, wakes worker as needed). // A full bounded inbox spills to a locked overflow queue; false means the // worker has stopped and the caller retains the handle. bool schedule(std::coroutine_handle<> handle); @@ -939,7 +939,9 @@ public: **Idle Behavior:** - Workers block efficiently on I/O poll (with eventfd wake support) when no tasks are available - Optional spin phase before blocking (configurable via `wait_strategy`) -- When a task is scheduled via `schedule()`, the worker is automatically woken +- Cross-thread submissions automatically wake a blocked worker. A burst shares + one outstanding eventfd notification; before blocking again, the worker + clears the wake claim and rechecks its external queues so no wake is lost - The bounded MPSC inbox remains the fast path; a locked overflow queue absorbs rare bursts without resuming worker-bound coroutines on submitter threads - Results in near-zero CPU usage (< 1%) when idle with default blocking strategy diff --git a/wiki/Performance-Tuning.md b/wiki/Performance-Tuning.md index 660354f6..e5f48905 100644 --- a/wiki/Performance-Tuning.md +++ b/wiki/Performance-Tuning.md @@ -45,24 +45,36 @@ Scaling efficiency depends on workload characteristics. Tasks with more computat ### Wake-up Mechanism -Elio uses an **eventfd embedded in each worker's I/O backend** (epoll/io_uring) for cross-thread notifications. This provides a single unified wait point — both I/O completions and task wake-ups unblock the same `poll()` call, eliminating the latency gap that exists with separate wait mechanisms. The eventfd counter deduplicates wakes, making unconditional wake safe and minimizing scheduling overhead. +Elio uses an **eventfd embedded in each worker's I/O backend** (epoll/io_uring) +for cross-thread notifications. This provides a single unified wait point: both +I/O completions and task wake-ups unblock the same `poll()` call, eliminating +the latency gap that exists with separate wait mechanisms. A worker-level +pending-wake claim lets a burst of submissions share one eventfd notification. ## Built-in Optimizations -### Unconditional Wake +### Coalesced Submission Wake -Workers track their idle state. Task submissions always trigger wake syscalls on cross-thread submit; the eventfd counter deduplicates, making unconditional wake safe: +The first cross-thread submission in a busy or blocked interval claims and +writes an eventfd wake. Later submissions still publish their queue entries but +share that outstanding notification: ```cpp // In worker_thread::schedule() if (inbox_->push(handle.address())) { - // Always wake on cross-thread submit. The eventfd dedupes via its - // counter — calling wake() on a busy worker just bumps the counter. - wake(); + wake_for_submission(); } ``` -The previous lazy wake optimization was removed due to a race condition that caused 10ms tail latency on weak hardware. +Before entering a blocking poll, the worker atomically clears the pending-wake +claim and rechecks both external queues. A submission published before that +clear is found by the recheck; a submission published after it claims a new +wake and interrupts the poll. This handshake preserves unconditional-wake +correctness while avoiding one `eventfd_write` syscall per task in a burst. + +This is deliberately not the previous idle-flag lazy-wake optimization. The +producer never decides that a wake is unnecessary from a sampled worker state; +that approach had a race which caused 10 ms tail latency on weak hardware. The MPSC ring remains the normal submission path. If it stays full after bounded retries, the worker accepts the handle through a locked overflow queue. @@ -86,15 +98,23 @@ local fast path on a draining worker. New independent tasks submitted through ### Unified Wake Mechanism -Each worker's I/O backend (epoll or io_uring) contains an embedded `eventfd`. When a task is submitted to a worker from another thread, the submitter writes to that worker's eventfd. Because the eventfd is registered with the same epoll/io_uring instance that handles I/O completions, both I/O events and task wake-ups unblock the same `poll()` call. +Each worker's I/O backend (epoll or io_uring) contains an embedded `eventfd`. +The first task submitted from another thread while no submission wake is +outstanding writes to that eventfd. Because the descriptor is registered with +the same epoll/io_uring instance that handles I/O completions, both I/O events +and task wake-ups unblock the same `poll()` call. This unified design has two key benefits: 1. **Single wait point.** A worker blocked on I/O poll is immediately woken by a cross-thread task submission. There is no separate condition variable or futex that could introduce a latency gap between "I/O ready" and "task ready" paths. -2. **Safe unconditional wake.** The eventfd counter deduplicates wakes — calling `wake()` on a busy worker just bumps the counter, which is consumed on the next poll return. This eliminates the race condition that existed with the previous lazy wake optimization. +2. **Safe coalescing.** Producers atomically share one outstanding submission + wake. The worker clears that claim and rechecks its external queues before + every blocking poll, closing the clear-versus-block lost-wake window. -The result is that cross-thread scheduling latency equals one `eventfd_write` plus one `epoll_wait`/`io_uring_enter` return — typically under 5 microseconds. +The first submission pays for one `eventfd_write`; additional tasks in the same +burst pay only for the atomic claim check. A blocked worker is still interrupted +immediately rather than waiting for a polling timeout. ### Wait Strategy @@ -625,7 +645,8 @@ Elio includes several benchmark tools: ```bash cmake --build build -# Quick benchmark - measures spawn, context switch, yield, reschedule +# Quick benchmark - measures spawn, context switch, yield, reschedule, +# and external submission bursts ./build/examples/quick_benchmark # Microbenchmarks - individual operation timing