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:
Claude 2026-02-13 09:38:10 +00:00
parent 93b1b1fa1e
commit 09d5c75c39
12 changed files with 285 additions and 203 deletions

View file

@ -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

View file

@ -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()))
}
} }

View file

@ -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};

View 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);
}
}

View file

@ -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]

View file

@ -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);

View file

@ -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,
});
}
} }

View file

@ -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]);
} }

View file

@ -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> {

View file

@ -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,14 +82,12 @@ 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 {
// Inbox or cross-worker: route through address map / inbox registry
match tc.address_map.lookup(&dest) { match tc.address_map.lookup(&dest) {
Some(wid) => { Some(wid) => {
tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg)); tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg));
@ -195,12 +100,13 @@ impl Worker {
} }
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)
if let Some(ext) = &mut self.worker_ext {
let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect(); let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect();
self.timers.gc_dead_intervals(&dead_addrs); 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> {

View file

@ -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 }

View file

@ -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};