feat: per-actor mailbox backpressure with configurable overflow
Add optional bounded mailboxes to prevent unbounded memory growth. MailboxOverflow enum: DropNewest (discard incoming) or DropOldest (evict oldest to make room). Default capacity=0 preserves unbounded behavior for full backward compatibility. messages_dropped counter added to WorkerStats for observability. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
7d00e65a0a
commit
265992c3db
7 changed files with 242 additions and 11 deletions
|
|
@ -2,7 +2,7 @@
|
||||||
|
|
||||||
## Current Stage: Phase 1 — Research + First Improvement Cycle
|
## Current Stage: Phase 1 — Research + First Improvement Cycle
|
||||||
|
|
||||||
### Status: Cycle 5 COMPLETE
|
### Status: Cycle 6 COMPLETE
|
||||||
|
|
||||||
## Plan Overview
|
## Plan Overview
|
||||||
1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
|
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)
|
- **Benchmarks**: 1 new group — `placement/spawn_under_load` (2t, 4t)
|
||||||
- **Result**: 60 tests pass, all workspace compiles
|
- **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
|
### Research Notes
|
||||||
- Full analysis in `CLAUDE/notes/research_synthesis.md`
|
- Full analysis in `CLAUDE/notes/research_synthesis.md`
|
||||||
- Baseline benchmarks in `CLAUDE/notes/baseline_benchmarks.md`
|
- Baseline benchmarks in `CLAUDE/notes/baseline_benchmarks.md`
|
||||||
|
|
@ -99,8 +117,9 @@
|
||||||
- [x] **Cycle 3: Adaptive backoff with thread parking** ✅
|
- [x] **Cycle 3: Adaptive backoff with thread parking** ✅
|
||||||
- [x] **Cycle 4: Enhanced benchmarks + bug-inspired tests** ✅
|
- [x] **Cycle 4: Enhanced benchmarks + bug-inspired tests** ✅
|
||||||
- [x] **Cycle 5: Work stealing research + load-aware placement** ✅
|
- [x] **Cycle 5: Work stealing research + load-aware placement** ✅
|
||||||
- [ ] **Cycle 6: Next improvement**
|
- [x] **Cycle 6: Mailbox backpressure** ✅
|
||||||
- Candidates: mailbox backpressure, actor recovery, LIFO slot optimization
|
- [ ] **Cycle 7: Next improvement**
|
||||||
|
- Candidates: actor recovery (factory restart), LIFO slot, VecDeque→ring buffer optimization
|
||||||
- Pick based on highest impact-to-effort ratio
|
- Pick based on highest impact-to-effort ratio
|
||||||
|
|
||||||
## Open Questions
|
## Open Questions
|
||||||
|
|
|
||||||
|
|
@ -120,7 +120,7 @@ until A finishes. Every other runtime studied prevents this:
|
||||||
|
|
||||||
### Swactor Weaknesses to Address
|
### Swactor Weaknesses to Address
|
||||||
- Box<dyn Any> downcast can fail silently → type mismatch tracking needed (have it)
|
- Box<dyn Any> 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
|
- Panicked actors permanently poisoned → no recovery path
|
||||||
- ~~Spin/sleep backoff wastes CPU~~ → FIXED: thread parking with instant wakeup (Cycle 3)
|
- ~~Spin/sleep backoff wastes CPU~~ → FIXED: thread parking with instant wakeup (Cycle 3)
|
||||||
- No supervision trees
|
- No supervision trees
|
||||||
|
|
|
||||||
|
|
@ -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.
|
/// The tunable settings for the runtime.
|
||||||
pub struct RuntimeConfig {
|
pub struct RuntimeConfig {
|
||||||
pub max_actors: usize,
|
pub max_actors: usize,
|
||||||
|
|
@ -33,6 +42,11 @@ pub struct RuntimeConfig {
|
||||||
/// Prevents a single actor with a large mailbox from starving others.
|
/// Prevents a single actor with a large mailbox from starving others.
|
||||||
/// `0` means unlimited (drain entire mailbox).
|
/// `0` means unlimited (drain entire mailbox).
|
||||||
pub actor_message_budget: usize,
|
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
|
/// 8kB for the `Box<..>` before counting the rest of the memory
|
||||||
|
|
@ -55,6 +69,8 @@ impl Default for RuntimeConfig {
|
||||||
num_threads: 1,
|
num_threads: 1,
|
||||||
backoff_policy: BackoffPolicy::default(),
|
backoff_policy: BackoffPolicy::default(),
|
||||||
actor_message_budget: DEFAULT_ACTOR_MESSAGE_BUDGET,
|
actor_message_budget: DEFAULT_ACTOR_MESSAGE_BUDGET,
|
||||||
|
default_mailbox_capacity: 0,
|
||||||
|
mailbox_overflow: MailboxOverflow::DropNewest,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -8,7 +8,7 @@ use std::time::Instant;
|
||||||
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message};
|
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message};
|
||||||
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, RuntimeConfig};
|
pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
|
||||||
use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId};
|
use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId};
|
||||||
use crate::stats::{StatsHook, WorkerStats};
|
use crate::stats::{StatsHook, WorkerStats};
|
||||||
// Re-export stats types so existing code using `runtime::*` still works
|
// Re-export stats types so existing code using `runtime::*` still works
|
||||||
|
|
@ -137,7 +137,14 @@ impl Runtime {
|
||||||
|
|
||||||
let stats = Arc::new(WorkerStats::new());
|
let stats = Arc::new(WorkerStats::new());
|
||||||
worker_stats.push(stats.clone());
|
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());
|
let placement = Placement::new(num_workers, worker_stats.clone());
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,8 @@ pub struct WorkerStats {
|
||||||
// Error counters
|
// Error counters
|
||||||
pub type_mismatches: AtomicU64,
|
pub type_mismatches: AtomicU64,
|
||||||
pub panics: 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 timing ring buffer (last N ticks, lock-free)
|
||||||
tick_timings: ArrayQueue<TickTiming>,
|
tick_timings: ArrayQueue<TickTiming>,
|
||||||
}
|
}
|
||||||
|
|
@ -45,6 +47,7 @@ impl WorkerStats {
|
||||||
inbox_sends: AtomicU64::new(0),
|
inbox_sends: AtomicU64::new(0),
|
||||||
type_mismatches: AtomicU64::new(0),
|
type_mismatches: AtomicU64::new(0),
|
||||||
panics: AtomicU64::new(0),
|
panics: AtomicU64::new(0),
|
||||||
|
messages_dropped: AtomicU64::new(0),
|
||||||
tick_timings: ArrayQueue::new(TICK_BUFFER_CAP),
|
tick_timings: ArrayQueue::new(TICK_BUFFER_CAP),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -79,6 +82,7 @@ impl WorkerStats {
|
||||||
inbox_sends: self.inbox_sends.load(Relaxed),
|
inbox_sends: self.inbox_sends.load(Relaxed),
|
||||||
type_mismatches: self.type_mismatches.load(Relaxed),
|
type_mismatches: self.type_mismatches.load(Relaxed),
|
||||||
panics: self.panics.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 inbox_sends: u64,
|
||||||
pub type_mismatches: u64,
|
pub type_mismatches: u64,
|
||||||
pub panics: u64,
|
pub panics: u64,
|
||||||
|
pub messages_dropped: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Per-actor snapshot transferred from worker to runtime (not serialized).
|
/// Per-actor snapshot transferred from worker to runtime (not serialized).
|
||||||
|
|
|
||||||
|
|
@ -8,6 +8,7 @@ use std::time::Instant;
|
||||||
|
|
||||||
use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx};
|
use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx};
|
||||||
use crate::channel::Receiver;
|
use crate::channel::Receiver;
|
||||||
|
use crate::config::MailboxOverflow;
|
||||||
use crate::delivery::{Envelope, TickContext, WorkerId};
|
use crate::delivery::{Envelope, TickContext, WorkerId};
|
||||||
use crate::stats::{ActorSnapshot, TickTiming, WorkerStats};
|
use crate::stats::{ActorSnapshot, TickTiming, WorkerStats};
|
||||||
use crate::Error;
|
use crate::Error;
|
||||||
|
|
@ -29,10 +30,12 @@ impl Worker {
|
||||||
transfer_rx: Receiver<Envelope>,
|
transfer_rx: Receiver<Envelope>,
|
||||||
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
|
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
|
||||||
stats: Arc<WorkerStats>,
|
stats: Arc<WorkerStats>,
|
||||||
|
default_mailbox_capacity: usize,
|
||||||
|
default_overflow_policy: MailboxOverflow,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
id,
|
id,
|
||||||
pool: ActorPool::new(),
|
pool: ActorPool::new(default_mailbox_capacity, default_overflow_policy),
|
||||||
transfer_rx,
|
transfer_rx,
|
||||||
spawn_rx,
|
spawn_rx,
|
||||||
stats,
|
stats,
|
||||||
|
|
@ -119,10 +122,14 @@ impl Worker {
|
||||||
let t5 = Instant::now();
|
let t5 = Instant::now();
|
||||||
|
|
||||||
// 6. Publish stats (skip entirely when idle to avoid allocation + mutex)
|
// 6. Publish stats (skip entirely when idle to avoid allocation + mutex)
|
||||||
|
let drops = self.pool.take_drops();
|
||||||
if did_work {
|
if did_work {
|
||||||
self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed);
|
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.total_mailbox_depth.store(self.pool.total_mailbox_depth(), Ordering::Relaxed);
|
||||||
self.stats.messages_processed.fetch_add(processed as u64, 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 {
|
if let Some(hook) = tc.stats_hook {
|
||||||
self.pool.mailbox_depths_into(&mut self.snapshot_buf);
|
self.pool.mailbox_depths_into(&mut self.snapshot_buf);
|
||||||
|
|
@ -238,34 +245,60 @@ struct ActorSlot {
|
||||||
poisoned: bool,
|
poisoned: bool,
|
||||||
last_msg_type: Option<&'static str>,
|
last_msg_type: Option<&'static str>,
|
||||||
messages_processed: u64,
|
messages_processed: u64,
|
||||||
|
/// Per-actor mailbox capacity. 0 = unbounded.
|
||||||
|
mailbox_capacity: usize,
|
||||||
|
overflow_policy: MailboxOverflow,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Per-worker actor storage. Owns per-actor mailboxes.
|
/// Per-worker actor storage. Owns per-actor mailboxes.
|
||||||
pub(crate) struct ActorPool {
|
pub(crate) struct ActorPool {
|
||||||
actors: HashMap<ActorAddress, ActorSlot>,
|
actors: HashMap<ActorAddress, ActorSlot>,
|
||||||
|
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 {
|
impl ActorPool {
|
||||||
pub fn new() -> Self {
|
pub fn new(default_mailbox_capacity: usize, default_overflow_policy: MailboxOverflow) -> Self {
|
||||||
Self {
|
Self {
|
||||||
actors: HashMap::new(),
|
actors: HashMap::new(),
|
||||||
|
default_mailbox_capacity,
|
||||||
|
default_overflow_policy,
|
||||||
|
drops_this_tick: 0,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
|
pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
|
||||||
|
let cap = self.default_mailbox_capacity;
|
||||||
|
let prealloc = if cap > 0 { cap.min(64) } else { 16 };
|
||||||
self.actors.insert(addr, ActorSlot {
|
self.actors.insert(addr, ActorSlot {
|
||||||
mailbox: VecDeque::with_capacity(16),
|
mailbox: VecDeque::with_capacity(prealloc),
|
||||||
actor,
|
actor,
|
||||||
poisoned: false,
|
poisoned: false,
|
||||||
last_msg_type: None,
|
last_msg_type: None,
|
||||||
messages_processed: 0,
|
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`.
|
/// 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<dyn Any + Send>) -> bool {
|
pub fn deliver(&mut self, addr: &ActorAddress, msg: Box<dyn Any + Send>) -> bool {
|
||||||
if let Some(slot) = self.actors.get_mut(addr) {
|
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);
|
slot.mailbox.push_back(msg);
|
||||||
true
|
true
|
||||||
} else {
|
} 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.
|
/// Tick all actors in the pool. Returns the number of messages processed.
|
||||||
///
|
///
|
||||||
/// Each actor processes up to `budget` messages per tick (0 = unlimited).
|
/// Each actor processes up to `budget` messages per tick (0 = unlimited).
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use swactor::actor::{ActorAddress, ActorInterface};
|
use swactor::actor::{ActorAddress, ActorInterface};
|
||||||
use swactor::runtime::{Ctx, Inbox, Runtime, RuntimeConfig};
|
use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig};
|
||||||
|
|
||||||
// ── Messages ────────────────────────────────────────────────────────────────
|
// ── 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::<Count>().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::<Done>().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::<Count>().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::<Count>().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");
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue