diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index 2246212..e03d2c2 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 18 COMPLETE +### Status: Cycle 19 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -158,6 +158,38 @@ - `supervisor_on_stop_kills_children` — supervisor shutdown cascades to children - **Result**: 138 tests pass (130 behavioral + 7 proptest + 1 doctest), zero warnings, full workspace compiles +### Cycle 19: Router (Actor Pool with Message Routing) +- **Research**: Cross-framework analysis of actor pool/router patterns: + - Erlang: poolboy (checkout/checkin), wpool (transparent forwarding, 6 strategies + custom) + - Akka: Router actors (Pool vs Group), 8 strategies (RoundRobin, Random, SmallestMailbox, + Balancing, Broadcast, ScatterGather, TailChopping, ConsistentHashing), Resizer for dynamic sizing + - Ractor: No built-in router (process groups only) + - Actix: SyncArbiter (shared queue, implicit work-stealing) + - Kameo: ActorPool (least-connections, auto-replace dead workers) + - Key finding: Router-as-actor with transparent forwarding (wpool/Akka style) is the best fit + - Decision: user-space actor like Supervisor, reusing monitor + handle_down for worker replacement +- **Implementation**: `Router` actor in `src/actor.rs` + - `RoutingStrategy::RoundRobin` — sequential circular distribution + - `RoutingStrategy::Random` — random worker selection via `get_random()` + - `RoutingStrategy::Broadcast` — clone message to all live workers + - Generic over `M: Message` (same Incoming type as workers) — transparent forwarding + - Workers spawned in `on_start`, monitored via `ctx.monitor()`, auto-replaced via `handle_down` + - Meltdown protection: `total_restarts > max_restarts` → `ctx.stop_self()` + - Cascading shutdown: `on_stop` sends stop signals to all workers + - Reuses `ActiveChild` struct from Supervisor (addr + monitor_ref) + - Factory: `Arc Result + Send + Sync>` + - SmallestMailbox deferred: requires runtime stats access not available in user-space + - ConsistentHashing deferred: requires hash_fn parameter, can add later as builder method +- **Tests**: 7 new behavioral tests + - `router_round_robin_distributes_across_workers` — 6 msgs to 3 workers, each gets 2 + - `router_broadcast_sends_to_all_workers` — 1 msg, all 3 workers receive + - `router_random_delivers_to_some_worker` — 30 msgs across 3 workers, at least 2 used + - `router_replaces_dead_worker` — panicked worker auto-replaced, pool size maintained + - `router_meltdown_after_max_restarts` — 3 deaths with max_restarts=2 → router stops + - `router_on_stop_kills_workers` — stopping router cascades to all workers + - `router_broadcast_multiple_messages_all_received` — 5 msgs × 3 workers = 15 received +- **Result**: 148 tests pass (140 behavioral + 7 proptest + 1 doctest), zero warnings + ### Cycle 18: OneForAll + RestForOne Supervisor Strategies - **Research**: Investigated SmallBox/InlineAny optimization (44% queue throughput improvement) but deferred due to unsafe code risk and 32+ call-site changes violating structural constraints. diff --git a/docs/runtime.md b/docs/runtime.md index f599d27..05655fe 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -239,6 +239,31 @@ restarts the full set in spec order. Already-dead children are handled immediate Meltdown: supervisor stops itself when total restarts exceed `max_restarts`. Cascading: supervisor stops all children in `on_stop`. +## Router — Actor Pool with Message Routing + +The `Router` actor manages a pool of identical workers and distributes +incoming messages across them. Callers send messages to the router's address, +and the router forwards them according to the configured strategy. + +``` + let router = Router::new( + RoutingStrategy::RoundRobin, + 5, // pool size + |ctx| ctx.spawn(MyWorker::new()), // worker factory + 10, // max restarts before meltdown + ); + let router_addr = rt.spawn(router)?; + rt.send_to(router_addr, WorkerMsg::DoWork(42))?; +``` + +Routing strategies: +- `RoundRobin`: sequential circular distribution +- `Random`: random worker selection +- `Broadcast`: clone message to all workers + +Workers are monitored and automatically replaced on failure. Meltdown +protection stops the router when total restarts exceed `max_restarts`. + ### handle_down Callback Any actor can override `handle_down` to react to monitored actor deaths diff --git a/src/actor.rs b/src/actor.rs index 2cae7ff..53ce703 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,4 +1,5 @@ use std::any::Any; +use std::marker::PhantomData; use std::sync::Arc; use crate::Error; @@ -707,3 +708,168 @@ impl ActorInterface for Supervisor { } } } + +// --------------------------------------------------------------------------- +// Router — pool of identical workers with configurable routing strategy +// --------------------------------------------------------------------------- + +/// Strategy for distributing messages across pool workers. +#[derive(Debug, Clone)] +pub enum RoutingStrategy { + /// Sequential round-robin distribution. + RoundRobin, + /// Random worker selection. + Random, + /// Send to all workers (message is cloned to each). + Broadcast, +} + +/// A router actor that manages a pool of identical workers and distributes +/// incoming messages across them according to a [`RoutingStrategy`]. +/// +/// Workers are spawned during `on_start`, monitored for failures, and +/// automatically replaced to maintain the target pool size. Meltdown +/// protection stops the router when total restarts exceed `max_restarts`. +/// +/// # Example +/// +/// ```ignore +/// let router = Router::new( +/// RoutingStrategy::RoundRobin, +/// 5, +/// |ctx| ctx.spawn(MyWorker::new()), +/// 10, +/// ); +/// let router_addr = rt.spawn(router)?; +/// rt.send_to(router_addr, WorkerMessage::DoWork(42))?; +/// ``` +pub struct Router { + strategy: RoutingStrategy, + pool_size: usize, + factory: Arc Result + Send + Sync>, + workers: Vec>, + rr_index: usize, + total_restarts: u32, + max_restarts: u32, + _marker: PhantomData, +} + +impl Router { + pub fn new( + strategy: RoutingStrategy, + pool_size: usize, + factory: impl Fn(&Ctx) -> Result + Send + Sync + 'static, + max_restarts: u32, + ) -> Self { + Self { + strategy, + pool_size, + factory: Arc::new(factory), + workers: (0..pool_size).map(|_| None).collect(), + rr_index: 0, + total_restarts: 0, + max_restarts, + _marker: PhantomData, + } + } + + fn start_worker(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> { + let addr = (self.factory)(ctx)?; + let mref = ctx.monitor(addr); + self.workers[idx] = Some(ActiveChild { + addr, + _monitor_ref: mref, + }); + Ok(()) + } + + fn find_worker_idx(&self, addr: ActorAddress) -> Option { + self.workers + .iter() + .position(|w| w.as_ref().map_or(false, |ac| ac.addr == addr)) + } + + fn live_workers(&self) -> Vec { + self.workers + .iter() + .filter_map(|w| w.as_ref().map(|ac| ac.addr)) + .collect() + } + + fn select_one(&mut self) -> Option { + let live = self.live_workers(); + if live.is_empty() { + return None; + } + match self.strategy { + RoutingStrategy::RoundRobin => { + let idx = self.rr_index % live.len(); + self.rr_index = self.rr_index.wrapping_add(1); + Some(live[idx]) + } + RoutingStrategy::Random => { + let mut buf = [0u8; 8]; + crate::get_random(&mut buf); + let r = u64::from_ne_bytes(buf) as usize; + Some(live[r % live.len()]) + } + RoutingStrategy::Broadcast => None, // handled separately + } + } +} + +impl ActorInterface for Router { + type Incoming = M; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: M) { + match self.strategy { + RoutingStrategy::Broadcast => { + let live = self.live_workers(); + for addr in live { + let _ = ctx.send(addr, msg.clone()); + } + } + _ => { + if let Some(addr) = self.select_one() { + let _ = ctx.send(addr, msg); + } + } + } + } + + fn on_start(&mut self, ctx: &Ctx) { + for idx in 0..self.pool_size { + if let Err(e) = self.start_worker(ctx, idx) { + eprintln!("swactor: router failed to start worker {idx}: {e}"); + } + } + } + + fn on_stop(&mut self, ctx: &Ctx) { + for child in self.workers.iter().flatten() { + let _ = ctx.stop_actor(child.addr); + } + } + + fn handle_down(&mut self, ctx: &Ctx, down: Down) { + let Some(idx) = self.find_worker_idx(down.addr) else { + return; + }; + self.workers[idx] = None; + + self.total_restarts += 1; + if self.total_restarts > self.max_restarts { + eprintln!( + "swactor: router reached max restarts ({}), shutting down", + self.max_restarts + ); + ctx.stop_self(); + return; + } + + if let Err(e) = self.start_worker(ctx, idx) { + eprintln!("swactor: router failed to restart worker {idx}: {e}"); + } + } +} diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index b08771d..8fc1596 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -2,8 +2,8 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use swactor::actor::{ - ActorAddress, ActorInterface, ChildSpec, Down, MonitorRef, RestartPolicy, StopReason, - Supervisor, SupervisorStrategy, + ActorAddress, ActorInterface, ChildSpec, Down, MonitorRef, RestartPolicy, Router, + RoutingStrategy, StopReason, Supervisor, SupervisorStrategy, }; use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig}; @@ -4230,3 +4230,302 @@ fn supervisor_on_stop_kills_children() { assert_eq!(rt.stats().workers[0].num_actors, 0); } + +// ── Router tests ───────────────────────────────────────────────────────────── + +#[test] +fn router_round_robin_distributes_across_workers() { + // Given a round-robin router with 3 workers + // When we send 6 messages + // Then each worker should receive exactly 2 messages + let rt = Runtime::new(RuntimeConfig::default()); + let collected = Arc::new(std::sync::Mutex::new(Vec::new())); + + struct Collector(Arc>>); + #[derive(Clone)] + struct Work(usize); + impl ActorInterface for Collector { + type Incoming = Work; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: Work) { + self.0.lock().unwrap().push((ctx.self_addr(), msg.0)); + } + } + + let c = collected.clone(); + let router = Router::::new( + RoutingStrategy::RoundRobin, + 3, + move |ctx| ctx.spawn(Collector(c.clone())), + 10, + ); + let router_addr = rt.spawn(router).unwrap(); + rt.tick(); // on_start spawns 3 workers + + for i in 0..6 { + rt.send_to(router_addr, Work(i)).unwrap(); + } + rt.tick(); // router receives 6 Work messages, forwards to workers + rt.tick(); // workers process their messages + + let data = collected.lock().unwrap(); + assert_eq!(data.len(), 6); + + // Count how many unique workers received messages + let mut per_worker = std::collections::HashMap::new(); + for (addr, _) in data.iter() { + *per_worker.entry(*addr).or_insert(0usize) += 1; + } + // All 3 workers should have received exactly 2 messages each + assert_eq!(per_worker.len(), 3); + for count in per_worker.values() { + assert_eq!(*count, 2); + } +} + +#[test] +fn router_broadcast_sends_to_all_workers() { + // Given a broadcast router with 3 workers + // When we send 1 message + // Then all 3 workers should receive it + let rt = Runtime::new(RuntimeConfig::default()); + let count = Arc::new(AtomicUsize::new(0)); + + struct Counter(Arc); + #[derive(Clone)] + struct Ping; + impl ActorInterface for Counter { + type Incoming = Ping; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: Ping) { + self.0.fetch_add(1, Ordering::Relaxed); + } + } + + let c = count.clone(); + let router = Router::::new( + RoutingStrategy::Broadcast, + 3, + move |ctx| ctx.spawn(Counter(c.clone())), + 10, + ); + let router_addr = rt.spawn(router).unwrap(); + rt.tick(); // on_start spawns workers + + rt.send_to(router_addr, Ping).unwrap(); + rt.tick(); // router broadcasts + rt.tick(); // workers process + + assert_eq!(count.load(Ordering::Relaxed), 3); +} + +#[test] +fn router_random_delivers_to_some_worker() { + // Given a random router with 3 workers + // When we send 30 messages + // Then at least 2 different workers should have received messages + let rt = Runtime::new(RuntimeConfig::default()); + let collected = Arc::new(std::sync::Mutex::new(Vec::new())); + + struct Collector(Arc>>); + #[derive(Clone)] + struct Work; + impl ActorInterface for Collector { + type Incoming = Work; + type Response = (); + fn handle(&mut self, ctx: &Ctx, _msg: Work) { + self.0.lock().unwrap().push(ctx.self_addr()); + } + } + + let c = collected.clone(); + let router = Router::::new( + RoutingStrategy::Random, + 3, + move |ctx| ctx.spawn(Collector(c.clone())), + 10, + ); + let router_addr = rt.spawn(router).unwrap(); + rt.tick(); + + for _ in 0..30 { + rt.send_to(router_addr, Work).unwrap(); + } + rt.tick(); + rt.tick(); + + let data = collected.lock().unwrap(); + assert_eq!(data.len(), 30); + + let unique: std::collections::HashSet<_> = data.iter().collect(); + // With 30 messages across 3 workers, probability of all going to 1 is vanishingly small + assert!(unique.len() >= 2, "expected at least 2 workers used, got {}", unique.len()); +} + +#[test] +fn router_replaces_dead_worker() { + // Given a router with 3 workers + // When one worker panics + // Then the router should spawn a replacement and messages continue to be delivered + let rt = Runtime::new(RuntimeConfig::default()); + let spawn_count = Arc::new(AtomicUsize::new(0)); + + struct PanicOnFirst { + first: bool, + } + #[derive(Clone)] + struct Work; + impl ActorInterface for PanicOnFirst { + type Incoming = Work; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: Work) { + if self.first { + self.first = false; + panic!("first message panic"); + } + } + } + + let sc = spawn_count.clone(); + let router = Router::::new( + RoutingStrategy::RoundRobin, + 3, + move |ctx| { + let n = sc.fetch_add(1, Ordering::Relaxed); + // Only the first worker panics on its first message + ctx.spawn(PanicOnFirst { first: n == 0 }) + }, + 10, + ); + let router_addr = rt.spawn(router).unwrap(); + rt.tick(); // spawn workers (3 spawned) + assert_eq!(spawn_count.load(Ordering::Relaxed), 3); + + // Send a message that will hit worker 0 (round-robin starts at 0) + rt.send_to(router_addr, Work).unwrap(); + rt.tick(); // router forwards to worker 0 + rt.tick(); // worker 0 panics + rt.tick(); // cleanup + Down delivered to router + rt.tick(); // router spawns replacement + rt.tick(); // replacement starts + + // Should have spawned 4 total (3 original + 1 replacement) + assert_eq!(spawn_count.load(Ordering::Relaxed), 4); + + // Verify all 3 slots are live — stats should show router + 3 workers + assert_eq!(rt.stats().workers[0].num_actors, 4); +} + +#[test] +fn router_meltdown_after_max_restarts() { + // Given a router with max_restarts=2 + // When 3 workers die in succession + // Then the router should stop itself + let rt = Runtime::new(RuntimeConfig::default()); + + struct AlwaysPanics; + #[derive(Clone)] + struct Work; + impl ActorInterface for AlwaysPanics { + type Incoming = Work; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: Work) { + panic!("always"); + } + } + + let router = Router::::new( + RoutingStrategy::RoundRobin, + 1, + |ctx| ctx.spawn(AlwaysPanics), + 2, // max 2 restarts + ); + let router_addr = rt.spawn(router).unwrap(); + rt.tick(); // on_start + + // Kill the worker 3 times (> max_restarts=2) + for _ in 0..3 { + rt.send_to(router_addr, Work).unwrap(); + for _ in 0..5 { + rt.tick(); + } + } + + // After 3 restarts, router should have shut down + for _ in 0..5 { + rt.tick(); + } + assert_eq!(rt.stats().workers[0].num_actors, 0); +} + +#[test] +fn router_on_stop_kills_workers() { + // Given a running router with 3 workers + // When the router is stopped + // Then all workers should also be stopped + let rt = Runtime::new(RuntimeConfig::default()); + + struct Dummy; + #[derive(Clone)] + struct Work; + impl ActorInterface for Dummy { + type Incoming = Work; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: Work) {} + } + + let router = Router::::new( + RoutingStrategy::RoundRobin, + 3, + |ctx| ctx.spawn(Dummy), + 10, + ); + let router_addr = rt.spawn(router).unwrap(); + rt.tick(); // on_start + assert_eq!(rt.stats().workers[0].num_actors, 4); // router + 3 workers + + rt.stop_actor(router_addr).unwrap(); + for _ in 0..5 { + rt.tick(); + } + + assert_eq!(rt.stats().workers[0].num_actors, 0); +} + +#[test] +fn router_broadcast_multiple_messages_all_received() { + // Given a broadcast router + // When we send 5 messages to 3 workers + // Then total received = 5 * 3 = 15 + let rt = Runtime::new(RuntimeConfig::default()); + let total = Arc::new(AtomicUsize::new(0)); + + struct Sink(Arc); + #[derive(Clone)] + struct Tick; + impl ActorInterface for Sink { + type Incoming = Tick; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: Tick) { + self.0.fetch_add(1, Ordering::Relaxed); + } + } + + let t = total.clone(); + let router = Router::::new( + RoutingStrategy::Broadcast, + 3, + move |ctx| ctx.spawn(Sink(t.clone())), + 10, + ); + let router_addr = rt.spawn(router).unwrap(); + rt.tick(); + + for _ in 0..5 { + rt.send_to(router_addr, Tick).unwrap(); + } + rt.tick(); // router broadcasts + rt.tick(); // workers process + + assert_eq!(total.load(Ordering::Relaxed), 15); +}