RFC 004 chunk 1: substrate — queue_len counter, n_spinning/park_seq/spinning state, spin_budget_cycles+max_spinners Config knobs, pub(crate) futex helpers. Behavioural no-op (feature off by default).
This commit is contained in:
+2
-2
@@ -242,7 +242,7 @@ impl<T> Drop for RawMutexGuard<'_, T> {
|
|||||||
// futex (x86-64 Linux; master is x86-only, see arm-port branch)
|
// 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.
|
// SAFETY: `state` is a valid, aligned u32 for the duration of the call.
|
||||||
// Spurious wakeups and EAGAIN (value already changed) are both handled by
|
// Spurious wakeups and EAGAIN (value already changed) are both handled by
|
||||||
// the caller's retry loop.
|
// 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.
|
// SAFETY: as above.
|
||||||
unsafe {
|
unsafe {
|
||||||
libc::syscall(
|
libc::syscall(
|
||||||
|
|||||||
+83
-2
@@ -118,8 +118,9 @@ use crate::timer::Timers;
|
|||||||
use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor};
|
use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor};
|
||||||
|
|
||||||
use std::sync::atomic::{
|
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::sync::{Arc, Mutex};
|
||||||
use std::thread;
|
use std::thread;
|
||||||
|
|
||||||
@@ -155,6 +156,11 @@ pub struct Config {
|
|||||||
stack_pool_cap: usize,
|
stack_pool_cap: usize,
|
||||||
max_actors: usize,
|
max_actors: usize,
|
||||||
wake_slot: bool,
|
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<u32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Config {
|
impl Config {
|
||||||
@@ -168,6 +174,8 @@ impl Config {
|
|||||||
stack_pool_cap: n * 4,
|
stack_pool_cap: n * 4,
|
||||||
max_actors: DEFAULT_MAX_ACTORS,
|
max_actors: DEFAULT_MAX_ACTORS,
|
||||||
wake_slot: false,
|
wake_slot: false,
|
||||||
|
spin_budget_cycles: 0,
|
||||||
|
max_spinners: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -185,6 +193,8 @@ impl Config {
|
|||||||
stack_pool_cap: max * 4,
|
stack_pool_cap: max * 4,
|
||||||
max_actors: DEFAULT_MAX_ACTORS,
|
max_actors: DEFAULT_MAX_ACTORS,
|
||||||
wake_slot: false,
|
wake_slot: false,
|
||||||
|
spin_budget_cycles: 0,
|
||||||
|
max_spinners: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -241,6 +251,25 @@ impl Config {
|
|||||||
self
|
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.
|
/// The number of scheduler threads this config resolves to.
|
||||||
pub fn resolved_thread_count(&self) -> usize {
|
pub fn resolved_thread_count(&self) -> usize {
|
||||||
if let Some(e) = self.exact {
|
if let Some(e) = self.exact {
|
||||||
@@ -265,6 +294,8 @@ impl Default for Config {
|
|||||||
stack_pool_cap: avail * 4,
|
stack_pool_cap: avail * 4,
|
||||||
max_actors: DEFAULT_MAX_ACTORS,
|
max_actors: DEFAULT_MAX_ACTORS,
|
||||||
wake_slot: false,
|
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
|
/// Incremented in `spawn` before the enqueue; decremented at the very end
|
||||||
/// of `finalize_actor`, after every wakeup finalize produces.
|
/// of `finalize_actor`, after every wakeup finalize produces.
|
||||||
pub(crate) live_actors: AtomicU32,
|
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.
|
/// Timer heap. Independent lock: never nested with any other.
|
||||||
pub(crate) timers: Mutex<Timers>,
|
pub(crate) timers: Mutex<Timers>,
|
||||||
/// IO subsystem. `None` between runs. Lock order: io before everything.
|
/// IO subsystem. `None` between runs. Lock order: io before everything.
|
||||||
@@ -521,6 +579,8 @@ impl RuntimeInner {
|
|||||||
stack_pool_cap: usize,
|
stack_pool_cap: usize,
|
||||||
max_actors: usize,
|
max_actors: usize,
|
||||||
wake_slot: bool,
|
wake_slot: bool,
|
||||||
|
spin_budget_cycles: u64,
|
||||||
|
max_spinners: u32,
|
||||||
) -> Arc<Self> {
|
) -> Arc<Self> {
|
||||||
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
|
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
|
||||||
let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
|
let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
|
||||||
@@ -531,6 +591,12 @@ impl RuntimeInner {
|
|||||||
slots,
|
slots,
|
||||||
free: RawMutex::new(free),
|
free: RawMutex::new(free),
|
||||||
live_actors: AtomicU32::new(0),
|
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()),
|
timers: Mutex::new(Timers::new()),
|
||||||
io: Mutex::new(None),
|
io: Mutex::new(None),
|
||||||
next_monitor_id: AtomicU64::new(0),
|
next_monitor_id: AtomicU64::new(0),
|
||||||
@@ -572,6 +638,10 @@ impl RuntimeInner {
|
|||||||
"enqueue of a pid not in (gen, Queued)"
|
"enqueue of a pid not in (gen, Queued)"
|
||||||
);
|
);
|
||||||
self.run_queue.push(pid);
|
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));
|
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.
|
/// Initialise the runtime with the given config. Returns a reusable handle.
|
||||||
pub fn init(config: Config) -> Runtime {
|
pub fn init(config: Config) -> Runtime {
|
||||||
let n = config.resolved_thread_count();
|
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 {
|
Runtime {
|
||||||
inner: RuntimeInner::new(
|
inner: RuntimeInner::new(
|
||||||
n,
|
n,
|
||||||
@@ -712,6 +786,8 @@ pub fn init(config: Config) -> Runtime {
|
|||||||
config.stack_pool_cap,
|
config.stack_pool_cap,
|
||||||
config.max_actors,
|
config.max_actors,
|
||||||
config.wake_slot,
|
config.wake_slot,
|
||||||
|
config.spin_budget_cycles,
|
||||||
|
max_spinners,
|
||||||
),
|
),
|
||||||
thread_count: n,
|
thread_count: n,
|
||||||
}
|
}
|
||||||
@@ -1207,7 +1283,12 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
|||||||
|
|
||||||
stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed);
|
stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed);
|
||||||
let pop = match inner.run_queue.pop() {
|
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 => {
|
None => {
|
||||||
// Termination does not lean on pop-None being a fence (with
|
// Termination does not lean on pop-None being a fence (with
|
||||||
// the ring queues it is only a snapshot). The argument is
|
// the ring queues it is only a snapshot). The argument is
|
||||||
|
|||||||
Reference in New Issue
Block a user