diff --git a/cranelift/runtime/src/netpoll.rs b/cranelift/runtime/src/netpoll.rs index 5696ae4c..5ceb1278 100644 --- a/cranelift/runtime/src/netpoll.rs +++ b/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, } +/// 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>` pops the @@ -159,6 +172,10 @@ struct Inner { waiters: HashMap<(RawFd, Interest), Vec>, fds: HashMap, deadlines: BinaryHeap>, + /// 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, 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) -> 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,7 +590,8 @@ 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) { // track which fds we touch so we can re-arm/disarm them once at the end. let mut touched: Vec = Vec::new(); @@ -522,30 +599,51 @@ fn collect_expired(reactor: &Reactor, inner: &mut Inner, now: Instant, to_wake: 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. diff --git a/cranelift/runtime/src/netpoll_fallback.rs b/cranelift/runtime/src/netpoll_fallback.rs index f9941fda..66f0c96c 100644 --- a/cranelift/runtime/src/netpoll_fallback.rs +++ b/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. diff --git a/cranelift/runtime/src/platform.rs b/cranelift/runtime/src/platform.rs index bf45ebd7..14a3ca06 100644 --- a/cranelift/runtime/src/platform.rs +++ b/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)); } diff --git a/docs/concurrency.md b/docs/concurrency.md index 58fc6a46..68b19dc9 100644 --- a/docs/concurrency.md +++ b/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 diff --git a/docs/limitations.md b/docs/limitations.md index d8d75ee5..1e318bd6 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -145,29 +145,25 @@ while the caller parks, and child processes park on the reactor too: pipe reads because a pipe is pollable, `wait` because linux gives out a `pidfd` that reports the exit. none of those stall anyone any more. -two things are left. the cheap end of `host_fs`, meaning `exists`, `size`, -`rename`, `mkdir` and removing a single file, still runs on the worker, because -each is one cached kernel lookup that costs about what handing it to another -thread would. on a slow network mount that reasoning does not hold and a task -doing one of them holds its worker until it returns. making that decision -adaptive rather than fixed is the open work. - -the other is `sleep`, and everything built on it. `time.delay` maps straight -to the blocking sleep, so a green task that sleeps holds its worker for the -duration — and the waiting loops built on delay hold one too: a `select` with -no ready arm sleeps a millisecond per probe until an arm fires, and the -`concurrent.after`/`ticker` workers and a context's deadline watcher do the -same. each slice is short and correctness is unaffected, but on a two-worker -host two idle selects can occupy both workers while they poll. the fix is a -real timer in the reactor — the deadline heap `netpoll` already keeps could -serve a pure-timeout wait with no fd — so a sleeping task parks the way a -socket read does. `PITH_GREEN=0` is the blunt workaround. +what is left is the cheap end of `host_fs`, meaning `exists`, `size`, +`rename`, `mkdir` and removing a single file, which still runs on the worker, +because each is one cached kernel lookup that costs about what handing it to +another thread would. on a slow network mount that reasoning does not hold and +a task doing one of them holds its worker until it returns. making that +decision adaptive rather than fixed is the open work. (`process.output` and the calls built on it — `run`, `text`, `output_checked`, `run_shell`, `output_shell`, plus `exec` and `exec_output` — used to be on this list for holding a worker while a child ran to completion; they now hand the whole spawn-drain-wait to a process pool of their own and the caller -parks, the same shape dns and file i/o use.) +parks, the same shape dns and file i/o use. `sleep` used to be on it too: +`time.delay` mapped to a blocking sleep, so a sleeping task held its worker +and an idle `select`, probing in a one-millisecond sleep loop, could pin a +whole worker to no work at all. a green task's sleep now registers a timer on +the reactor's deadline heap — the same sweep that times out socket waits — and +parks the way a socket read does, which carries `select`'s idle probing, the +`concurrent.after`/`ticker` workers, and a context's deadline watcher along +with it.) the calling task pays for that. a file call made from inside a task now costs a thread handoff it did not before, so a task reading a small cached file in a diff --git a/tests/cases/test_green_sleep_parks.pith b/tests/cases/test_green_sleep_parks.pith new file mode 100644 index 00000000..17156d46 --- /dev/null +++ b/tests/cases/test_green_sleep_parks.pith @@ -0,0 +1,47 @@ +# a sleeping task must not hold its worker. under the green backend a +# `time.delay` parks the task on the reactor's timer heap, so with a single +# worker (the corpus runs PITH_GREEN_WORKERS=1) a runnable task still gets the +# cpu while two spawned-earlier tasks sleep. before that fix the first sleeper +# blocked the only worker and "sleeper-a" finished first here. + +import std.time as time + +fn sleeper(name: String, ms: Int, done: Channel[String]): + time.delay(ms) + done.send(name) + +fn busy(done: Channel[String]): + mut total := 0 + mut i := 0 + while i < 200000: + total = total + i + i = i + 1 + if total > 0: + done.send("busy") + +fn main() -> Int!: + start := time.mono_millis() + done := Channel[String](3) + + # the sleepers are spawned first on purpose: with one worker they reach the + # cpu first, and only parking (not blocking) lets the busy task overtake. + a := spawn sleeper("sleeper-a", 250, done) + b := spawn sleeper("sleeper-b", 700, done) + c := spawn busy(done) + + print("first finished: {done.recv()?}") + print("second finished: {done.recv()?}") + print("third finished: {done.recv()?}") + + await a + await b + await c + + elapsed := time.mono_millis() - start + # the longest sleeper is 700ms, so anything shorter means delay came back + # early. the two sleeps overlapping (700ms total, not 950ms serialized) + # proves the timer heap serves concurrent sleepers; the bound is loose so a + # slow ci box does not trip it. + print("slept the full duration: {elapsed >= 700}") + print("sleepers overlapped: {elapsed < 940}") + return 0 diff --git a/tests/cases/test_sleep_edge_durations.pith b/tests/cases/test_sleep_edge_durations.pith new file mode 100644 index 00000000..3345cef2 --- /dev/null +++ b/tests/cases/test_sleep_edge_durations.pith @@ -0,0 +1,36 @@ +# a sleep has to mean what it says at the edges. +# +# a negative duration used to hang: the runtime cast the millisecond count to +# an unsigned width, so -5 became roughly 584 million years and looked exactly +# like a deadlock. zero and negative both return immediately now, on either +# backend, and a batch of sleepers overlaps instead of serializing — which is +# the property that makes a sleeping task cheap. + +import std.time as time + +fn napper() -> Int: + time.delay(120) + return 1 + +fn main(): + start := time.mono_millis() + time.delay(0) + time.delay(0 - 5) + time.delay(0 - 1000000) + print("edge_durations_return_promptly={time.mono_millis() - start < 100}") + + # a sleeper that is the only live task still gets woken + print("solo_sleeper_woke={await spawn napper()}") + + # eight sleepers on a two-worker pool: if a sleep held its worker these + # would run four batches deep, so the elapsed time separates the two. + mut tasks := [] + mut i := 0 + while i < 8: + tasks.push(spawn napper()) + i = i + 1 + batch := time.mono_millis() + mut woke := 0 + for t in tasks: + woke = woke + (await t) + print("all_woke={woke} overlapped={time.mono_millis() - batch < 400}") diff --git a/tests/expected/test_green_sleep_parks.txt b/tests/expected/test_green_sleep_parks.txt new file mode 100644 index 00000000..469abcc6 --- /dev/null +++ b/tests/expected/test_green_sleep_parks.txt @@ -0,0 +1,5 @@ +first finished: busy +second finished: sleeper-a +third finished: sleeper-b +slept the full duration: true +sleepers overlapped: true diff --git a/tests/expected/test_sleep_edge_durations.txt b/tests/expected/test_sleep_edge_durations.txt new file mode 100644 index 00000000..84e7a3f7 --- /dev/null +++ b/tests/expected/test_sleep_edge_durations.txt @@ -0,0 +1,3 @@ +edge_durations_return_promptly=true +solo_sleeper_woke=1 +all_woke=8 overlapped=true