diff --git a/src/raw_mutex.rs b/src/raw_mutex.rs index 628c898..59d8cd2 100644 --- a/src/raw_mutex.rs +++ b/src/raw_mutex.rs @@ -242,7 +242,7 @@ impl Drop for RawMutexGuard<'_, T> { // futex (x86-64 Linux; master is x86-only, see arm-port branch) // --------------------------------------------------------------------------- -fn futex_wait(state: &AtomicU32, expected: u32) { +pub(crate) fn futex_wait(state: &AtomicU32, expected: u32) { // SAFETY: `state` is a valid, aligned u32 for the duration of the call. // Spurious wakeups and EAGAIN (value already changed) are both handled by // the caller's retry loop. @@ -257,7 +257,7 @@ fn futex_wait(state: &AtomicU32, expected: u32) { } } -fn futex_wake(state: &AtomicU32, n: i32) { +pub(crate) fn futex_wake(state: &AtomicU32, n: i32) { // SAFETY: as above. unsafe { libc::syscall( diff --git a/src/runtime.rs b/src/runtime.rs index dc1d9a2..bdbe277 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -118,8 +118,9 @@ use crate::timer::Timers; use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor}; use std::sync::atomic::{ - AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering, + fence, AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering, }; +use crate::raw_mutex::{futex_wait, futex_wake}; use std::sync::{Arc, Mutex}; use std::thread; @@ -155,6 +156,11 @@ pub struct Config { stack_pool_cap: usize, max_actors: usize, wake_slot: bool, + /// RFC 004: TSC cycles a scheduler spins before parking. `0` (default) + /// disables spinning entirely — the historical park-immediately behaviour. + spin_budget_cycles: u64, + /// RFC 004: cap on concurrent spinners. `None` resolves to N/2 at init. + max_spinners: Option, } impl Config { @@ -168,6 +174,8 @@ impl Config { stack_pool_cap: n * 4, max_actors: DEFAULT_MAX_ACTORS, wake_slot: false, + spin_budget_cycles: 0, + max_spinners: None, } } @@ -185,6 +193,8 @@ impl Config { stack_pool_cap: max * 4, max_actors: DEFAULT_MAX_ACTORS, wake_slot: false, + spin_budget_cycles: 0, + max_spinners: None, } } @@ -241,6 +251,25 @@ impl Config { self } + /// RFC 004: TSC cycles an idle scheduler spins searching for work before + /// parking on the futex. The idle-CPU-vs-wake-latency dial, in the same + /// unit as [`timeslice_cycles`](Self::timeslice_cycles) (uses `rdtsc()`). + /// `0` (the default) disables spinning and the futex park entirely, + /// recovering the historical `thread::sleep` idle behaviour exactly. + pub fn spin_budget_cycles(mut self, n: u64) -> Self { + self.spin_budget_cycles = n; + self + } + + /// RFC 004: cap on the number of schedulers that may spin concurrently. + /// Default (`None`) resolves to N/2 at init. Capping at half keeps half + /// the pool hot and parks the rest — the half-load case is then the easy + /// case. Ignored when `spin_budget_cycles == 0`. + pub fn max_spinners(mut self, n: u32) -> Self { + self.max_spinners = Some(n); + self + } + /// The number of scheduler threads this config resolves to. pub fn resolved_thread_count(&self) -> usize { if let Some(e) = self.exact { @@ -265,6 +294,8 @@ impl Default for Config { stack_pool_cap: avail * 4, max_actors: DEFAULT_MAX_ACTORS, wake_slot: false, + spin_budget_cycles: 0, + max_spinners: None, } } } @@ -484,6 +515,33 @@ pub(crate) struct RuntimeInner { /// Incremented in `spawn` before the enqueue; decremented at the very end /// of `finalize_actor`, after every wakeup finalize produces. pub(crate) live_actors: AtomicU32, + /// RFC 004: lock-free shared-run-queue depth. `+1` in `enqueue` (the sole + /// `run_queue.push` site — covers spawn-publish, unpark, and slot + /// displacement), `-1` on a successful pop. This is the poll source the + /// idle spinners read; it never takes the queue mutex. Slot-parked work is + /// deliberately NOT counted: it is thread-local and only its owning thread + /// can resume it, so it is not work a spinning sibling could grab. + pub(crate) queue_len: AtomicU64, + /// RFC 004: schedulers currently *searching* for work (spinning) — the + /// middle state between running and parked. The submit rule loads this to + /// decide whether a wake is needed (a live spinner will catch the work + /// with no syscall). Capped at `max_spinners`. + pub(crate) n_spinning: AtomicU32, + /// RFC 004: the scheduler park futex word. A monotonic sequence: a waker + /// does `fetch_add(1) + futex_wake`, a parker captures the value and + /// `futex_wait`s on it, so a bump landing in the arm→wait gap returns the + /// wait immediately (EAGAIN) instead of sleeping — the lost-wakeup guard. + /// One shared word; `futex_wake(1)` wakes exactly one parked scheduler. + pub(crate) park_seq: AtomicU32, + /// RFC 004: `spin_budget_cycles > 0`. When false the feature is fully off — + /// the idle path is the historical `thread::sleep` park, `n_spinning` and + /// `park_seq` are never touched, and the submit rule is a no-op. This is + /// what makes `spin_budget_cycles = 0` recover today's behaviour exactly. + pub(crate) spinning: bool, + /// RFC 004: TSC cycles a scheduler searches before parking (`Config`). + pub(crate) spin_budget_cycles: u64, + /// RFC 004: cap on concurrent spinners (`Config`, default N/2). + pub(crate) max_spinners: u32, /// Timer heap. Independent lock: never nested with any other. pub(crate) timers: Mutex, /// IO subsystem. `None` between runs. Lock order: io before everything. @@ -521,6 +579,8 @@ impl RuntimeInner { stack_pool_cap: usize, max_actors: usize, wake_slot: bool, + spin_budget_cycles: u64, + max_spinners: u32, ) -> Arc { let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect(); let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect(); @@ -531,6 +591,12 @@ impl RuntimeInner { slots, free: RawMutex::new(free), live_actors: AtomicU32::new(0), + queue_len: AtomicU64::new(0), + n_spinning: AtomicU32::new(0), + park_seq: AtomicU32::new(0), + spinning: spin_budget_cycles > 0, + spin_budget_cycles, + max_spinners, timers: Mutex::new(Timers::new()), io: Mutex::new(None), next_monitor_id: AtomicU64::new(0), @@ -572,6 +638,10 @@ impl RuntimeInner { "enqueue of a pid not in (gen, Queued)" ); 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 + // wake rule that pairs with this lands in chunk 2.) + self.queue_len.fetch_add(1, Ordering::Release); crate::te!(crate::trace::Event::Enqueue(pid)); } @@ -704,6 +774,10 @@ pub struct Runtime { /// Initialise the runtime with the given config. Returns a reusable handle. pub fn init(config: Config) -> Runtime { let n = config.resolved_thread_count(); + // RFC 004: default spinner cap is N/2 (capping at half keeps half the pool + // hot, parks the rest). With N=1 this is 0 — a lone thread has no + // cross-thread handoff to hide, so it never spins. + let max_spinners = config.max_spinners.unwrap_or((n / 2) as u32); Runtime { inner: RuntimeInner::new( n, @@ -712,6 +786,8 @@ pub fn init(config: Config) -> Runtime { config.stack_pool_cap, config.max_actors, config.wake_slot, + config.spin_budget_cycles, + max_spinners, ), thread_count: n, } @@ -1207,7 +1283,12 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed); let pop = match inner.run_queue.pop() { - Some(pid) => Pop::Got(pid), + Some(pid) => { + // RFC 004: mirror the enqueue increment. Release pairs with + // the spinners' Acquire load of queue_len. + inner.queue_len.fetch_sub(1, Ordering::Release); + Pop::Got(pid) + } None => { // Termination does not lean on pop-None being a fence (with // the ring queues it is only a snapshot). The argument is