diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index aca55df..04f3594 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 5 COMPLETE +### Status: Cycle 6 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -89,6 +89,24 @@ - **Benchmarks**: 1 new group — `placement/spawn_under_load` (2t, 4t) - **Result**: 60 tests pass, all workspace compiles +### Cycle 6: Mailbox Backpressure +- **Research**: Compared backpressure across Erlang (unbounded, pobox), Actix (cap 16, do_send bypass), + Kameo (cap 64, bounded), Tokio mpsc (bounded, permit pattern), Go channels (blocking) + - Consensus: bounded by default, configurable overflow policy +- **Implementation**: Per-actor bounded mailboxes with configurable overflow + - Added `MailboxOverflow` enum: `DropNewest` (discard incoming) and `DropOldest` (evict oldest) + - Added `default_mailbox_capacity` and `mailbox_overflow` to `RuntimeConfig` + - Default: capacity=0 (unbounded) — 100% backward compatible + - `ActorSlot` stores per-actor capacity and policy (from runtime defaults) + - `deliver()` enforces bounds; dropped messages tracked via `drops_this_tick` counter + - `messages_dropped: AtomicU64` added to `WorkerStats` and `WorkerInfo` +- **Tests**: 4 new tests + - `bounded_mailbox_drop_newest_caps_at_capacity` — 50 msgs, cap 10 → only 10 delivered + - `bounded_mailbox_drop_oldest_keeps_newest` — 10 msgs, cap 5 → newest 5 kept + - `unbounded_mailbox_delivers_all_messages` — backward compatibility check + - `bounded_mailbox_refills_after_processing` — cap 5, process, refill works +- **Result**: 64 tests pass, all workspace compiles + ### Research Notes - Full analysis in `CLAUDE/notes/research_synthesis.md` - Baseline benchmarks in `CLAUDE/notes/baseline_benchmarks.md` @@ -99,8 +117,9 @@ - [x] **Cycle 3: Adaptive backoff with thread parking** ✅ - [x] **Cycle 4: Enhanced benchmarks + bug-inspired tests** ✅ - [x] **Cycle 5: Work stealing research + load-aware placement** ✅ -- [ ] **Cycle 6: Next improvement** - - Candidates: mailbox backpressure, actor recovery, LIFO slot optimization +- [x] **Cycle 6: Mailbox backpressure** ✅ +- [ ] **Cycle 7: Next improvement** + - Candidates: actor recovery (factory restart), LIFO slot, VecDeque→ring buffer optimization - Pick based on highest impact-to-effort ratio ## Open Questions diff --git a/CLAUDE/notes/research_synthesis.md b/CLAUDE/notes/research_synthesis.md index 23336d8..5024bb9 100644 --- a/CLAUDE/notes/research_synthesis.md +++ b/CLAUDE/notes/research_synthesis.md @@ -120,7 +120,7 @@ until A finishes. Every other runtime studied prevents this: ### Swactor Weaknesses to Address - Box downcast can fail silently → type mismatch tracking needed (have it) -- No backpressure: senders never block → unbounded queue growth under sustained load +- ~~No backpressure~~ → FIXED: optional bounded mailbox with DropNewest/DropOldest (Cycle 6) - Panicked actors permanently poisoned → no recovery path - ~~Spin/sleep backoff wastes CPU~~ → FIXED: thread parking with instant wakeup (Cycle 3) - No supervision trees diff --git a/src/config.rs b/src/config.rs index 7ac08a0..4772bd4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -23,6 +23,15 @@ impl Default for BackoffPolicy { } } +/// What to do when a bounded mailbox is full. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MailboxOverflow { + /// Drop the incoming message (newest). The message is silently discarded. + DropNewest, + /// Drop the oldest message in the queue to make room for the new one. + DropOldest, +} + /// The tunable settings for the runtime. pub struct RuntimeConfig { pub max_actors: usize, @@ -33,6 +42,11 @@ pub struct RuntimeConfig { /// Prevents a single actor with a large mailbox from starving others. /// `0` means unlimited (drain entire mailbox). pub actor_message_budget: usize, + /// Default per-actor mailbox capacity. `0` means unbounded (no limit). + /// When non-zero, `mailbox_overflow` controls what happens when the mailbox is full. + pub default_mailbox_capacity: usize, + /// Overflow policy for bounded mailboxes. Ignored when `default_mailbox_capacity` is 0. + pub mailbox_overflow: MailboxOverflow, } /// 8kB for the `Box<..>` before counting the rest of the memory @@ -55,6 +69,8 @@ impl Default for RuntimeConfig { num_threads: 1, backoff_policy: BackoffPolicy::default(), actor_message_budget: DEFAULT_ACTOR_MESSAGE_BUDGET, + default_mailbox_capacity: 0, + mailbox_overflow: MailboxOverflow::DropNewest, } } } diff --git a/src/runtime.rs b/src/runtime.rs index 8e9b562..4fea2cb 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -8,7 +8,7 @@ use std::time::Instant; use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}; use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works -pub use crate::config::{BackoffPolicy, RuntimeConfig}; +pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig}; use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; use crate::stats::{StatsHook, WorkerStats}; // Re-export stats types so existing code using `runtime::*` still works @@ -137,7 +137,14 @@ impl Runtime { let stats = Arc::new(WorkerStats::new()); worker_stats.push(stats.clone()); - workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats)); + workers.push(Worker::new( + WorkerId(i), + transfer_rx, + spawn_rx, + stats, + config.default_mailbox_capacity, + config.mailbox_overflow, + )); } let placement = Placement::new(num_workers, worker_stats.clone()); diff --git a/src/stats.rs b/src/stats.rs index 2b49db5..f6fb097 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -30,6 +30,8 @@ pub struct WorkerStats { // Error counters pub type_mismatches: AtomicU64, pub panics: AtomicU64, + /// Messages dropped due to mailbox overflow (bounded mailbox policy). + pub messages_dropped: AtomicU64, // Tick timing ring buffer (last N ticks, lock-free) tick_timings: ArrayQueue, } @@ -45,6 +47,7 @@ impl WorkerStats { inbox_sends: AtomicU64::new(0), type_mismatches: AtomicU64::new(0), panics: AtomicU64::new(0), + messages_dropped: AtomicU64::new(0), tick_timings: ArrayQueue::new(TICK_BUFFER_CAP), } } @@ -79,6 +82,7 @@ impl WorkerStats { inbox_sends: self.inbox_sends.load(Relaxed), type_mismatches: self.type_mismatches.load(Relaxed), panics: self.panics.load(Relaxed), + messages_dropped: self.messages_dropped.load(Relaxed), } } } @@ -96,6 +100,7 @@ pub struct WorkerInfo { pub inbox_sends: u64, pub type_mismatches: u64, pub panics: u64, + pub messages_dropped: u64, } /// Per-actor snapshot transferred from worker to runtime (not serialized). diff --git a/src/worker.rs b/src/worker.rs index 5537c09..7c1773d 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -8,6 +8,7 @@ use std::time::Instant; use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx}; use crate::channel::Receiver; +use crate::config::MailboxOverflow; use crate::delivery::{Envelope, TickContext, WorkerId}; use crate::stats::{ActorSnapshot, TickTiming, WorkerStats}; use crate::Error; @@ -29,10 +30,12 @@ impl Worker { transfer_rx: Receiver, spawn_rx: Receiver<(ActorAddress, Box)>, stats: Arc, + default_mailbox_capacity: usize, + default_overflow_policy: MailboxOverflow, ) -> Self { Self { id, - pool: ActorPool::new(), + pool: ActorPool::new(default_mailbox_capacity, default_overflow_policy), transfer_rx, spawn_rx, stats, @@ -119,10 +122,14 @@ impl Worker { let t5 = Instant::now(); // 6. Publish stats (skip entirely when idle to avoid allocation + mutex) + let drops = self.pool.take_drops(); if did_work { self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); self.stats.total_mailbox_depth.store(self.pool.total_mailbox_depth(), Ordering::Relaxed); self.stats.messages_processed.fetch_add(processed as u64, Ordering::Relaxed); + if drops > 0 { + self.stats.messages_dropped.fetch_add(drops as u64, Ordering::Relaxed); + } if let Some(hook) = tc.stats_hook { self.pool.mailbox_depths_into(&mut self.snapshot_buf); @@ -238,34 +245,60 @@ struct ActorSlot { poisoned: bool, last_msg_type: Option<&'static str>, messages_processed: u64, + /// Per-actor mailbox capacity. 0 = unbounded. + mailbox_capacity: usize, + overflow_policy: MailboxOverflow, } /// Per-worker actor storage. Owns per-actor mailboxes. pub(crate) struct ActorPool { actors: HashMap, + default_mailbox_capacity: usize, + default_overflow_policy: MailboxOverflow, + /// Messages dropped this tick due to mailbox overflow. Reset after publishing to stats. + drops_this_tick: usize, } impl ActorPool { - pub fn new() -> Self { + pub fn new(default_mailbox_capacity: usize, default_overflow_policy: MailboxOverflow) -> Self { Self { actors: HashMap::new(), + default_mailbox_capacity, + default_overflow_policy, + drops_this_tick: 0, } } pub fn insert(&mut self, addr: ActorAddress, actor: Box) { + let cap = self.default_mailbox_capacity; + let prealloc = if cap > 0 { cap.min(64) } else { 16 }; self.actors.insert(addr, ActorSlot { - mailbox: VecDeque::with_capacity(16), + mailbox: VecDeque::with_capacity(prealloc), actor, poisoned: false, last_msg_type: None, messages_processed: 0, + mailbox_capacity: self.default_mailbox_capacity, + overflow_policy: self.default_overflow_policy, }); } /// Deliver a type-erased message to the actor at `addr`. - /// Returns `true` if the actor exists (message is queued; type check deferred to tick). + /// Returns `true` if the actor exists (message handled or dropped; type check deferred to tick). pub fn deliver(&mut self, addr: &ActorAddress, msg: Box) -> bool { if let Some(slot) = self.actors.get_mut(addr) { + if slot.mailbox_capacity > 0 && slot.mailbox.len() >= slot.mailbox_capacity { + match slot.overflow_policy { + MailboxOverflow::DropNewest => { + self.drops_this_tick += 1; + return true; + } + MailboxOverflow::DropOldest => { + slot.mailbox.pop_front(); + self.drops_this_tick += 1; + } + } + } slot.mailbox.push_back(msg); true } else { @@ -273,6 +306,11 @@ impl ActorPool { } } + /// Take and reset the drop counter for this tick. + pub fn take_drops(&mut self) -> usize { + std::mem::replace(&mut self.drops_this_tick, 0) + } + /// Tick all actors in the pool. Returns the number of messages processed. /// /// Each actor processes up to `budget` messages per tick (0 = unlimited). diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index 81d108f..ea3e273 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -2,7 +2,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use swactor::actor::{ActorAddress, ActorInterface}; -use swactor::runtime::{Ctx, Inbox, Runtime, RuntimeConfig}; +use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig}; // ── Messages ──────────────────────────────────────────────────────────────── @@ -1847,3 +1847,149 @@ fn load_aware_placement_falls_back_to_round_robin_on_fresh_runtime() { ); } } + +// ── Mailbox Backpressure Tests ───────────────────────────────────────────── + +/// Given a runtime with bounded mailboxes (capacity=10, DropNewest), +/// when 50 messages are sent to an actor before any ticks, +/// then only the first 10 are delivered and the rest are dropped. +#[test] +fn bounded_mailbox_drop_newest_caps_at_capacity() { + let rt = Runtime::new(RuntimeConfig { + default_mailbox_capacity: 10, + mailbox_overflow: MailboxOverflow::DropNewest, + ..Default::default() + }); + + let inbox = rt.new_inbox::().unwrap(); + let addr = rt.spawn(CounterActor { count: 0 }).unwrap(); + + // Send 50 messages — only first 10 should be queued + for _ in 0..50 { + let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() }); + } + + // Tick enough times to process all queued messages + for _ in 0..20 { + rt.tick(); + } + + // Count replies — should be exactly 10 (the mailbox capacity) + let mut replies = 0; + while inbox.try_recv().is_some() { + replies += 1; + } + assert_eq!(replies, 10, "should deliver exactly mailbox_capacity messages"); + + // Stats should show drops + let stats = rt.stats(); + let total_drops: u64 = stats.workers.iter().map(|w| w.messages_dropped).sum(); + assert_eq!(total_drops, 40, "40 messages should have been dropped"); +} + +/// Given a runtime with bounded mailboxes (capacity=5, DropOldest), +/// when 10 messages are sent before any tick, +/// then only the 5 most recent messages are delivered. +#[test] +fn bounded_mailbox_drop_oldest_keeps_newest() { + let rt = Runtime::new(RuntimeConfig { + default_mailbox_capacity: 5, + mailbox_overflow: MailboxOverflow::DropOldest, + ..Default::default() + }); + + let inbox = rt.new_inbox::().unwrap(); + let addr = rt.spawn(DoubleActor).unwrap(); + + // Send messages with values 0..10. DoubleActor replies Done(value * 2). + // With DropOldest and capacity 5, messages 0-4 should be dropped as 5-9 arrive. + for i in 0..10 { + let _ = rt.send_to(addr, Forward { + value: i, + reply_to: *inbox.addr(), + }); + } + + for _ in 0..10 { + rt.tick(); + } + + // Collect all replies + let mut replies = Vec::new(); + while let Some(Done(v)) = inbox.try_recv() { + replies.push(v); + } + + assert_eq!(replies.len(), 5, "should deliver exactly 5 messages"); + // The 5 most recent: values 5,6,7,8,9 → doubled: 10,12,14,16,18 + assert_eq!(replies, vec![10, 12, 14, 16, 18], "should keep the newest messages"); +} + +/// Given a runtime with unbounded mailboxes (capacity=0, the default), +/// when many messages are sent, +/// then all are delivered (backward compatibility). +#[test] +fn unbounded_mailbox_delivers_all_messages() { + let rt = Runtime::new(RuntimeConfig::default()); + + let inbox = rt.new_inbox::().unwrap(); + let addr = rt.spawn(CounterActor { count: 0 }).unwrap(); + + for _ in 0..200 { + let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() }); + } + + for _ in 0..50 { + rt.tick(); + } + + let mut replies = 0; + while inbox.try_recv().is_some() { + replies += 1; + } + assert_eq!(replies, 200, "all 200 messages should be delivered with unbounded mailbox"); + + let stats = rt.stats(); + let total_drops: u64 = stats.workers.iter().map(|w| w.messages_dropped).sum(); + assert_eq!(total_drops, 0, "no drops with unbounded mailbox"); +} + +/// Given bounded mailboxes with budget, when an actor processes messages +/// and frees mailbox space, then new messages should be accepted on subsequent ticks. +#[test] +fn bounded_mailbox_refills_after_processing() { + let rt = Runtime::new(RuntimeConfig { + default_mailbox_capacity: 5, + actor_message_budget: 5, + mailbox_overflow: MailboxOverflow::DropNewest, + ..Default::default() + }); + + let inbox = rt.new_inbox::().unwrap(); + let addr = rt.spawn(CounterActor { count: 0 }).unwrap(); + + // Send first batch of 5 — fills mailbox exactly + for _ in 0..5 { + let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() }); + } + + // Tick to process all 5 (budget=5, capacity=5) + rt.tick(); + + // Send second batch of 5 — mailbox is empty, so all 5 should be accepted + for _ in 0..5 { + let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() }); + } + + rt.tick(); + + let mut replies = 0; + while inbox.try_recv().is_some() { + replies += 1; + } + assert_eq!(replies, 10, "all 10 messages across 2 batches should be processed"); + + let stats = rt.stats(); + let total_drops: u64 = stats.workers.iter().map(|w| w.messages_dropped).sum(); + assert_eq!(total_drops, 0, "no drops when mailbox drains between batches"); +}