feat: move TimerWheel from core to std via WorkerExtension
Add WorkerExtension trait (on_tick, handle_request, gc_dead) as a per-worker counterpart to the shared RuntimeExtension. This enables per-worker state like timer wheels to live outside core. - TimerWheel, TimerRequest, CloneMsg → crates/std/src/timer_wheel.rs - CtxTimers extension trait replaces Ctx::send_after_ticks/send_interval_ticks - schedule_timer → generic post_worker_request on ContextInner - Worker.timers → Worker.worker_ext (Option<Box<dyn WorkerExtension>>) - StdExtension factory creates TimerWheel per worker Core: 2440→2320 (−120). worker.rs: 696→599 (−97). Cumulative: 2636→2320 (−316, −12.0%). worker.rs: 845→599 (−29.1%). Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski
This commit is contained in:
parent
fd8f4b2a05
commit
8342c16de4
12 changed files with 285 additions and 203 deletions
|
|
@ -2,6 +2,7 @@ use swactor::actor::{ActorAddress, ActorInterface, Ctx, Message, MonitorRef};
|
||||||
use swactor::Error;
|
use swactor::Error;
|
||||||
|
|
||||||
use crate::StdExtension;
|
use crate::StdExtension;
|
||||||
|
use crate::timer_wheel::{CloneMsg, TimerRequest};
|
||||||
|
|
||||||
fn get_ext<'a>(ctx: &'a Ctx) -> &'a StdExtension {
|
fn get_ext<'a>(ctx: &'a Ctx) -> &'a StdExtension {
|
||||||
ctx.extension()
|
ctx.extension()
|
||||||
|
|
@ -96,6 +97,43 @@ impl CtxWatching for Ctx<'_> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Timer extension for [`Ctx`].
|
||||||
|
///
|
||||||
|
/// Provides `send_after_ticks` / `send_interval_ticks` via the per-worker
|
||||||
|
/// [`TimerWheel`](crate::timer_wheel::TimerWheel).
|
||||||
|
pub trait CtxTimers {
|
||||||
|
/// Schedule a one-shot timer: deliver `msg` to `addr` after `ticks` worker ticks.
|
||||||
|
///
|
||||||
|
/// The message is delivered as a normal mailbox message during the fire tick,
|
||||||
|
/// before `tick_all` processes messages. The timer is tick-counted (deterministic),
|
||||||
|
/// not wall-clock based.
|
||||||
|
fn send_after_ticks<M: Message>(&self, addr: ActorAddress, msg: M, ticks: u64);
|
||||||
|
|
||||||
|
/// Schedule a repeating timer: deliver a clone of `msg` to `addr` every `period` ticks.
|
||||||
|
///
|
||||||
|
/// The first delivery happens after `period` ticks. The message is cloned for each
|
||||||
|
/// delivery. The timer continues until the target actor is stopped/poisoned.
|
||||||
|
fn send_interval_ticks<M: Message>(&self, addr: ActorAddress, msg: M, period: u64);
|
||||||
|
}
|
||||||
|
|
||||||
|
impl CtxTimers for Ctx<'_> {
|
||||||
|
fn send_after_ticks<M: Message>(&self, addr: ActorAddress, msg: M, ticks: u64) {
|
||||||
|
self.raw_inner().post_worker_request(Box::new(TimerRequest::Once {
|
||||||
|
dest: addr,
|
||||||
|
msg: Box::new(msg),
|
||||||
|
ticks,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
fn send_interval_ticks<M: Message>(&self, addr: ActorAddress, msg: M, period: u64) {
|
||||||
|
self.raw_inner().post_worker_request(Box::new(TimerRequest::Interval {
|
||||||
|
dest: addr,
|
||||||
|
msg: Box::new(msg) as Box<dyn CloneMsg>,
|
||||||
|
period,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Group extension for [`Ctx`].
|
/// Group extension for [`Ctx`].
|
||||||
///
|
///
|
||||||
/// Provides `join_group`, `leave_group`, `publish`, and `group_members` via
|
/// Provides `join_group`, `leave_group`, `publish`, and `group_members` via
|
||||||
|
|
|
||||||
|
|
@ -1,11 +1,12 @@
|
||||||
use std::any::Any;
|
use std::any::Any;
|
||||||
|
|
||||||
use swactor::actor::{ActorAddress, Down, ExitReason, StopReason};
|
use swactor::actor::{ActorAddress, Down, ExitReason, StopReason};
|
||||||
use swactor::extension::RuntimeExtension;
|
use swactor::extension::{RuntimeExtension, WorkerExtension};
|
||||||
|
|
||||||
use crate::group_registry::GroupRegistry;
|
use crate::group_registry::GroupRegistry;
|
||||||
use crate::monitor_registry::MonitorRegistry;
|
use crate::monitor_registry::MonitorRegistry;
|
||||||
use crate::name_registry::NameRegistry;
|
use crate::name_registry::NameRegistry;
|
||||||
|
use crate::timer_wheel::TimerWheel;
|
||||||
use crate::watch_registry::WatchRegistry;
|
use crate::watch_registry::WatchRegistry;
|
||||||
|
|
||||||
/// Standard library extension — provides naming, monitoring, watching, and group registries.
|
/// Standard library extension — provides naming, monitoring, watching, and group registries.
|
||||||
|
|
@ -80,4 +81,8 @@ impl RuntimeExtension for StdExtension {
|
||||||
fn as_any(&self) -> &dyn Any {
|
fn as_any(&self) -> &dyn Any {
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn create_worker_extension(&self) -> Option<Box<dyn WorkerExtension>> {
|
||||||
|
Some(Box::new(TimerWheel::new()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -4,6 +4,7 @@ pub mod name_registry;
|
||||||
pub mod monitor_registry;
|
pub mod monitor_registry;
|
||||||
pub mod watch_registry;
|
pub mod watch_registry;
|
||||||
pub mod group_registry;
|
pub mod group_registry;
|
||||||
|
pub(crate) mod timer_wheel;
|
||||||
mod extension;
|
mod extension;
|
||||||
mod ctx_ext;
|
mod ctx_ext;
|
||||||
mod runtime_ext;
|
mod runtime_ext;
|
||||||
|
|
@ -11,5 +12,5 @@ mod runtime_ext;
|
||||||
pub use supervisor::{ChildSpec, RestartPolicy, Supervisor, SupervisorStrategy};
|
pub use supervisor::{ChildSpec, RestartPolicy, Supervisor, SupervisorStrategy};
|
||||||
pub use router::{Router, RoutingStrategy};
|
pub use router::{Router, RoutingStrategy};
|
||||||
pub use extension::StdExtension;
|
pub use extension::StdExtension;
|
||||||
pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups, CtxWatching};
|
pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups, CtxWatching, CtxTimers};
|
||||||
pub use runtime_ext::{RuntimeNaming, RuntimeGroups, RuntimeWatching};
|
pub use runtime_ext::{RuntimeNaming, RuntimeGroups, RuntimeWatching};
|
||||||
|
|
|
||||||
148
crates/std/src/timer_wheel.rs
Normal file
148
crates/std/src/timer_wheel.rs
Normal file
|
|
@ -0,0 +1,148 @@
|
||||||
|
use std::any::Any;
|
||||||
|
|
||||||
|
use swactor::actor::{ActorAddress, Message};
|
||||||
|
use swactor::extension::WorkerExtension;
|
||||||
|
|
||||||
|
// ─── Cloneable Message Trait ────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Type-erased cloneable message for interval timers.
|
||||||
|
/// Since `Message: Clone`, all actor messages implement this.
|
||||||
|
pub(crate) trait CloneMsg: Send {
|
||||||
|
fn clone_boxed(&self) -> Box<dyn Any + Send>;
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<M: Message> CloneMsg for M {
|
||||||
|
fn clone_boxed(&self) -> Box<dyn Any + Send> {
|
||||||
|
Box::new(self.clone())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─── Timer Request ──────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Timer request from a handler, queued for processing after tick_all.
|
||||||
|
pub(crate) enum TimerRequest {
|
||||||
|
/// One-shot: deliver `msg` to `dest` after `ticks` worker ticks.
|
||||||
|
Once {
|
||||||
|
dest: ActorAddress,
|
||||||
|
msg: Box<dyn Any + Send>,
|
||||||
|
ticks: u64,
|
||||||
|
},
|
||||||
|
/// Repeating: deliver a clone of `msg` to `dest` every `period` ticks.
|
||||||
|
Interval {
|
||||||
|
dest: ActorAddress,
|
||||||
|
msg: Box<dyn CloneMsg>,
|
||||||
|
period: u64,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─── Timer Wheel ────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
struct OnceTimer {
|
||||||
|
fire_at: u64,
|
||||||
|
dest: ActorAddress,
|
||||||
|
msg: Box<dyn Any + Send>,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct IntervalTimer {
|
||||||
|
next_fire: u64,
|
||||||
|
period: u64,
|
||||||
|
dest: ActorAddress,
|
||||||
|
msg: Box<dyn CloneMsg>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Per-worker tick-counting timer wheel.
|
||||||
|
///
|
||||||
|
/// Timers are deterministic (tick-counted, not wall-clock). One-shot timers
|
||||||
|
/// fire once and are consumed; interval timers fire repeatedly every N ticks.
|
||||||
|
pub struct TimerWheel {
|
||||||
|
current_tick: u64,
|
||||||
|
once_timers: Vec<OnceTimer>,
|
||||||
|
interval_timers: Vec<IntervalTimer>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl TimerWheel {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
current_tick: 0,
|
||||||
|
once_timers: Vec::new(),
|
||||||
|
interval_timers: Vec::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Advance the tick counter and collect all due timer messages.
|
||||||
|
fn fire(&mut self) -> Vec<(ActorAddress, Box<dyn Any + Send>)> {
|
||||||
|
self.current_tick += 1;
|
||||||
|
let tick = self.current_tick;
|
||||||
|
let mut result = Vec::new();
|
||||||
|
|
||||||
|
// Fire one-shot timers (swap-remove for O(1) removal)
|
||||||
|
let mut i = 0;
|
||||||
|
while i < self.once_timers.len() {
|
||||||
|
if self.once_timers[i].fire_at <= tick {
|
||||||
|
let timer = self.once_timers.swap_remove(i);
|
||||||
|
result.push((timer.dest, timer.msg));
|
||||||
|
} else {
|
||||||
|
i += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fire interval timers
|
||||||
|
for timer in &mut self.interval_timers {
|
||||||
|
if timer.next_fire <= tick {
|
||||||
|
let msg = timer.msg.clone_boxed();
|
||||||
|
result.push((timer.dest, msg));
|
||||||
|
timer.next_fire = tick + timer.period;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
result
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Remove interval timers whose target was just removed from the worker.
|
||||||
|
fn gc_dead_intervals(&mut self, dead: &[ActorAddress]) {
|
||||||
|
if dead.is_empty() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
self.interval_timers
|
||||||
|
.retain(|t| !dead.iter().any(|d| *d == t.dest));
|
||||||
|
}
|
||||||
|
|
||||||
|
fn add_once(&mut self, dest: ActorAddress, msg: Box<dyn Any + Send>, ticks: u64) {
|
||||||
|
self.once_timers.push(OnceTimer {
|
||||||
|
fire_at: self.current_tick + ticks,
|
||||||
|
dest,
|
||||||
|
msg,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
fn add_interval(&mut self, dest: ActorAddress, msg: Box<dyn CloneMsg>, period: u64) {
|
||||||
|
let period = period.max(1); // prevent zero-period infinite loop
|
||||||
|
self.interval_timers.push(IntervalTimer {
|
||||||
|
next_fire: self.current_tick + period,
|
||||||
|
period,
|
||||||
|
dest,
|
||||||
|
msg,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl WorkerExtension for TimerWheel {
|
||||||
|
fn on_tick(&mut self) -> Vec<(ActorAddress, Box<dyn Any + Send>)> {
|
||||||
|
self.fire()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn handle_request(&mut self, request: Box<dyn Any + Send>) {
|
||||||
|
if let Ok(req) = request.downcast::<TimerRequest>() {
|
||||||
|
match *req {
|
||||||
|
TimerRequest::Once { dest, msg, ticks } => self.add_once(dest, msg, ticks),
|
||||||
|
TimerRequest::Interval { dest, msg, period } => {
|
||||||
|
self.add_interval(dest, msg, period)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn gc_dead(&mut self, dead: &[ActorAddress]) {
|
||||||
|
self.gc_dead_intervals(dead);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -11,6 +11,7 @@ cargo-fuzz = true
|
||||||
libfuzzer-sys = { version = "0.4", features = ["arbitrary-derive"] }
|
libfuzzer-sys = { version = "0.4", features = ["arbitrary-derive"] }
|
||||||
arbitrary = { version = "1", features = ["derive"] }
|
arbitrary = { version = "1", features = ["derive"] }
|
||||||
swactor = { path = "..", default-features = true }
|
swactor = { path = "..", default-features = true }
|
||||||
|
swactor-std = { path = "../crates/std" }
|
||||||
|
|
||||||
# Prevent this from interfering with workspaces
|
# Prevent this from interfering with workspaces
|
||||||
[workspace]
|
[workspace]
|
||||||
|
|
|
||||||
|
|
@ -7,9 +7,12 @@ use std::sync::OnceLock;
|
||||||
use arbitrary::Arbitrary;
|
use arbitrary::Arbitrary;
|
||||||
use libfuzzer_sys::fuzz_target;
|
use libfuzzer_sys::fuzz_target;
|
||||||
|
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
use swactor::actor::{ActorAddress, ActorInterface};
|
use swactor::actor::{ActorAddress, ActorInterface};
|
||||||
use swactor::config::RuntimeConfig;
|
use swactor::config::RuntimeConfig;
|
||||||
use swactor::runtime::{Ctx, Inbox, Runtime};
|
use swactor::runtime::{Ctx, Inbox, Runtime};
|
||||||
|
use swactor_std::{CtxTimers, StdExtension};
|
||||||
|
|
||||||
// ─── Run Logging ────────────────────────────────────────────────────────────
|
// ─── Run Logging ────────────────────────────────────────────────────────────
|
||||||
// FUZZ_LOG=1 → trace every run
|
// FUZZ_LOG=1 → trace every run
|
||||||
|
|
@ -881,7 +884,8 @@ fuzz_target!(|input: FuzzInput| {
|
||||||
num_threads: 1,
|
num_threads: 1,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
let rt = Runtime::new(config);
|
let rt = Runtime::new(config)
|
||||||
|
.with_extension(Arc::new(StdExtension::new()));
|
||||||
let mut state = FuzzState::new(rt, tracing);
|
let mut state = FuzzState::new(rt, tracing);
|
||||||
|
|
||||||
let scenario_limit = input.scenarios.len().min(64);
|
let scenario_limit = input.scenarios.len().min(64);
|
||||||
|
|
|
||||||
63
src/actor.rs
63
src/actor.rs
|
|
@ -199,47 +199,19 @@ pub struct Down {
|
||||||
/// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`.
|
/// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`.
|
||||||
pub(crate) struct StopSignal;
|
pub(crate) struct StopSignal;
|
||||||
|
|
||||||
/// Type-erased cloneable message for interval timers.
|
|
||||||
/// Since `Message: Clone`, all actor messages can implement this.
|
|
||||||
pub(crate) trait CloneMsg: Send {
|
|
||||||
fn clone_boxed(&self) -> Box<dyn Any + Send>;
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<M: Message> CloneMsg for M {
|
|
||||||
fn clone_boxed(&self) -> Box<dyn Any + Send> {
|
|
||||||
Box::new(self.clone())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Timer request from a handler, queued for processing after tick_all.
|
|
||||||
pub(crate) enum TimerRequest {
|
|
||||||
/// One-shot: deliver `msg` to `dest` after `ticks` worker ticks.
|
|
||||||
Once {
|
|
||||||
dest: ActorAddress,
|
|
||||||
msg: Box<dyn Any + Send>,
|
|
||||||
ticks: u64,
|
|
||||||
},
|
|
||||||
/// Repeating: deliver a clone of `msg` to `dest` every `period` ticks.
|
|
||||||
Interval {
|
|
||||||
dest: ActorAddress,
|
|
||||||
msg: Box<dyn CloneMsg>,
|
|
||||||
period: u64,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Object-safe inner trait for sending type-erased messages.
|
/// Object-safe inner trait for sending type-erased messages.
|
||||||
///
|
///
|
||||||
/// Minimal core interface: send, spawn, stop, timers, and extension access.
|
/// Minimal core interface: send, spawn, stop, and extension access.
|
||||||
/// Registry methods (naming, monitoring, groups) are provided by extension
|
/// Registry methods (naming, monitoring, groups) and timer scheduling
|
||||||
/// traits in `swactor-std`.
|
/// are provided by extension traits in `swactor-std`.
|
||||||
#[allow(private_interfaces)]
|
#[allow(private_interfaces)]
|
||||||
pub trait ContextInner {
|
pub trait ContextInner {
|
||||||
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
|
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
|
||||||
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>);
|
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>);
|
||||||
/// Request graceful stop for an actor. Takes effect after the current message.
|
/// Request graceful stop for an actor. Takes effect after the current message.
|
||||||
fn request_stop(&self, addr: ActorAddress);
|
fn request_stop(&self, addr: ActorAddress);
|
||||||
/// Schedule a timer (one-shot or interval).
|
/// Post a request to the per-worker extension (e.g., timer scheduling).
|
||||||
fn schedule_timer(&self, request: TimerRequest);
|
fn post_worker_request(&self, request: Box<dyn Any + Send>);
|
||||||
/// Access the runtime extension (if installed).
|
/// Access the runtime extension (if installed).
|
||||||
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension>;
|
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension>;
|
||||||
}
|
}
|
||||||
|
|
@ -303,29 +275,4 @@ impl<'a> Ctx<'a> {
|
||||||
self.inner.send_any(addr, Box::new(StopSignal))
|
self.inner.send_any(addr, Box::new(StopSignal))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Schedule a one-shot timer: deliver `msg` to `addr` after `ticks` worker ticks.
|
|
||||||
///
|
|
||||||
/// The message is delivered as a normal mailbox message during the fire tick,
|
|
||||||
/// before `tick_all` processes messages. The timer is tick-counted (deterministic),
|
|
||||||
/// not wall-clock based.
|
|
||||||
pub fn send_after_ticks<M: Message>(&self, addr: ActorAddress, msg: M, ticks: u64) {
|
|
||||||
self.inner.schedule_timer(TimerRequest::Once {
|
|
||||||
dest: addr,
|
|
||||||
msg: Box::new(msg),
|
|
||||||
ticks,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Schedule a repeating timer: deliver a clone of `msg` to `addr` every `period` ticks.
|
|
||||||
///
|
|
||||||
/// The first delivery happens after `period` ticks. The message is cloned for each
|
|
||||||
/// delivery. The timer continues until the target actor is stopped/poisoned.
|
|
||||||
pub fn send_interval_ticks<M: Message>(&self, addr: ActorAddress, msg: M, period: u64) {
|
|
||||||
self.inner.schedule_timer(TimerRequest::Interval {
|
|
||||||
dest: addr,
|
|
||||||
msg: Box::new(msg),
|
|
||||||
period,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -25,4 +25,29 @@ pub trait RuntimeExtension: Send + Sync {
|
||||||
|
|
||||||
/// Downcast support for Ctx extension traits.
|
/// Downcast support for Ctx extension traits.
|
||||||
fn as_any(&self) -> &dyn Any;
|
fn as_any(&self) -> &dyn Any;
|
||||||
|
|
||||||
|
/// Create a per-worker extension instance. Called once per worker during init.
|
||||||
|
///
|
||||||
|
/// Unlike `RuntimeExtension` (shared across all workers), each worker owns
|
||||||
|
/// its own `WorkerExtension` instance for per-worker state like timer wheels.
|
||||||
|
fn create_worker_extension(&self) -> Option<Box<dyn WorkerExtension>> {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Per-worker extension state, created by [`RuntimeExtension::create_worker_extension`].
|
||||||
|
///
|
||||||
|
/// Each worker owns its own instance. Core calls these methods during tick phases:
|
||||||
|
/// - `on_tick`: phase 2.5 — before tick_all, returns messages to deliver
|
||||||
|
/// - `handle_request`: phase 5.5 — processes deferred requests from handlers
|
||||||
|
/// - `gc_dead`: after cleanup_dead — removes state for dead actors
|
||||||
|
pub trait WorkerExtension: Send {
|
||||||
|
/// Called each tick before tick_all. Returns messages to deliver.
|
||||||
|
fn on_tick(&mut self) -> Vec<(ActorAddress, Box<dyn Any + Send>)>;
|
||||||
|
|
||||||
|
/// Process a deferred request posted during handle() via `post_worker_request`.
|
||||||
|
fn handle_request(&mut self, request: Box<dyn Any + Send>);
|
||||||
|
|
||||||
|
/// Clean up state for dead actors.
|
||||||
|
fn gc_dead(&mut self, dead: &[ActorAddress]);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,7 @@ use std::thread::{self, JoinHandle};
|
||||||
use std::thread::Thread;
|
use std::thread::Thread;
|
||||||
use crate::Instant;
|
use crate::Instant;
|
||||||
|
|
||||||
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal, TimerRequest};
|
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal};
|
||||||
use crate::channel::{Receiver, Sender};
|
use crate::channel::{Receiver, Sender};
|
||||||
// Re-export config types so existing code using `runtime::RuntimeConfig` still works
|
// Re-export config types so existing code using `runtime::RuntimeConfig` still works
|
||||||
pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
|
pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
|
||||||
|
|
@ -245,6 +245,12 @@ impl Runtime {
|
||||||
///
|
///
|
||||||
/// Must be called before `run()` or `tick()`.
|
/// Must be called before `run()` or `tick()`.
|
||||||
pub fn with_extension(mut self, ext: Arc<dyn RuntimeExtension>) -> Self {
|
pub fn with_extension(mut self, ext: Arc<dyn RuntimeExtension>) -> Self {
|
||||||
|
// Create per-worker extensions (e.g., timer wheels)
|
||||||
|
for worker in self.tick_workers.get_mut().iter_mut() {
|
||||||
|
if let Some(wext) = ext.create_worker_extension() {
|
||||||
|
worker.worker_ext = Some(wext);
|
||||||
|
}
|
||||||
|
}
|
||||||
self.extension = Some(ext);
|
self.extension = Some(ext);
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
@ -503,11 +509,10 @@ impl ContextInner for Runtime {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn schedule_timer(&self, _request: TimerRequest) {
|
fn post_worker_request(&self, _request: Box<dyn Any + Send>) {
|
||||||
// Timers are per-worker and tick-counted; scheduling from outside
|
// Worker requests (e.g., timers) are per-worker; posting from outside
|
||||||
// a worker context (e.g., rt.spawn() callback) is not supported.
|
// a worker context (e.g., rt.spawn() callback) is not supported.
|
||||||
// Use rt.send_to() with a delay loop instead.
|
eprintln!("swactor: post_worker_request called outside worker context — ignored");
|
||||||
eprintln!("swactor: schedule_timer called outside worker context — ignored");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn extension(&self) -> Option<&dyn RuntimeExtension> {
|
fn extension(&self) -> Option<&dyn RuntimeExtension> {
|
||||||
|
|
|
||||||
169
src/worker.rs
169
src/worker.rs
|
|
@ -6,107 +6,14 @@ use std::sync::Arc;
|
||||||
use std::thread;
|
use std::thread;
|
||||||
use crate::Instant;
|
use crate::Instant;
|
||||||
|
|
||||||
use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopReason, StopSignal, TimerRequest};
|
use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, StopReason, StopSignal};
|
||||||
use crate::channel::Receiver;
|
use crate::channel::Receiver;
|
||||||
use crate::config::MailboxOverflow;
|
use crate::config::MailboxOverflow;
|
||||||
use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId};
|
use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId};
|
||||||
use crate::stats::{ActorSnapshot, TickTiming, WorkerStats};
|
use crate::stats::{ActorSnapshot, TickTiming, WorkerStats};
|
||||||
use crate::Error;
|
use crate::Error;
|
||||||
|
|
||||||
// ─── Per-Worker Timer Wheel ─────────────────────────────────────────────────
|
use crate::extension::WorkerExtension;
|
||||||
|
|
||||||
struct OnceTimer {
|
|
||||||
fire_at: u64,
|
|
||||||
dest: ActorAddress,
|
|
||||||
msg: Box<dyn Any + Send>,
|
|
||||||
}
|
|
||||||
|
|
||||||
struct IntervalTimer {
|
|
||||||
next_fire: u64,
|
|
||||||
period: u64,
|
|
||||||
dest: ActorAddress,
|
|
||||||
msg: Box<dyn CloneMsg>,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Per-worker tick-counting timer wheel.
|
|
||||||
///
|
|
||||||
/// Timers are deterministic (tick-counted, not wall-clock). One-shot timers
|
|
||||||
/// fire once and are consumed; interval timers fire repeatedly every N ticks.
|
|
||||||
struct TimerWheel {
|
|
||||||
current_tick: u64,
|
|
||||||
once_timers: Vec<OnceTimer>,
|
|
||||||
interval_timers: Vec<IntervalTimer>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl TimerWheel {
|
|
||||||
fn new() -> Self {
|
|
||||||
Self {
|
|
||||||
current_tick: 0,
|
|
||||||
once_timers: Vec::new(),
|
|
||||||
interval_timers: Vec::new(),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Advance the tick counter and collect all due timer messages.
|
|
||||||
/// Returns the messages to be routed by the caller (may target local or remote actors/inboxes).
|
|
||||||
fn fire(&mut self) -> Vec<(ActorAddress, Box<dyn Any + Send>)> {
|
|
||||||
self.current_tick += 1;
|
|
||||||
let tick = self.current_tick;
|
|
||||||
let mut result = Vec::new();
|
|
||||||
|
|
||||||
// Fire one-shot timers (swap-remove for O(1) removal)
|
|
||||||
let mut i = 0;
|
|
||||||
while i < self.once_timers.len() {
|
|
||||||
if self.once_timers[i].fire_at <= tick {
|
|
||||||
let timer = self.once_timers.swap_remove(i);
|
|
||||||
result.push((timer.dest, timer.msg));
|
|
||||||
} else {
|
|
||||||
i += 1;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Fire interval timers
|
|
||||||
for timer in &mut self.interval_timers {
|
|
||||||
if timer.next_fire <= tick {
|
|
||||||
let msg = timer.msg.clone_boxed();
|
|
||||||
result.push((timer.dest, msg));
|
|
||||||
timer.next_fire = tick + timer.period;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
result
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Remove interval timers whose target was just removed from the worker.
|
|
||||||
/// Only GCs timers for addresses in `dead` — inboxes and cross-worker actors
|
|
||||||
/// are not in the local pool but are still valid targets.
|
|
||||||
fn gc_dead_intervals(&mut self, dead: &[ActorAddress]) {
|
|
||||||
if dead.is_empty() {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
self.interval_timers.retain(|t| !dead.iter().any(|d| *d == t.dest));
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Add a one-shot timer.
|
|
||||||
fn add_once(&mut self, dest: ActorAddress, msg: Box<dyn Any + Send>, ticks: u64) {
|
|
||||||
self.once_timers.push(OnceTimer {
|
|
||||||
fire_at: self.current_tick + ticks,
|
|
||||||
dest,
|
|
||||||
msg,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Add an interval timer. First fire is after `period` ticks.
|
|
||||||
fn add_interval(&mut self, dest: ActorAddress, msg: Box<dyn CloneMsg>, period: u64) {
|
|
||||||
let period = period.max(1); // prevent zero-period infinite loop
|
|
||||||
self.interval_timers.push(IntervalTimer {
|
|
||||||
next_fire: self.current_tick + period,
|
|
||||||
period,
|
|
||||||
dest,
|
|
||||||
msg,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// ─── Worker ─────────────────────────────────────────────────────────────────
|
// ─── Worker ─────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
@ -119,8 +26,8 @@ pub(crate) struct Worker {
|
||||||
stats: Arc<WorkerStats>,
|
stats: Arc<WorkerStats>,
|
||||||
/// Reusable scratch buffer for building per-actor snapshots.
|
/// Reusable scratch buffer for building per-actor snapshots.
|
||||||
snapshot_buf: Vec<ActorSnapshot>,
|
snapshot_buf: Vec<ActorSnapshot>,
|
||||||
/// Per-worker tick-counting timer wheel.
|
/// Per-worker extension (e.g., timer wheel). Created by RuntimeExtension factory.
|
||||||
timers: TimerWheel,
|
pub(crate) worker_ext: Option<Box<dyn WorkerExtension>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Worker {
|
impl Worker {
|
||||||
|
|
@ -139,7 +46,7 @@ impl Worker {
|
||||||
spawn_rx,
|
spawn_rx,
|
||||||
stats,
|
stats,
|
||||||
snapshot_buf: Vec::new(),
|
snapshot_buf: Vec::new(),
|
||||||
timers: TimerWheel::new(),
|
worker_ext: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -175,32 +82,31 @@ impl Worker {
|
||||||
}
|
}
|
||||||
let t2 = Instant::now();
|
let t2 = Instant::now();
|
||||||
|
|
||||||
// 2.5. Fire due timers → deliver to mailboxes before tick_all
|
// 2.5. Fire per-worker extension (e.g., timers) → deliver before tick_all
|
||||||
let timer_msgs = self.timers.fire();
|
if let Some(ext) = &mut self.worker_ext {
|
||||||
for (dest, msg) in timer_msgs {
|
for (dest, msg) in ext.on_tick() {
|
||||||
if self.pool.contains(&dest) {
|
if self.pool.contains(&dest) {
|
||||||
// Same-worker: deliver directly to actor's mailbox
|
self.pool.deliver(&dest, msg);
|
||||||
self.pool.deliver(&dest, msg);
|
} else {
|
||||||
} else {
|
match tc.address_map.lookup(&dest) {
|
||||||
// Inbox or cross-worker: route through address map / inbox registry
|
Some(wid) => {
|
||||||
match tc.address_map.lookup(&dest) {
|
tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg));
|
||||||
Some(wid) => {
|
crate::runtime::notify_worker(tc.worker_threads, wid.as_usize());
|
||||||
tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg));
|
}
|
||||||
crate::runtime::notify_worker(tc.worker_threads, wid.as_usize());
|
None => {
|
||||||
}
|
let _ = tc.inbox_registry.try_deliver(dest, msg);
|
||||||
None => {
|
}
|
||||||
let _ = tc.inbox_registry.try_deliver(dest, msg);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
did_work = true;
|
||||||
}
|
}
|
||||||
did_work = true;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 3. Tick all actors with WorkerContext
|
// 3. Tick all actors with WorkerContext
|
||||||
let pending_local: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
|
let pending_local: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
|
||||||
RefCell::new(Vec::new());
|
RefCell::new(Vec::new());
|
||||||
let stop_requests: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
let stop_requests: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
||||||
let timer_requests: RefCell<Vec<TimerRequest>> = RefCell::new(Vec::new());
|
let worker_requests: RefCell<Vec<Box<dyn Any + Send>>> = RefCell::new(Vec::new());
|
||||||
|
|
||||||
let processed;
|
let processed;
|
||||||
{
|
{
|
||||||
|
|
@ -209,7 +115,7 @@ impl Worker {
|
||||||
tc,
|
tc,
|
||||||
pending_local: &pending_local,
|
pending_local: &pending_local,
|
||||||
stop_requests: &stop_requests,
|
stop_requests: &stop_requests,
|
||||||
timer_requests: &timer_requests,
|
worker_requests: &worker_requests,
|
||||||
stats: &self.stats,
|
stats: &self.stats,
|
||||||
};
|
};
|
||||||
processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget, &stop_requests);
|
processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget, &stop_requests);
|
||||||
|
|
@ -245,15 +151,10 @@ impl Worker {
|
||||||
self.pool.deliver(&addr, msg);
|
self.pool.deliver(&addr, msg);
|
||||||
}
|
}
|
||||||
|
|
||||||
// 5.5. Process timer requests from handlers
|
// 5.5. Process worker extension requests from handlers (e.g., timer scheduling)
|
||||||
for request in timer_requests.into_inner() {
|
if let Some(ext) = &mut self.worker_ext {
|
||||||
match request {
|
for request in worker_requests.into_inner() {
|
||||||
TimerRequest::Once { dest, msg, ticks } => {
|
ext.handle_request(request);
|
||||||
self.timers.add_once(dest, msg, ticks);
|
|
||||||
}
|
|
||||||
TimerRequest::Interval { dest, msg, period } => {
|
|
||||||
self.timers.add_interval(dest, msg, period);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -308,14 +209,14 @@ impl Worker {
|
||||||
let cleanup_pending: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
|
let cleanup_pending: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
|
||||||
RefCell::new(Vec::new());
|
RefCell::new(Vec::new());
|
||||||
let cleanup_stops: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
let cleanup_stops: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
||||||
let cleanup_timers: RefCell<Vec<TimerRequest>> = RefCell::new(Vec::new());
|
let cleanup_requests: RefCell<Vec<Box<dyn Any + Send>>> = RefCell::new(Vec::new());
|
||||||
let dead = {
|
let dead = {
|
||||||
let cleanup_ctx = WorkerContext {
|
let cleanup_ctx = WorkerContext {
|
||||||
worker_id: self.id,
|
worker_id: self.id,
|
||||||
tc,
|
tc,
|
||||||
pending_local: &cleanup_pending,
|
pending_local: &cleanup_pending,
|
||||||
stop_requests: &cleanup_stops,
|
stop_requests: &cleanup_stops,
|
||||||
timer_requests: &cleanup_timers,
|
worker_requests: &cleanup_requests,
|
||||||
stats: &self.stats,
|
stats: &self.stats,
|
||||||
};
|
};
|
||||||
let dead = self.pool.cleanup_dead(&cleanup_ctx);
|
let dead = self.pool.cleanup_dead(&cleanup_ctx);
|
||||||
|
|
@ -362,9 +263,11 @@ impl Worker {
|
||||||
self.pool.deliver(&addr, msg);
|
self.pool.deliver(&addr, msg);
|
||||||
}
|
}
|
||||||
|
|
||||||
// GC orphaned interval timers for actors that were just removed
|
// GC per-worker extension state for dead actors (e.g., orphaned interval timers)
|
||||||
let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect();
|
if let Some(ext) = &mut self.worker_ext {
|
||||||
self.timers.gc_dead_intervals(&dead_addrs);
|
let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect();
|
||||||
|
ext.gc_dead(&dead_addrs);
|
||||||
|
}
|
||||||
|
|
||||||
did_work
|
did_work
|
||||||
}
|
}
|
||||||
|
|
@ -408,7 +311,7 @@ struct WorkerContext<'a> {
|
||||||
tc: &'a TickContext<'a>,
|
tc: &'a TickContext<'a>,
|
||||||
pending_local: &'a RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>>,
|
pending_local: &'a RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>>,
|
||||||
stop_requests: &'a RefCell<Vec<ActorAddress>>,
|
stop_requests: &'a RefCell<Vec<ActorAddress>>,
|
||||||
timer_requests: &'a RefCell<Vec<TimerRequest>>,
|
worker_requests: &'a RefCell<Vec<Box<dyn Any + Send>>>,
|
||||||
stats: &'a WorkerStats,
|
stats: &'a WorkerStats,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -445,8 +348,8 @@ impl ContextInner for WorkerContext<'_> {
|
||||||
self.stop_requests.borrow_mut().push(addr);
|
self.stop_requests.borrow_mut().push(addr);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn schedule_timer(&self, request: TimerRequest) {
|
fn post_worker_request(&self, request: Box<dyn Any + Send>) {
|
||||||
self.timer_requests.borrow_mut().push(request);
|
self.worker_requests.borrow_mut().push(request);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> {
|
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> {
|
||||||
|
|
|
||||||
|
|
@ -8,9 +8,12 @@ use std::collections::HashMap;
|
||||||
use proptest::prelude::*;
|
use proptest::prelude::*;
|
||||||
use proptest_state_machine::{prop_state_machine, ReferenceStateMachine, StateMachineTest};
|
use proptest_state_machine::{prop_state_machine, ReferenceStateMachine, StateMachineTest};
|
||||||
|
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
use swactor::actor::{ActorAddress, ActorInterface};
|
use swactor::actor::{ActorAddress, ActorInterface};
|
||||||
use swactor::config::{MailboxOverflow, RuntimeConfig};
|
use swactor::config::{MailboxOverflow, RuntimeConfig};
|
||||||
use swactor::runtime::{Ctx, Inbox, Runtime};
|
use swactor::runtime::{Ctx, Inbox, Runtime};
|
||||||
|
use swactor_std::{CtxTimers, StdExtension};
|
||||||
|
|
||||||
// ─── Shared Actor Types ────────────────────────────────────────────────────
|
// ─── Shared Actor Types ────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
@ -141,7 +144,8 @@ proptest! {
|
||||||
/// One-shot timer fires at exactly the right tick for any delay.
|
/// One-shot timer fires at exactly the right tick for any delay.
|
||||||
#[test]
|
#[test]
|
||||||
fn one_shot_timer_fires_at_correct_tick(delay in 1u64..20) {
|
fn one_shot_timer_fires_at_correct_tick(delay in 1u64..20) {
|
||||||
let rt = Runtime::new(RuntimeConfig::default());
|
let rt = Runtime::new(RuntimeConfig::default())
|
||||||
|
.with_extension(Arc::new(StdExtension::new()));
|
||||||
let inbox = rt.new_inbox::<Ping>().unwrap();
|
let inbox = rt.new_inbox::<Ping>().unwrap();
|
||||||
|
|
||||||
struct TimerActor { target: ActorAddress, delay: u64 }
|
struct TimerActor { target: ActorAddress, delay: u64 }
|
||||||
|
|
@ -175,7 +179,8 @@ proptest! {
|
||||||
/// Interval timer fires at correct periodic ticks for any period.
|
/// Interval timer fires at correct periodic ticks for any period.
|
||||||
#[test]
|
#[test]
|
||||||
fn interval_timer_fires_at_correct_period(period in 1u64..10) {
|
fn interval_timer_fires_at_correct_period(period in 1u64..10) {
|
||||||
let rt = Runtime::new(RuntimeConfig::default());
|
let rt = Runtime::new(RuntimeConfig::default())
|
||||||
|
.with_extension(Arc::new(StdExtension::new()));
|
||||||
let inbox = rt.new_inbox::<Ping>().unwrap();
|
let inbox = rt.new_inbox::<Ping>().unwrap();
|
||||||
|
|
||||||
struct IntervalActor { target: ActorAddress, period: u64 }
|
struct IntervalActor { target: ActorAddress, period: u64 }
|
||||||
|
|
|
||||||
|
|
@ -3,8 +3,8 @@ use std::sync::Arc;
|
||||||
|
|
||||||
use swactor::actor::{ActorAddress, ActorInterface, Down, MonitorRef, StopReason};
|
use swactor::actor::{ActorAddress, ActorInterface, Down, MonitorRef, StopReason};
|
||||||
use swactor_std::{
|
use swactor_std::{
|
||||||
ChildSpec, CtxGroups, CtxMonitoring, CtxNaming, RestartPolicy, Router, RoutingStrategy,
|
ChildSpec, CtxGroups, CtxMonitoring, CtxNaming, CtxTimers, RestartPolicy, Router,
|
||||||
RuntimeGroups, RuntimeNaming, StdExtension, Supervisor, SupervisorStrategy,
|
RoutingStrategy, RuntimeGroups, RuntimeNaming, StdExtension, Supervisor, SupervisorStrategy,
|
||||||
};
|
};
|
||||||
use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig};
|
use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig};
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue