diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index 44f8ec8..8cabeb8 100644 --- a/CLAUDE/notes/progress.md +++ b/CLAUDE/notes/progress.md @@ -2,7 +2,7 @@ ## Current Stage: Phase 1 — Research + First Improvement Cycle -### Status: Cycle 9 COMPLETE +### Status: Cycle 10 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -124,6 +124,37 @@ - `stop_nonexistent_actor_returns_error` — stop on bad address returns Err - **Result**: 82 tests pass, all workspace compiles +### Cycle 10: Actor Timers (Tick-Counting) +- **Research**: Studied timer/scheduling patterns across Erlang (timer:send_after, erlang:start_timer), + Akka (scheduleOnce, scheduler), Actix (ctx.run_later, ctx.run_interval), Kameo (tokio::time::sleep), + Tokio (tokio::time), Go (time.After, time.NewTicker) + - Also researched priority messages (REJECTED: lifecycle hooks cover 95% of use cases) + - Also researched SmallBox optimization (DEFERRED: measure allocation cost first) + - Key finding: per-worker tick-counting is ideal for swactor's synchronous model (deterministic) +- **Implementation**: Per-worker `TimerWheel` with deterministic tick-based scheduling + - `OnceTimer`: fire once at `fire_at` tick, consumed after firing + - `IntervalTimer`: fire every `period` ticks, message cloned via `CloneMsg` trait + - `CloneMsg` trait: type-erased clone for interval timer messages (blanket impl for `Message`) + - `TimerRequest` enum: `Once { dest, msg, ticks }` | `Interval { dest, msg, period }` + - `ctx.send_after_ticks(addr, msg, ticks)` — one-shot timer API + - `ctx.send_interval_ticks(addr, msg, period)` — interval timer API + - Phase 2.5 in tick_once: fire due timers, route through full delivery system + (pool.deliver for local actors, transfer_txs for cross-worker, inbox_registry for inboxes) + - Phase 5.5: drain timer requests from handler buffer into TimerWheel + - GC: interval timers for removed actors cleaned up after cleanup_dead + - `schedule_timer` on Runtime's ContextInner: no-op with warning (timers are per-worker only) +- **Bug fixed**: `gc_dead_intervals` was over-aggressive — removed timers for ANY address not + in the local pool, including inboxes and cross-worker actors. Fixed to only GC timers for + addresses in the `dead` set from cleanup_dead. +- **Tests**: 6 new tests + - `one_shot_timer_fires_after_n_ticks` — timer with delay=3 fires on tick 4 + - `handler_can_schedule_one_shot_timer` — timer scheduled from handler, fires correctly + - `one_shot_timer_fires_only_once` — consumed after firing, doesn't repeat + - `interval_timer_fires_repeatedly` — period=2, fires every 2 ticks (3 fires verified) + - `interval_timer_cleaned_up_when_actor_dies` — GC removes orphaned timers + - `timer_with_zero_delay_fires_next_tick` — delay=0 fires on next tick +- **Result**: 88 tests pass, all workspace compiles, zero warnings + ### Cycle 8: Dead Actor Cleanup (Memory Leak Fix) - **Research**: Audited remaining improvement gaps, found ActorPool and AddressMap both leak permanently poisoned actors. Known class of bug in Akka (#22990), CAF (#420). @@ -194,8 +225,10 @@ - [x] **Cycle 7: Actor recovery (factory restart)** ✅ - [x] **Cycle 8: Dead actor cleanup** ✅ - [x] **Cycle 9: Lifecycle hooks + graceful stop** ✅ -- [ ] **Cycle 10: Next improvement** - - Candidates: priority messages, actor timers, SmallBox optimization, property-based tests +- [x] **Cycle 10: Actor timers (tick-counting)** ✅ +- [ ] **Cycle 11: Next improvement** + - Candidates: SmallBox optimization, property-based tests, named actors/registry, actor groups + - Priority messages REJECTED (lifecycle hooks cover 95% of cases) - LIFO slot rejected (0-5% benefit for typical workloads, not worth complexity) ## Open Questions diff --git a/src/actor.rs b/src/actor.rs index af9581e..8e151d8 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -141,12 +141,43 @@ where /// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`. 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; +} + +impl CloneMsg for M { + fn clone_boxed(&self) -> Box { + 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, + ticks: u64, + }, + /// Repeating: deliver a clone of `msg` to `dest` every `period` ticks. + Interval { + dest: ActorAddress, + msg: Box, + period: u64, + }, +} + /// Object-safe inner trait for sending type-erased messages. +#[allow(private_interfaces)] pub trait ContextInner { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; fn spawn_any(&self, addr: ActorAddress, actor: Box); /// Request graceful stop for an actor. Takes effect after the current message. fn request_stop(&self, addr: ActorAddress); + /// Schedule a timer (one-shot or interval). + fn schedule_timer(&self, request: TimerRequest); } /// Actor syscall interface — passed to `ActorInterface::handle()`. @@ -193,6 +224,31 @@ impl<'a> Ctx<'a> { self.inner.request_stop(self.self_addr); } + /// 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(&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(&self, addr: ActorAddress, msg: M, period: u64) { + self.inner.schedule_timer(TimerRequest::Interval { + dest: addr, + msg: Box::new(msg), + period, + }); + } + /// Spawn a restartable actor. On panic, recreated via `factory` up to /// `max_restarts` times before permanent poisoning. pub fn spawn_restartable( diff --git a/src/delivery.rs b/src/delivery.rs index 0d08ef8..37b12b8 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -161,6 +161,7 @@ impl InboxRegistry { } /// Check if an address is registered without consuming a message. + #[cfg(feature = "transport")] pub fn contains(&self, addr: &ActorAddress) -> bool { self.senders.read().unwrap().contains_key(addr) } diff --git a/src/runtime.rs b/src/runtime.rs index 583c453..0cf44c1 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -5,7 +5,7 @@ use std::sync::{Arc, OnceLock}; use std::thread::{self, JoinHandle, Thread}; use std::time::Instant; -use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal}; +use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal, TimerRequest}; use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig}; @@ -419,6 +419,7 @@ pub(crate) fn notify_worker(threads: &[OnceLock], wid: usize) { } } +#[allow(private_interfaces)] impl ContextInner for Runtime { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { match self.address_map.lookup(&addr) { @@ -448,4 +449,11 @@ impl ContextInner for Runtime { notify_worker(&self.worker_threads, wid.as_usize()); } } + + fn schedule_timer(&self, _request: TimerRequest) { + // Timers are per-worker and tick-counted; scheduling from outside + // a worker context (e.g., rt.spawn() callback) is not supported. + // Use rt.send_to() with a delay loop instead. + eprintln!("swactor: schedule_timer called outside worker context — ignored"); + } } diff --git a/src/worker.rs b/src/worker.rs index 4925fb9..37ca80d 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -6,13 +6,110 @@ use std::sync::Arc; use std::thread; use std::time::Instant; -use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, StopSignal}; +use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopSignal, TimerRequest}; use crate::channel::Receiver; use crate::config::MailboxOverflow; use crate::delivery::{Envelope, TickContext, WorkerId}; use crate::stats::{ActorSnapshot, TickTiming, WorkerStats}; use crate::Error; +// ─── Per-Worker Timer Wheel ───────────────────────────────────────────────── + +struct OnceTimer { + fire_at: u64, + dest: ActorAddress, + msg: Box, +} + +struct IntervalTimer { + next_fire: u64, + period: u64, + dest: ActorAddress, + msg: Box, +} + +/// 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, + interval_timers: Vec, +} + +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)> { + 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, 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, 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 ───────────────────────────────────────────────────────────────── + /// A worker owns a set of actors and runs them in a loop. pub(crate) struct Worker { pub(crate) id: WorkerId, @@ -22,6 +119,8 @@ pub(crate) struct Worker { stats: Arc, /// Reusable scratch buffer for building per-actor snapshots. snapshot_buf: Vec, + /// Per-worker tick-counting timer wheel. + timers: TimerWheel, } impl Worker { @@ -40,6 +139,7 @@ impl Worker { spawn_rx, stats, snapshot_buf: Vec::new(), + timers: TimerWheel::new(), } } @@ -75,10 +175,32 @@ impl Worker { } let t2 = Instant::now(); + // 2.5. Fire due timers → deliver to mailboxes before tick_all + let timer_msgs = self.timers.fire(); + for (dest, msg) in timer_msgs { + if self.pool.contains(&dest) { + // Same-worker: deliver directly to actor's mailbox + self.pool.deliver(&dest, msg); + } else { + // Inbox or cross-worker: route through address map / inbox registry + match tc.address_map.lookup(&dest) { + Some(wid) => { + 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); + } + } + } + did_work = true; + } + // 3. Tick all actors with WorkerContext let pending_local: RefCell)>> = RefCell::new(Vec::new()); let stop_requests: RefCell> = RefCell::new(Vec::new()); + let timer_requests: RefCell> = RefCell::new(Vec::new()); let processed; { @@ -87,6 +209,7 @@ impl Worker { tc, pending_local: &pending_local, stop_requests: &stop_requests, + timer_requests: &timer_requests, stats: &self.stats, }; processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget, &stop_requests); @@ -121,6 +244,18 @@ impl Worker { for (addr, msg) in pending { self.pool.deliver(&addr, msg); } + + // 5.5. Process timer requests from handlers + for request in timer_requests.into_inner() { + match request { + TimerRequest::Once { dest, msg, ticks } => { + self.timers.add_once(dest, msg, ticks); + } + TimerRequest::Interval { dest, msg, period } => { + self.timers.add_interval(dest, msg, period); + } + } + } let t5 = Instant::now(); // 6. Publish stats (skip entirely when idle to avoid allocation + mutex) @@ -172,12 +307,14 @@ impl Worker { let cleanup_pending: RefCell)>> = RefCell::new(Vec::new()); let cleanup_stops: RefCell> = RefCell::new(Vec::new()); - { + let cleanup_timers: RefCell> = RefCell::new(Vec::new()); + let dead = { let cleanup_ctx = WorkerContext { worker_id: self.id, tc, pending_local: &cleanup_pending, stop_requests: &cleanup_stops, + timer_requests: &cleanup_timers, stats: &self.stats, }; let dead = self.pool.cleanup_dead(&cleanup_ctx); @@ -189,12 +326,16 @@ impl Worker { self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); did_work = true; } - } + dead + }; // Deliver any messages sent during on_stop callbacks for (addr, msg) in cleanup_pending.into_inner() { self.pool.deliver(&addr, msg); } + // GC orphaned interval timers for actors that were just removed + self.timers.gc_dead_intervals(&dead); + did_work } @@ -237,6 +378,7 @@ struct WorkerContext<'a> { tc: &'a TickContext<'a>, pending_local: &'a RefCell)>>, stop_requests: &'a RefCell>, + timer_requests: &'a RefCell>, stats: &'a WorkerStats, } @@ -272,6 +414,10 @@ impl ContextInner for WorkerContext<'_> { fn request_stop(&self, addr: ActorAddress) { self.stop_requests.borrow_mut().push(addr); } + + fn schedule_timer(&self, request: TimerRequest) { + self.timer_requests.borrow_mut().push(request); + } } struct ActorSlot { @@ -463,6 +609,10 @@ impl ActorPool { self.actors.len() } + pub fn contains(&self, addr: &ActorAddress) -> bool { + self.actors.contains_key(addr) + } + pub fn total_mailbox_depth(&self) -> usize { self.actors.values().map(|slot| slot.mailbox.len()).sum() } diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index 9939b43..d529fc3 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -2663,3 +2663,216 @@ fn stop_nonexistent_actor_returns_error() { let result = rt.stop_actor(fake_addr); assert!(result.is_err(), "stop_actor on nonexistent address should return Err"); } + +// ── Timer Helpers ───────────────────────────────────────────────────────── + +/// Actor that schedules a one-shot timer in on_start: sends a Ping to target after N ticks. +struct TimerStartActor { + target: ActorAddress, + delay_ticks: u64, +} + +impl ActorInterface for TimerStartActor { + type Incoming = Ping; + type Response = Pong; + + fn on_start(&mut self, ctx: &Ctx) { + ctx.send_after_ticks(self.target, Ping { reply_to: ctx.self_addr() }, self.delay_ticks); + } + + fn handle(&mut self, _ctx: &Ctx, _msg: Ping) {} +} + +/// Actor that schedules a one-shot timer when it receives a Forward message. +struct DelayEchoActor; + +impl ActorInterface for DelayEchoActor { + type Incoming = Forward; + type Response = Done; + + fn handle(&mut self, ctx: &Ctx, msg: Forward) { + ctx.send_after_ticks(msg.reply_to, Done(msg.value), 3); + } +} + +/// Actor that schedules an interval timer on start: sends Ping every N ticks. +struct HeartbeatActor { + target: ActorAddress, + period: u64, +} + +impl ActorInterface for HeartbeatActor { + type Incoming = Ping; + type Response = Pong; + + fn on_start(&mut self, ctx: &Ctx) { + ctx.send_interval_ticks(self.target, Ping { reply_to: ctx.self_addr() }, self.period); + } + + fn handle(&mut self, _ctx: &Ctx, _msg: Ping) {} +} + +// ── Timer Tests ─────────────────────────────────────────────────────────── + +/// Given an actor that schedules a one-shot timer in on_start, +/// when enough ticks pass, +/// then the timer message is delivered to the target. +#[test] +fn one_shot_timer_fires_after_n_ticks() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let _timer_actor = rt.spawn(TimerStartActor { + target: *inbox.addr(), + delay_ticks: 3, + }).unwrap(); + + // Tick 1: on_start schedules timer (fire_at = current_tick + 3 = 4) + // Timer fires when current_tick >= fire_at, so after tick 4 completes + rt.tick(); // tick 1: on_start, timer scheduled + assert!(inbox.try_recv().is_none(), "no delivery before delay"); + + rt.tick(); // tick 2 + assert!(inbox.try_recv().is_none(), "no delivery on tick 2"); + + rt.tick(); // tick 3 + assert!(inbox.try_recv().is_none(), "no delivery on tick 3"); + + rt.tick(); // tick 4: timer fires + let msg = inbox.try_recv(); + assert!(msg.is_some(), "timer message delivered after 3-tick delay"); +} + +/// Given an actor that schedules a one-shot timer from a message handler, +/// when enough ticks pass after the triggering message, +/// then the delayed response arrives. +#[test] +fn handler_can_schedule_one_shot_timer() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let addr = rt.spawn(DelayEchoActor).unwrap(); + + let _ = rt.send_to(addr, Forward { value: 42, reply_to: *inbox.addr() }); + rt.tick(); // process Forward, schedule timer (delay=3) + + assert!(inbox.try_recv().is_none(), "no immediate reply"); + + rt.tick(); // tick 2 + rt.tick(); // tick 3 + assert!(inbox.try_recv().is_none(), "not yet"); + + rt.tick(); // tick 4: timer fires + let reply = inbox.try_recv(); + assert_eq!(reply, Some(Done(42)), "delayed reply arrives after 3 ticks"); +} + +/// Given a one-shot timer, +/// when it fires, +/// then it does NOT fire again on subsequent ticks (consumed). +#[test] +fn one_shot_timer_fires_only_once() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let _timer_actor = rt.spawn(TimerStartActor { + target: *inbox.addr(), + delay_ticks: 1, + }).unwrap(); + + rt.tick(); // on_start schedules timer + rt.tick(); // timer fires + assert!(inbox.try_recv().is_some(), "first fire"); + + // Subsequent ticks should NOT fire again + for _ in 0..5 { rt.tick(); } + assert!(inbox.try_recv().is_none(), "one-shot does not repeat"); +} + +/// Given an interval timer with period 2, +/// when multiple ticks pass, +/// then the timer fires repeatedly every 2 ticks. +#[test] +fn interval_timer_fires_repeatedly() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let _heartbeat = rt.spawn(HeartbeatActor { + target: *inbox.addr(), + period: 2, + }).unwrap(); + + rt.tick(); // tick 1: on_start, interval scheduled (next_fire = current + 2 = 3) + assert!(inbox.try_recv().is_none(), "no fire on tick 1"); + + rt.tick(); // tick 2 + assert!(inbox.try_recv().is_none(), "no fire on tick 2"); + + rt.tick(); // tick 3: first fire + assert!(inbox.try_recv().is_some(), "fire on tick 3"); + + rt.tick(); // tick 4 + assert!(inbox.try_recv().is_none(), "no fire on tick 4"); + + rt.tick(); // tick 5: second fire + assert!(inbox.try_recv().is_some(), "fire on tick 5"); + + rt.tick(); // tick 6 + assert!(inbox.try_recv().is_none(), "no fire on tick 6"); + + rt.tick(); // tick 7: third fire + assert!(inbox.try_recv().is_some(), "fire on tick 7"); +} + +/// Given an interval timer targeting an actor that gets stopped, +/// when the actor is removed, +/// then the interval timer is cleaned up (no orphan timers). +#[test] +fn interval_timer_cleaned_up_when_actor_dies() { + let rt = Runtime::new(RuntimeConfig::default()); + let _inbox = rt.new_inbox::().unwrap(); + + // Heartbeat sends to a counter that we'll kill + let counter_addr = rt.spawn(CounterActor { count: 0 }).unwrap(); + + // HeartbeatActor sends Ping to counter every tick + let _hb = rt.spawn(HeartbeatActor { + target: counter_addr, + period: 1, + }).unwrap(); + + // Let it run a few ticks + for _ in 0..3 { rt.tick(); } + + // Stop the counter + rt.stop_actor(counter_addr).unwrap(); + for _ in 0..5 { rt.tick(); } + + // Counter is gone, interval timer should be GC'd. + // No crash, no leak — just verifying it doesn't panic. + let stats = rt.stats(); + // Only the heartbeat actor should remain + assert_eq!(stats.workers[0].num_actors, 1); +} + +/// Given a timer with delay 0, +/// when the next tick fires, +/// then the message is delivered immediately on the next tick. +#[test] +fn timer_with_zero_delay_fires_next_tick() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let _timer_actor = rt.spawn(TimerStartActor { + target: *inbox.addr(), + delay_ticks: 0, + }).unwrap(); + + rt.tick(); // on_start schedules timer with delay=0 + // Timer requests are processed after tick_all (phase 5.5) + // Timer fires on the NEXT tick (phase 2.5) + assert!(inbox.try_recv().is_none(), "not yet — timer fires next tick"); + + rt.tick(); // timer fires + assert!(inbox.try_recv().is_some(), "zero-delay timer fires on next tick"); +}