diff --git a/Cargo.toml b/Cargo.toml index dbbb6f0..7b458d8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,17 +25,16 @@ codegen-units = 1 [[bench]] name = "primes" harness = false - + [[bench]] name = "multi_scheduler" harness = false - + [[bench]] name = "general" harness = false - + [[bench]] name = "smarm_favored" harness = false - diff --git a/benches/baseline.json b/benches/baseline.json index fbe9ccd..4a72d5b 100644 --- a/benches/baseline.json +++ b/benches/baseline.json @@ -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 } } } \ No newline at end of file diff --git a/src/runtime.rs b/src/runtime.rs index b4b49ac..68259aa 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -72,6 +72,7 @@ pub struct Config { exact: Option, 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, pub(crate) timers: Timers, pub(crate) io: Option, - /// 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>, } 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 { - 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>, + /// 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 { + fn new(thread_count: usize, alloc_interval: u32, timeslice_cycles: u64, stack_pool_cap: usize) -> Arc { 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, 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, 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)) — 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), + Idle { + next_deadline: Option, + io_outstanding: u32, + wake_fd: Option, + }, + 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, 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); } diff --git a/src/scheduler.rs b/src/scheduler.rs index 8c6a2d2..eae3dda 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -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) { pub fn run(f: F) { crate::runtime::init(crate::runtime::Config::exact(1)).run(f); } - - -