From a2ef45922885dd687e877d49e8d2a246fc6bb563 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 16 Aug 2026 17:27:38 +0400 Subject: [PATCH] perf(engine): park idle core drivers instead of self-wake spinning The core driver re-armed itself via wake_by_ref() after every poll, an unconditional spin at scheduler speed per worker for the engine's lifetime. The provisioning-reconciler-demo supervisor + 3 node children burned ~750% CPU idle; now ~50% (demo churn), wakeup latency for work delivered to an idle worker bounded by the idle interval. - ExecutionBackend::core_idle_poll() (default Duration::ZERO = previous immediate re-arm) lets a backend opt its drivers into idle parking. - TokioConfig::core_idle_poll (default 500us) configures it for the Tokio backend; from_runtime adopts the default. - CoreDriver: busy tick (or zero interval) re-arms immediately; idle tick arms one backend timer and parks. Every poll still runs exactly one try_tick, so stepping-backend semantics are unchanged. The backend is held Weak and touched only on idle transitions; if the engine is gone the driver parks until the substrate cancels it. - Contract tests: a parked driver observes a late external (engine-invisible) send within the idle interval; the backend reports its configured interval. --- crates/engine/src/backend.rs | 31 ++++++++- crates/engine/src/core_driver.rs | 95 ++++++++++++++++++++------ crates/engine/src/engine.rs | 12 ++-- crates/engine/src/lib.rs | 4 +- crates/engine/src/time.rs | 4 +- crates/engine/src/tokio.rs | 36 ++++++++-- crates/engine/tests/common/mod.rs | 2 - crates/engine/tests/engine_contract.rs | 95 ++++++++++++++++++++++---- crates/engine/tests/engine_unit.rs | 22 +++--- 9 files changed, 229 insertions(+), 72 deletions(-) diff --git a/crates/engine/src/backend.rs b/crates/engine/src/backend.rs index 3e669aa..47c51a3 100644 --- a/crates/engine/src/backend.rs +++ b/crates/engine/src/backend.rs @@ -33,6 +33,16 @@ pub trait ExecutionBackend: Send + Sync + 'static { fn now(&self) -> EngineInstant; /// Report the substrate's advertised capabilities. fn capabilities(&self) -> Capabilities; + /// How long the core driver parks between ticks when its worker is idle. + /// + /// `Duration::ZERO` (the default) re-arms the driver immediately after + /// every tick — a poll loop at scheduler speed. Backends with real timers + /// return a small interval so an idle core parks instead of spinning; + /// newly delivered work is observed within one interval. Every poll still + /// runs one tick, so this only bounds idle wakeup latency. + fn core_idle_poll(&self) -> Duration { + Duration::ZERO + } } /// Capabilities an execution backend advertises. @@ -51,14 +61,29 @@ pub struct Capabilities { impl Capabilities { /// Convenience: only baseline task execution. - pub const TASKS_ONLY: Self = Self { tasks: true, timers: false, blocking: false, io: false }; + pub const TASKS_ONLY: Self = Self { + tasks: true, + timers: false, + blocking: false, + io: false, + }; /// Convenience: every capability. - pub const ALL: Self = Self { tasks: true, timers: true, blocking: true, io: true }; + pub const ALL: Self = Self { + tasks: true, + timers: true, + blocking: true, + io: true, + }; /// Convenience: no capabilities. Reported by an [`EngineHandle`](crate::EngineHandle) /// whose owning engine has been dropped (ENGINE_SPEC.md). - pub const NONE: Self = Self { tasks: false, timers: false, blocking: false, io: false }; + pub const NONE: Self = Self { + tasks: false, + timers: false, + blocking: false, + io: false, + }; /// Whether `self` satisfies every capability marked `true` in `required`. /// diff --git a/crates/engine/src/core_driver.rs b/crates/engine/src/core_driver.rs index 25a2da4..027d648 100644 --- a/crates/engine/src/core_driver.rs +++ b/crates/engine/src/core_driver.rs @@ -1,32 +1,42 @@ //! Core driving loop. //! //! Substrate-neutral: each driver owns one core worker. A poll runs one -//! [`Worker::try_tick`] — synchronous, returns immediately — then re-schedules -//! itself by waking its own waker. There is no backend reference on the hot path, -//! no per-turn boxed yield, no inbox-wake or readiness mechanism, and no -//! `has_work()` gate; later polls observe newly delivered messages +//! [`Worker::try_tick`] — synchronous, returns immediately — then re-arms +//! itself. While its worker keeps doing work the driver re-arms immediately by +//! waking its own waker; once a tick finds no work the driver parks on a +//! backend timer for the backend's idle interval instead of spinning at +//! scheduler speed. There is no inbox-wake or readiness mechanism; the next +//! poll — self-wake or timer — observes newly delivered messages //! (ENGINE_SPEC.md). use std::pin::Pin; -use std::sync::Arc; +use std::sync::{Arc, Weak}; use std::task::{Context, Poll}; +use std::time::Duration; -use crate::backend::{BoxTask, ExecutionBackend}; +use crate::backend::{BoxTask, BoxTimer, ExecutionBackend}; use swactor::worker::Worker; - /// One core-driving loop for one worker. /// -/// Each poll runs one [`Worker::try_tick`] — synchronous, returns immediately — -/// then re-arms itself via `cx.waker().wake_by_ref()` and returns `Pending`. -/// Rescheduling through the waker hands control back to the substrate -/// scheduler between ticks, so other engine work progresses. The driver holds -/// no backend reference and allocates nothing per turn (ENGINE_SPEC.md). +/// Each poll runs one [`Worker::try_tick`] — synchronous, returns immediately. +/// The re-arm policy depends on the outcome and the backend's +/// [`core_idle_poll`](ExecutionBackend::core_idle_poll): /// -/// On a real executor (e.g. Tokio) `wake_by_ref` re-enqueues the task rather -/// than re-polling inline, so other ready tasks run between ticks. On the -/// stepping test backend the waker is a no-op and each `step` re-polls every -/// task, so one `step` still advances the driver by exactly one tick. +/// - The worker did work, or the interval is zero: re-arm immediately via +/// `cx.waker().wake_by_ref()`, handing control back to the substrate +/// scheduler between ticks so other engine work progresses. +/// - The worker was idle: arm one backend timer for the idle interval and park +/// on it. The timer firing re-polls, so newly delivered work is observed +/// within one interval. +/// +/// Either way every poll ticks exactly once. The driver allocates nothing per +/// busy turn and touches the backend only on idle transitions, never on the +/// busy path (ENGINE_SPEC.md §8). +/// +/// On the stepping test backend the default interval is zero and the waker is +/// a no-op; each `step` re-polls every task, so one `step` still advances the +/// driver by exactly one tick. /// /// # Non-reentrancy /// Each worker is moved into exactly one driver, and `try_tick` runs to @@ -37,6 +47,12 @@ use swactor::worker::Worker; /// [`Worker`]: swactor::worker::Worker struct CoreDriver { worker: Worker, + /// Weak backend reference, upgraded only when arming an idle timer. + backend: Weak, + /// Idle park interval; zero means re-arm immediately after every tick. + idle_poll: Duration, + /// The armed idle timer, present only while parked. + idle_timer: Option, } // Core drivers are the engine's sole core-progression path; this is the one @@ -48,11 +64,40 @@ impl Future for CoreDriver { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { // `CoreDriver` is `Unpin`; it contains no self-referential state. let this = self.get_mut(); - this.worker.try_tick(); - // Re-arm immediately: the substrate scheduler redispatches this task, - // yielding to other engine work between ticks. No boxed yield is - // allocated per turn and no backend reference is retained. - cx.waker().wake_by_ref(); + let did_work = this.worker.try_tick(); + + if did_work || this.idle_poll.is_zero() { + // Busy (or the backend has no idle parking): drop any armed + // timer and re-arm immediately. The substrate scheduler + // redispatches this task, yielding to other engine work between + // ticks. + this.idle_timer = None; + cx.waker().wake_by_ref(); + return Poll::Pending; + } + + // Idle: park on a backend timer instead of spinning. + if this.idle_timer.is_none() { + let Some(backend) = this.backend.upgrade() else { + // Engine dropped: the substrate cancels this task on backend + // drop, so park rather than spin. + return Poll::Pending; + }; + this.idle_timer = Some(backend.timer(this.idle_poll)); + } + // Poll the armed timer to register `cx` with it; when it fires the + // task is woken, re-polled, and ticks again. + if this + .idle_timer + .as_mut() + .unwrap() + .as_mut() + .poll(cx) + .is_ready() + { + this.idle_timer = None; + cx.waker().wake_by_ref(); + } Poll::Pending } } @@ -62,8 +107,14 @@ impl Future for CoreDriver { /// One task is allocated per worker at engine construction and runs for the /// engine's lifetime; the substrate cancels it when the backend is dropped. pub(crate) fn install(workers: Vec, backend: &Arc) { + let idle_poll = backend.core_idle_poll(); for worker in workers { - let driver: BoxTask = Box::pin(CoreDriver { worker }); + let driver: BoxTask = Box::pin(CoreDriver { + worker, + backend: Arc::downgrade(backend), + idle_poll, + idle_timer: None, + }); backend.spawn(driver); } } diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index 7e7dbbf..2ee5552 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -4,10 +4,9 @@ use std::sync::{Arc, Weak}; use std::time::Duration; use crate::backend::{Capabilities, EngineError, ExecutionBackend}; -use crate::time::{EngineInstant, Interval, Timer, Timeout}; +use crate::time::{EngineInstant, Interval, Timeout, Timer}; use swactor::runtime::{Runtime, RuntimeParts}; - /// The composite engine: retains a configured core runtime handle and its /// execution backend, and owns one core-driving loop per worker. /// @@ -28,10 +27,7 @@ impl Engine { /// the engine owns every worker and is their sole driver. Construction fails /// if `backend` does not advertise a capability the engine requires (at /// minimum, `tasks`). - pub fn new( - parts: RuntimeParts, - backend: impl ExecutionBackend, - ) -> Result { + pub fn new(parts: RuntimeParts, backend: impl ExecutionBackend) -> Result { let backend: Arc = Arc::new(backend); if !backend.capabilities().tasks { return Err(EngineError::MissingRequiredCapability); @@ -109,7 +105,9 @@ impl EngineHandle { /// fires. pub fn timer(&self, delay: Duration) -> Timer { match self.backend() { - Some(backend) => Timer { inner: backend.timer(delay) }, + Some(backend) => Timer { + inner: backend.timer(delay), + }, None => Timer::closed(), } } diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 8af9f5c..aeee854 100644 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -14,9 +14,7 @@ mod time; #[cfg(feature = "tokio")] mod tokio; -pub use backend::{ - BoxTask, BoxTimer, BoxWork, Capabilities, EngineError, ExecutionBackend, -}; +pub use backend::{BoxTask, BoxTimer, BoxWork, Capabilities, EngineError, ExecutionBackend}; pub use engine::{Engine, EngineHandle}; pub use stepping::SteppingBackend; pub use time::{Elapsed, EngineInstant, Interval, Timeout, Timer}; diff --git a/crates/engine/src/time.rs b/crates/engine/src/time.rs index 65329ab..5e9185c 100644 --- a/crates/engine/src/time.rs +++ b/crates/engine/src/time.rs @@ -59,7 +59,9 @@ impl Timer { /// dropped, so a handle can still produce a [`Timer`] without keeping the /// backend alive (ENGINE_SPEC.md). pub(crate) fn closed() -> Self { - Timer { inner: Box::pin(std::future::pending()) } + Timer { + inner: Box::pin(std::future::pending()), + } } } diff --git a/crates/engine/src/tokio.rs b/crates/engine/src/tokio.rs index 42de05d..4dc2f27 100644 --- a/crates/engine/src/tokio.rs +++ b/crates/engine/src/tokio.rs @@ -18,11 +18,20 @@ use crate::time::EngineInstant; pub struct TokioConfig { /// Number of async worker threads backing the runtime. pub worker_threads: usize, + /// How long a core driver parks between ticks while its worker is idle. + /// + /// Bounds the wakeup latency for work delivered to an idle worker + /// (an external send, transport delivery, process output). Busy workers + /// never park. `Duration::ZERO` restores the always-immediate re-arm. + pub core_idle_poll: Duration, } impl Default for TokioConfig { fn default() -> Self { - Self { worker_threads: 2 } + Self { + worker_threads: 2, + core_idle_poll: Duration::from_micros(500), + } } } @@ -33,6 +42,7 @@ impl Default for TokioConfig { /// [`EngineHandle`](crate::EngineHandle). pub struct TokioBackend { pub(crate) runtime: tokio::runtime::Runtime, + core_idle_poll: Duration, } impl TokioBackend { @@ -52,12 +62,19 @@ impl TokioBackend { .enable_all() .build() .map_err(|e| EngineError::BackendSetup(e.to_string()))?; - Ok(Self { runtime }) + Ok(Self { + runtime, + core_idle_poll: config.core_idle_poll, + }) } /// Adopt a caller-tuned Tokio runtime, moving it into engine ownership. + /// Core drivers park for the default idle interval. pub fn from_runtime(runtime: tokio::runtime::Runtime) -> Self { - Self { runtime } + Self { + runtime, + core_idle_poll: TokioConfig::default().core_idle_poll, + } } } @@ -87,7 +104,9 @@ impl ExecutionBackend for TokioBackend { } fn now(&self) -> EngineInstant { - EngineInstant { instant: Instant::now() } + EngineInstant { + instant: Instant::now(), + } } fn capabilities(&self) -> Capabilities { @@ -102,6 +121,10 @@ impl ExecutionBackend for TokioBackend { io: true, } } + + fn core_idle_poll(&self) -> Duration { + self.core_idle_poll + } } /// A `tokio::time::sleep` whose construction is deferred to first poll. @@ -120,7 +143,10 @@ struct LazySleep { impl LazySleep { fn new(delay: Duration) -> Self { - Self { delay: Some(delay), inner: None } + Self { + delay: Some(delay), + inner: None, + } } } diff --git a/crates/engine/tests/common/mod.rs b/crates/engine/tests/common/mod.rs index 24d7054..ddf7b42 100644 --- a/crates/engine/tests/common/mod.rs +++ b/crates/engine/tests/common/mod.rs @@ -27,12 +27,10 @@ pub fn runtime_parts_with_workers(worker_count: usize) -> (RuntimeParts, Runtime runtime_parts(config) } - pub fn default_parts() -> RuntimeParts { RuntimeParts::new(RuntimeConfig::default()) } - // ── Probe message ─────────────────────────────────────────────────────────── /// A minimal message delivered to probe actors. diff --git a/crates/engine/tests/engine_contract.rs b/crates/engine/tests/engine_contract.rs index 8cc4226..82adb15 100644 --- a/crates/engine/tests/engine_contract.rs +++ b/crates/engine/tests/engine_contract.rs @@ -14,11 +14,10 @@ mod common; use common::*; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize}; use std::sync::atomic::Ordering::SeqCst; +use std::sync::atomic::{AtomicBool, AtomicUsize}; use std::time::Duration; - use swactor_engine::{Engine, TokioBackend, TokioConfig}; /// Outer deadline shared across tests: generous enough to absorb scheduler @@ -60,9 +59,7 @@ fn engine_drives_core_without_application_ticks() { let _engine = Engine::new(parts, backend).expect("construct engine"); // Deliver AFTER engine construction: a later tick must observe it. - runtime - .send_to(addr, Probe) - .expect("deliver probe message"); + runtime.send_to(addr, Probe).expect("deliver probe message"); assert!( wait_for(|| received.load(SeqCst) >= 1, DEADLINE), @@ -116,9 +113,7 @@ fn actor_ticks_and_supporting_work_both_progress() { }); // Deliver an actor message while the supporting work is still active. - runtime - .send_to(addr, Probe) - .expect("deliver probe message"); + runtime.send_to(addr, Probe).expect("deliver probe message"); assert!( wait_for( @@ -204,8 +199,11 @@ fn blocking_work_does_not_stop_actor_ticks() { }) .expect("spawn probe actor"); - let backend = TokioBackend::new(TokioConfig { worker_threads: 1 }) - .expect("build tokio backend"); + let backend = TokioBackend::new(TokioConfig { + worker_threads: 1, + ..Default::default() + }) + .expect("build tokio backend"); let engine = Engine::new(parts, backend).expect("construct engine"); let handle = engine.handle(); @@ -222,9 +220,7 @@ fn blocking_work_does_not_stop_actor_ticks() { }); // Deliver an actor message while the blocking work remains blocked. - runtime - .send_to(addr, Probe) - .expect("deliver probe message"); + runtime.send_to(addr, Probe).expect("deliver probe message"); assert!( wait_for(|| received.load(SeqCst) >= 1, DEADLINE), @@ -265,8 +261,7 @@ fn engine_timer_fires() { let _ = tx.send(()); }); - rx.recv_timeout(DEADLINE) - .expect("engine timer must fire"); + rx.recv_timeout(DEADLINE).expect("engine timer must fire"); } // ── 7.9 ────────────────────────────────────────────────────────────────────── @@ -362,3 +357,73 @@ fn engine_adopts_caller_tuned_tokio_runtime() { "engine must drive core through an adopted runtime" ); } + +// ── 7.12 ────────────────────────────────────────────────────────────────────── + +#[test] +fn idle_core_observes_late_external_work_within_idle_interval() { + // A core driver parks on a timer once its worker goes idle. Work can then + // arrive through paths the engine cannot see (an external send from this + // thread). The parked driver must wake within its idle interval and tick. + // A driver that parks and never re-arms would hang this test. + let (parts, runtime) = default_runtime_parts(); + let received = Arc::new(AtomicUsize::new(0)); + let addr = runtime + .spawn(RecordingProbe { + received: received.clone(), + }) + .expect("spawn probe actor"); + + let idle_poll = Duration::from_millis(50); + let backend = TokioBackend::new(TokioConfig { + core_idle_poll: idle_poll, + ..Default::default() + }) + .expect("build tokio backend"); + let _engine = Engine::new(parts, backend).expect("construct engine"); + + // First message proves the driver ran before parking. + runtime.send_to(addr, Probe).expect("deliver first probe"); + assert!( + wait_for(|| received.load(SeqCst) >= 1, DEADLINE), + "driver must process work before going idle" + ); + // Long enough for every worker driver to observe no work and park. + std::thread::sleep(Duration::from_millis(300)); + + // External send from outside the engine: invisible to any engine-side + // wake path, observed only by the parked driver's timer. + runtime.send_to(addr, Probe).expect("deliver late probe"); + + let start = std::time::Instant::now(); + assert!( + wait_for(|| received.load(SeqCst) >= 2, DEADLINE), + "a parked core driver must still observe external work" + ); + // The idle timer is free-running (armed once per idle transition, re-armed + // on each firing), so the send lands at a random phase within one + // interval: observed latency is uniform in [0, idle_poll). Bound it + // generously to absorb scheduler jitter without asserting the phase. + let elapsed = start.elapsed(); + assert!( + elapsed < idle_poll * 10, + "late work observed in {elapsed:?}, far beyond the idle interval" + ); +} + +#[test] +fn tokio_backend_reports_configured_core_idle_poll() { + use swactor_engine::ExecutionBackend; + + let configured = Duration::from_millis(5); + let backend = TokioBackend::new(TokioConfig { + core_idle_poll: configured, + ..Default::default() + }) + .expect("build tokio backend"); + assert_eq!( + backend.core_idle_poll(), + configured, + "backend must report its configured idle interval" + ); +} diff --git a/crates/engine/tests/engine_unit.rs b/crates/engine/tests/engine_unit.rs index 148d8fb..f1f58de 100644 --- a/crates/engine/tests/engine_unit.rs +++ b/crates/engine/tests/engine_unit.rs @@ -9,14 +9,11 @@ mod common; use common::*; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize}; use std::sync::atomic::Ordering::SeqCst; +use std::sync::atomic::{AtomicBool, AtomicUsize}; use std::time::Duration; - -use swactor_engine::{ - Capabilities, Engine, EngineError, ExecutionBackend, SteppingBackend, -}; +use swactor_engine::{Capabilities, Engine, EngineError, ExecutionBackend, SteppingBackend}; #[cfg(feature = "tokio")] use swactor_engine::{TokioBackend, TokioConfig}; @@ -305,9 +302,7 @@ fn stepping_engine_installs_one_driver_per_worker() { #[test] fn stepping_engine_drives_every_worker() { let (parts, runtime) = runtime_parts_with_workers(3); - let counters: Vec<_> = (0..3) - .map(|_| Arc::new(AtomicUsize::new(0))) - .collect(); + let counters: Vec<_> = (0..3).map(|_| Arc::new(AtomicUsize::new(0))).collect(); let addrs: Vec<_> = counters .iter() .map(|received| { @@ -391,10 +386,7 @@ fn stepping_core_and_supporting_work_both_progress() { backend.step(); } - assert!( - steps.load(SeqCst) >= 10, - "supporting work must finish" - ); + assert!(steps.load(SeqCst) >= 10, "supporting work must finish"); assert!( received.load(SeqCst) >= 1, "actor message must be processed" @@ -580,7 +572,9 @@ fn dropping_engine_releases_backend_even_with_live_handles() { let parts = default_parts(); let engine = Engine::new( parts, - SentinelBackend { sentinel: sentinel.clone() }, + SentinelBackend { + sentinel: sentinel.clone(), + }, ) .expect("tasks capability present"); let handle = engine.handle(); @@ -632,4 +626,4 @@ fn handle_used_after_engine_drop_degrades_gracefully() { handle.spawn_blocking(|| {}); let _never_fires = handle.timer(Duration::from_secs(1)); let _never_ticks = handle.interval(Duration::from_secs(1)); -} \ No newline at end of file +}