rt: add EventLoop and LocalEventLoop driven by a host event loop

[guybedford]

Motivation

This adds an event loop form of the current_thread runtime that exposes what a host needs to integrate with its own event loop.

This is needed for any programs embedded when an event loop already owns the thread, for example libuv in a Node.js addon, KJ, a GUI toolkit, a JavaScript host under WebAssembly. In these environments the runtime must never block so the host needs to wait on the runtime's behalf and call it back when it has work.

Solution

Under tokio_unstable:

impl Builder {
    /// Current-thread only; errors otherwise.
    pub fn build_event_loop(&mut self) -> io::Result<EventLoop>;
    pub fn build_local_event_loop(&mut self, options: LocalOptions) -> io::Result<LocalEventLoop>;

    /// Invoked for a remote wakeup (a spawn, a wake from another thread,
    /// a nearer timer registered elsewhere) that arrives while the loop is
    /// not being ticked or parked: once per idle period. Not for I/O
    /// readiness on a runtime with the I/O driver; the host waits on
    /// `event_fd` (or in `park`) for that. An `Unparker::unpark` always
    /// reaches it, as it carries no work for `next_sleep` to report.
    ///
    /// Runs inside the runtime's scheduling and timer internals: it must
    /// not call into the runtime, and should post a tick to the host.
    pub fn on_work_available<F>(&mut self, f: F) -> &mut Self
    where
        F: Fn() + Send + Sync + 'static;

    /// Give the event loop an `event_fd` even without an I/O driver, by
    /// allocating a standalone wakeup descriptor. With the I/O driver the
    /// reactor's own descriptor serves and nothing is allocated.
    #[cfg(unix)]
    pub fn enable_event_fd(&mut self) -> &mut Self;
}

impl EventLoop {
    pub fn handle(&self) -> &Handle;
    pub fn enter(&self) -> EnterGuard<'_>;

    /// Poll the I/O driver without blocking, then poll up to `max_tasks`
    /// ready tasks. Returns the number of tasks polled.
    ///
    /// `max_tasks == 0` drains the driver without running any tasks. The
    /// builder's `event_interval` has no role; `max_tasks` bounds each batch.
    /// Panics if called from within a runtime, including one of this loop's
    /// own tasks.
    pub fn tick(&mut self, max_tasks: usize) -> usize;

    /// How long the host may sleep before calling `tick` again.
    ///
    /// * `Some(ZERO)` — work is ready now (local or injected); tick again.
    /// * `Some(d)` — nothing until a timer fires in `d`, or the loop is woken.
    /// * `None` — nothing scheduled; wait indefinitely for a wakeup.
    ///
    /// A wakeup during a tick is reported here, not by the callback.
    pub fn next_sleep(&self) -> Option<Duration>;

    /// A descriptor readable whenever the loop has work other than a timer.
    ///
    /// With the I/O driver, the reactor's `epoll`/`kqueue` fd: level-readable
    /// for I/O readiness and signals, and through the driver's waker for
    /// spawns, cross-thread wakes, nearer timers and unparks. Without it,
    /// the `enable_event_fd` descriptor, with the same semantics, or `None`.
    /// Timers never make it readable, so sleep on it with `next_sleep` as
    /// the timeout.
    #[cfg(unix)]
    pub fn event_fd(&self) -> Option<BorrowedFd<'_>>;

    /// Block until there is work, a timer expires, `timeout` elapses, or
    /// `Unparker::unpark` is called. May return spuriously.
    ///
    /// Sleeps for `min(timeout, next_sleep())`, dispatching readiness and
    /// due timers to their tasks but running none. Panics if called from
    /// within a runtime.
    pub fn park(&mut self, timeout: Option<Duration>);

    /// Wakes a current or future `park` from any thread.
    pub fn unparker(&self) -> Unparker;

    pub fn shutdown_background(self);
    pub fn shutdown_timeout(self, duration: Duration);
}

impl LocalEventLoop {
    // ... the same, plus:
    pub fn spawn_local<F>(&self, future: F) -> JoinHandle<F::Output>
    where
        F: Future + 'static;
}

EventLoop is Send + Sync over a Runtime, LocalEventLoop is !Send over a LocalRuntime.

Hosts can decide on the type of integration - a descriptor-only host is correct without ever reading next_sleep for readiness, since the descriptor stays level-readable from a wakeup until a tick drains it. Similarly, a callback host ticks until next_sleep is not zero before waiting, since wakeups during a tick are reported there.

All remote wake paths - schedule, the block_on waker, timer re-registration, Unparker - go through the driver handle's unpark, which is where the host notification lives as a Signal on the driver handle. It only needs to notify once per idle period, so wakes are coalesced, and while the loop is being ticked or parked the work a wake carries will be found by the tick itself, so the callback is held back for that window. That holding-back is what creates a lost-wake race (a stale busy read on one side, an empty inject queue on the other). Both sides take the inject queue lock, so its order rules the race out, not the atomics - see signal.rs. loom_current_thread::event_loop models it along with an unpark racing a tick (R-loom-current-thread).

The first commit is independent of the event loop and affects every current_thread runtime, so it is split out for separate consideration: While testing the event loop, every in-runtime sleep showed up as a spurious wakeup: registering a timer from within the scheduler's own context was unparking the driver. But that thread holds the driver, so it can't be parked while registering, and when it does park the time driver reads the wheel's next expiration under the lock - the unpark was never doing anything except costing a waker syscall and an empty driver turn. Multi-thread is unchanged since the registering worker there may not hold the driver.

Windows IOCP support is left as a possible follow-on, if we get handle support from Mio via tokio-rs/mio#2023. Without that, and for hosts without IO that don't want to enable event_fd, on_work_available is supported perfectly fine for wakeups and park for I/O.

Tests

rt_event_loop drives both loops as a host would: over poll(2) on event_fd with next_sleep as the timeout, over park, and over the callback. It pins when the host is and isn't notified (spawns, cross-thread and cross-loop wakes, I/O readiness, nearer timers, timers as timeouts only), what a tick runs (max_tasks batches, leftovers via next_sleep), park's return conditions, the standalone descriptor, the callback's once-per-idle-period rule, and the usual lifecycle cases (drop, shutdown, panics, nested ticks, auto traits). loom_current_thread::event_loop models the wakeup protocol.

[Darksonn]

Would it be possible to split this into two PRs so that there's just one PR that adds a generic EventLoop utility without any emscripten specific code?

[guybedford]

Would it be possible to split this into two PRs so that there's just one PR that adds a generic EventLoop utility without any emscripten specific code?

I've gone ahead and updated this PR to just create the new EventLoop and LocalEventLoop, while supporting a custom EventLoopHost trait for hosts, but without actually including the implementation for that.

Since this no longer depends on the Emscripten PRs, this is now directly on main and no longer a draft PR. I am posting the follow-on Emscripten implementation based to this now.

[guybedford]

Thank you @mcollina for trying this native integration out in libuv for Node.js - this is really valuable implementation feedback to have at such an early stage.

I was initially a bit concerned about Windows support for your suggestion, but was able to verify that this works under Windows IOCP with one change to Mio in tokio-rs/mio#2023.

With this, I've refactored the approach to expose the handle shape you proposed:

• LocalEventLoop is AsRawFd on Unix and AsRawHandle on Windows
• next_timeout() -> Option<Duration> is now provided to extract the current shortest timeout
• drive() now returns a boolean indicating if more work remains
• block_on() is removed

This now provides all of the necessary primitives to integrate into a libuv event loop for Node.js addons - please do test it out and share any feedback further.

In terms of performance, note you can tune the event_internal (batch size per drive) to see what gives the best numbers. Suggestions re defaults here very welcome too.

[mcollina]

Thanks @guybedford, this works. I tested 6e9537b on Linux with a prototype inside our Node.js addon, and the new API covers everything we need:

• Integration: uv_poll_start on as_raw_fd() and a one-shot uv_timer from next_timeout(). Both callbacks call drive() while it returns true, and we cap the drives per wake so libuv still gets a turn. Nothing else was needed, and the PR's rt_event_loop tests pass here.
• Tip worth documenting: wake-ups that happen inside drive() (for example JS completing requests from a task) go straight onto the local run queue. From outside the drive, they go through the inject queue and the driver's waker, which costs a syscall and another trip around the loop.
• event_interval: no measurable difference across 16–1024 for us, so the default seems fine.
• Results: the thread hop is gone, and single-core throughput and tail latency both improved noticeably over our current multi-thread design.

I haven't tested macOS or Windows yet. Happy to keep testing as this evolves.

[Darksonn]

Sorry for being so long to review this. Here is an API design I sketched a while ago, and I think it'd be useful to compare the designs for differences.

impl Builder {
    /// Current-thread only; errors otherwise.
    pub fn build_event_loop(&mut self) -> io::Result<EventLoop>;
    pub fn build_local_event_loop(&mut self) -> io::Result<LocalEventLoop>;

    /// Invoked when the loop transitions from "nothing to do" to "has work".
    ///
    /// Called for remote wakeups of the runtime, but not for IO events on
    /// a runtime with IO enabled.
    pub fn on_work_available<F>(&mut self, f: F) -> &mut Self
    where
        F: Fn() + Send + Sync + 'static;

    /// Give the event loop an `event_fd` even when built without an I/O
    /// driver, by allocating a standalone eventfd for wakeups.
    #[cfg(unix)]
    pub fn enable_event_fd(&mut self) -> &mut Self;
}

impl EventLoop {
    pub fn handle(&self) -> &Handle;
    pub fn enter(&self) -> EnterGuard<'_>;

    /// Poll the I/O driver without blocking, then poll up to `max_tasks`
    /// ready tasks. Returns the number of tasks polled.
    ///
    /// `max_tasks == 0` drains the driver without running any tasks.
    /// Panics if called from within a task owned by this loop.
    pub fn tick(&mut self, max_tasks: usize) -> usize;

    /// How long the caller may sleep before calling `tick` again.
    ///
    /// * `Some(ZERO)` — work is ready now; call `tick` again.
    /// * `Some(d)` — nothing until a timer fires in `d`, or the loop is woken.
    /// * `None` — nothing scheduled; wait indefinitely for a wake.
    pub fn next_sleep(&self) -> Option<Duration>;

    /// A descriptor that becomes readable whenever the loop has work.
    ///
    /// Might be the epoll fd, an io-uring fd, an eventfd or other similar fd.
    ///
    /// The fd becomes ready any time there is runnable work, except for
    /// timers, so sleeping on this fd should be done using `next_sleep` as
    /// the timeout.
    #[cfg(unix)]
    pub fn event_fd(&self) -> Option<BorrowedFd<'_>>;

    /// Block until there is work, a timer expires, `timeout` elapses, or
    /// `Unparker::unpark` is called. May return spuriously.
    ///
    /// Sleeps for `min(timeout, next_sleep())`. Does not tick.
    /// Panics if called from within a task owned by this loop.
    ///
    /// On wasm this suspends the current JSPI activation rather than blocking
    /// the thread.
    pub fn park(&mut self, timeout: Option<Duration>);

    /// When the `unpark()` method on this `Unparker` is called, a call to `park`
    /// will return.
    pub fn unparker(&self) -> Unparker;

    pub fn shutdown_background(self);
    pub fn shutdown_timeout(self, duration: Duration);
}

impl LocalEventLoop {
    // ... same, plus:
    pub fn spawn_local<F>(&self, future: F) -> JoinHandle<F::Output>
    where
        F: Future + 'static;
}

[Darksonn]

In particular, it looks like your design implies that the event loop always has an fd available for readiness, which causes complications on windows. I'm not sure we need to always provide an fd for runtimes without IO.

[guybedford]

@Darksonn wow this is a really nicely worked through design as well. Most things seem to adapt pretty straightforwardly in the above.

A couple of the main differences AFAICT are:

1. enable_event_fd() allows wakeups with IO disabled
2. on_work_available supports even non-FD callbacks and seems to effectively be the Windows fallback, but without full IO support.

For Windows support - for feature parity I think we would still want the IOCP handle exposed. The purist in me would prefer that mirroring, but that would require tokio-rs/mio#2023.

If we did have that Windows support, then on_work_available remains a layer for non-event-fd support, but is that layer strictly necessary?

FWIW Emscripten has landed epoll and epoll callbacks now such that a full IO event_fd would be enough of a primitive allowing us to even skip #8484 entirely and move that out to userland runtime integration instead.

So the question seems to me really one of scope, and including if tokio-rs/mio#2023 lands or not.

[Darksonn]

I think it makes sense to avoid the dependency on tokio-rs/mio#2023 for this PR, and consider it as a follow-up. You can file a feature request issue to track it.

I think having on_work_available makes sense as a general utility, as it allows the non-io usecase to hook into a parent executor that uses futexes for sleeping rather than fds.

[Darksonn]

Note CI issues are fixed if you rebase on master.

[guybedford]

@Darksonn I've updated this PR now to your design now, which I've verified is compatible with the previous one. Ready for review.

[Master-Norna]

Thanks for working through the host-loop API. I focused on the callback notification protocol and found the two independent issues below. The upstream rt_event_loop suite passes locally (23 tests on Windows, 35 on Linux), as do the Windows runtime/timer regression suites I ran (200 tests); those passing tests do not exercise these notification interleavings. These comments are not a review of the entire API or of the macOS reactor path.

[Master-Norna]

The lost-wake proof in signal.rs relies on the host taking the inject queue mutex after Signal::end, but this path calls Inject::is_empty, which only loads Shared::len with Acquire. The existing Loom test instead reads an always-locked queue, so it supplies synchronization that next_sleep does not have.

I extended the model to use the production queue's Release length updates, Acquire empty checks, and no-I/O parker notification. Loom finds a wake neither consumed by the tick, observed by next_sleep, nor delivered to the callback. I also reproduced it in a crate-internal probe using the real Signal, Inject, and ParkThread, not just the simplified queue model. Taking the queue mutex for the host check eliminates this counterexample in the isolated model. This matters for a callback-only host sleeping on its own queue: the internal parker notification cannot wake that host.

Could we give the host check a real synchronization edge, or change the signal handshake so it supplies one, and make the Loom model exercise that production protocol?

[guybedford]

Yes the claim in the comment of signal.rs that the host takes the inject lock in next_sleep was incorrect here. Inject::is_empty as a lock-free Acquire resulted in no common ordering, and this was missed in Loom modelling the queue as a Mutex.

This is now managed through ordered pending.swap(false, AcqRel) in end() so that every write to pending is now an RMW with on modification order:

• if end's swap reads a wakeup's true, the host sees that wakeup's push and next_sleep reports it
• if wake's swap reads end's false, it sees busy = false and runs the callback.

begin() also has a pending swap where a plain store would be correct, since Loom reports a false positive here pending tokio-rs/loom#282.

New tests are included for these cases.

[Master-Norna]

There is also a separate coalescing race, even with a locked host queue check:

1. Remote wake A queues a task, then pauses before signal.wake().
2. The tick consumes A and reaches end()'s pending.store(false).
3. A resumes: pending.swap(true) returns false, but busy is still true, so no callback runs.
4. end() clears busy; the host finds no remaining work and may sleep.
5. New wake B sees pending == true and returns without notifying the host.

The single-wake Loom assertion still passes because A was consumed; extending it across the next idle period catches B's missing notification. I confirmed this against the unchanged Signal, independently of the lock-free queue-check issue above. Could we make the busy-to-idle transition and pending reset one coherent protocol, and add a two-period regression model?

[guybedford]

This is now resolved with end() clearing busy first, then handling pending so that a wakeup landing between the two can no longer be left pending without having seen the host idle.

A new test, wake_after_tick_is_never_lost covers this "two-period" model, which fails before the fix.