RFC 004 chunk 2: spin-then-park idle policy — submit rule (Dekker-paired wake), CAS-capped spinner enlistment, bounded spin on queue_len, futex park with lost-wakeup + shutdown guards, last-spinner handoff, AllDone broadcast. Gated on spin_budget_cycles>0; default-off path unchanged (full suite green).

src/runtime.rs

@@ -639,12 +639,126 @@ impl RuntimeInner {
         );
         self.run_queue.push(pid);
         // RFC 004: maintain the lock-free depth the idle spinners poll. Release
-        // so a spinner's Acquire load observes the pushed entry. (The submit
+        // so a spinner's Acquire load observes the pushed entry.
-        // wake rule that pairs with this lands in chunk 2.)
         self.queue_len.fetch_add(1, Ordering::Release);
+        // RFC 004 submit rule: ensure a searcher will see this work. A live
+        // spinner (n_spinning > 0) claims it with no syscall — the common,
+        // cheap path; only when nobody is searching do we pay a wake. The
+        // SeqCst fence forbids reordering the n_spinning load ahead of the
+        // queue_len store above. Paired with the parker's symmetric
+        // (decrement n_spinning; SeqCst fence; re-check queue_len), it
+        // guarantees at least one of {we observe the spinner, the parker
+        // observes this work} — the Dekker property that keeps work from
+        // stranding with every scheduler asleep. Loom does not model the
+        // futex path (see sync_shim.rs), so this is correct by construction.
+        if self.spinning {
+            fence(Ordering::SeqCst);
+            if self.n_spinning.load(Ordering::Acquire) == 0 {
+                self.unpark_one_scheduler();
+            }
+        }
         crate::te!(crate::trace::Event::Enqueue(pid));
     }
 
+    /// RFC 004: wake exactly one parked scheduler. Bumping the futex sequence
+    /// before the wake closes the arm→wait race: a parker that already
+    /// captured the old sequence fails its `futex_wait` value-compare and
+    /// returns (EAGAIN) instead of sleeping through the wake. Only called when
+    /// no scheduler is searching, so it is off the common path.
+    pub(crate) fn unpark_one_scheduler(&self) {
+        self.park_seq.fetch_add(1, Ordering::Release);
+        futex_wake(&self.park_seq, 1);
+    }
+
+    /// RFC 004: release ALL futex-parked schedulers (termination). Required in
+    /// a no-io runtime, where there is no wake pipe to fall back on: a sibling
+    /// parked indefinitely on the futex would otherwise sleep past program
+    /// end. Each woken thread re-runs the idle verdict, reaches AllDone, and
+    /// returns — idempotent, mirroring the io wake-pipe terminal broadcast.
+    pub(crate) fn unpark_all_schedulers(&self) {
+        self.park_seq.fetch_add(1, Ordering::Release);
+        futex_wake(&self.park_seq, i32::MAX);
+    }
+
+    /// RFC 004: the no-timer / no-io idle policy. Spin on `queue_len` for the
+    /// configured budget (if under the spinner cap), then park on the futex.
+    /// Returns when the caller should loop back and re-attempt a pop. Only
+    /// invoked when `self.spinning` (budget > 0); with budget 0 the caller
+    /// keeps the historical `thread::sleep` park.
+    fn idle_spin_then_park(&self) {
+        // 1. Try to enlist as a spinner, strictly honouring the cap (CAS, so
+        //    `n_spinning` never exceeds `max_spinners` even transiently).
+        let mut spinning = false;
+        loop {
+            let c = self.n_spinning.load(Ordering::Acquire);
+            if c >= self.max_spinners {
+                break; // cap met (or 0): skip the spin, park directly
+            }
+            if self
+                .n_spinning
+                .compare_exchange_weak(c, c + 1, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
+                spinning = true;
+                break;
+            }
+        }
+
+        if spinning {
+            let start = crate::preempt::rdtsc();
+            loop {
+                if self.queue_len.load(Ordering::Acquire) > 0 {
+                    // Found work. Leave the spinner set; if we were the LAST
+                    // spinner (1 → 0), wake one parked sibling before going
+                    // heads-down (rule 4) so a burst still arriving is still
+                    // observed by a searcher.
+                    let prev = self.n_spinning.fetch_sub(1, Ordering::AcqRel);
+                    if prev == 1 {
+                        fence(Ordering::SeqCst);
+                        if self.queue_len.load(Ordering::Acquire) > 0 {
+                            self.unpark_one_scheduler();
+                        }
+                    }
+                    return; // back to the pop
+                }
+                if crate::preempt::rdtsc().wrapping_sub(start) >= self.spin_budget_cycles {
+                    break; // budget exhausted: park
+                }
+                core::hint::spin_loop();
+            }
+            // Budget exhausted: leave the spinner set, then park (below). The
+            // decrement must precede the park's queue re-check — that is the
+            // parker half of the submit-rule Dekker pairing.
+            self.n_spinning.fetch_sub(1, Ordering::AcqRel);
+        }
+
+        // 2. Park on the futex. Three guards close every lost-wakeup / hang:
+        //    - SeqCst fence + queue_len re-check: the Dekker pairing with the
+        //      submit rule (which loads n_spinning AFTER publishing the work),
+        //      so a submit that declined to wake us because it saw us spinning
+        //      is observed here instead.
+        //    - capture park_seq BEFORE the final re-check: a submit-wake
+        //      landing in the gap bumps the sequence, so futex_wait's
+        //      value-compare returns immediately rather than sleeping.
+        //    - live_actors re-check: the shutdown guard. In this arm io_out is
+        //      0 by construction (the io>0 case is handled by a sibling arm),
+        //      so live_actors == 0 IS the AllDone verdict; if termination is
+        //      racing our park we loop back and reach AllDone instead of
+        //      sleeping past program end. (unpark_all_schedulers covers the
+        //      already-parked case from the other side.)
+        fence(Ordering::SeqCst);
+        let observed = self.park_seq.load(Ordering::Acquire);
+        if self.queue_len.load(Ordering::Acquire) > 0 {
+            return;
+        }
+        if self.live_actors.load(Ordering::Acquire) == 0 {
+            return;
+        }
+        futex_wait(&self.park_seq, observed);
+        // Woken by a wake, a stale-sequence EAGAIN, or spuriously: loop back
+        // and re-pop. The verdict re-runs from scratch, so any wake is benign.
+    }
+
     /// Make `pid` runnable if it is parked; coalesce or defer otherwise.
     /// The runtime-internal core of `scheduler::unpark`. WILDCARD wake:
     /// consumes the epoch but does not check it — reserved for terminal
@@ -1330,6 +1444,13 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                     if let Some(io) = inner.io.lock().unwrap().as_ref() {
                         io.wake();
                     }
+                    // RFC 004: a no-io runtime has no wake pipe; futex-parked
+                    // siblings must be released or shutdown hangs. Broadcast;
+                    // each re-runs the verdict and reaches AllDone itself. The
+                    // sequence bump also EAGAINs any sibling mid-park.
+                    if inner.spinning {
+                        inner.unpark_all_schedulers();
+                    }
                     return;
                 }
                 Pop::Idle { io_outstanding, wake_fd } => {
@@ -1355,7 +1476,16 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                             crate::io::drain_wake_pipe(fd);
                         }
                         _ => {
-                            thread::sleep(std::time::Duration::from_micros(100));
+                            // RFC 004: the no-timer / no-io idle arm — the
+                            // ~100µs thread::sleep worst case this RFC targets.
+                            // With spinning enabled, spin-before-park on a
+                            // wakeable futex; with budget 0 the historical
+                            // sleep is preserved exactly.
+                            if inner.spinning {
+                                inner.idle_spin_then_park();
+                            } else {
+                                thread::sleep(std::time::Duration::from_micros(100));
+                            }
                         }
                     }
                     continue;