Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: reduce scheduler mutex contention + stack pool

Browse Source 
Three changes, each independently measured, landed together.

Fuse pre-resume mutex acquisitions
-----------------------------------
The schedule loop previously took the shared lock three separate times
per actor resume: once to pop the run queue, once to read the stack
pointer, and once to pop the first-resume closure. These are now a
single acquisition that returns everything needed to resume the actor.

As a side effect, pending_closures was changed from a Vec<(Pid, Closure)>
with O(n) linear scan to a Vec<Option<Closure>> indexed by slot index,
making first-closure lookup O(1).

Move stack allocation outside the shared lock
----------------------------------------------
Stack::new() (mmap + mprotect) was previously called inside with_shared,
stalling every other scheduler thread for the duration of two syscalls on
every spawn. It now runs before the lock is acquired.

Stack pool
----------
Rather than munmap-ing a stack when an actor finishes and mmap-ing a fresh
one on the next spawn, stacks are now recycled through a per-Runtime pool
(Mutex<Vec<Stack>>). finalize_actor extracts the stack from the Actor
before clearing the slot and pushes it to the pool outside the shared lock.
spawn_under pops from the pool before falling back to Stack::new().

The pool is unbounded for now (shrink policy TBD) but capped at
stack_pool_cap stacks on return, defaulting to thread_count * 4.
The cap is configurable via Config::stack_pool_cap(n).

Results (24-thread, against stored baseline)
--------------------------------------------
chained_spawn      smarm 1-thread:   9763 → 261 µs   (-97%)
chained_spawn      smarm 24-thread: 23562 → 838 µs   (-96%)
ping_pong_oneshot  smarm 1-thread:  18409 → 742 µs   (-96%)
ping_pong_oneshot  smarm 24-thread: 44596 → 1425 µs  (-97%)
catch_unwind_panics smarm 24-thread: 267812 → 124094 µs (-54%)
fan_out_compute    smarm 24-thread:   2839 → 2226 µs  (-22%)

Tokio regressions in the checker output are baseline measurement drift;
no tokio code was changed.
This commit is contained in:
smarm
2026-05-25 23:28:11 +02:00
parent 64be3f23ac
commit 72f5d38e5d
4 changed files with 216 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
-1
View File
@@ -38,4 +38,3 @@ harness = false
name = "smarm_favored"
harness = false
benches/baseline.json
+83 -83
View File
@@ -2,185 +2,185 @@
"chained_spawn": {
"smarm 1-thread": {
"result": 1000,
"median": 9763,
"min": 8852,
"max": 10520
"median": 273,
"min": 261,
"max": 292
},
"smarm 24-thread": {
"result": 1000,
"median": 23562,
"min": 20515,
"max": 24398
"median": 837,
"min": 788,
"max": 913
},
"tokio current_thread": {
"result": 1000,
"median": 70,
"median": 68,
"min": 61,
"max": 75
"max": 113
},
"tokio multi-thread": {
"result": 1000,
"median": 187,
"min": 158,
"max": 228
"median": 200,
"min": 176,
"max": 230
}
},
"yield_many": {
"smarm 1-thread": {
"result": 200000,
"median": 21704,
"min": 20353,
"max": 22053
"median": 19465,
"min": 19299,
"max": 20007
},
"smarm 24-thread": {
"result": 200000,
"median": 180752,
"min": 148576,
"max": 187771
"median": 163921,
"min": 155700,
"max": 167921
},
"tokio current_thread": {
"result": 200000,
"median": 5816,
"min": 5567,
"max": 6278
"median": 5022,
"min": 5001,
"max": 5530
},
"tokio multi-thread": {
"result": 200000,
"median": 8259,
"min": 7760,
"max": 9610
"median": 8341,
"min": 7412,
"max": 9199
}
},
"fan_out_compute": {
"smarm 1-thread": {
"result": 33860,
"median": 14009,
"min": 13834,
"max": 14274
"median": 14096,
"min": 13982,
"max": 17406
},
"smarm 24-thread": {
"result": 33860,
"median": 2839,
"min": 2702,
"max": 3019
"median": 2475,
"min": 2264,
"max": 2589
},
"tokio current_thread": {
"result": 33860,
"median": 13083,
"min": 12634,
"max": 15737
"median": 12640,
"min": 12548,
"max": 15881
},
"tokio multi-thread": {
"result": 33860,
"median": 1538,
"min": 1434,
"max": 2201
"median": 1483,
"min": 1395,
"max": 1710
}
},
"ping_pong_oneshot": {
"smarm 1-thread": {
"result": 1000,
"median": 18409,
"min": 16381,
"max": 21858
"median": 801,
"min": 720,
"max": 929
},
"smarm 24-thread": {
"result": 1000,
"median": 44596,
"min": 41804,
"max": 48324
"median": 1449,
"min": 1292,
"max": 1530
},
"tokio current_thread": {
"result": 1000,
"median": 494,
"min": 465,
"max": 602
"median": 398,
"min": 369,
"max": 407
},
"tokio multi-thread": {
"result": 1000,
"median": 10907,
"min": 9582,
"max": 12430
"median": 10497,
"min": 9238,
"max": 11961
}
},
"deep_recursion": {
"smarm 1-thread": {
"result": 1,
"median": 123,
"min": 108,
"max": 169
"median": 136,
"min": 117,
"max": 163
},
"smarm 24-thread": {
"result": 1,
"median": 633,
"min": 558,
"max": 783
"median": 743,
"min": 654,
"max": 845
},
"tokio current_thread": {
"result": 1,
"median": 12,
"min": 11,
"max": 14
"median": 14,
"min": 14,
"max": 16
},
"tokio multi-thread": {
"result": 1,
"median": 52,
"min": 42,
"max": 55
"median": 61,
"min": 47,
"max": 94
}
},
"yield_in_hot_loop": {
"smarm 1-thread": {
"result": 1000000,
"median": 82273,
"min": 81870,
"max": 86041
"median": 83281,
"min": 80264,
"max": 85697
},
"tokio current_thread": {
"result": 1000000,
"median": 75353,
"min": 71841,
"max": 77573
"median": 75538,
"min": 69842,
"max": 78823
}
},
"uncontended_channel": {
"smarm 1-thread": {
"result": 1000000,
"median": 9390,
"min": 9331,
"max": 12504
"median": 9440,
"min": 9393,
"max": 13168
},
"tokio current_thread": {
"result": 1000000,
"median": 17007,
"min": 16933,
"max": 17657
"median": 17817,
"min": 16926,
"max": 18572
}
},
"catch_unwind_panics": {
"smarm 1-thread": {
"result": 10000,
"median": 144807,
"min": 138473,
"max": 153664
"median": 133840,
"min": 130425,
"max": 139709
},
"smarm 24-thread": {
"result": 10000,
"median": 267812,
"min": 229587,
"max": 288329
"median": 128729,
"min": 108091,
"max": 132700
},
"tokio current_thread": {
"result": 10000,
"median": 11084,
"min": 10904,
"max": 12041
"median": 10681,
"min": 10158,
"max": 11644
},
"tokio multi-thread": {
"result": 10000,
"median": 6029,
"min": 5054,
"max": 7216
"median": 6700,
"min": 2778,
"max": 7453
}
}
}
src/runtime.rs
+114 -50
View File
@@ -72,6 +72,7 @@ pub struct Config {
exact: Option<usize>,
alloc_interval: u32,
timeslice_cycles: u64,
stack_pool_cap: usize,
}
impl Config {
@@ -82,6 +83,7 @@ impl Config {
min: n, max: n, exact: Some(n),
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
stack_pool_cap: n * 4,
}
}
@@ -96,6 +98,7 @@ impl Config {
min, max, exact,
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
stack_pool_cap: max * 4,
}
}
@@ -116,6 +119,14 @@ impl Config {
self
}
/// Maximum number of stacks kept in the pool for reuse across spawns.
/// A larger cap reduces `mmap`/`munmap` syscalls at the cost of idle memory.
/// Default: `thread_count * 4`.
pub fn stack_pool_cap(mut self, n: usize) -> Self {
self.stack_pool_cap = n;
self
}
/// The number of scheduler threads this config resolves to.
pub fn resolved_thread_count(&self) -> usize {
if let Some(e) = self.exact {
@@ -137,6 +148,7 @@ impl Default for Config {
min: 1, max: avail, exact: None,
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
stack_pool_cap: avail * 4,
}
}
}
@@ -244,8 +256,10 @@ pub(crate) struct SharedState {
pub(crate) root_pid: Option<Pid>,
pub(crate) timers: Timers,
pub(crate) io: Option<IoThread>,
/// Closures awaiting their first resume, keyed by Pid.
pub(crate) pending_closures: Vec<(Pid, Closure)>,
/// Closures awaiting their first resume, indexed by slot index.
/// `pending_closures[idx]` is `Some` iff the actor at slot `idx` has not
/// yet been resumed for the first time. Grows on demand; never shrinks.
pub(crate) pending_closures: Vec<Option<Closure>>,
}
impl SharedState {
@@ -257,7 +271,7 @@ impl SharedState {
root_pid: None,
timers: Timers::new(),
io: None,
pending_closures: Vec::new(),
pending_closures: Vec::new(), // indexed by slot index
}
}
@@ -283,8 +297,9 @@ impl SharedState {
}
pub(crate) fn pop_pending_closure(&mut self, pid: Pid) -> Option<Closure> {
let pos = self.pending_closures.iter().position(|(p, _)| *p == pid)?;
Some(self.pending_closures.swap_remove(pos).1)
self.pending_closures
.get_mut(pid.index() as usize)
.and_then(|cell| cell.take())
}
}
@@ -304,10 +319,15 @@ pub(crate) struct RuntimeInner {
/// Preemption knobs, written into each scheduler thread's locals on startup.
pub(crate) alloc_interval: u32,
pub(crate) timeslice_cycles: u64,
/// Recycled stacks waiting to be reused by the next spawn.
/// Grows like a Vec (doubles capacity); shrink policy is TBD.
pub(crate) stack_pool: Mutex<Vec<crate::stack::Stack>>,
/// Maximum number of stacks to retain in the pool.
pub(crate) stack_pool_cap: usize,
}
impl RuntimeInner {
fn new(thread_count: usize, alloc_interval: u32, timeslice_cycles: u64) -> Arc<Self> {
fn new(thread_count: usize, alloc_interval: u32, timeslice_cycles: u64, stack_pool_cap: usize) -> Arc<Self> {
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
Arc::new(Self {
shared: Mutex::new(SharedState::new()),
@@ -317,6 +337,8 @@ impl RuntimeInner {
sleeping: AtomicU32::new(0),
alloc_interval,
timeslice_cycles,
stack_pool: Mutex::new(Vec::new()),
stack_pool_cap,
})
}
@@ -346,7 +368,7 @@ pub struct Runtime {
pub fn init(config: Config) -> Runtime {
let n = config.resolved_thread_count();
Runtime {
inner: RuntimeInner::new(n, config.alloc_interval, config.timeslice_cycles),
inner: RuntimeInner::new(n, config.alloc_interval, config.timeslice_cycles, config.stack_pool_cap),
thread_count: n,
}
}
@@ -514,15 +536,28 @@ fn finalize_actor(inner: &Arc<RuntimeInner>, pid: Pid, outcome: Outcome) {
),
};
let (waiters, supervisor_pid) = inner.with_shared(|s| {
let (waiters, supervisor_pid, recycled_stack) = inner.with_shared(|s| {
let slot = s.slot_mut(pid).expect("finalize_actor: slot vanished");
let sup = slot.actor.as_ref().map(|a| a.supervisor);
// Extract the stack before clearing the actor so we can recycle it
// into the pool *outside* the shared lock.
let stack = slot.actor.take().map(|a| a.stack);
slot.outcome = Some(joiner_outcome);
slot.state = State::Done;
slot.actor = None;
(std::mem::take(&mut slot.waiters), sup)
(std::mem::take(&mut slot.waiters), sup, stack)
});
// Return the stack to the pool outside the shared lock. The pool grows
// like a Vec (doubles capacity); if we are already at the cap we just
// drop the stack (munmap) instead of retaining it.
if let Some(stack) = recycled_stack {
let mut pool = inner.stack_pool.lock().unwrap();
if pool.len() < inner.stack_pool_cap {
pool.push(stack);
}
// else: drop here → munmap, same as before
}
// Deliver to supervisor.
if let Some(sup) = supervisor_pid {
let sender = inner.with_shared(|s| {
@@ -640,47 +675,84 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
} // drain_guard drops here
// ----------------------------------------------------------------
// 2. Pop a runnable actor from the shared queue.
// 2 + 3. Pop a runnable actor and grab everything we need to resume
// it — all in a single mutex acquisition.
//
// Returns:
// Some((pid, sp, Option<first-resume closure>)) — got work
// None with all_clear=true — done, exit
// None with all_clear=false — idle, wait
// ----------------------------------------------------------------
let pid = match inner.with_shared(|s| {
enum PopResult {
Work(Pid, usize, Option<Closure>),
Idle {
next_deadline: Option<std::time::Instant>,
io_outstanding: u32,
wake_fd: Option<std::os::fd::RawFd>,
},
AllDone,
}
// SAFETY: the raw pointer is the actor's stack pointer, valid for the
// lifetime of the actor's Slot. It is only dereferenced by
// switch_to_actor() on this same thread immediately after this block.
unsafe impl Send for PopResult {} // closure is Send; usize sp is plain data
let pop = inner.with_shared(|s| {
let len = s.run_queue.len() as u64;
stats.run_queue_len.store(len, Ordering::Relaxed);
s.run_queue.pop_front()
}) {
Some(p) => {
crate::te!(crate::trace::Event::Dequeue(p));
p
}
None => {
// Queue was empty when we popped. Re-examine under the lock to
// decide whether to exit or wait. All four conditions must hold
// simultaneously before we exit:
if let Some(pid) = s.run_queue.pop_front() {
// Happy path: got a PID, grab SP and optional first closure.
let sp = s.slot(pid)
.and_then(|slot| slot.actor.as_ref().map(|a| a.sp));
match sp {
Some(sp) => {
let closure = s.pop_pending_closure(pid);
PopResult::Work(pid, sp, closure)
}
None => {
// Stale PID (actor already finalized). Treat as idle
// so the outer loop retries immediately.
PopResult::Idle {
next_deadline: None,
io_outstanding: 0,
wake_fd: None,
}
}
}
} else {
// Queue was empty. Re-examine to decide exit vs wait.
// All four conditions must hold simultaneously before we exit:
// 1. run queue is still empty
// 2. no live actors (nothing parked, nothing mid-finalize)
// 3. no pending timers
// 4. no outstanding IO
// If any is non-zero we keep spinning — "check the fridge is
// empty before you leave for the airport".
let (next_deadline, io_outstanding, wake_fd, all_clear) =
inner.with_shared(|s| {
let next = s.timers.peek_deadline();
let (out, fd) = match s.io.as_ref() {
Some(io) => (
io.outstanding + io.waiters.len() as u32,
Some(io.wake_fd()),
),
None => (0, None),
};
let live = s.slots.iter().filter(|slot| slot.actor.is_some()).count();
let queue_empty = s.run_queue.is_empty();
let all_clear = queue_empty && live == 0 && next.is_none() && out == 0;
(next, out, fd, all_clear)
});
let next = s.timers.peek_deadline();
let (out, fd) = match s.io.as_ref() {
Some(io) => (
io.outstanding + io.waiters.len() as u32,
Some(io.wake_fd()),
),
None => (0, None),
};
let live = s.slots.iter().filter(|slot| slot.actor.is_some()).count();
let all_clear = s.run_queue.is_empty() && live == 0
&& next.is_none() && out == 0;
if all_clear {
return;
PopResult::AllDone
} else {
PopResult::Idle { next_deadline: next, io_outstanding: out, wake_fd: fd }
}
}
});
let (pid, sp, first_closure) = match pop {
PopResult::Work(pid, sp, closure) => {
crate::te!(crate::trace::Event::Dequeue(pid));
(pid, sp, closure)
}
PopResult::AllDone => return,
PopResult::Idle { next_deadline, io_outstanding, wake_fd } => {
// Something is still in flight. Sleep on the appropriate source
// to avoid hammering the mutex; the loop will retry on wake.
match (next_deadline, wake_fd) {
@@ -712,17 +784,9 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
// ----------------------------------------------------------------
// 3. Resume the actor.
// ----------------------------------------------------------------
let sp = match inner.with_shared(|s| {
s.slot(pid).and_then(|slot| slot.actor.as_ref().map(|a| a.sp))
}) {
Some(sp) => sp,
None => {
continue; // stale pid
}
};
// First resume: move the closure into the trampoline's thread-local.
if let Some(b) = inner.with_shared(|s| s.pop_pending_closure(pid)) {
if let Some(b) = first_closure {
set_current_actor_box(b);
}
src/scheduler.rs
+16 -7
View File
@@ -132,13 +132,20 @@ pub fn spawn(f: impl FnOnce() + Send + 'static) -> JoinHandle {
}
pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHandle {
// Try to reuse a stack from the pool; fall back to a fresh mmap if empty.
// Allocation happens before taking the shared lock so any syscall doesn't
// stall other scheduler threads.
let stack = with_runtime(|inner| inner.stack_pool.lock().unwrap().pop())
.unwrap_or_else(|| {
crate::stack::Stack::new(crate::runtime::ACTOR_STACK_SIZE)
.expect("stack allocation failed")
});
let sp = init_actor_stack(stack.top(), crate::actor::trampoline);
let pid = with_runtime(|inner| {
inner.with_shared(|s| {
let (idx, gen) = s.allocate_slot();
let pid = Pid::new(idx, gen);
let stack = crate::stack::Stack::new(crate::runtime::ACTOR_STACK_SIZE)
.expect("stack allocation failed");
let sp = init_actor_stack(stack.top(), crate::actor::trampoline);
let slot = &mut s.slots[idx as usize];
slot.actor = Some(crate::actor::Actor { pid, stack, sp, supervisor });
slot.state = crate::runtime::State::Runnable;
@@ -149,7 +156,12 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHa
slot.pending_unpark = false;
slot.pending_io_result = None;
s.run_queue.push_back(pid);
s.pending_closures.push((pid, Box::new(f) as crate::runtime::Closure));
// Grow the closures vec to cover this slot index, then store.
let idx = idx as usize;
if s.pending_closures.len() <= idx {
s.pending_closures.resize_with(idx + 1, || None);
}
s.pending_closures[idx] = Some(Box::new(f) as crate::runtime::Closure);
crate::te!(crate::trace::Event::Spawn { parent: supervisor, child: pid });
crate::te!(crate::trace::Event::Enqueue(pid));
pid
@@ -344,6 +356,3 @@ pub fn register_supervisor_channel(pid: Pid, sender: Sender<Signal>) {
pub fn run<F: FnOnce() + Send + 'static>(f: F) {
crate::runtime::init(crate::runtime::Config::exact(1)).run(f);
}