feat: identity hashing for ActorAddress hot-path optimization (Cycle 20)
Replace SipHash with identity hasher on all hot-path HashMaps keyed by ActorAddress. Since addresses are crypto-random, the first 8 bytes serve as an excellent hash directly. Microbenchmarks show 1.9-4.6x lookup speedup depending on map size. - Custom Hash impl for ActorAddress (8-byte write_u64 instead of 32) - AddrHasher/AddrBuildHasher identity hasher in delivery.rs - AddrMap<V>/AddrSet type aliases used in 7 HashMap sites - Stop-requests is_empty() short-circuit in tick_all inner loop - 3 new behavioral tests, component microbenchmarks Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Developer
parent
9ad90aed12
commit
e91308720b
9 changed files with 639 additions and 502 deletions
102
CLAUDE/notes/dispatch_comparison.md
Normal file
102
CLAUDE/notes/dispatch_comparison.md
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
# Dispatch Model Comparison: Stakker vs Actix vs Swactor
|
||||
|
||||
## Stakker — Closure-Based Dispatch (No HashMap, No Downcast)
|
||||
|
||||
**Architecture**: Single-threaded, synchronous actor runtime (like swactor). Messages are `FnOnce` closures, not typed structs.
|
||||
|
||||
**Key Design**:
|
||||
- `call!` macro turns method calls into closures: `call!([actor], method(arg1, arg2))` generates a `FnOnce` that directly calls the method on the actor
|
||||
- The closure captures the actor reference and method pointer — dispatch is a **direct function call**, not a downcast
|
||||
- Closures stored in a **flat heterogeneous FnOnce queue** (byte Vec) — no per-message heap allocation
|
||||
- Compiler can inline the closure, reducing message handling to "a single branch to optimised inlined code"
|
||||
|
||||
**Why It's Fast**:
|
||||
- Zero `Box<dyn Any>` allocation per message
|
||||
- Zero `TypeId` downcast per message
|
||||
- Zero HashMap lookup per message (actors addressed by direct `ActorOwn<A>` references, not opaque addresses)
|
||||
- Queue is a flat contiguous memory region — excellent cache locality
|
||||
|
||||
**Tradeoffs**:
|
||||
- Single-threaded only (no multi-worker routing)
|
||||
- No location-transparent addresses (can't route to remote actors)
|
||||
- Uses `unsafe` code by default for the FnOnce queue
|
||||
|
||||
## Actix — Vtable Dispatch (No HashMap, No Downcast)
|
||||
|
||||
**Architecture**: Async actor framework on tokio. Each actor is a pollable Future on an Arbiter thread.
|
||||
|
||||
**Key Design**:
|
||||
- `Addr<A>` wraps an `AddressSender<A>` — a direct channel reference, not an address in a map
|
||||
- Messages wrapped as `Box<dyn EnvelopeProxy<A>>` — vtable dispatch, not `Box<dyn Any>` downcast
|
||||
- Custom Vyukov lock-free MPSC queue (single `AtomicPtr::swap` for push)
|
||||
- Default mailbox capacity: 16
|
||||
|
||||
**Why It's Fast**:
|
||||
- `Addr<A>` is a direct channel reference — zero HashMap lookup per send
|
||||
- `EnvelopeProxy` vtable call — one virtual dispatch, no TypeId comparison
|
||||
- Vyukov MPSC queue — lock-free push, minimal atomic operations
|
||||
- `do_send()` bypasses backpressure for internal messages
|
||||
|
||||
**Tradeoffs**:
|
||||
- Requires async runtime (tokio dependency)
|
||||
- `Addr<A>` is typed — can't send different message types without `Recipient<M>` adaptation
|
||||
- `do_send()` silently drops messages to closed actors (no error feedback)
|
||||
|
||||
## Swactor — Type-Erased Dispatch with HashMap Routing
|
||||
|
||||
**Architecture**: Synchronous tick-based runtime with multi-worker support. Messages are `Box<dyn Any + Send>`.
|
||||
|
||||
**Message Send Path** (per-message costs):
|
||||
1. `Box::new(msg)` — heap allocation (~10-15ns)
|
||||
2. `address_map.lookup(&addr)` — RwLock read + HashMap get with 32-byte key
|
||||
3. Push to VecDeque or crossbeam queue
|
||||
|
||||
**Message Process Path** (per-message costs):
|
||||
1. `slot.mailbox.pop_front()` — VecDeque pop
|
||||
2. `msg.downcast::<Incoming>()` — TypeId comparison (~1ns)
|
||||
3. `catch_unwind(|| actor.handle_any(ctx, msg))` — unwind setup (~3-5ns)
|
||||
|
||||
**Why It's Slower on Paper**:
|
||||
- HashMap lookup on every send AND every deliver (2 lookups per message)
|
||||
- ActorAddress is 32 bytes — expensive to hash (SipHash: ~15ns for 32 bytes)
|
||||
- `Box<dyn Any>` heap allocation on every send
|
||||
|
||||
**Why This Architecture Exists**:
|
||||
- Location-transparent 32-byte addresses enable multi-worker routing, transport layer, and distribution
|
||||
- Type erasure enables heterogeneous mailboxes and type-agnostic forwarding
|
||||
- HashMap enables O(1) address → worker lookup for any actor, from any thread
|
||||
|
||||
## Optimization Applied: Identity Hashing
|
||||
|
||||
**Problem**: Every HashMap operation hashes 32 bytes of ActorAddress with SipHash.
|
||||
|
||||
**Solution** (two-layer):
|
||||
1. Custom `Hash` impl on ActorAddress — only hashes first 8 bytes via `write_u64` (SipHash on 8 bytes is ~3x faster than 32 bytes)
|
||||
2. Identity hasher (`AddrHasher`) on hot-path HashMaps — uses the 8-byte hash value directly as the bucket index, skipping SipHash entirely
|
||||
|
||||
**Microbenchmark Results** (same-process A/B comparison, reliable):
|
||||
|
||||
| Operation | SipHash (8-byte) | Identity | Speedup |
|
||||
|-----------|------------------|----------|---------|
|
||||
| Hash | 1.35 ns | 0.67 ns | 2.0x |
|
||||
| Lookup (100 entries) | 29.8 ns | 15.5 ns | 1.9x |
|
||||
| Lookup (1000 entries) | 38.0 ns | 8.2 ns | 4.6x |
|
||||
| Insert 1000 | 48.2 µs | 21.1 µs | 2.3x |
|
||||
|
||||
**Why It's Safe**: ActorAddress bytes come from `getrandom` — cryptographically random, providing excellent uniform distribution without additional mixing.
|
||||
|
||||
## Summary Table
|
||||
|
||||
| Aspect | Stakker | Actix | Swactor (optimized) |
|
||||
|--------|---------|-------|---------------------|
|
||||
| Message type | `FnOnce` closure | `Box<dyn EnvelopeProxy>` | `Box<dyn Any + Send>` |
|
||||
| Dispatch | Direct call / inlined | Vtable call | TypeId downcast |
|
||||
| Address lookup | None (direct ref) | None (direct channel) | Identity-hash HashMap |
|
||||
| Per-msg alloc | None (flat queue) | Box (MPSC node) | Box (heap + VecDeque) |
|
||||
| Multi-thread | No | Yes (tokio) | Yes (worker threads) |
|
||||
| Location transparency | No | No | Yes (32-byte address) |
|
||||
| Remote transport | No | No (without extras) | Yes (pluggable) |
|
||||
|
||||
## Key Insight
|
||||
|
||||
Stakker and Actix are faster because they avoid the per-message HashMap lookup entirely — addresses are direct references, not opaque identifiers that need routing. Swactor pays for HashMap routing because its 32-byte addresses enable multi-worker distribution and remote transport. The identity hasher minimizes this cost without changing the architecture.
|
||||
|
|
@ -1,487 +1,62 @@
|
|||
# Progress Log
|
||||
|
||||
## Current Stage: Phase 1 — Research + First Improvement Cycle
|
||||
## Current Stage: Cycle 20 — Hot-Path Performance (Identity Hashing)
|
||||
|
||||
### Status: Cycle 19 COMPLETE
|
||||
### Status: COMPLETE
|
||||
|
||||
## Plan Overview
|
||||
1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
|
||||
2. **Phase 1**: Broad survey + interleaved improvements
|
||||
3. **Phase 2**: Deeper improvements based on findings
|
||||
4. **Phase 3**: Testing methodology improvements
|
||||
5. **Phase 4**: Final evaluation & documentation
|
||||
### Research
|
||||
- **Stakker**: Closure-based dispatch. `call!` macro generates `FnOnce` closures pushed to a flat byte Vec queue. No HashMap, no Box<dyn Any>, no downcast. Direct method calls that the compiler can inline. Single-threaded only.
|
||||
- **Actix**: Vtable dispatch via `Box<dyn EnvelopeProxy<A>>`. `Addr<A>` is a direct channel reference (no HashMap lookup). Custom Vyukov lock-free MPSC queue. Default mailbox capacity 16.
|
||||
- **Key insight**: Both avoid HashMap entirely by using direct references. Swactor needs HashMaps for location-transparent 32-byte addresses (multi-worker routing + transport). The optimization is to minimize HashMap cost, not eliminate it.
|
||||
- Full analysis: `CLAUDE/notes/dispatch_comparison.md`
|
||||
|
||||
## Completed This Session
|
||||
### Implementation
|
||||
1. **Custom `Hash` for ActorAddress** (`src/actor.rs`) — only hashes first 8 bytes instead of 32. All HashMaps using ActorAddress benefit automatically.
|
||||
2. **Identity hasher** (`src/delivery.rs`) — `AddrHasher`/`AddrBuildHasher` that passes the 8-byte hash value through as the bucket index directly, skipping SipHash.
|
||||
3. **Hot-path HashMap replacement** — `AddrMap<V>` type alias used in:
|
||||
- `AddressMap.inner` (delivery.rs) — on every send
|
||||
- `ActorPool.actors` (worker.rs) — on every deliver and tick_all
|
||||
- `InboxRegistry.senders` (delivery.rs)
|
||||
- `MonitorRegistry.monitors` (delivery.rs)
|
||||
- `NameRegistry.reverse` (delivery.rs)
|
||||
- `GroupRegistry.memberships` + `AddrSet` for group member sets (delivery.rs)
|
||||
- `TransportRouter.routes` (transport.rs)
|
||||
4. **Stop-requests optimization** (worker.rs) — `is_empty()` short-circuit before linear scan in inner message loop
|
||||
|
||||
### Cycle 1: Fairness (Message Budget)
|
||||
- **Research**: Studied ractor, tokio, Erlang/OTP BEAM, Linux CFS/EEVDF, libuv
|
||||
- **Finding**: `tick_all` drained ENTIRE mailbox per actor per tick — critical fairness bug
|
||||
- BEAM uses 4000 reduction budget, tokio uses 128-op cooperative budget
|
||||
- Swactor had zero budget — one hot actor could starve all others on same worker
|
||||
- **Implementation**: Added `actor_message_budget` to `RuntimeConfig` (default: 64)
|
||||
- Modified `tick_all` to break after `budget` messages per actor
|
||||
- `budget=0` means unlimited (backward compatible)
|
||||
- **Tests**: 3 new fairness tests (hot_actor_does_not_starve_cold_actor, unlimited_budget_drains_all, budget_messages_drain_across_multiple_ticks)
|
||||
- **Benchmarks**: Added fairness benchmark group (cold_latency_under_pressure, throughput_by_budget)
|
||||
- **Fixes**: Updated RuntimeConfig struct literals across crates (python, runtime-dashboard, mt_benchmarks)
|
||||
- **Result**: 45 tests pass (42 original + 3 new), all workspace crates compile
|
||||
### Microbenchmark Results (reliable, same-process A/B)
|
||||
| Operation | SipHash | Identity | Speedup |
|
||||
|-----------|---------|----------|---------|
|
||||
| Hash | 1.35 ns | 0.67 ns | 2.0x |
|
||||
| Lookup/100 | 29.8 ns | 15.5 ns | 1.9x |
|
||||
| Lookup/1000 | 38.0 ns | 8.2 ns | 4.6x |
|
||||
| Insert 1000 | 48.2 µs | 21.1 µs | 2.3x |
|
||||
|
||||
### Cycle 2: Stress Tests, Benchmarks, Research Expansion
|
||||
- **Research**: Added Kameo and Actix analysis to synthesis
|
||||
- Actix uses custom Vyukov lock-free MPSC queue (why it's fastest)
|
||||
- Kameo has dual bounded/unbounded mailbox, default capacity 64
|
||||
- Both use vtable dispatch (not Box<dyn Any> downcast)
|
||||
- Actix has 256-message assertion guard (validates our budget approach)
|
||||
- **Stress tests**: 6 new tests
|
||||
- `message_ordering_preserved_under_budget` — FIFO order with budget=8
|
||||
- `mt_stress_many_senders_one_receiver` — 50 senders × 100 msgs, 4 threads
|
||||
- `mt_stress_concurrent_spawn_and_send` — 200 concurrent spawn+send, 4 threads
|
||||
- `mt_chain_spawning_under_load` — 50-level chain across 2 workers
|
||||
- `mt_panic_isolation_under_load` — 10 panicking + 10 healthy actors, 4 threads
|
||||
- `sustained_throughput_does_not_drop_messages` — 10 batches × 100 msgs
|
||||
- **Benchmarks**: 2 new benchmark groups
|
||||
- `msg_size`: throughput and send_latency by message size (8B-4KB)
|
||||
- `contention`: fanin (1-100 senders to 1 sink), cross_worker (1-4 threads)
|
||||
- **Result**: 51 tests pass (42 original + 3 fairness + 6 stress), all workspace compiles
|
||||
Note: End-to-end benchmarks unreliable in sandbox (55% variation between identical runs). Microbenchmarks confirmed significant hash/lookup improvement.
|
||||
|
||||
### Cycle 3: Thread Parking (Adaptive Backoff)
|
||||
- **Implementation**: Replaced `thread::sleep` with `thread::park_timeout` in worker run loop
|
||||
- Workers register `thread::current()` via `OnceLock<Thread>` on startup
|
||||
- `send_to` and `spawn` call `Thread::unpark()` on target worker
|
||||
- Cross-worker sends from `WorkerContext` also unpark target
|
||||
- Zero new dependencies (uses `std::sync::OnceLock` + `std::thread::park_timeout`)
|
||||
- **Design source**: Tokio's parker state machine, Linux NO_HZ adaptive ticks
|
||||
- **Benefits**: Parked workers wake instantly when work arrives (vs waiting for sleep timer)
|
||||
- Reduces idle-to-active latency from up to 1ms to near-zero
|
||||
- No overhead on hot path — `unpark()` is no-op if thread isn't parked
|
||||
- **Tests**: 1 new test (`mt_parked_worker_wakes_on_send`)
|
||||
- **Result**: 52 tests pass (51 + 1 new), all workspace compiles
|
||||
### Tests
|
||||
- 143 behavioral tests pass (140 existing + 3 new)
|
||||
- 7 proptest pass
|
||||
- New tests:
|
||||
- `many_actors_all_receive_correct_messages` — 200 actors, verifies no misrouting from identity hasher
|
||||
- `ring_routing_unchanged_after_hasher_optimization` — 100-actor chain, verifies address_map correctness
|
||||
- `stop_self_with_pending_messages_still_works` — verifies stop_requests optimization correctness
|
||||
|
||||
### Cycle 4: Shutdown Fix + Bug-Inspired Tests
|
||||
- **Shutdown improvement**: `shutdown()` now unparks all workers for immediate exit
|
||||
- Previously, parked workers wouldn't notice shutdown until park_timeout expired
|
||||
- **Bug-inspired tests** (5 new, from competitor bug reports):
|
||||
- `stats_snapshot_is_read_only` — from ractor #310 (destructive get_children)
|
||||
- `stats_under_load_do_not_interfere_with_processing` — stats don't affect msg processing
|
||||
- `shutdown_wakes_parked_workers_immediately` — validates fast shutdown with parking
|
||||
- `mt_send_after_run_delivers_to_running_actors` — from kameo #185 (startup delivery)
|
||||
- `budget_respected_even_with_self_sends` — from actix #515 (mailbox bypass)
|
||||
- **Result**: 57 tests pass, all workspace compiles
|
||||
|
||||
### Cycle 5: Work Stealing Research + Load-Aware Placement
|
||||
- **Research**: Deep analysis of work stealing in Tokio, Go, BEAM, ForkJoinPool
|
||||
- Tokio: fixed 256-slot ring, steal-half, LIFO slot (3-use starvation cap), N/2 searcher limit
|
||||
- Go: M:N scheduler, runnext + 256-slot local queue, steal-half, 4 tries with random permutation
|
||||
- BEAM: unique dual approach — reactive stealing + proactive migration via check_balance()
|
||||
- ForkJoinPool: owner LIFO / thief FIFO deque, even/odd queue indexing
|
||||
- **Feasibility analysis**: Full actor migration IS mechanically possible (ActorSlot is Send), but:
|
||||
- Requires push-based donation (ActorPool not Sync → no pull stealing)
|
||||
- 1-tick message loss window during migration
|
||||
- Significant complexity for uncertain benefit
|
||||
- **Implementation**: Load-aware placement replaces blind round-robin
|
||||
- `Placement::next_worker()` now reads per-worker stats (num_actors + mailbox_depth)
|
||||
- Scan starts from rotating position → round-robin when all stats equal (initial burst)
|
||||
- O(N) relaxed atomic loads per spawn, trivial for N≤8 workers
|
||||
- **Tests**: 3 new tests
|
||||
- `load_aware_placement_prefers_lighter_worker` — imbalanced load biases toward lighter worker
|
||||
- `load_aware_placement_single_worker_degrades_gracefully` — single-thread works correctly
|
||||
- `load_aware_placement_falls_back_to_round_robin_on_fresh_runtime` — even distribution before ticks
|
||||
- **Benchmarks**: 1 new group — `placement/spawn_under_load` (2t, 4t)
|
||||
- **Result**: 60 tests pass, all workspace compiles
|
||||
|
||||
### Cycle 9: Lifecycle Hooks + Graceful Stop
|
||||
- **Research**: Cross-framework lifecycle analysis (Erlang init/terminate, Akka preStart/postStop,
|
||||
Actix started/stopping/stopped, Kameo on_start/on_stop/on_panic, Ractor pre_start/post_stop,
|
||||
Stakker state-based Prep/Ready/Zombie, CAF on_exit)
|
||||
- Also researched graceful stop across Erlang (gen_server:stop, exit, kill), Akka (stop, PoisonPill,
|
||||
Kill, gracefulStop), Actix (ctx.stop, Running::Stop), Kameo (stop_gracefully, kill), Go (context.Done)
|
||||
- Key finding: most frameworks have on_stop NOT called on panic (state may be corrupt)
|
||||
- Key finding: self-stop should be immediate (after current message), external stop is queued
|
||||
- **Implementation**: Lifecycle hooks + dual-mode graceful stop
|
||||
- `ActorInterface::on_start()` and `on_stop()` — default no-ops, backward compatible
|
||||
- `AnyActor::on_start()`/`on_stop()` forwarded from `Actor<A>` impl
|
||||
- `ctx.stop_self()` — immediate stop after current message via `request_stop` buffer
|
||||
- `runtime.stop_actor(addr)` — external stop via StopSignal message (PoisonPill semantics)
|
||||
- `ActorSlot` gains `started: bool` and `stopping: bool` flags
|
||||
- `on_start` called in tick_all before first message; panic in on_start → immediate poison
|
||||
- `on_stop` called in cleanup_dead for stopping (not poisoned) actors, wrapped in catch_unwind
|
||||
- Restarted actors get `started=false` so on_start fires again on fresh instance
|
||||
- `stops: AtomicU64` added to WorkerStats and WorkerInfo
|
||||
- `ContextInner::request_stop()` method for same-worker immediate stop
|
||||
- Phase 7 cleanup_dead now handles both poisoned AND stopping actors, with on_stop context
|
||||
- **Tests**: 12 new tests
|
||||
- `on_start_called_before_first_message` — on_start fires on first tick, before messages
|
||||
- `on_start_called_per_actor` — 5 actors each get one on_start call
|
||||
- `on_start_panic_poisons_actor` — panic in on_start → poisoned, no messages processed
|
||||
- `actor_can_stop_self` — 5 msgs sent, stops after 3, only 3 processed, on_stop called
|
||||
- `runtime_can_stop_actor` — external stop via runtime, on_stop called, actor removed
|
||||
- `send_to_stopped_actor_returns_error` — stopped actor gone from address map
|
||||
- `stop_vs_panic_tracked_separately_in_stats` — stops and panics counted independently
|
||||
- `on_stop_can_send_messages` — farewell message sent during on_stop is delivered
|
||||
- `on_start_called_again_after_restart` — restartable actor gets on_start on fresh instance
|
||||
- `external_stop_is_queued_after_pending_messages` — PoisonPill semantics for external stop
|
||||
- `external_stop_before_new_messages_prevents_processing` — stop before send blocks msgs
|
||||
- `stop_nonexistent_actor_returns_error` — stop on bad address returns Err
|
||||
- **Result**: 82 tests pass, all workspace compiles
|
||||
|
||||
### Cycle 17: Supervision Trees (handle_down + Supervisor Actor)
|
||||
- **Research**: Cross-framework supervision analysis — Erlang (one_for_one/all/rest, child specs, intensity/period),
|
||||
Akka (SupervisorStrategy, Resume/Restart/Stop/Escalate, BackoffSupervisor), Ractor (SupervisionEvent,
|
||||
ractor-supervisor crate), Bastion (hierarchy, redundancy groups), CAF (no built-in supervisor, monitor-based)
|
||||
- Key finding: swactor has all building blocks (monitor, spawn_restartable, lifecycle hooks, Down messages)
|
||||
- Decision: Supervisor as a user-space actor built on existing primitives (like Ractor base crate)
|
||||
- handle_down callback enables any actor to react to monitored deaths without making Down the Incoming type
|
||||
- **Implementation**: Three features added to `src/actor.rs`:
|
||||
1. **`handle_down` callback on ActorInterface** — default no-op, called when monitored actor dies
|
||||
and actor's Incoming type is NOT Down. Implemented via second downcast attempt in `handle_any`.
|
||||
Fully backward-compatible: actors with `Incoming = Down` still receive via `handle()`.
|
||||
2. **`ctx.stop_actor(addr)`** — send graceful stop to another actor from handler context.
|
||||
Uses StopSignal through normal message routing (PoisonPill semantics).
|
||||
3. **`Supervisor` actor** — manages child actors with configurable restart policies:
|
||||
- `SupervisorStrategy::OneForOne` — only failed child is restarted
|
||||
- `RestartPolicy::Permanent` — always restart
|
||||
- `RestartPolicy::Transient` — restart only on Panicked, not Normal
|
||||
- `RestartPolicy::Temporary` — never restart
|
||||
- `ChildSpec` with id, restart policy, and factory closure `Fn(&Ctx) -> Result<ActorAddress>`
|
||||
- Meltdown detection: stops itself when `total_restarts > max_restarts`
|
||||
- Cascading shutdown: on_stop sends stop signals to all living children
|
||||
- **Tests**: 10 new behavioral tests
|
||||
- `handle_down_receives_death_notification` — handle_down callback fires on monitored death
|
||||
- `handle_down_skipped_when_incoming_is_down` — backward compat: Incoming=Down uses handle()
|
||||
- `ctx_stop_actor_stops_target` — one actor can stop another via ctx.stop_actor()
|
||||
- `supervisor_restarts_permanent_child_on_panic` — panic → restart (OneForOne + Permanent)
|
||||
- `supervisor_does_not_restart_transient_child_on_normal_stop` — Normal stop → no restart
|
||||
- `supervisor_restarts_transient_child_on_panic` — Panicked → restart (Transient)
|
||||
- `supervisor_never_restarts_temporary_child` — Temporary → never restart
|
||||
- `supervisor_meltdown_after_max_restarts` — exceeding max_restarts stops supervisor
|
||||
- `supervisor_one_for_one_only_restarts_failed_child` — multi-child, only crashed child affected
|
||||
- `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<M>` 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<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + 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.
|
||||
Chose to extend Supervisor with remaining Erlang-style strategies instead.
|
||||
- **Implementation**: Extended `Supervisor` in `src/actor.rs` with coordinated restart strategies:
|
||||
- `SupervisorStrategy::OneForAll` — all children restarted when one fails
|
||||
- `SupervisorStrategy::RestForOne` — failed child + all children after it (in spec order) restarted
|
||||
- `SupervisorPhase` state machine: `Normal` (steady state) | `Stopping { awaiting, restart_set }` (coordinated)
|
||||
- During coordinated restart: supervisor stops living siblings, waits for all Down confirmations,
|
||||
then restarts the full restart set in spec order
|
||||
- `begin_coordinated_restart(ctx, indices)` — sends stop signals, transitions to Stopping phase
|
||||
- `finish_restart(ctx)` — called when all awaiting Downs received, restarts from spec order
|
||||
- `check_intensity()` factored out for restart budget checking
|
||||
- Already-dead children are handled: if all targets are already dead, immediate restart (no Stopping phase)
|
||||
- **Tests**: 3 new behavioral tests
|
||||
- `supervisor_one_for_all_restarts_all_on_single_failure` — one child panics, all 3 get new addresses
|
||||
- `supervisor_rest_for_one_restarts_rest_after_failed` — child_b panics, child_a unchanged, child_c restarted
|
||||
- `supervisor_one_for_all_waits_for_all_downs_before_restart` — verifies coordinated shutdown completes before restart
|
||||
- **Result**: 141 tests pass (133 behavioral + 7 proptest + 1 doctest), zero warnings, full workspace compiles
|
||||
|
||||
### Cycle 16: Benchmark New Features
|
||||
- **Scope**: Added benchmark group for features from Cycles 12-15 (named actors, groups, monitors, ask)
|
||||
- **New benchmarks** (5 total in `registry` group):
|
||||
- `named_spawn_lookup` — spawn_named + where_is roundtrip: **~2.4µs** (vs bare spawn 1.9µs → +0.5µs overhead for name registration)
|
||||
- `where_is_100_names` — lookup in 100-name registry: **~9.0µs** (includes setup overhead)
|
||||
- `group_publish/{10,50,100}` — broadcast to N members: 4.8µs/15.5µs/60µs (linear with O(N) clones)
|
||||
- `monitor_setup` — monitor + stop + cleanup: **~13.4µs**
|
||||
- `ask_roundtrip` — ask + recv_ticking: **~4.5µs** (vs manual roundtrip 3.0µs → +1.5µs for inbox creation)
|
||||
- **Analysis**: All registry operations are efficient. Named lookup adds <1µs over bare spawn.
|
||||
Ask adds ~50% overhead vs manual inbox pattern (acceptable for convenience). Group publish
|
||||
scales linearly — expected for O(N) message cloning. No optimization needed.
|
||||
- **Result**: All benchmarks run cleanly, 127 tests pass, zero warnings
|
||||
|
||||
### Cycle 15: Ask Pattern (Request-Response)
|
||||
- **Research**: Studied ask/call/request-response patterns across Erlang gen_server:call (From + reply),
|
||||
Akka ask (temporary actor + Future), Ractor call (RpcReplyPort), Kameo ask (async + Reply trait),
|
||||
xactor Handler (return value auto-routing)
|
||||
- Key finding: swactor's synchronous tick model requires explicit reply_to, not implicit routing
|
||||
- Decision: convenience wrapper over existing inbox pattern, not implicit auto-reply
|
||||
- **Implementation**: `Ask<R>` struct + `Runtime::ask()` method
|
||||
- `Ask<R>`: wraps `Inbox<R>` with `try_recv()` and `recv_ticking(rt, max_ticks)`
|
||||
- `rt.ask(addr, |reply_to| Msg { reply_to })` — creates inbox, builds message, sends, returns Ask<R>
|
||||
- `ask.recv_ticking(&rt, max_ticks)` — ticks until response or timeout (single-threaded only)
|
||||
- `ask.try_recv()` — poll without ticking (works in both modes)
|
||||
- `ask.reply_addr()` — access inbox address for manual use
|
||||
- Purely sugar over `new_inbox → send_to → tick → try_recv` pattern
|
||||
- Zero changes to ContextInner or ActorInterface — no implicit auto-reply magic
|
||||
- **Tests**: 5 new behavioral tests
|
||||
- `ask_recv_ticking_returns_response` — basic PingPong ask roundtrip
|
||||
- `ask_multiple_times_tracks_state` — 3 sequential asks to CounterActor
|
||||
- `ask_timeout_when_no_response` — ask dead actor → timeout error
|
||||
- `ask_try_recv_returns_none_before_tick` — poll before tick → None, after tick → Some
|
||||
- `ask_reply_addr_is_accessible` — reply address is valid
|
||||
- **Result**: 127 tests pass (120 behavioral + 7 proptest), all workspace compiles, zero warnings
|
||||
|
||||
### Cycle 14: Actor Groups (Pub-Sub)
|
||||
- **Research**: Studied group/pub-sub patterns across Erlang pg (scopes, join/leave/get_members),
|
||||
Akka DistributedPubSub (mediator, topics), Ractor pg (join/leave/broadcast), Bastion (Dispatcher),
|
||||
Redis pub/sub (channels, patterns)
|
||||
- Common patterns: auto-cleanup on death, at-most-once delivery, string-based naming,
|
||||
flat groups (not hierarchical), lazy creation/deletion
|
||||
- Decision: Erlang pg-style flat groups, string keys, auto-cleanup, RwLock<HashMap> pattern
|
||||
- **Implementation**: `GroupRegistry` in delivery.rs with forward + reverse maps
|
||||
- `groups: RwLock<HashMap<String, HashSet<ActorAddress>>>` — group→members
|
||||
- `memberships: RwLock<HashMap<ActorAddress, HashSet<String>>>` — actor→groups (reverse for cleanup)
|
||||
- Groups auto-create on first join, auto-delete when empty
|
||||
- Runtime API: `join_group(addr, name)`, `leave_group(addr, name)`, `publish_to(group, msg)`,
|
||||
`group_members(group)`, `groups()`
|
||||
- Ctx API: `join_group(name)`, `leave_group(name)`, `publish(group, msg)`, `group_members(group)`
|
||||
- `publish` clones at the typed level (Message: Clone), sends to each member via normal routing
|
||||
- Auto-cleanup: `group_registry.cleanup(&addr)` in cleanup_dead phase removes dead actor from all groups
|
||||
- ContextInner extended: `join_group()`, `leave_group()`, `group_members()` (publish is Ctx-level only)
|
||||
- **Tests**: 9 new behavioral tests
|
||||
- `group_members_returns_joined_actors` — join + query
|
||||
- `empty_group_returns_no_members` — nonexistent group → empty
|
||||
- `publish_broadcasts_to_all_members` — 2 members, both receive
|
||||
- `leave_group_stops_receiving_publishes` — leave → excluded from broadcast
|
||||
- `dead_actor_auto_removed_from_group` — stop → removed from group
|
||||
- `actor_removed_from_all_groups_on_death` — multi-group membership cleanup
|
||||
- `empty_group_auto_deleted` — last member leaves → group removed from groups()
|
||||
- `ctx_join_group_from_handler` — join via on_start
|
||||
- `ctx_publish_broadcasts_from_handler` — publish via handler
|
||||
- **Result**: 122 tests pass (115 behavioral + 7 proptest), all workspace compiles, zero warnings
|
||||
|
||||
### Cycle 13: Actor Monitoring / Death Watch
|
||||
- **Research**: Studied monitoring across Erlang (monitor/2, DOWN messages), Akka (watch/Terminated),
|
||||
Ractor (link, SupervisionEvent), Actix (none), Kameo (link, on_link_died callback)
|
||||
- Key finding: Erlang's unidirectional monitor + message delivery is the best fit for swactor
|
||||
(reuses existing type-erased handler, zero trait changes, composable)
|
||||
- Callbacks (Ractor/Kameo style) rejected: would require adding to AnyActor/ActorInterface traits
|
||||
- Bidirectional links deferred: can layer on top of monitors later
|
||||
- **Implementation**: `MonitorRegistry` in delivery.rs + `Down`/`StopReason`/`MonitorRef` in actor.rs
|
||||
- `MonitorRegistry`: `RwLock<HashMap<ActorAddress, Vec<(MonitorRef, ActorAddress)>>>` (watched→watchers)
|
||||
+ reverse `RwLock<HashMap<MonitorRef, ActorAddress>>` for O(1) demonitor
|
||||
- `MonitorRef(u64)`: unique token from `AtomicU64` counter
|
||||
- `Down { addr: ActorAddress, reason: StopReason }`: delivered as normal mailbox message
|
||||
- `StopReason`: `Normal` (graceful stop) | `Panicked` (panic, not restartable)
|
||||
- `ctx.monitor(target)` → `MonitorRef` — subscribe to death notifications
|
||||
- `ctx.demonitor(mref)` — cancel a subscription
|
||||
- `cleanup_dead` now returns `Vec<(ActorAddress, StopReason)>` instead of `Vec<ActorAddress>`
|
||||
- After cleanup_dead: iterate dead actors, take_monitors from registry, route Down through normal
|
||||
delivery (pool.deliver for same-worker, transfer_txs for cross-worker, inbox_registry for inboxes)
|
||||
- Dead watcher cleanup: `remove_watcher()` strips monitor subscriptions for dead watchers
|
||||
- Multiple monitors of same target produce independent notifications (stacking, like Erlang)
|
||||
- **Tests**: 7 new behavioral tests
|
||||
- `monitor_notifies_on_graceful_stop` — Down{reason: Normal} on stop
|
||||
- `monitor_notifies_on_panic` — Down{reason: Panicked} on panic
|
||||
- `multiple_watchers_all_notified` — two watchers both get Down
|
||||
- `demonitor_cancels_notification` — demonitor → no Down delivered
|
||||
- `dead_watcher_does_not_receive_down` — dead watcher's monitors cleaned up
|
||||
- `down_delivered_to_external_inbox` — Down forwarded through inbox
|
||||
- `stacked_monitors_produce_multiple_notifications` — two monitors on same target → two Downs
|
||||
- **Result**: 113 tests pass (106 behavioral + 7 proptest), all workspace compiles, zero warnings
|
||||
|
||||
### Cycle 12: Named Actor Registry
|
||||
- **Research**: Studied named actor/service discovery across Erlang (register/2, whereis/1, global, pg),
|
||||
Actix (Registry, SystemRegistry — TypeId keys), Bastion (hierarchy-based), Ractor (String keys, DashMap,
|
||||
global static), xactor (TypeId singleton), Akka (Receptionist, ServiceKey[T])
|
||||
- Key findings: TypeId keys (Actix/xactor) don't fit swactor's type-erased model; global static
|
||||
(Ractor) breaks multi-runtime scenarios; Erlang's register/whereis is the gold standard
|
||||
- Decision: String keys, RwLock<HashMap> (matches existing AddressMap/InboxRegistry pattern),
|
||||
per-runtime scope, error on collision, auto-unregister on death
|
||||
- **Implementation**: `NameRegistry` in delivery.rs with forward + reverse maps
|
||||
- `NameRegistry`: `RwLock<HashMap<String, ActorAddress>>` + `RwLock<HashMap<ActorAddress, String>>`
|
||||
- Forward map for O(1) name→addr lookup, reverse map for O(1) addr→name cleanup
|
||||
- Added to `Runtime` as `Arc<NameRegistry>`, threaded through `TickContext`
|
||||
- Runtime API: `spawn_named(name, actor)`, `where_is(name)`, `unregister(name)`, `registered_names()`
|
||||
- Ctx API: `spawn_named(name, actor)`, `where_is(name)` — usable from inside handlers
|
||||
- `ContextInner` trait extended: `where_is()` + `register_name()` (private, supports both Runtime and WorkerContext)
|
||||
- Auto-unregister on death: `cleanup_dead` phase calls `name_registry.unregister_by_addr()` for each dead actor
|
||||
- Name reservation is immediate (before spawn queue push) — prevents TOCTOU race
|
||||
- Collision returns `Err("Name already registered")` — original binding preserved
|
||||
- **Tests**: 11 new behavioral tests
|
||||
- `named_actor_lookup_returns_spawn_address` — spawn_named → where_is roundtrip
|
||||
- `named_actor_receives_messages_via_lookup` — send to looked-up address works
|
||||
- `duplicate_name_returns_error` — collision error, original preserved
|
||||
- `where_is_returns_none_for_unknown_name` — nonexistent name → None
|
||||
- `name_auto_unregistered_on_actor_death` — stop_actor → name freed
|
||||
- `name_can_be_reused_after_actor_death` — death → respawn with same name
|
||||
- `name_auto_unregistered_on_panic` — panic → name freed
|
||||
- `registered_names_lists_all` — all registered names returned
|
||||
- `manual_unregister_frees_name_but_actor_lives` — unregister doesn't kill actor
|
||||
- `ctx_where_is_resolves_inside_handler` — where_is from handler context
|
||||
- `ctx_spawn_named_registers_from_handler` — spawn_named from handler context
|
||||
- **Result**: 106 tests pass (99 behavioral + 7 proptest), all workspace compiles, zero warnings
|
||||
|
||||
### Cycle 11: Property-Based Testing (proptest + fuzz extension)
|
||||
- **Research**: Studied testing approaches across tokio (loom), Erlang (PropEr, QuickCheck, Concuerror),
|
||||
Rust property-based testing (proptest vs quickcheck), cargo-fuzz, and actor-specific testing patterns.
|
||||
- Ranked approaches: #1 proptest-state-machine (perfect fit for deterministic ticks),
|
||||
#2 extend cargo-fuzz, #3 simple proptest, #4 shuttle, #5 loom, #6 DST
|
||||
- Also researched remaining feature gaps: named actors, monitoring/death watch, groups, ask pattern
|
||||
- **Implementation**: Property-based testing suite with proptest-state-machine
|
||||
- Added `proptest` and `proptest-state-machine` to dev-dependencies
|
||||
- New test file: `tests/proptest_runtime.rs` with 7 tests:
|
||||
- `fifo_ordering_for_any_message_sequence` — FIFO preserved for 1-100 random messages
|
||||
- `budget_limits_per_actor_processing` — budget caps per-tick processing for 2-10 actors
|
||||
- `one_shot_timer_fires_at_correct_tick` — timer with delay 1-20 fires at exact right tick
|
||||
- `interval_timer_fires_at_correct_period` — period 1-10, verifies 3 consecutive fires
|
||||
- `bounded_mailbox_never_exceeds_capacity` — capacity 1-20, 1-200 messages, never exceeds
|
||||
- `spawn_n_actors_all_tracked` — 1-50 actors, all unique, all in stats
|
||||
- `swactor_state_machine` — stateful property test: random Spawn/Send/Tick/Stop/CheckStats
|
||||
sequences (up to 40 transitions, 128 cases), verifies runtime invariants after each step
|
||||
- State machine test defines SwactorModel (reference) vs SwactorTest (SUT) with:
|
||||
- Reference model: HashMap<id, alive> tracking expected actor lifecycle
|
||||
- Invariants checked after every transition: worker count, actor placement, mailbox safety
|
||||
- Automatic shrinking finds minimal failing sequences
|
||||
- Extended fuzz targets (fuzz_runtime.rs) with 4 new RawAction variants:
|
||||
- `StopActor` — graceful stop via runtime.stop_actor
|
||||
- `SpawnRestartable` — spawn_restartable with configurable max_restarts
|
||||
- `ScheduleTimer` — one-shot timer via TimerSchedulerActor
|
||||
- `ScheduleInterval` — interval timer via IntervalSchedulerActor
|
||||
- Added 3 new actor types to fuzz: TimerSchedulerActor, IntervalSchedulerActor, RestartableEchoActor
|
||||
- **Bug found**: State machine test immediately caught invariant mismatch: address map tracks spawned
|
||||
actors immediately, but per-worker num_actors lags until first tick. Fixed invariant to use <= check.
|
||||
- **Result**: 95 tests pass (88 behavioral + 7 proptest), fuzz targets compile, zero warnings
|
||||
|
||||
### 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).
|
||||
- **Implementation**: Automatic cleanup of poisoned actors after tick_all
|
||||
- `AddressMap::remove()` added to delivery.rs
|
||||
- `ActorPool::cleanup_dead()` collects and removes poisoned actors, returns their addresses
|
||||
- Phase 7 in tick_once: cleanup_dead → remove from address_map → update num_actors stat
|
||||
- Re-publish num_actors after cleanup so stats immediately reflect removal
|
||||
- **Behavior change**: Sends to poisoned actors now return Err (address not found) instead of
|
||||
silently discarding. This is better — callers learn the actor is gone.
|
||||
- **Tests**: 2 new tests + 2 existing tests updated
|
||||
- `dead_actor_cleaned_up_from_stats_and_address_map` — good actor persists, bad actor removed
|
||||
- `bulk_dead_actor_cleanup` — 20 panicked actors all cleaned up
|
||||
- Updated `send_to_poisoned_actor_is_a_silent_black_hole` — now asserts send returns Err
|
||||
- Updated `poisoned_actor_messages_not_counted_as_processed` — sends fail to cleaned-up actor
|
||||
- **Result**: 70 tests pass, all workspace compiles
|
||||
|
||||
### Cycle 7: Actor Recovery (Factory Restart)
|
||||
- **Research**: Deep analysis of supervision/recovery across Erlang (supervision trees, restart intensity),
|
||||
Akka (Resume/Restart/Stop/Escalate), Kameo (on_panic hook), Actix (Supervised trait), Ractor (SupervisionEvent)
|
||||
- Erlang: fresh process via factory (MFA tuple), mailbox lost, PID changes
|
||||
- Akka: replace internals but keep ActorRef stable, mailbox preserved (docs say this is usually wrong)
|
||||
- Kameo: on_panic(&mut self) — risky with corrupt state after panic
|
||||
- Decision: factory-based restart (Erlang-style), safest approach
|
||||
- **Implementation**: `spawn_restartable(actor, factory, max_restarts)` on Runtime and Ctx
|
||||
- `Actor<A>` expanded from tuple struct to named fields: inner, restart_factory, max_restarts, restart_count
|
||||
- `AnyActor::try_restart(&self)` trait method (default None, backward compatible)
|
||||
- Factory stored as `Arc<dyn Fn() -> A + Send + Sync>` — cloned into fresh Actor on restart
|
||||
- `tick_all` panic handler: try_restart before poisoning, clear mailbox, fresh state
|
||||
- `restarts` counter added to `WorkerStats` and `WorkerInfo`
|
||||
- **Safety**: Factory fields are "cold" (never touched by handle_any), safe to read after catch_unwind
|
||||
- **Tests**: 4 new tests
|
||||
- `restartable_actor_recovers_after_panic` — basic restart works
|
||||
- `restartable_actor_resets_state_on_restart` — fresh state post-restart
|
||||
- `restartable_actor_respects_max_restarts` — 2 restarts then permanent poison
|
||||
- `non_restartable_actor_still_poisons_on_panic` — backward compatibility
|
||||
- **Result**: 68 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`
|
||||
- Constraints in `CLAUDE/notes/constraints.md`
|
||||
### Files Modified
|
||||
- `src/actor.rs` — Custom Hash impl for ActorAddress (8-byte)
|
||||
- `src/delivery.rs` — AddrHasher, AddrBuildHasher, AddrMap, AddrSet types; 5 HashMap replacements
|
||||
- `src/worker.rs` — ActorPool.actors AddrMap; stop_requests optimization
|
||||
- `src/transport.rs` — TransportRouter.routes AddrMap
|
||||
- `Cargo.toml` — Added hasher_benchmarks bench entry
|
||||
- `benches/hasher_benchmarks.rs` — New: component-level microbenchmarks
|
||||
- `tests/runtime_api.rs` — 3 new behavioral tests
|
||||
- `CLAUDE/notes/dispatch_comparison.md` — New: Stakker/Actix/Swactor analysis
|
||||
|
||||
## Next Steps
|
||||
- [x] **Cycle 2: Stress testing + property-based tests** ✅
|
||||
- [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** ✅
|
||||
- [x] **Cycle 6: Mailbox backpressure** ✅
|
||||
- [x] **Cycle 7: Actor recovery (factory restart)** ✅
|
||||
- [x] **Cycle 8: Dead actor cleanup** ✅
|
||||
- [x] **Cycle 9: Lifecycle hooks + graceful stop** ✅
|
||||
- [x] **Cycle 10: Actor timers (tick-counting)** ✅
|
||||
- [x] **Cycle 11: Property-based testing (proptest-state-machine + fuzz extension)** ✅
|
||||
- [ ] **Cycle 12: Next improvement**
|
||||
- Candidates: named actors/registry (small effort, high value), actor monitoring/death watch,
|
||||
actor groups/pub-sub, SmallBox optimization
|
||||
- Priority messages REJECTED (lifecycle hooks cover 95% of cases)
|
||||
- LIFO slot rejected (0-5% benefit for typical workloads, not worth complexity)
|
||||
- Profile the `Box::new(msg)` allocation cost — SmallBox/inline storage could eliminate heap alloc for small messages
|
||||
- Investigate VecDeque mailbox alternative (slab-allocated ring buffer)
|
||||
- Consider `enum_dispatch` pattern for avoiding `dyn Any` downcast (would require API changes)
|
||||
- Benchmark on dedicated hardware (sandbox too noisy for reliable end-to-end measurement)
|
||||
|
||||
## Open Questions
|
||||
- Should budget be configurable per-actor (not just per-runtime)?
|
||||
- Is 64 the right default budget? Benchmarks show budget=32 slightly faster for throughput
|
||||
- ~~Thread parking: notification mechanism~~ RESOLVED: OnceLock<Thread> + unpark()
|
||||
- Should load-aware placement weight mailbox depth more than actor count?
|
||||
- LIFO slot for same-worker sends: worth the complexity?
|
||||
|
||||
## Blockers
|
||||
- (none)
|
||||
- Is 8 bytes sufficient for the identity hash? (Yes — 2^64 from crypto-random bytes)
|
||||
- Should we provide a `with_hasher` public API for users? (No — internal optimization only)
|
||||
|
|
|
|||
|
|
@ -43,6 +43,10 @@ harness = false
|
|||
name = "mt_benchmarks"
|
||||
harness = false
|
||||
|
||||
[[bench]]
|
||||
name = "hasher_benchmarks"
|
||||
harness = false
|
||||
|
||||
[[example]]
|
||||
name = "tcp_ping_pong"
|
||||
required-features = ["transport"]
|
||||
|
|
|
|||
164
benches/hasher_benchmarks.rs
Normal file
164
benches/hasher_benchmarks.rs
Normal file
|
|
@ -0,0 +1,164 @@
|
|||
use criterion::{criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion};
|
||||
use std::collections::HashMap;
|
||||
use std::hash::{BuildHasher, Hash, Hasher};
|
||||
use swactor::actor::ActorAddress;
|
||||
|
||||
// ─── Reproduce the identity hasher for benchmarking ─────────────────────────
|
||||
// (The real one is pub(crate) in delivery.rs — recreate here for bench access)
|
||||
|
||||
struct AddrHasher(u64);
|
||||
|
||||
impl Hasher for AddrHasher {
|
||||
#[inline]
|
||||
fn finish(&self) -> u64 {
|
||||
self.0
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn write(&mut self, _bytes: &[u8]) {}
|
||||
|
||||
#[inline]
|
||||
fn write_u64(&mut self, i: u64) {
|
||||
self.0 = i;
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
struct AddrBuildHasher;
|
||||
|
||||
impl BuildHasher for AddrBuildHasher {
|
||||
type Hasher = AddrHasher;
|
||||
#[inline]
|
||||
fn build_hasher(&self) -> AddrHasher {
|
||||
AddrHasher(0)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Helpers ────────────────────────────────────────────────────────────────
|
||||
|
||||
fn random_addresses(n: usize) -> Vec<ActorAddress> {
|
||||
(0..n).map(|_| ActorAddress::new_random()).collect()
|
||||
}
|
||||
|
||||
// ─── Hash benchmarks ────────────────────────────────────────────────────────
|
||||
|
||||
fn bench_hash(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("hash");
|
||||
|
||||
let addr = ActorAddress::new_random();
|
||||
|
||||
// Default hasher (SipHash) — hashes all 32 bytes via derived Hash,
|
||||
// but our custom Hash impl only writes 8 bytes
|
||||
group.bench_function("siphash_custom_hash", |b| {
|
||||
b.iter(|| {
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
addr.hash(&mut hasher);
|
||||
std::hint::black_box(hasher.finish())
|
||||
})
|
||||
});
|
||||
|
||||
// Identity hasher — reads the u64 from our custom Hash impl directly
|
||||
group.bench_function("identity", |b| {
|
||||
b.iter(|| {
|
||||
let mut hasher = AddrHasher(0);
|
||||
addr.hash(&mut hasher);
|
||||
std::hint::black_box(hasher.finish())
|
||||
})
|
||||
});
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
// ─── Lookup benchmarks ─────────────────────────────────────────────────────
|
||||
|
||||
fn bench_lookup(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("lookup");
|
||||
|
||||
for &size in &[100, 1000] {
|
||||
let addrs = random_addresses(size);
|
||||
let lookup_targets: Vec<ActorAddress> = addrs.iter().cloned().collect();
|
||||
|
||||
// SipHash HashMap (but with our custom 8-byte Hash impl)
|
||||
let sip_map: HashMap<ActorAddress, usize> =
|
||||
addrs.iter().enumerate().map(|(i, a)| (*a, i)).collect();
|
||||
|
||||
group.bench_with_input(
|
||||
BenchmarkId::new("siphash", size),
|
||||
&size,
|
||||
|b, _| {
|
||||
let mut idx = 0;
|
||||
b.iter(|| {
|
||||
let addr = &lookup_targets[idx % lookup_targets.len()];
|
||||
idx += 1;
|
||||
std::hint::black_box(sip_map.get(addr))
|
||||
})
|
||||
},
|
||||
);
|
||||
|
||||
// Identity HashMap
|
||||
let identity_map: HashMap<ActorAddress, usize, AddrBuildHasher> = {
|
||||
let mut m = HashMap::with_capacity_and_hasher(size, AddrBuildHasher);
|
||||
for (i, a) in addrs.iter().enumerate() {
|
||||
m.insert(*a, i);
|
||||
}
|
||||
m
|
||||
};
|
||||
|
||||
group.bench_with_input(
|
||||
BenchmarkId::new("identity", size),
|
||||
&size,
|
||||
|b, _| {
|
||||
let mut idx = 0;
|
||||
b.iter(|| {
|
||||
let addr = &lookup_targets[idx % lookup_targets.len()];
|
||||
idx += 1;
|
||||
std::hint::black_box(identity_map.get(addr))
|
||||
})
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
// ─── Insert benchmarks ─────────────────────────────────────────────────────
|
||||
|
||||
fn bench_insert(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("insert");
|
||||
|
||||
let addrs = random_addresses(1000);
|
||||
|
||||
group.bench_function("siphash", |b| {
|
||||
b.iter_batched(
|
||||
|| addrs.clone(),
|
||||
|addrs| {
|
||||
let mut m: HashMap<ActorAddress, usize> = HashMap::with_capacity(addrs.len());
|
||||
for (i, a) in addrs.iter().enumerate() {
|
||||
m.insert(*a, i);
|
||||
}
|
||||
std::hint::black_box(m.len())
|
||||
},
|
||||
BatchSize::SmallInput,
|
||||
)
|
||||
});
|
||||
|
||||
group.bench_function("identity", |b| {
|
||||
b.iter_batched(
|
||||
|| addrs.clone(),
|
||||
|addrs| {
|
||||
let mut m: HashMap<ActorAddress, usize, AddrBuildHasher> =
|
||||
HashMap::with_capacity_and_hasher(addrs.len(), AddrBuildHasher);
|
||||
for (i, a) in addrs.iter().enumerate() {
|
||||
m.insert(*a, i);
|
||||
}
|
||||
std::hint::black_box(m.len())
|
||||
},
|
||||
BatchSize::SmallInput,
|
||||
)
|
||||
});
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
criterion_group!(benches, bench_hash, bench_lookup, bench_insert);
|
||||
criterion_main!(benches);
|
||||
15
src/actor.rs
15
src/actor.rs
|
|
@ -39,10 +39,23 @@ pub trait ActorInterface: 'static + Send {
|
|||
/// A unique address for this actor. 32 bytes is overkill for a small application,
|
||||
/// but most systems are powerful, and this allows us to create a global map of
|
||||
/// actor processes in the future, without worrying about collision.
|
||||
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
|
||||
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
|
||||
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
|
||||
pub struct ActorAddress(pub [u8; 32]);
|
||||
|
||||
/// Custom Hash: only hash the first 8 bytes since all 32 are random.
|
||||
/// SipHash on 8 bytes is ~3x faster than on 32 bytes, with identical
|
||||
/// collision properties (2^64 possible values from cryptographic randomness).
|
||||
impl std::hash::Hash for ActorAddress {
|
||||
#[inline]
|
||||
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
|
||||
// SAFETY: ActorAddress is always 32 bytes, so [..8] is valid.
|
||||
state.write_u64(u64::from_ne_bytes(
|
||||
self.0[..8].try_into().unwrap(),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for ActorAddress {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
for b in &self.0[..8] {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
use std::any::Any;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::hash::{BuildHasher, Hasher};
|
||||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, OnceLock, RwLock};
|
||||
use std::thread::Thread;
|
||||
|
|
@ -10,6 +11,55 @@ use crate::config::RuntimeConfig;
|
|||
use crate::stats::WorkerStats;
|
||||
use crate::Error;
|
||||
|
||||
// ─── Identity Hasher for ActorAddress ───────────────────────────────────────
|
||||
|
||||
/// Identity hasher for ActorAddress keys.
|
||||
///
|
||||
/// ActorAddress contains 32 cryptographically random bytes. The custom `Hash`
|
||||
/// impl on ActorAddress writes only the first 8 bytes as a `u64`. This hasher
|
||||
/// passes that u64 through as the hash value directly — no mixing, no SipHash.
|
||||
///
|
||||
/// This is safe because the input is already random (uniform distribution),
|
||||
/// so additional mixing would be redundant.
|
||||
pub(crate) struct AddrHasher(u64);
|
||||
|
||||
impl Hasher for AddrHasher {
|
||||
#[inline]
|
||||
fn finish(&self) -> u64 {
|
||||
self.0
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn write(&mut self, _bytes: &[u8]) {
|
||||
// Unused — ActorAddress::hash calls write_u64 directly.
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn write_u64(&mut self, i: u64) {
|
||||
self.0 = i;
|
||||
}
|
||||
}
|
||||
|
||||
/// BuildHasher for creating AddrHasher instances.
|
||||
#[derive(Default, Clone)]
|
||||
pub(crate) struct AddrBuildHasher;
|
||||
|
||||
impl BuildHasher for AddrBuildHasher {
|
||||
type Hasher = AddrHasher;
|
||||
|
||||
#[inline]
|
||||
fn build_hasher(&self) -> AddrHasher {
|
||||
AddrHasher(0)
|
||||
}
|
||||
}
|
||||
|
||||
/// HashMap optimized for ActorAddress keys.
|
||||
/// Uses identity hashing since ActorAddress bytes are already random.
|
||||
pub(crate) type AddrMap<V> = HashMap<ActorAddress, V, AddrBuildHasher>;
|
||||
|
||||
/// HashSet optimized for ActorAddress keys.
|
||||
pub(crate) type AddrSet = HashSet<ActorAddress, AddrBuildHasher>;
|
||||
|
||||
// ─── Address Map Types ───────────────────────────────────────────────────────
|
||||
|
||||
/// Identifies a worker thread.
|
||||
|
|
@ -26,13 +76,13 @@ impl WorkerId {
|
|||
///
|
||||
/// `RwLock<HashMap>` — zero contention for parallel reads, write-rare (only on spawn).
|
||||
pub(crate) struct AddressMap {
|
||||
inner: RwLock<HashMap<ActorAddress, WorkerId>>,
|
||||
inner: RwLock<AddrMap<WorkerId>>,
|
||||
}
|
||||
|
||||
impl AddressMap {
|
||||
pub fn with_capacity(cap: usize) -> Self {
|
||||
Self {
|
||||
inner: RwLock::new(HashMap::with_capacity(cap)),
|
||||
inner: RwLock::new(HashMap::with_capacity_and_hasher(cap, AddrBuildHasher)),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -146,13 +196,13 @@ impl<M: Message> SenderT for Sender<M> {
|
|||
|
||||
/// Registry of external inboxes — replaces the Router's role for non-actor receivers.
|
||||
pub(crate) struct InboxRegistry {
|
||||
senders: RwLock<HashMap<ActorAddress, Arc<dyn SenderT>>>,
|
||||
senders: RwLock<AddrMap<Arc<dyn SenderT>>>,
|
||||
}
|
||||
|
||||
impl InboxRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
senders: RwLock::new(HashMap::new()),
|
||||
senders: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -209,14 +259,14 @@ pub(crate) struct TickContext<'a> {
|
|||
/// read-often (lookup). A reverse map enables O(1) cleanup on actor death.
|
||||
pub(crate) struct NameRegistry {
|
||||
names: RwLock<HashMap<String, ActorAddress>>,
|
||||
reverse: RwLock<HashMap<ActorAddress, String>>,
|
||||
reverse: RwLock<AddrMap<String>>,
|
||||
}
|
||||
|
||||
impl NameRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
names: RwLock::new(HashMap::new()),
|
||||
reverse: RwLock::new(HashMap::new()),
|
||||
reverse: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -264,7 +314,7 @@ impl NameRegistry {
|
|||
/// Write-rare (monitor/demonitor/death), read at cleanup time.
|
||||
pub(crate) struct MonitorRegistry {
|
||||
/// watched_addr → [(mref, watcher_addr)]
|
||||
monitors: RwLock<HashMap<ActorAddress, Vec<(MonitorRef, ActorAddress)>>>,
|
||||
monitors: RwLock<AddrMap<Vec<(MonitorRef, ActorAddress)>>>,
|
||||
/// mref → watched_addr (for O(1) demonitor)
|
||||
ref_to_target: RwLock<HashMap<MonitorRef, ActorAddress>>,
|
||||
next_ref: AtomicU64,
|
||||
|
|
@ -273,7 +323,7 @@ pub(crate) struct MonitorRegistry {
|
|||
impl MonitorRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
monitors: RwLock::new(HashMap::new()),
|
||||
monitors: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
|
||||
ref_to_target: RwLock::new(HashMap::new()),
|
||||
next_ref: AtomicU64::new(1),
|
||||
}
|
||||
|
|
@ -341,16 +391,16 @@ impl MonitorRegistry {
|
|||
/// Groups are created lazily on first join and removed when empty.
|
||||
pub(crate) struct GroupRegistry {
|
||||
/// group_name → set of member addresses
|
||||
groups: RwLock<HashMap<String, HashSet<ActorAddress>>>,
|
||||
groups: RwLock<HashMap<String, AddrSet>>,
|
||||
/// actor_addr → set of group names (reverse map for O(G) cleanup on death)
|
||||
memberships: RwLock<HashMap<ActorAddress, HashSet<String>>>,
|
||||
memberships: RwLock<AddrMap<HashSet<String>>>,
|
||||
}
|
||||
|
||||
impl GroupRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
groups: RwLock::new(HashMap::new()),
|
||||
memberships: RwLock::new(HashMap::new()),
|
||||
memberships: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -358,7 +408,7 @@ impl GroupRegistry {
|
|||
pub fn join(&self, group: String, addr: ActorAddress) {
|
||||
self.groups.write().unwrap()
|
||||
.entry(group.clone())
|
||||
.or_default()
|
||||
.or_insert_with(|| HashSet::with_hasher(AddrBuildHasher))
|
||||
.insert(addr);
|
||||
self.memberships.write().unwrap()
|
||||
.entry(addr)
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ use std::collections::HashMap;
|
|||
use std::sync::{Arc, RwLock};
|
||||
|
||||
use crate::actor::{ActorAddress, Message};
|
||||
use crate::delivery::{AddrBuildHasher, AddrMap};
|
||||
use crate::Error;
|
||||
|
||||
// ─── Codec ──────────────────────────────────────────────────────────────────
|
||||
|
|
@ -150,13 +151,13 @@ impl CodecRegistry {
|
|||
|
||||
/// Maps remote actor addresses to their [`Transport`].
|
||||
pub struct TransportRouter {
|
||||
routes: RwLock<HashMap<ActorAddress, Arc<dyn Transport>>>,
|
||||
routes: RwLock<AddrMap<Arc<dyn Transport>>>,
|
||||
}
|
||||
|
||||
impl TransportRouter {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
routes: RwLock::new(HashMap::new()),
|
||||
routes: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ use std::time::Instant;
|
|||
use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopReason, StopSignal, TimerRequest};
|
||||
use crate::channel::Receiver;
|
||||
use crate::config::MailboxOverflow;
|
||||
use crate::delivery::{Envelope, TickContext, WorkerId};
|
||||
use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId};
|
||||
use crate::stats::{ActorSnapshot, TickTiming, WorkerStats};
|
||||
use crate::Error;
|
||||
|
||||
|
|
@ -493,7 +493,7 @@ struct ActorSlot {
|
|||
|
||||
/// Per-worker actor storage. Owns per-actor mailboxes.
|
||||
pub(crate) struct ActorPool {
|
||||
actors: HashMap<ActorAddress, ActorSlot>,
|
||||
actors: AddrMap<ActorSlot>,
|
||||
default_mailbox_capacity: usize,
|
||||
default_overflow_policy: MailboxOverflow,
|
||||
/// Messages dropped this tick due to mailbox overflow. Reset after publishing to stats.
|
||||
|
|
@ -503,7 +503,7 @@ pub(crate) struct ActorPool {
|
|||
impl ActorPool {
|
||||
pub fn new(default_mailbox_capacity: usize, default_overflow_policy: MailboxOverflow) -> Self {
|
||||
Self {
|
||||
actors: HashMap::new(),
|
||||
actors: HashMap::with_hasher(AddrBuildHasher),
|
||||
default_mailbox_capacity,
|
||||
default_overflow_policy,
|
||||
drops_this_tick: 0,
|
||||
|
|
@ -591,11 +591,15 @@ impl ActorPool {
|
|||
continue;
|
||||
}
|
||||
// Check if on_start requested stop
|
||||
if stop_requests.borrow().contains(&addr) {
|
||||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
continue;
|
||||
{
|
||||
let stops = stop_requests.borrow();
|
||||
if !stops.is_empty() && stops.contains(&addr) {
|
||||
drop(stops);
|
||||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -646,11 +650,15 @@ impl ActorPool {
|
|||
actor_count += 1;
|
||||
|
||||
// Check if handler requested self-stop (via ctx.stop_self())
|
||||
if stop_requests.borrow().contains(&addr) {
|
||||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
break;
|
||||
{
|
||||
let stops = stop_requests.borrow();
|
||||
if !stops.is_empty() && stops.contains(&addr) {
|
||||
drop(stops);
|
||||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if budget > 0 && actor_count >= budget {
|
||||
|
|
|
|||
|
|
@ -4529,3 +4529,223 @@ fn router_broadcast_multiple_messages_all_received() {
|
|||
|
||||
assert_eq!(total.load(Ordering::Relaxed), 15);
|
||||
}
|
||||
|
||||
// ── Identity Hasher Correctness ────────────────────────────────────────────
|
||||
|
||||
/// Given: 200 actors each expecting a unique numbered message
|
||||
/// When: Each actor receives its number and replies with (self_addr, number)
|
||||
/// Then: All 200 replies match — no message was misrouted by the identity hasher
|
||||
#[test]
|
||||
fn many_actors_all_receive_correct_messages() {
|
||||
#[derive(Clone)]
|
||||
struct NumberedMsg {
|
||||
n: usize,
|
||||
reply_to: ActorAddress,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct NumberedReply {
|
||||
from: ActorAddress,
|
||||
n: usize,
|
||||
}
|
||||
|
||||
struct NumberedActor;
|
||||
|
||||
impl ActorInterface for NumberedActor {
|
||||
type Incoming = NumberedMsg;
|
||||
type Response = ();
|
||||
fn handle(&mut self, ctx: &Ctx, msg: NumberedMsg) {
|
||||
let _ = ctx.send(
|
||||
msg.reply_to,
|
||||
NumberedReply {
|
||||
from: ctx.self_addr(),
|
||||
n: msg.n,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let rt = Runtime::new(RuntimeConfig {
|
||||
max_actors: 300,
|
||||
channel_buffer_size: 1024,
|
||||
num_threads: 1,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let inbox = rt.new_inbox::<NumberedReply>().unwrap();
|
||||
let inbox_addr = *inbox.addr();
|
||||
|
||||
// Spawn 200 actors
|
||||
let mut addrs = Vec::new();
|
||||
for _ in 0..200 {
|
||||
addrs.push(rt.spawn(NumberedActor).unwrap());
|
||||
}
|
||||
rt.tick(); // on_start
|
||||
|
||||
// Send unique numbered message to each
|
||||
for (i, addr) in addrs.iter().enumerate() {
|
||||
rt.send_to(
|
||||
*addr,
|
||||
NumberedMsg {
|
||||
n: i,
|
||||
reply_to: inbox_addr,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
rt.tick(); // process + reply
|
||||
rt.tick(); // deliver replies
|
||||
|
||||
// Verify all 200 replies
|
||||
let mut replies: Vec<NumberedReply> = Vec::new();
|
||||
while let Some(reply) = inbox.try_recv() {
|
||||
replies.push(reply);
|
||||
}
|
||||
|
||||
assert_eq!(replies.len(), 200, "should receive exactly 200 replies");
|
||||
|
||||
// Verify each reply came from the correct actor with the correct number
|
||||
for (i, addr) in addrs.iter().enumerate() {
|
||||
let reply = replies.iter().find(|r| r.n == i);
|
||||
assert!(
|
||||
reply.is_some(),
|
||||
"missing reply for actor #{i}"
|
||||
);
|
||||
assert_eq!(
|
||||
reply.unwrap().from, *addr,
|
||||
"reply #{i} came from wrong actor"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Given: A 100-actor ring where each actor forwards to the next
|
||||
/// When: A message enters the ring and traverses all 100 hops
|
||||
/// Then: The message completes the full circuit (address_map lookups all correct)
|
||||
#[test]
|
||||
fn ring_routing_unchanged_after_hasher_optimization() {
|
||||
#[derive(Clone)]
|
||||
struct RingHop {
|
||||
hops_remaining: usize,
|
||||
final_dest: ActorAddress,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct RingDone(usize); // total hops completed
|
||||
|
||||
struct RingNode {
|
||||
next: ActorAddress,
|
||||
}
|
||||
|
||||
impl ActorInterface for RingNode {
|
||||
type Incoming = RingHop;
|
||||
type Response = ();
|
||||
fn handle(&mut self, ctx: &Ctx, msg: RingHop) {
|
||||
if msg.hops_remaining == 0 {
|
||||
let _ = ctx.send(msg.final_dest, RingDone(100));
|
||||
} else {
|
||||
let _ = ctx.send(
|
||||
self.next,
|
||||
RingHop {
|
||||
hops_remaining: msg.hops_remaining - 1,
|
||||
final_dest: msg.final_dest,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let rt = Runtime::new(RuntimeConfig {
|
||||
max_actors: 200,
|
||||
channel_buffer_size: 1024,
|
||||
num_threads: 1,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let inbox = rt.new_inbox::<RingDone>().unwrap();
|
||||
let inbox_addr = *inbox.addr();
|
||||
|
||||
// Build chain backwards: last node sends to inbox, first node receives
|
||||
let mut addrs = Vec::new();
|
||||
let mut next = inbox_addr;
|
||||
for _ in (0..100).rev() {
|
||||
let node = RingNode { next };
|
||||
let addr = rt.spawn(node).unwrap();
|
||||
addrs.push(addr);
|
||||
next = addr;
|
||||
}
|
||||
addrs.reverse(); // addrs[0] is start of chain
|
||||
|
||||
rt.tick(); // on_start
|
||||
|
||||
// Inject message at the start
|
||||
rt.send_to(
|
||||
addrs[0],
|
||||
RingHop {
|
||||
hops_remaining: 99,
|
||||
final_dest: inbox_addr,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// Tick enough times for the message to traverse all 100 actors
|
||||
// (each tick processes one hop via pending_local delivery)
|
||||
for _ in 0..110 {
|
||||
rt.tick();
|
||||
}
|
||||
|
||||
let result = inbox.try_recv();
|
||||
assert!(result.is_some(), "ring message should complete all 100 hops");
|
||||
assert_eq!(result.unwrap(), RingDone(100));
|
||||
}
|
||||
|
||||
/// Given: An actor that calls ctx.stop_self() upon receiving a trigger message
|
||||
/// When: The trigger is sent, then 5 more messages are sent, then ticked
|
||||
/// Then: The actor is removed, only messages before stop are processed
|
||||
#[test]
|
||||
fn stop_self_with_pending_messages_still_works() {
|
||||
let processed = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Msg(bool); // true = trigger stop
|
||||
|
||||
struct StopOnTrigger(Arc<AtomicUsize>);
|
||||
|
||||
impl ActorInterface for StopOnTrigger {
|
||||
type Incoming = Msg;
|
||||
type Response = ();
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Msg) {
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
if msg.0 {
|
||||
ctx.stop_self();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let rt = Runtime::new(RuntimeConfig::default());
|
||||
let p = processed.clone();
|
||||
let addr = rt.spawn(StopOnTrigger(p)).unwrap();
|
||||
rt.tick(); // on_start
|
||||
|
||||
// Send: 2 normal, 1 trigger, 5 more normal
|
||||
rt.send_to(addr, Msg(false)).unwrap();
|
||||
rt.send_to(addr, Msg(false)).unwrap();
|
||||
rt.send_to(addr, Msg(true)).unwrap(); // stop trigger
|
||||
rt.send_to(addr, Msg(false)).unwrap();
|
||||
rt.send_to(addr, Msg(false)).unwrap();
|
||||
rt.send_to(addr, Msg(false)).unwrap();
|
||||
rt.send_to(addr, Msg(false)).unwrap();
|
||||
rt.send_to(addr, Msg(false)).unwrap();
|
||||
|
||||
rt.tick(); // process messages — stops after trigger
|
||||
rt.tick(); // cleanup
|
||||
|
||||
// Only 3 messages should be processed (2 normal + 1 trigger)
|
||||
assert_eq!(
|
||||
processed.load(Ordering::Relaxed),
|
||||
3,
|
||||
"should process exactly the messages up to and including the stop trigger"
|
||||
);
|
||||
|
||||
// Subsequent sends should fail
|
||||
assert!(rt.send_to(addr, Msg(false)).is_err());
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue