park a sleeping green task in the reactor instead of holding its worker

cranelift/runtime/src/netpoll.rs

@@ -42,9 +42,13 @@
 //!   the last waiter leaving does `EPOLL_CTL_DEL`. keeping epoll state a pure
 //!   function of the waiter lists is what makes the arm/disarm bookkeeping easy to
 //!   reason about.
-//! - `deadlines`: a min-heap of `(instant, seq, fd, interest)` for waits that
-//!   carry a finite timeout. the reactor sets each `epoll_wait` timeout to the
-//!   nearest deadline and, on expiry, wakes the task with a timeout result.
+//! - `deadlines`: a min-heap of timestamped entries for waits that carry a
+//!   finite timeout. the reactor sets each `epoll_wait` timeout to the nearest
+//!   deadline and, on expiry, wakes the task with a timeout result. an entry
+//!   targets either an fd wait or a pure timer.
+//! - `timers`: tasks sleeping with no fd at all (`sleep_task`, behind
+//!   `pith_sleep`). a sleep is a deadline wait minus the fd: same heap, same
+//!   sweep, nothing armed in epoll.
 //!
 //! ## the eventfd nudge
 //!
@@ -119,12 +123,21 @@ struct Waiter {
     outcome: Arc<AtomicI32>,
 }
 
+/// what a heap entry resolves when it expires: an fd wait that carried a
+/// timeout, or a pure timer with no fd at all (a green task sleeping). the two
+/// are distinct variants — rather than a sentinel fd smuggled through the io
+/// shape — so the fd-arming code (`reconcile_fd`, `desired_mask`, `on_close`)
+/// can never be handed a value that is not a real fd.
+enum DeadlineTarget {
+    Io { fd: RawFd, interest: Interest },
+    Timer,
+}
+
 /// a finite-timeout registration, ordered by instant for the min-heap.
 struct Deadline {
     at: Instant,
     seq: u64,
-    fd: RawFd,
-    interest: Interest,
+    target: DeadlineTarget,
 }
 
 // order deadlines by time (then seq) so a `BinaryHeap<Reverse<Deadline>>` pops the
@@ -159,6 +172,10 @@ struct Inner {
     waiters: HashMap<(RawFd, Interest), Vec<Waiter>>,
     fds: HashMap<RawFd, FdReg>,
     deadlines: BinaryHeap<std::cmp::Reverse<Deadline>>,
+    /// sleeping tasks, keyed by the seq of their heap entry. a timer has no
+    /// (fd, interest) to key on, and giving it its own map keeps it out of
+    /// `waiters` — where everything is fd-shaped — entirely.
+    timers: HashMap<u64, Waiter>,
     next_seq: u64,
 }
 
@@ -168,6 +185,7 @@ impl Inner {
             waiters: HashMap::new(),
             fds: HashMap::new(),
             deadlines: BinaryHeap::new(),
+            timers: HashMap::new(),
             next_seq: 0,
         }
     }
@@ -362,8 +380,7 @@ pub(crate) fn wait_io(fd: RawFd, read: bool, timeout_ms: i64, task: usize) -> i6
             inner.deadlines.push(std::cmp::Reverse(Deadline {
                 at,
                 seq,
-                fd,
-                interest,
+                target: DeadlineTarget::Io { fd, interest },
             }));
         }
         reconcile_fd(reactor, &mut inner, fd);
@@ -393,6 +410,65 @@ pub(crate) fn wait_io(fd: RawFd, read: bool, timeout_ms: i64, task: usize) -> i6
     }
 }
 
+/// park the current green task for `ms` milliseconds without holding its worker.
+/// this is `wait_io` minus the fd: register a deadline, nudge the reactor if it
+/// is now the nearest one, and suspend; the deadline sweep wakes us. must be
+/// called only from inside a green task (the caller checks
+/// `green::current_task`); `task` is that task's slab id.
+///
+/// a zero or negative duration returns immediately — there is nothing to wait
+/// for, so nothing is registered.
+pub(crate) fn sleep_task(ms: i64, task: usize) {
+    if ms <= 0 {
+        return;
+    }
+    ensure_started();
+    let reactor = reactor();
+    let outcome = Arc::new(AtomicI32::new(PENDING));
+    let at = Instant::now() + Duration::from_millis(ms as u64);
+
+    let should_nudge = register_timer(reactor, at, task, &outcome);
+    if should_nudge {
+        nudge(reactor);
+    }
+
+    // park until the sweep marks us done. unlike `wait_io` — where a spurious
+    // wake is safe because the task re-runs its syscall — a sleep has no
+    // syscall to act as the source of truth, so an early resume must go back
+    // to sleep: our timer entry is still registered and the sweep will wake us
+    // again. once the outcome is terminal the entry is gone and we return.
+    while outcome.load(AtomicOrdering::Acquire) == PENDING {
+        green::park_current();
+    }
+}
+
+/// the registration half of `sleep_task`, split out so it can be exercised
+/// without a live coroutine to suspend. pushes a timer deadline and its waiter
+/// under the lock; returns whether the reactor must be nudged (only when this
+/// deadline is nearer than whatever it is currently sleeping on).
+fn register_timer(reactor: &Reactor, at: Instant, task: usize, outcome: &Arc<AtomicI32>) -> bool {
+    let mut inner = lock_inner(reactor);
+    let seq = inner.next_seq;
+    inner.next_seq += 1;
+
+    let nudge = inner.deadlines.peek().map_or(true, |top| at < top.0.at);
+
+    inner.timers.insert(
+        seq,
+        Waiter {
+            task,
+            seq,
+            outcome: outcome.clone(),
+        },
+    );
+    inner.deadlines.push(std::cmp::Reverse(Deadline {
+        at,
+        seq,
+        target: DeadlineTarget::Timer,
+    }));
+    nudge
+}
+
 /// the reactor thread: wait on epoll, wake the tasks whose fds are ready or whose
 /// deadlines expired, and re-arm what is left. never resumes a coroutine.
 fn reactor_loop() {
@@ -514,38 +590,60 @@ fn take_ready(inner: &mut Inner, fd: RawFd, interest: Interest, to_wake: &mut Ve
 
 /// pop every deadline at or before `now`, mark the matching still-pending waiter
 /// timed out, remove it, and queue it to wake. a stale heap entry (its waiter
-/// already fired ready and was removed) simply matches nothing.
+/// already fired ready and was removed) simply matches nothing. a timer entry
+/// has no fd, so it wakes its sleeper directly and never touches epoll.
 fn collect_expired(reactor: &Reactor, inner: &mut Inner, now: Instant, to_wake: &mut Vec<usize>) {
     // track which fds we touch so we can re-arm/disarm them once at the end.
     let mut touched: Vec<RawFd> = Vec::new();
     while let Some(top) = inner.deadlines.peek() {
         if top.0.at > now {
             break;
         }
-        let Deadline {
-            seq, fd, interest, ..
-        } = inner.deadlines.pop().unwrap().0;
-
-        if let Some(waiters) = inner.waiters.get_mut(&(fd, interest)) {
-            if let Some(pos) = waiters.iter().position(|w| w.seq == seq) {
-                let w = waiters.remove(pos);
-                if w
-                    .outcome
-                    .compare_exchange(
-                        PENDING,
-                        TIMED_OUT,
-                        AtomicOrdering::AcqRel,
-                        AtomicOrdering::Acquire,
-                    )
-                    .is_ok()
-                {
-                    to_wake.push(w.task);
-                }
-                if waiters.is_empty() {
-                    inner.waiters.remove(&(fd, interest));
+        let Deadline { seq, target, .. } = inner.deadlines.pop().unwrap().0;
+
+        match target {
+            DeadlineTarget::Timer => {
+                // a sleep expiring is its success case; the same PENDING->
+                // terminal compare-exchange keeps the resolve-once discipline
+                // even though nothing else currently races for a timer.
+                if let Some(w) = inner.timers.remove(&seq) {
+                    if w
+                        .outcome
+                        .compare_exchange(
+                            PENDING,
+                            TIMED_OUT,
+                            AtomicOrdering::AcqRel,
+                            AtomicOrdering::Acquire,
+                        )
+                        .is_ok()
+                    {
+                        to_wake.push(w.task);
+                    }
                 }
-                if !touched.contains(&fd) {
-                    touched.push(fd);
+            }
+            DeadlineTarget::Io { fd, interest } => {
+                if let Some(waiters) = inner.waiters.get_mut(&(fd, interest)) {
+                    if let Some(pos) = waiters.iter().position(|w| w.seq == seq) {
+                        let w = waiters.remove(pos);
+                        if w
+                            .outcome
+                            .compare_exchange(
+                                PENDING,
+                                TIMED_OUT,
+                                AtomicOrdering::AcqRel,
+                                AtomicOrdering::Acquire,
+                            )
+                            .is_ok()
+                        {
+                            to_wake.push(w.task);
+                        }
+                        if waiters.is_empty() {
+                            inner.waiters.remove(&(fd, interest));
+                        }
+                        if !touched.contains(&fd) {
+                            touched.push(fd);
+                        }
+                    }
                 }
             }
         }
@@ -639,14 +737,15 @@ mod tests {
         heap.push(std::cmp::Reverse(Deadline {
             at: now + Duration::from_millis(50),
             seq: 1,
-            fd: 3,
-            interest: Interest::Read,
+            target: DeadlineTarget::Io {
+                fd: 3,
+                interest: Interest::Read,
+            },
         }));
         heap.push(std::cmp::Reverse(Deadline {
             at: now + Duration::from_millis(10),
             seq: 2,
-            fd: 4,
-            interest: Interest::Write,
+            target: DeadlineTarget::Timer,
         }));
         // the nearer deadline (10ms) must come out first.
         assert_eq!(heap.pop().unwrap().0.seq, 2);
@@ -721,6 +820,66 @@ mod tests {
         assert!(inner.fds.get(&fd).is_none());
     }
 
+    // the two timer tests share the one process-global registry, so they must
+    // not interleave with each other (cargo runs tests concurrently).
+    static TIMER_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+    // a timer registration must land in the timers map and the deadline heap,
+    // and — once the sweep runs past it — resolve to TIMED_OUT with the entry
+    // gone and no fd touched. driven by hand (register_timer + collect_expired)
+    // because a real sleep_task needs a live coroutine to suspend into.
+    #[test]
+    fn timer_registers_expires_and_wakes() {
+        let _guard = TIMER_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
+        let reactor = reactor();
+        let outcome = Arc::new(AtomicI32::new(PENDING));
+        let at = Instant::now() + Duration::from_millis(5);
+        register_timer(reactor, at, usize::MAX, &outcome);
+
+        // identify our entry by its shared outcome cell, not by map emptiness:
+        // the registry is global and other tests may hold entries of their own.
+        let ours = |inner: &Inner| {
+            inner
+                .timers
+                .values()
+                .any(|w| Arc::ptr_eq(&w.outcome, &outcome))
+        };
+
+        let mut to_wake = Vec::new();
+        {
+            let mut inner = lock_inner(reactor);
+            assert!(ours(&inner));
+
+            // not due yet: the sweep leaves it alone.
+            collect_expired(reactor, &mut inner, at - Duration::from_millis(1), &mut to_wake);
+            assert!(to_wake.is_empty());
+            assert_eq!(outcome.load(AtomicOrdering::Acquire), PENDING);
+
+            // due: it resolves to TIMED_OUT, queues the task, and cleans up.
+            collect_expired(reactor, &mut inner, at, &mut to_wake);
+            assert!(!ours(&inner));
+        }
+        assert_eq!(to_wake, vec![usize::MAX]);
+        assert_eq!(outcome.load(AtomicOrdering::Acquire), TIMED_OUT);
+    }
+
+    // the nudge decision: a timer nearer than the current top asks for one, a
+    // farther timer rides on the reactor's existing sleep. asserted relative to
+    // a timer of our own, since the global registry may hold other entries.
+    #[test]
+    fn timer_nudges_only_when_nearest() {
+        let _guard = TIMER_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
+        let reactor = reactor();
+        let far = Instant::now() + Duration::from_secs(600);
+        let outcome = Arc::new(AtomicI32::new(PENDING));
+        register_timer(reactor, far, usize::MAX, &outcome);
+        assert!(register_timer(reactor, far - Duration::from_secs(60), usize::MAX, &outcome));
+        assert!(!register_timer(reactor, far + Duration::from_secs(60), usize::MAX, &outcome));
+        // the far-future entries left behind are inert: nothing sweeps them
+        // until long after every test in this process has finished, and their
+        // task id (usize::MAX) makes any eventual wake a no-op.
+    }
+
     // on_close must only close out a *pending* wait: a waiter a readiness or
     // timeout already claimed keeps its terminal outcome, so a close racing a
     // fire never double-resolves the wait.

cranelift/runtime/src/netpoll_fallback.rs

@@ -31,6 +31,18 @@ pub(crate) fn wait_io(fd: RawFd, read: bool, timeout_ms: i64, _task: usize) -> i
     crate::fdio::poll_wait(fd as i64, events, timeout_ms)
 }
 
+/// sleep for `ms` milliseconds. `task` is the caller's green slab id, unused
+/// here: with no reactor there is no timer heap to park the task on, so the
+/// worker OS thread blocks for the duration — the same degradation socket
+/// waits take on this platform (see the module comment). a zero or negative
+/// duration returns immediately, matching the linux reactor's contract.
+pub(crate) fn sleep_task(ms: i64, _task: usize) {
+    if ms <= 0 {
+        return;
+    }
+    std::thread::sleep(std::time::Duration::from_millis(ms as u64));
+}
+
 /// no-op: without a reactor there is no per-fd registration to tear down. the
 /// linux build cleans up parked waiters here; a `poll` wait owns its fd for the
 /// length of one call and leaves nothing behind.

cranelift/runtime/src/platform.rs

@@ -18,9 +18,27 @@ pub extern "C" fn pith_exit(code: i64) {
     std::process::exit(code as i32);
 }
 
-/// Sleep for given number of milliseconds
+/// Sleep for given number of milliseconds.
+///
+/// from inside a green task this parks the *task* on the reactor's timer heap
+/// and gives the worker back — a sleeping task must not stall every other task
+/// pinned to its worker (`select`'s idle probe sleeps in a loop, so before
+/// this two idle selects could occupy a whole two-worker pool). the guard is
+/// the same one `dns.rs` uses: `current_task()` is `None` on the main thread,
+/// on any plain OS thread, and on the whole os-thread backend, and there
+/// blocking the thread is exactly right, so that path is unchanged. on
+/// non-linux there is no reactor and `netpoll`'s fallback blocks the worker,
+/// consistent with how socket waits degrade there.
 #[no_mangle]
 pub extern "C" fn pith_sleep(ms: i64) {
+    // a negative duration is not a very long one: `ms as u64` turns -5 into
+    // about 584 million years, which is indistinguishable from a hang. clamp
+    // first, so both paths below see a duration that means what it says.
+    let ms = ms.max(0);
+    if let Some(task) = crate::concurrency::green::current_task() {
+        crate::netpoll::sleep_task(ms, task);
+        return;
+    }
     std::thread::sleep(std::time::Duration::from_millis(ms as u64));
 }
 

docs/concurrency.md

@@ -195,8 +195,11 @@ handoff between two tasks on the same worker is a userspace switch with no
 kernel involved. socket i/o goes through an epoll reactor: a read or write
 that would block parks the task on the reactor and frees the worker to run
 something else, so the whole net stack — raw tcp, tls, http/2, grpc — yields
-the same way without any code of its own. that reactor is also why the default
-is linux-only; see "which backend to use".
+the same way without any code of its own. sleeping parks there too: a
+`time.delay` from inside a task registers a timer on the reactor's deadline
+heap instead of blocking the worker, so a task sleeping out a backoff — or a
+`select` idling between probes — costs nothing but its own time. that reactor
+is also why the default is linux-only; see "which backend to use".
 
 name resolution and file i/o get there by a different route. `getaddrinfo` is
 synchronous and there is nothing to poll, and a regular file is always reported
@@ -286,8 +289,9 @@ threads, and `PITH_GREEN=1` opts in.
 
 the reason for that split is the reactor. it is epoll and eventfd, so it is
 linux-only; macos and the bsds compile a stand-in with no reactor at all, and a
-green task waiting on a socket there blocks its worker for as long as the wait
-lasts. green is still correct on those platforms and you can turn it on, but an
+green task waiting on a socket there — or sleeping — blocks its worker for as
+long as the wait lasts. green is still correct on those platforms and you can
+turn it on, but an
 i/o-heavy server on macos wants os threads until there is a kqueue sibling.
 
 the caveats that come with the linux default all come from one fact: a green