Compare commits

...

1 commit

Author SHA1 Message Date
Developer
812826bb0a feat: extract swactor-std crate with RuntimeExtension hook pattern
Debloat the core swactor crate by moving higher-level features to a new
swactor-std crate, leaving core with only the essential actor primitives
(spawn, send, stop, timers, lifecycle hooks).

Phase 1: Move Supervisor + Router to swactor-std
- Supervisor (ChildSpec, RestartPolicy, SupervisorStrategy) and Router
  (RoutingStrategy) extracted to crates/std/
- ~465 lines removed from src/actor.rs

Phase 2: Remove restart fields from Actor<A>
- Actor<A> slimmed to { inner: A } — no restart_factory, max_restarts
- Removed spawn_restartable from Ctx and Runtime
- Simplified panic handler: always poison, never inline restart
- Supervision-based restart via Supervisor in swactor-std

Phase 3: RuntimeExtension trait + registry extraction
- New src/extension.rs: RuntimeExtension trait (on_actor_death,
  cleanup_dead, as_any) — single hybrid hook for lifecycle events
- ContextInner slimmed from 11 methods to 5 (removed 7 registry
  methods, added extension())
- Ctx slimmed: removed all registry methods, added extension() accessor
- Runtime: removed 3 registry fields + 9 methods, added
  with_extension() builder + extension() accessor
- TickContext: replaced 3 registry fields with extension hook
- tick_once phase 7: uses ext.on_actor_death() + ext.cleanup_dead()
- Moved NameRegistry, MonitorRegistry, GroupRegistry from delivery.rs
  to swactor-std
- StdExtension wraps the 3 registries, implements RuntimeExtension
- Extension traits: CtxMonitoring, CtxNaming, CtxGroups on Ctx;
  RuntimeNaming, RuntimeGroups on Runtime
- Down, StopReason, MonitorRef, handle_down remain in core
- AddrMap/AddrSet/AddrBuildHasher made public, re-exported
- MonitorRef::from_raw(u64) added for cross-crate construction

All 140 tests pass. Benchmarks updated.

Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski
2026-02-13 06:59:29 +00:00
28 changed files with 1299 additions and 1802 deletions

View file

@ -1,49 +0,0 @@
Plan:
You are to improve this codebase via:
- investigating similar codebases
- identifying and summarizing their design decisions when compared to swactor:
- runtime engine
- benchmarking
- testing
- overall performance
- etc.
- implementing improvements based on your anaylsis
Workflow:
- Read `CLAUDE/TASK.md` and `CLAUDE/notes/progress.md`
- Identify what stage you are on.
- Read and update yourself as necessary.
- Proceed to accomplishing the next task as written in `progress.md`
- For each attempt at any step, keep a record. If you reach attempt 3, step back, document, and try something else.
- When done, because attempt limit or task success:
- update `progress.md` with:
- Completed this session
- Next steps (specific, actionable)
- Open Questions
- Blockers
- make a commit
- compress your context and start the loop again
Style:
- Do not add to existing modules in the root swactor `src/` they should stay as they are. You may modify but not change module structure.
- Integration tests in `tests/`, benchmark code in `benches/`
- cap execution time at 2 minutes max for fuzz, or benchmarks, or single test suite
- if they take too long, refactor and break up into logical modules
- You may modify these as you wish, so long as logical 'coverage' does not decline.
- Report all your changes to architecture with changes to the `docs/` items
- all notes you wish to keep across iterations shall go in the `CLAUDE/notes/` folder
Example loop (not restrictive, feel free to ignore if prudent):
- Pick a related codebase and a concept to execute (benchmarks, test coverage, engine performance under various scenarios)
- compare to swactor
- make analysis
- implement plan
- execute
- evaluate
- if satisfied, pick a new codebase and/or concept. If not, repeat from step 'compare to swactor'
Before git commit:
- all `cargo test` passes, including feature gated material
- if a test fails, investigate do not ignore or delete
- You can combine tests but not skip code paths or delete them for active code
- if a fix takes > 3 attempts, log and move on

View file

@ -1,32 +0,0 @@
# Baseline Benchmarks (pre-improvements)
## Latency (single-threaded)
| Benchmark | Time |
|-----------|------|
| spawn | 1.28 µs |
| message_roundtrip | 2.24 µs |
| send_fire_and_forget | 1.50 µs |
| inbox_creation | 1.63 µs |
## Throughput (single-threaded)
| Benchmark | Time | Throughput |
|-----------|------|-----------|
| single_actor/100 | 15.0 µs | 6.65 Melem/s |
| single_actor/1000 | 57.5 µs | 17.4 Melem/s |
| single_actor/10000 | 474.6 µs | 21.1 Melem/s |
| multi_actor/10x100 | 80.6 µs | 12.4 Melem/s |
| multi_actor/100x100 | 610.6 µs | 16.4 Melem/s |
| multi_actor/100x1000 | 6.10 ms | 16.4 Melem/s |
| ring/10 | 6.3 µs | 1.75 Melem/s |
| ring/100 | 99.9 µs | 1.01 Melem/s |
| ring/500 | 919.0 µs | 545 Kelem/s |
| spawn/100 | 33.5 µs | 2.99 Melem/s |
| spawn/1000 | 326.2 µs | 3.07 Melem/s |
| spawn/5000 | 1.69 ms | 2.96 Melem/s |
## Key Observations
- Single-actor throughput scales well: 6.65M → 21.1M msgs/s as batch size grows (amortized overhead)
- Multi-actor throughput lower due to iteration overhead across actors
- Ring throughput degrades with ring size (expected: each message traverses more actors)
- Spawn throughput steady at ~3M/s regardless of batch size
- Message roundtrip latency: 2.24µs (spawn + deliver + process + reply + deliver)

View file

@ -1,30 +0,0 @@
# Task Constraints (from user)
## Scope of Study
- **Broad survey**: Not just Rust actor frameworks — include:
- Rust: ractor, actix, kameo, coerce, stakker, xactor, bastion
- Non-actor runtimes: tokio, C++ node/libuv event loop
- OS-level: `process` scheduling/logic in operating systems
- Classic actor systems: Erlang/OTP, Akka/Pekko (JVM)
- Any widely-used, well-reputed system
## Priority & Approach
- **Interleaved**: Pick a topic → analyze competitors → benchmark swactor → improve → repeat
- **Also improve testing methodology and coverage** based on analysis
- **Look at bug report histories** of competitor projects for insights
- Behavioral tests only (Given/When/Then), no white-box/structural tests
## Code Structure Rules
- **src/ is frozen**: No new files, no new modules, no structural changes. Only modify existing files in-place.
- **No new dependencies** on the root crate (swactor's Cargo.toml)
- May add new crates to `crates/` but they must NOT be pulled into `src/`
- Cap at ~5 new crates — if approaching that, prune back
- Integration tests in `tests/`, benchmarks in `benches/`
- Benchmark execution capped at 2 minutes max
- All notes go in `CLAUDE/notes/`
- Report architecture changes in `docs/`
## Commit Rules
- All `cargo test` must pass (including feature-gated)
- Never skip/delete tests for active code
- If a fix takes >3 attempts, log and move on

View file

@ -1,102 +0,0 @@
# 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.

View file

@ -1,62 +0,0 @@
# Progress Log
## Current Stage: Cycle 20 — Hot-Path Performance (Identity Hashing)
### Status: COMPLETE
### 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`
### 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
### 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 |
Note: End-to-end benchmarks unreliable in sandbox (55% variation between identical runs). Microbenchmarks confirmed significant hash/lookup improvement.
### 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
### 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
- 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
- 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)

View file

@ -1,185 +0,0 @@
# Research Synthesis: Competitor Analysis
## Frameworks Studied
1. **Ractor** (Rust) — async task-per-actor on tokio
2. **Tokio** (Rust) — work-stealing async runtime
3. **Erlang/OTP BEAM** — reduction-counted preemptive scheduler
4. **Linux CFS/EEVDF** — vruntime fairness, work stealing, adaptive ticks
5. **libuv/Node.js** — single-threaded event loop with phase-based execution
## Critical Finding: Swactor Fairness Bug
`tick_all` in `worker.rs` drains the ENTIRE mailbox for each actor before moving to the next:
```rust
while let Some(msg) = slot.mailbox.pop_front() {
// processes ALL messages for actor A before moving to actor B
}
```
If actor A has 10,000 queued messages, all other actors on the same worker are completely starved
until A finishes. Every other runtime studied prevents this:
- **BEAM**: 4000 reductions per process, then preempt
- **Tokio**: 128-256 operation cooperative budget per task
- **libuv**: Round-robin across handlers; no single handler drains completely
- **Linux CFS**: vruntime-based fairness; time slices enforced
## Ranked Improvement Opportunities
### P0: Per-Actor Message Budget (Fairness)
- **Impact**: Prevents starvation; critical for production workloads
- **Effort**: Small — modify `tick_all` loop in `worker.rs`
- **Source**: BEAM reductions, tokio coop budget
- **Design**: Process up to N messages per actor per tick, configurable via RuntimeConfig
### P1: Improved Benchmarks
- **Impact**: Can't improve what you can't measure
- **Effort**: Medium — new benchmark scenarios
- **Source**: Ractor benchmarks, tokio benchmarks
- **New scenarios needed**:
- Fairness: imbalanced load (1 hot actor + 99 cold actors)
- Message size sensitivity (8B, 64B, 256B, 1KB)
- Contention: many-to-one fanin
- Latency percentiles (p50, p99, p999)
- Cross-worker vs same-worker message delivery
### P2: Better Testing Coverage
- **Impact**: Catches regressions, validates fairness guarantees
- **Effort**: Medium
- **Source**: BEAM testing patterns, tokio Loom
- **New tests needed**:
- Fairness: hot actor doesn't starve cold actors
- Backpressure: tiny buffer under load
- Concurrent spawn+send races
- Multi-threaded delivery guarantees
### P3: Adaptive Backoff with Thread Parking
- **Impact**: Better latency under varying load; power savings
- **Effort**: Medium — modify `run` loop in `worker.rs`
- **Source**: Tokio parker, Linux NO_HZ
- **Design**: Replace spinning with condvar-based parking; use notification to wake
### P4: Work Stealing (Future)
- **Impact**: Dynamic load balancing
- **Effort**: Large — significant architectural change
- **Source**: Tokio steal-half, BEAM migration plans
- **Note**: Would require stealing actors between workers, which changes ownership
## Key Design Comparisons
| Dimension | Swactor | Ractor | Tokio | BEAM |
|-----------|---------|--------|-------|------|
| Scheduling | Sync tick | Async task-per-actor | Work-stealing | Reduction preemption |
| Fairness | None (drain all) | N/A (1 task = 1 actor) | Coop budget (128) | 4000 reductions |
| Backpressure | Bounded crossbeam ring | None (unbounded) | Bounded MPSC | Off-heap mailbox |
| Work stealing | None | Tokio handles it | Steal-half, N/2 searchers | Steal + migrate |
| Panic handling | catch_unwind + poison | AssertUnwindSafe + supervisor | N/A | Process isolation |
| Message passing | Box<dyn Any> downcast | Box<dyn Any> downcast | Typed channels | Term copying |
## Ractor Bug History Lessons
- Destructive `get_children()` — snapshot methods must not mutate state
- OutputPort silent drops — bounded channels need explicit backpressure, not silent overflow
- Remote actor latency regression — cross-runtime messaging needs careful ordering
- Memory bloat per actor — each channel/structure per actor adds up at scale
## Tokio Patterns to Adopt
1. LIFO slot for same-worker sends (cache locality)
2. Searcher count limiting (N/2 max) for cross-worker stealing
3. Steal-half strategy (amortize overhead)
4. Global queue interval checking (reduce contention)
5. Loom-style testing for lock-free code
6. Single allocation per actor context (hot/cold layout)
## Additional Frameworks Studied (Cycle 2)
### Kameo (v0.19)
- Fully async on tokio, one task per actor
- Dual mailbox: bounded (default 64) or unbounded tokio mpsc channels
- Proper backpressure via bounded mpsc sender blocking
- Typed signals (no Box<dyn Any>) — vtable dispatch, no downcast failures
- Erlang-style links for supervision (`on_link_died`)
- `on_panic` hook can restart actor (vs swactor's permanent poisoning)
- Bugs: deadlocks in link establishment, leaked ActorRef preventing stop
### Actix (v0.13)
- Context-as-Future model — each actor is a single pollable Future on an Arbiter
- **Custom Vyukov lock-free MPSC queue** (not tokio channels) — push is single atomic_swap
- Default mailbox capacity: 16 (tiny!)
- `do_send()` bypasses capacity for internal notifications
- Mailbox has 256-message assertion guard (similar to our budget approach!)
- vtable dispatch via `Box<dyn EnvelopeProxy<A>>` — no Any downcast
- SyncArbiter: crossbeam_channel thread pool for blocking actors
- WHY FAST: custom MPSC queue, no async overhead for message processing,
same-thread actors avoid cross-thread coordination, SmallVec for futures
### Swactor Advantages (confirmed)
- Synchronous tick model: deterministic, no async overhead, simulation-friendly
- Hybrid channel: bounded ring + unbounded overflow = no message loss
- Per-actor message budget: validated by BEAM (4000 reds), tokio (128 ops), actix (256 assert)
- No tokio dependency: could run on bare metal
- Detailed per-phase timing stats (6-phase TickTiming)
### Swactor Weaknesses to Address
- Box<dyn Any> downcast can fail silently → type mismatch tracking needed (have it)
- ~~No backpressure~~ → FIXED: optional bounded mailbox with DropNewest/DropOldest (Cycle 6)
- ~~Panicked actors permanently poisoned~~ → FIXED: factory-based restart with max_restarts limit (Cycle 7)
- ~~Spin/sleep backoff wastes CPU~~ → FIXED: thread parking with instant wakeup (Cycle 3)
- ~~No lifecycle hooks~~ → FIXED: on_start/on_stop with default no-ops (Cycle 9)
- ~~No graceful stop~~ → FIXED: ctx.stop_self() (immediate) + runtime.stop_actor() (queued) (Cycle 9)
- No supervision trees (factory restart + lifecycle hooks are steps toward this)
## Lifecycle Hooks Deep Dive (Cycle 9)
| Framework | on_start | on_stop | on_panic | Self-stop | External stop |
|-----------|----------|---------|----------|-----------|---------------|
| Erlang | init/1 | terminate/2 (NOT on crash) | N/A | {stop,Reason,State} | gen_server:stop |
| Akka | preStart | postStop (always) | preRestart | context.stop(self) | PoisonPill / stop |
| Actix | started | stopped | N/A | ctx.stop() | addr.do_send(Stop) |
| Kameo | on_start | on_stop | on_panic | Context::stop() | stop_gracefully/kill |
| Ractor | pre_start | post_stop (NOT on kill/panic) | N/A | stop() | Signal::Kill |
| **Swactor** | **on_start** | **on_stop (NOT on panic)** | N/A | **ctx.stop_self()** | **runtime.stop_actor()** |
Design decisions:
- on_stop NOT called on panic (matches Erlang/Ractor — corrupt state is unsafe)
- ctx.stop_self() is immediate (after current message) via request_stop buffer
- runtime.stop_actor() uses StopSignal message (PoisonPill semantics — queued after existing msgs)
- on_start panics → immediate poison (no restart attempted — init failure is fatal)
- Restarted actors get on_start called again on fresh instance
## Work Stealing Deep Dive (Cycle 5)
### Cross-Runtime Comparison
| Aspect | Tokio | Go | BEAM | ForkJoinPool |
|--------|-------|-----|------|-------------|
| Queue | Fixed 256-slot ring | 256-slot ring + runnext | Per-priority linked | Growable array deque |
| Steal granularity | Half victim's queue | Half victim's runq | Individual processes | One task at a time |
| LIFO fast-path | Dedicated slot (3-use cap) | runnext (stealable 4th try) | None | Owner pops from top |
| Global queue | Mutex intrusive list | Checked 1/61 ticks | Per-priority migration | Even-indexed submit queues |
| Searcher limit | N/2 workers | GOMAXPROCS/2 | N/A (proactive migration) | Idle stack in ctl field |
| Balance strategy | Reactive steal | Reactive steal | **Proactive migration** + reactive | Reactive scan |
| Load compaction | No (spread) | No (spread) | **Yes** (min schedulers) | No (spread) |
### Key Patterns
1. **LIFO slot**: Every runtime has one. Improves cache locality by running the recipient immediately after the sender. Tokio caps at 3 consecutive uses to prevent starvation.
2. **Steal-half**: Tokio and Go both steal half the victim's queue. This amortizes the overhead of cross-thread coordination — O(1) per stolen item instead of O(1) per steal.
3. **N/2 searcher limit**: Both Tokio and Go cap concurrent searchers to prevent thundering herd. Without it, all N workers scanning causes O(N²) cache-line bouncing.
4. **BEAM's migration**: Unique dual approach — reactive stealing when idle, proactive migration via periodic `check_balance()` that computes migration paths based on average max queue length.
### Feasibility for Swactor
- **Full actor migration**: Mechanically possible (ActorSlot is Send), but has 1-tick message loss window and requires push-based donation (ActorPool not Sync → no pull stealing)
- **Message stealing without actors**: Impossible — actor IS the state, messages without the actor are meaningless
- **Transfer queue snooping**: Pointless without actor migration
- **Load-aware placement** ✅ IMPLEMENTED: Placement reads per-worker stats to bias toward lighter workers, with round-robin fallback when stats are equal
### Decision: Load-Aware Placement over Work Stealing
Chose load-aware placement because:
- Zero correctness risk (no message loss, no ordering changes)
- O(N) atomic loads per spawn (trivial for N≤8 workers)
- Handles the primary source of imbalance: uneven spawn distribution
- Full work stealing deferred — would require migration channels, address map coordination, and forwarding tombstones

9
Cargo.lock generated
View file

@ -1489,9 +1489,18 @@ dependencies = [
"proptest", "proptest",
"proptest-state-machine", "proptest-state-machine",
"serde", "serde",
"swactor-std",
"tracing", "tracing",
] ]
[[package]]
name = "swactor-std"
version = "0.1.0"
dependencies = [
"getrandom 0.2.17",
"swactor",
]
[[package]] [[package]]
name = "syn" name = "syn"
version = "2.0.114" version = "2.0.114"

View file

@ -1,5 +1,5 @@
[workspace] [workspace]
members = [".", "crates/python", "crates/wasm", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/simulation-dashboard"] members = [".", "crates/python", "crates/wasm", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/simulation-dashboard", "crates/std"]
exclude = ["tools/depgraph"] exclude = ["tools/depgraph"]
[package] [package]
@ -34,6 +34,7 @@ crossbeam-utils = "0.8.21"
criterion = { version = "0.5", features = ["html_reports"] } criterion = { version = "0.5", features = ["html_reports"] }
proptest = "1" proptest = "1"
proptest-state-machine = "0.3" proptest-state-machine = "0.3"
swactor-std = { path = "crates/std" }
[[bench]] [[bench]]
name = "runtime_benchmarks" name = "runtime_benchmarks"

View file

@ -1,11 +1,13 @@
use criterion::{ use criterion::{
criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion, Throughput,
}; };
use std::sync::Arc;
use swactor::{ use swactor::{
actor::{ActorAddress, ActorInterface}, actor::{ActorAddress, ActorInterface},
config::RuntimeConfig, config::RuntimeConfig,
runtime::{Ctx, Runtime}, runtime::{Ctx, Runtime},
}; };
use swactor_std::{RuntimeGroups, RuntimeNaming, StdExtension};
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Helper // Helper
@ -615,7 +617,8 @@ fn registry_benchmarks(c: &mut Criterion) {
b.iter_batched( b.iter_batched(
|| { || {
counter += 1; counter += 1;
let rt = Runtime::new(make_config(1_000, 1_000)); let rt = Runtime::new(make_config(1_000, 1_000))
.with_extension(Arc::new(StdExtension::new()));
(rt, counter) (rt, counter)
}, },
|(rt, i)| { |(rt, i)| {
@ -632,7 +635,8 @@ fn registry_benchmarks(c: &mut Criterion) {
group.bench_function("where_is_100_names", |b| { group.bench_function("where_is_100_names", |b| {
b.iter_batched( b.iter_batched(
|| { || {
let rt = Runtime::new(make_config(1_000, 1_000)); let rt = Runtime::new(make_config(1_000, 1_000))
.with_extension(Arc::new(StdExtension::new()));
for i in 0..100 { for i in 0..100 {
rt.spawn_named(format!("actor-{i}"), NoopActor).unwrap(); rt.spawn_named(format!("actor-{i}"), NoopActor).unwrap();
} }
@ -655,7 +659,8 @@ fn registry_benchmarks(c: &mut Criterion) {
|b, &members| { |b, &members| {
b.iter_batched( b.iter_batched(
|| { || {
let rt = Runtime::new(make_config(members + 100, members * 10)); let rt = Runtime::new(make_config(members + 100, members * 10))
.with_extension(Arc::new(StdExtension::new()));
for _ in 0..members { for _ in 0..members {
let addr = rt.spawn(SinkActor).unwrap(); let addr = rt.spawn(SinkActor).unwrap();
rt.join_group(addr, "bench-group"); rt.join_group(addr, "bench-group");

12
crates/std/Cargo.toml Normal file
View file

@ -0,0 +1,12 @@
[package]
name = "swactor-std"
version = "0.1.0"
edition = "2024"
[features]
default = ["getrandom"]
getrandom = ["dep:getrandom"]
[dependencies]
swactor = { path = "../.." }
getrandom = { version = "0.2", optional = true }

113
crates/std/src/ctx_ext.rs Normal file
View file

@ -0,0 +1,113 @@
use swactor::actor::{ActorAddress, ActorInterface, Ctx, Message, MonitorRef};
use swactor::Error;
use crate::StdExtension;
fn get_ext<'a>(ctx: &'a Ctx) -> &'a StdExtension {
ctx.extension()
.expect("StdExtension not installed — use Runtime::with_extension()")
.as_any()
.downcast_ref::<StdExtension>()
.expect("Extension is not StdExtension")
}
/// Monitoring extension for [`Ctx`].
///
/// Provides `monitor` / `demonitor` via the [`StdExtension`] monitor registry.
pub trait CtxMonitoring {
/// Subscribe to death notifications from `target`. Returns a [`MonitorRef`]
/// that can be used to cancel the subscription.
fn monitor(&self, target: ActorAddress) -> MonitorRef;
/// Cancel a monitor subscription.
fn demonitor(&self, mref: MonitorRef);
}
impl CtxMonitoring for Ctx<'_> {
fn monitor(&self, target: ActorAddress) -> MonitorRef {
get_ext(self).monitor_registry.register(self.self_addr(), target)
}
fn demonitor(&self, mref: MonitorRef) {
get_ext(self).monitor_registry.deregister(mref);
}
}
/// Naming extension for [`Ctx`].
///
/// Provides `where_is`, `register_name`, and `spawn_named` via the [`StdExtension`]
/// name registry.
pub trait CtxNaming {
/// Look up an actor address by its registered name.
fn where_is(&self, name: &str) -> Option<ActorAddress>;
/// Register a name for the given address.
fn register_name(&self, name: impl Into<String>, addr: ActorAddress) -> Result<(), Error>;
/// Spawn an actor with a registered name, returning its address.
fn spawn_named<A: ActorInterface>(&self, name: impl Into<String>, actor: A) -> Result<ActorAddress, Error>;
}
impl CtxNaming for Ctx<'_> {
fn where_is(&self, name: &str) -> Option<ActorAddress> {
get_ext(self).name_registry.lookup(name)
}
fn register_name(&self, name: impl Into<String>, addr: ActorAddress) -> Result<(), Error> {
get_ext(self).name_registry.register(name.into(), addr)
}
fn spawn_named<A: ActorInterface>(&self, name: impl Into<String>, actor: A) -> Result<ActorAddress, Error> {
let name = name.into();
let addr = self.spawn(actor)?;
if let Err(e) = get_ext(self).name_registry.register(name, addr) {
let _ = self.stop_actor(addr);
return Err(e);
}
Ok(addr)
}
}
/// Group extension for [`Ctx`].
///
/// Provides `join_group`, `leave_group`, `publish`, and `group_members` via
/// the [`StdExtension`] group registry.
pub trait CtxGroups {
/// Add this actor to a named group.
fn join_group(&self, group: impl Into<String>);
/// Remove this actor from a named group.
fn leave_group(&self, group: &str);
/// Broadcast a message to all members of a named group.
/// Returns the number of messages successfully enqueued.
fn publish<M: Message>(&self, group: &str, msg: M) -> usize;
/// Return all members of a named group.
fn group_members(&self, group: &str) -> Vec<ActorAddress>;
}
impl CtxGroups for Ctx<'_> {
fn join_group(&self, group: impl Into<String>) {
get_ext(self).group_registry.join(group.into(), self.self_addr());
}
fn leave_group(&self, group: &str) {
get_ext(self).group_registry.leave(group, &self.self_addr());
}
fn publish<M: Message>(&self, group: &str, msg: M) -> usize {
let members = get_ext(self).group_registry.members(group);
let mut count = 0;
for member in &members {
if self.send(*member, msg.clone()).is_ok() {
count += 1;
}
}
count
}
fn group_members(&self, group: &str) -> Vec<ActorAddress> {
get_ext(self).group_registry.members(group)
}
}

View file

@ -0,0 +1,62 @@
use std::any::Any;
use swactor::actor::{ActorAddress, Down, StopReason};
use swactor::extension::RuntimeExtension;
use crate::group_registry::GroupRegistry;
use crate::monitor_registry::MonitorRegistry;
use crate::name_registry::NameRegistry;
/// Standard library extension — provides naming, monitoring, and group registries.
///
/// Install on a `Runtime` via `runtime.with_extension(Arc::new(StdExtension::new()))`.
pub struct StdExtension {
pub(crate) name_registry: NameRegistry,
pub(crate) monitor_registry: MonitorRegistry,
pub(crate) group_registry: GroupRegistry,
}
impl StdExtension {
pub fn new() -> Self {
Self {
name_registry: NameRegistry::new(),
monitor_registry: MonitorRegistry::new(),
group_registry: GroupRegistry::new(),
}
}
}
impl Default for StdExtension {
fn default() -> Self {
Self::new()
}
}
impl RuntimeExtension for StdExtension {
fn on_actor_death(
&self,
dead: &[(ActorAddress, StopReason)],
) -> Vec<(ActorAddress, Box<dyn Any + Send>)> {
let mut notifications = Vec::new();
for &(addr, reason) in dead {
let watchers = self.monitor_registry.take_monitors(&addr);
for (_mref, watcher) in watchers {
let down = Down { addr, reason };
notifications.push((watcher, Box::new(down) as Box<dyn Any + Send>));
}
}
notifications
}
fn cleanup_dead(&self, dead: &[ActorAddress]) {
for addr in dead {
self.name_registry.unregister_by_addr(addr);
self.group_registry.cleanup(addr);
self.monitor_registry.remove_watcher(addr);
}
}
fn as_any(&self) -> &dyn Any {
self
}
}

View file

@ -0,0 +1,81 @@
use std::collections::{HashMap, HashSet};
use std::sync::RwLock;
use swactor::actor::ActorAddress;
use swactor::{AddrBuildHasher, AddrMap, AddrSet};
/// Actor groups (pub-sub). Actors join/leave named groups; messages can be
/// broadcast to all members of a group.
///
/// Groups are created lazily on first join and removed when empty.
pub struct GroupRegistry {
/// group_name → set of member addresses
groups: RwLock<HashMap<String, AddrSet>>,
/// actor_addr → set of group names (reverse map for O(G) cleanup on death)
memberships: RwLock<AddrMap<HashSet<String>>>,
}
impl GroupRegistry {
pub fn new() -> Self {
Self {
groups: RwLock::new(HashMap::new()),
memberships: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
}
}
/// Add an actor to a named group. Group is created if it doesn't exist.
pub fn join(&self, group: String, addr: ActorAddress) {
self.groups.write().unwrap()
.entry(group.clone())
.or_insert_with(|| HashSet::with_hasher(AddrBuildHasher))
.insert(addr);
self.memberships.write().unwrap()
.entry(addr)
.or_default()
.insert(group);
}
/// Remove an actor from a named group. Empty groups are auto-deleted.
pub fn leave(&self, group: &str, addr: &ActorAddress) {
let mut groups = self.groups.write().unwrap();
if let Some(members) = groups.get_mut(group) {
members.remove(addr);
if members.is_empty() {
groups.remove(group);
}
}
drop(groups);
if let Some(membership) = self.memberships.write().unwrap().get_mut(addr) {
membership.remove(group);
}
}
/// Return all members of a group.
pub fn members(&self, group: &str) -> Vec<ActorAddress> {
self.groups.read().unwrap()
.get(group)
.map(|s| s.iter().copied().collect())
.unwrap_or_default()
}
/// Remove a dead actor from all its groups.
pub fn cleanup(&self, addr: &ActorAddress) {
let group_names = self.memberships.write().unwrap().remove(addr);
if let Some(names) = group_names {
let mut groups = self.groups.write().unwrap();
for name in names {
if let Some(members) = groups.get_mut(&name) {
members.remove(addr);
if members.is_empty() {
groups.remove(&name);
}
}
}
}
}
/// Return all active group names.
pub fn group_names(&self) -> Vec<String> {
self.groups.read().unwrap().keys().cloned().collect()
}
}

14
crates/std/src/lib.rs Normal file
View file

@ -0,0 +1,14 @@
mod supervisor;
mod router;
pub mod name_registry;
pub mod monitor_registry;
pub mod group_registry;
mod extension;
mod ctx_ext;
mod runtime_ext;
pub use supervisor::{ChildSpec, RestartPolicy, Supervisor, SupervisorStrategy};
pub use router::{Router, RoutingStrategy};
pub use extension::StdExtension;
pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups};
pub use runtime_ext::{RuntimeNaming, RuntimeGroups};

View file

@ -0,0 +1,79 @@
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::RwLock;
use swactor::actor::{ActorAddress, MonitorRef};
use swactor::{AddrBuildHasher, AddrMap};
/// Tracks monitor subscriptions: watched actor → list of (MonitorRef, watcher address).
///
/// Write-rare (monitor/demonitor/death), read at cleanup time.
pub struct MonitorRegistry {
/// watched_addr → [(mref, watcher_addr)]
monitors: RwLock<AddrMap<Vec<(MonitorRef, ActorAddress)>>>,
/// mref → watched_addr (for O(1) demonitor)
ref_to_target: RwLock<HashMap<MonitorRef, ActorAddress>>,
next_ref: AtomicU64,
}
impl MonitorRegistry {
pub fn new() -> Self {
Self {
monitors: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
ref_to_target: RwLock::new(HashMap::new()),
next_ref: AtomicU64::new(1),
}
}
/// Register a monitor: `watcher` wants to know when `target` dies.
pub fn register(&self, watcher: ActorAddress, target: ActorAddress) -> MonitorRef {
let id = self.next_ref.fetch_add(1, Ordering::Relaxed);
let mref = MonitorRef::from_raw(id);
self.monitors.write().unwrap()
.entry(target)
.or_default()
.push((mref, watcher));
self.ref_to_target.write().unwrap().insert(mref, target);
mref
}
/// Cancel a monitor by its ref.
pub fn deregister(&self, mref: MonitorRef) {
if let Some(target) = self.ref_to_target.write().unwrap().remove(&mref) {
let mut monitors = self.monitors.write().unwrap();
if let Some(watchers) = monitors.get_mut(&target) {
watchers.retain(|(r, _)| *r != mref);
if watchers.is_empty() {
monitors.remove(&target);
}
}
}
}
/// Remove and return all monitors for a dead actor.
pub fn take_monitors(&self, target: &ActorAddress) -> Vec<(MonitorRef, ActorAddress)> {
let watchers = self.monitors.write().unwrap().remove(target).unwrap_or_default();
let mut ref_map = self.ref_to_target.write().unwrap();
for (mref, _) in &watchers {
ref_map.remove(mref);
}
watchers
}
/// Remove all monitor subscriptions where `addr` is the watcher (dead watcher cleanup).
pub fn remove_watcher(&self, addr: &ActorAddress) {
let mut monitors = self.monitors.write().unwrap();
let mut ref_map = self.ref_to_target.write().unwrap();
monitors.retain(|_target, watchers| {
watchers.retain(|(mref, watcher)| {
if watcher == addr {
ref_map.remove(mref);
false
} else {
true
}
});
!watchers.is_empty()
});
}
}

View file

@ -0,0 +1,59 @@
use std::collections::HashMap;
use std::sync::RwLock;
use swactor::actor::ActorAddress;
use swactor::{AddrBuildHasher, AddrMap};
/// Named actor registry — maps human-readable names to actor addresses.
///
/// `RwLock<HashMap>` — same pattern as `AddressMap`. Write-rare (spawn/death),
/// read-often (lookup). A reverse map enables O(1) cleanup on actor death.
pub struct NameRegistry {
names: RwLock<HashMap<String, ActorAddress>>,
reverse: RwLock<AddrMap<String>>,
}
impl NameRegistry {
pub fn new() -> Self {
Self {
names: RwLock::new(HashMap::new()),
reverse: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
}
}
/// Register a name → address mapping. Returns `Err` if the name is already taken.
pub fn register(&self, name: String, addr: ActorAddress) -> Result<(), swactor::Error> {
let mut names = self.names.write().unwrap();
if names.contains_key(&name) {
return Err(swactor::Error::from("Name already registered"));
}
names.insert(name.clone(), addr);
drop(names);
self.reverse.write().unwrap().insert(addr, name);
Ok(())
}
/// Look up an actor address by name.
pub fn lookup(&self, name: &str) -> Option<ActorAddress> {
self.names.read().unwrap().get(name).copied()
}
/// Unregister a name, returning the address it was bound to.
pub fn unregister(&self, name: &str) -> Option<ActorAddress> {
let addr = self.names.write().unwrap().remove(name)?;
self.reverse.write().unwrap().remove(&addr);
Some(addr)
}
/// Remove a name by address (called on actor death for auto-cleanup).
pub fn unregister_by_addr(&self, addr: &ActorAddress) {
if let Some(name) = self.reverse.write().unwrap().remove(addr) {
self.names.write().unwrap().remove(&name);
}
}
/// Return all registered names.
pub fn registered_names(&self) -> Vec<String> {
self.names.read().unwrap().keys().cloned().collect()
}
}

169
crates/std/src/router.rs Normal file
View file

@ -0,0 +1,169 @@
use std::marker::PhantomData;
use std::sync::Arc;
use swactor::actor::{ActorAddress, ActorInterface, Ctx, Down, Message};
use swactor::Error;
use crate::supervisor::ActiveChild;
use crate::CtxMonitoring;
/// Strategy for distributing messages across pool workers.
#[derive(Debug, Clone)]
pub enum RoutingStrategy {
/// Sequential round-robin distribution.
RoundRobin,
/// Random worker selection.
Random,
/// Send to all workers (message is cloned to each).
Broadcast,
}
/// A router actor that manages a pool of identical workers and distributes
/// incoming messages across them according to a [`RoutingStrategy`].
///
/// Workers are spawned during `on_start`, monitored for failures, and
/// automatically replaced to maintain the target pool size. Meltdown
/// protection stops the router when total restarts exceed `max_restarts`.
///
/// # Example
///
/// ```ignore
/// let router = Router::new(
/// RoutingStrategy::RoundRobin,
/// 5,
/// |ctx| ctx.spawn(MyWorker::new()),
/// 10,
/// );
/// let router_addr = rt.spawn(router)?;
/// rt.send_to(router_addr, WorkerMessage::DoWork(42))?;
/// ```
pub struct Router<M: Message> {
strategy: RoutingStrategy,
pool_size: usize,
factory: Arc<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync>,
workers: Vec<Option<ActiveChild>>,
rr_index: usize,
total_restarts: u32,
max_restarts: u32,
_marker: PhantomData<M>,
}
impl<M: Message> Router<M> {
pub fn new(
strategy: RoutingStrategy,
pool_size: usize,
factory: impl Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync + 'static,
max_restarts: u32,
) -> Self {
Self {
strategy,
pool_size,
factory: Arc::new(factory),
workers: (0..pool_size).map(|_| None).collect(),
rr_index: 0,
total_restarts: 0,
max_restarts,
_marker: PhantomData,
}
}
fn start_worker(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> {
let addr = (self.factory)(ctx)?;
let mref = ctx.monitor(addr);
self.workers[idx] = Some(ActiveChild {
addr,
_monitor_ref: mref,
});
Ok(())
}
fn find_worker_idx(&self, addr: ActorAddress) -> Option<usize> {
self.workers
.iter()
.position(|w| w.as_ref().map_or(false, |ac| ac.addr == addr))
}
fn live_workers(&self) -> Vec<ActorAddress> {
self.workers
.iter()
.filter_map(|w| w.as_ref().map(|ac| ac.addr))
.collect()
}
fn select_one(&mut self) -> Option<ActorAddress> {
let live = self.live_workers();
if live.is_empty() {
return None;
}
match self.strategy {
RoutingStrategy::RoundRobin => {
let idx = self.rr_index % live.len();
self.rr_index = self.rr_index.wrapping_add(1);
Some(live[idx])
}
RoutingStrategy::Random => {
let mut buf = [0u8; 8];
getrandom::getrandom(&mut buf).expect("getrandom failed");
let r = u64::from_ne_bytes(buf) as usize;
Some(live[r % live.len()])
}
RoutingStrategy::Broadcast => None, // handled separately
}
}
}
impl<M: Message> ActorInterface for Router<M> {
type Incoming = M;
type Response = ();
fn handle(&mut self, ctx: &Ctx, msg: M) {
match self.strategy {
RoutingStrategy::Broadcast => {
let live = self.live_workers();
for addr in live {
let _ = ctx.send(addr, msg.clone());
}
}
_ => {
if let Some(addr) = self.select_one() {
let _ = ctx.send(addr, msg);
}
}
}
}
fn on_start(&mut self, ctx: &Ctx) {
for idx in 0..self.pool_size {
if let Err(e) = self.start_worker(ctx, idx) {
eprintln!("swactor: router failed to start worker {idx}: {e}");
}
}
}
fn on_stop(&mut self, ctx: &Ctx) {
for child in self.workers.iter().flatten() {
let _ = ctx.stop_actor(child.addr);
}
}
fn handle_down(&mut self, ctx: &Ctx, down: Down) {
let Some(idx) = self.find_worker_idx(down.addr) else {
return;
};
self.workers[idx] = None;
self.total_restarts += 1;
if self.total_restarts > self.max_restarts {
eprintln!(
"swactor: router reached max restarts ({}), shutting down",
self.max_restarts
);
ctx.stop_self();
return;
}
if let Err(e) = self.start_worker(ctx, idx) {
eprintln!("swactor: router failed to restart worker {idx}: {e}");
}
}
}

View file

@ -0,0 +1,106 @@
use swactor::actor::{ActorAddress, ActorInterface, Message};
use swactor::runtime::Runtime;
use swactor::Error;
use crate::StdExtension;
fn get_ext(rt: &Runtime) -> &StdExtension {
rt.extension()
.expect("StdExtension not installed — use Runtime::with_extension()")
.as_any()
.downcast_ref::<StdExtension>()
.expect("Extension is not StdExtension")
}
/// Naming extension for [`Runtime`].
///
/// Provides `spawn_named`, `where_is`, `unregister`, and `registered_names`
/// via the [`StdExtension`] name registry.
pub trait RuntimeNaming {
/// Spawn an actor with a registered name, returning its address.
fn spawn_named<A: ActorInterface>(&self, name: impl Into<String>, actor: A) -> Result<ActorAddress, Error>;
/// Look up an actor address by its registered name.
fn where_is(&self, name: &str) -> Option<ActorAddress>;
/// Unregister a name. Returns the address it was bound to, or `None`.
fn unregister(&self, name: &str) -> Option<ActorAddress>;
/// Return all currently registered actor names.
fn registered_names(&self) -> Vec<String>;
}
impl RuntimeNaming for Runtime {
fn spawn_named<A: ActorInterface>(&self, name: impl Into<String>, actor: A) -> Result<ActorAddress, Error> {
let name = name.into();
let addr = self.spawn(actor)?;
if let Err(e) = get_ext(self).name_registry.register(name, addr) {
let _ = self.stop_actor(addr);
return Err(e);
}
Ok(addr)
}
fn where_is(&self, name: &str) -> Option<ActorAddress> {
get_ext(self).name_registry.lookup(name)
}
fn unregister(&self, name: &str) -> Option<ActorAddress> {
get_ext(self).name_registry.unregister(name)
}
fn registered_names(&self) -> Vec<String> {
get_ext(self).name_registry.registered_names()
}
}
/// Group extension for [`Runtime`].
///
/// Provides `join_group`, `leave_group`, `publish_to`, `group_members`,
/// and `groups` via the [`StdExtension`] group registry.
pub trait RuntimeGroups {
/// Add an actor to a named group. The group is created if it doesn't exist.
fn join_group(&self, addr: ActorAddress, group: impl Into<String>);
/// Remove an actor from a named group. Empty groups are auto-deleted.
fn leave_group(&self, addr: ActorAddress, group: &str);
/// Broadcast a message to all members of a named group.
/// Returns the number of messages successfully enqueued.
fn publish_to<M: Message>(&self, group: &str, msg: M) -> usize;
/// Return all current members of a named group.
fn group_members(&self, group: &str) -> Vec<ActorAddress>;
/// Return all active group names.
fn groups(&self) -> Vec<String>;
}
impl RuntimeGroups for Runtime {
fn join_group(&self, addr: ActorAddress, group: impl Into<String>) {
get_ext(self).group_registry.join(group.into(), addr);
}
fn leave_group(&self, addr: ActorAddress, group: &str) {
get_ext(self).group_registry.leave(group, &addr);
}
fn publish_to<M: Message>(&self, group: &str, msg: M) -> usize {
let members = get_ext(self).group_registry.members(group);
let mut count = 0;
for member in &members {
if self.send_to(*member, msg.clone()).is_ok() {
count += 1;
}
}
count
}
fn group_members(&self, group: &str) -> Vec<ActorAddress> {
get_ext(self).group_registry.members(group)
}
fn groups(&self) -> Vec<String> {
get_ext(self).group_registry.group_names()
}
}

View file

@ -0,0 +1,301 @@
use std::sync::Arc;
use swactor::actor::{ActorAddress, ActorInterface, Ctx, Down, MonitorRef, StopReason};
use swactor::Error;
use crate::CtxMonitoring;
/// How a child should be restarted when it dies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestartPolicy {
/// Always restart, regardless of stop reason.
Permanent,
/// Restart only on abnormal exit (Panicked). Normal stops are final.
Transient,
/// Never restart. The child is removed on any exit.
Temporary,
}
/// Strategy for handling child failures.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SupervisorStrategy {
/// Only restart the failed child. Other children are unaffected.
OneForOne,
/// Terminate all children and restart them all in spec order.
OneForAll,
/// Terminate children started after the failed child, then restart
/// the failed child and all terminated children in spec order.
RestForOne,
}
/// Specification for a supervised child actor.
///
/// The `start` closure is called with `&Ctx` and should spawn the child actor
/// (typically via `ctx.spawn()`). The supervisor monitors the returned address
/// and applies the restart policy when the child dies.
pub struct ChildSpec {
/// Unique identifier for this child.
pub id: String,
/// How to restart this child.
pub restart: RestartPolicy,
/// Factory to spawn the child. Called with `&Ctx`, returns the child's address.
pub start: Arc<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync>,
}
impl ChildSpec {
pub fn new(
id: impl Into<String>,
restart: RestartPolicy,
start: impl Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync + 'static,
) -> Self {
Self {
id: id.into(),
restart,
start: Arc::new(start),
}
}
}
/// Tracked state for an active child within a supervisor or router.
pub(crate) struct ActiveChild {
pub(crate) addr: ActorAddress,
pub(crate) _monitor_ref: MonitorRef,
}
/// Internal phase for coordinating multi-child restart (OneForAll, RestForOne).
///
/// In `Normal` phase, the supervisor processes Down messages and applies the strategy.
/// When a coordinated restart is needed, it transitions to `Stopping` (sends stop
/// signals, waits for Down confirmations) then restarts all affected children.
enum SupervisorPhase {
/// Normal operation — process Down messages and apply strategy.
Normal,
/// Waiting for children to confirm death before restarting.
Stopping {
/// Children we're still waiting for Down confirmation.
awaiting: Vec<ActorAddress>,
/// Spec indices to restart once all confirmations received.
restart_set: Vec<usize>,
},
}
/// A supervisor actor that manages child actors according to a restart strategy.
///
/// Children are spawned during `on_start`. When a child dies, the supervisor
/// receives a [`Down`] notification via [`ActorInterface::handle_down`] and
/// applies the configured strategy and restart policy.
///
/// # Strategies
///
/// - **OneForOne**: Only the failed child is restarted.
/// - **OneForAll**: All children are stopped, then all restarted in spec order.
/// - **RestForOne**: The failed child and all children started after it are
/// stopped, then restarted in spec order.
///
/// # Restart Intensity
///
/// The supervisor tracks total restarts. When `total_restarts > max_restarts`,
/// the supervisor stops itself (meltdown protection), escalating the failure
/// to its own supervisor if one exists.
///
/// # Example
///
/// ```ignore
/// let sup = Supervisor::new(
/// SupervisorStrategy::OneForOne,
/// 5, // max 5 restarts before meltdown
/// vec![
/// ChildSpec::new("worker", RestartPolicy::Permanent, |ctx| {
/// ctx.spawn(MyWorker::new())
/// }),
/// ],
/// );
/// let sup_addr = rt.spawn(sup)?;
/// ```
pub struct Supervisor {
strategy: SupervisorStrategy,
max_restarts: u32,
specs: Vec<ChildSpec>,
children: Vec<Option<ActiveChild>>,
total_restarts: u32,
phase: SupervisorPhase,
}
impl Supervisor {
pub fn new(
strategy: SupervisorStrategy,
max_restarts: u32,
specs: Vec<ChildSpec>,
) -> Self {
let children = (0..specs.len()).map(|_| None).collect();
Self {
strategy,
max_restarts,
specs,
children,
total_restarts: 0,
phase: SupervisorPhase::Normal,
}
}
fn start_child(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> {
let addr = (self.specs[idx].start)(ctx)?;
let mref = ctx.monitor(addr);
self.children[idx] = Some(ActiveChild {
addr,
_monitor_ref: mref,
});
Ok(())
}
fn find_child_idx(&self, addr: ActorAddress) -> Option<usize> {
self.children
.iter()
.position(|c| c.as_ref().map_or(false, |ac| ac.addr == addr))
}
/// Check meltdown intensity — returns true if we should stop.
fn check_intensity(&mut self) -> bool {
self.total_restarts += 1;
self.total_restarts > self.max_restarts
}
/// Try to finish the coordinated restart: restart all children in `restart_set`.
fn finish_restart(&mut self, ctx: &Ctx) {
let restart_set = match &mut self.phase {
SupervisorPhase::Stopping { restart_set, .. } => {
std::mem::take(restart_set)
}
_ => return,
};
self.phase = SupervisorPhase::Normal;
for idx in restart_set {
if let Err(e) = self.start_child(ctx, idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[idx].id, e
);
}
}
}
/// Begin a coordinated restart for the given spec indices.
/// Stops any living children in the set, then waits for their Down messages.
fn begin_coordinated_restart(&mut self, ctx: &Ctx, restart_indices: Vec<usize>) {
let mut awaiting = Vec::new();
for &idx in &restart_indices {
if let Some(child) = self.children[idx].take() {
let _ = ctx.stop_actor(child.addr);
awaiting.push(child.addr);
}
}
if awaiting.is_empty() {
// All children already dead — restart immediately.
for idx in &restart_indices {
if let Err(e) = self.start_child(ctx, *idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[*idx].id, e
);
}
}
} else {
self.phase = SupervisorPhase::Stopping {
awaiting,
restart_set: restart_indices,
};
}
}
}
impl ActorInterface for Supervisor {
type Incoming = ();
type Response = ();
fn handle(&mut self, _ctx: &Ctx, _msg: ()) {}
fn on_start(&mut self, ctx: &Ctx) {
for idx in 0..self.specs.len() {
if let Err(e) = self.start_child(ctx, idx) {
eprintln!(
"swactor: supervisor failed to start child '{}': {}",
self.specs[idx].id, e
);
}
}
}
fn on_stop(&mut self, ctx: &Ctx) {
for child in self.children.iter().flatten() {
let _ = ctx.stop_actor(child.addr);
}
}
fn handle_down(&mut self, ctx: &Ctx, down: Down) {
// During coordinated restart: track Down confirmations.
if matches!(self.phase, SupervisorPhase::Stopping { .. }) {
// Clear from children tracking
if let Some(idx) = self.find_child_idx(down.addr) {
self.children[idx] = None;
}
// Remove from awaiting list
if let SupervisorPhase::Stopping { awaiting, .. } = &mut self.phase {
awaiting.retain(|a| *a != down.addr);
}
let done = matches!(&self.phase,
SupervisorPhase::Stopping { awaiting, .. } if awaiting.is_empty());
if done {
self.finish_restart(ctx);
}
return;
}
// Normal phase: handle child death.
let Some(idx) = self.find_child_idx(down.addr) else {
return;
};
self.children[idx] = None;
let should_restart = match self.specs[idx].restart {
RestartPolicy::Permanent => true,
RestartPolicy::Transient => down.reason == StopReason::Panicked,
RestartPolicy::Temporary => false,
};
if !should_restart {
return;
}
if self.check_intensity() {
eprintln!(
"swactor: supervisor reached max restarts ({}), shutting down",
self.max_restarts
);
ctx.stop_self();
return;
}
match self.strategy {
SupervisorStrategy::OneForOne => {
if let Err(e) = self.start_child(ctx, idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[idx].id, e
);
}
}
SupervisorStrategy::OneForAll => {
// Stop all other living children, then restart all in order.
let restart_indices: Vec<usize> = (0..self.specs.len()).collect();
self.begin_coordinated_restart(ctx, restart_indices);
}
SupervisorStrategy::RestForOne => {
// Stop children after the failed one, then restart failed + rest.
let restart_indices: Vec<usize> = (idx..self.specs.len()).collect();
self.begin_coordinated_restart(ctx, restart_indices);
}
}
}
}

View file

@ -173,14 +173,14 @@ impl ActorInterface for IntervalSchedulerActor {
fn handle(&mut self, _ctx: &Ctx, _msg: FuzzMsg) {} fn handle(&mut self, _ctx: &Ctx, _msg: FuzzMsg) {}
} }
/// Restartable echo: panics on value=0, otherwise echoes. /// Panicking echo: panics on value=0, otherwise echoes.
struct RestartableEchoActor; struct PanickingEchoActor;
impl ActorInterface for RestartableEchoActor { impl ActorInterface for PanickingEchoActor {
type Incoming = FuzzMsg; type Incoming = FuzzMsg;
type Response = (); type Response = ();
fn handle(&mut self, ctx: &Ctx, msg: FuzzMsg) { fn handle(&mut self, ctx: &Ctx, msg: FuzzMsg) {
if msg.value == 0 { if msg.value == 0 {
panic!("fuzz: intentional panic for restart test"); panic!("fuzz: intentional panic");
} }
if let Some(idx) = msg.reply_to_idx { if let Some(idx) = msg.reply_to_idx {
let reply = FuzzMsg { value: msg.value, reply_to_idx: None }; let reply = FuzzMsg { value: msg.value, reply_to_idx: None };
@ -252,8 +252,8 @@ enum RawAction {
TickN { n: u8 }, TickN { n: u8 },
/// Graceful stop an actor /// Graceful stop an actor
StopActor { actor_idx: u8 }, StopActor { actor_idx: u8 },
/// Spawn a restartable echo actor (max_restarts = n) /// Spawn a panicking echo actor (panics on value=0)
SpawnRestartable { max_restarts: u8 }, SpawnPanicking,
/// Schedule a one-shot timer from an actor to an inbox /// Schedule a one-shot timer from an actor to an inbox
ScheduleTimer { delay: u8 }, ScheduleTimer { delay: u8 },
/// Schedule an interval timer from an actor to an inbox /// Schedule an interval timer from an actor to an inbox
@ -813,17 +813,12 @@ impl FuzzState {
self.log(format_args!("[STOP] {label}")); self.log(format_args!("[STOP] {label}"));
} }
} }
RawAction::SpawnRestartable { max_restarts } => { RawAction::SpawnPanicking => {
let restarts = (*max_restarts).min(5) as u32; if let Ok(addr) = self.runtime.spawn(PanickingEchoActor) {
if let Ok(addr) = self.runtime.spawn_restartable(
RestartableEchoActor,
|| RestartableEchoActor,
restarts,
) {
let id = self.actors.len(); let id = self.actors.len();
self.actors.push((addr, ActorKind::Echo)); self.actors.push((addr, ActorKind::Echo));
self.total_spawned += 1; self.total_spawned += 1;
self.log(format_args!("[SPAWN] Restartable(max={restarts}) -> actor#{id}")); self.log(format_args!("[SPAWN] PanickingEcho -> actor#{id}"));
} }
} }
RawAction::ScheduleTimer { delay } => { RawAction::ScheduleTimer { delay } => {

View file

@ -1,6 +1,4 @@
use std::any::Any; use std::any::Any;
use std::marker::PhantomData;
use std::sync::Arc;
use crate::Error; use crate::Error;
@ -75,33 +73,11 @@ impl ActorAddress {
/// The actor process as represented in the Runtime — thin wrapper around user state. /// The actor process as represented in the Runtime — thin wrapper around user state.
pub struct Actor<A: ActorInterface> { pub struct Actor<A: ActorInterface> {
inner: A, inner: A,
/// Factory for creating fresh instances on restart. None = not restartable.
restart_factory: Option<Arc<dyn Fn() -> A + Send + Sync>>,
max_restarts: u32,
restart_count: u32,
} }
impl<A: ActorInterface> Actor<A> { impl<A: ActorInterface> Actor<A> {
pub fn new(inner: A) -> Self { pub fn new(inner: A) -> Self {
Self { Self { inner }
inner,
restart_factory: None,
max_restarts: 0,
restart_count: 0,
}
}
pub fn new_restartable(
inner: A,
factory: Arc<dyn Fn() -> A + Send + Sync>,
max_restarts: u32,
) -> Self {
Self {
inner,
restart_factory: Some(factory),
max_restarts,
restart_count: 0,
}
} }
} }
@ -111,12 +87,6 @@ impl<A: ActorInterface> Actor<A> {
pub trait AnyActor: Send { pub trait AnyActor: Send {
fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) -> Option<&'static str>; fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) -> Option<&'static str>;
/// Attempt to create a fresh instance for restart after panic.
/// Returns `None` if restart is not supported or restart limit exceeded.
fn try_restart(&self) -> Option<Box<dyn AnyActor>> {
None
}
/// Called once after spawn, before first message. See [`ActorInterface::on_start`]. /// Called once after spawn, before first message. See [`ActorInterface::on_start`].
fn on_start(&mut self, _ctx: &Ctx) {} fn on_start(&mut self, _ctx: &Ctx) {}
@ -145,20 +115,6 @@ where
} }
} }
fn try_restart(&self) -> Option<Box<dyn AnyActor>> {
let factory = self.restart_factory.as_ref()?;
if self.restart_count >= self.max_restarts {
return None;
}
let fresh = factory();
Some(Box::new(Actor {
inner: fresh,
restart_factory: Some(factory.clone()),
max_restarts: self.max_restarts,
restart_count: self.restart_count + 1,
}))
}
fn on_start(&mut self, ctx: &Ctx) { fn on_start(&mut self, ctx: &Ctx) {
self.inner.on_start(ctx); self.inner.on_start(ctx);
} }
@ -174,6 +130,13 @@ where
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct MonitorRef(pub(crate) u64); pub struct MonitorRef(pub(crate) u64);
impl MonitorRef {
/// Construct a MonitorRef from a raw id. Used by extension crates.
pub fn from_raw(id: u64) -> Self {
Self(id)
}
}
/// Reason an actor was removed from the runtime. /// Reason an actor was removed from the runtime.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum StopReason { pub enum StopReason {
@ -228,6 +191,10 @@ pub(crate) enum TimerRequest {
} }
/// Object-safe inner trait for sending type-erased messages. /// Object-safe inner trait for sending type-erased messages.
///
/// Minimal core interface: send, spawn, stop, timers, and extension access.
/// Registry methods (naming, monitoring, groups) are provided by extension
/// traits in `swactor-std`.
#[allow(private_interfaces)] #[allow(private_interfaces)]
pub trait ContextInner { pub trait ContextInner {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>; fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
@ -236,26 +203,15 @@ pub trait ContextInner {
fn request_stop(&self, addr: ActorAddress); fn request_stop(&self, addr: ActorAddress);
/// Schedule a timer (one-shot or interval). /// Schedule a timer (one-shot or interval).
fn schedule_timer(&self, request: TimerRequest); fn schedule_timer(&self, request: TimerRequest);
/// Look up an actor address by registered name. /// Access the runtime extension (if installed).
fn where_is(&self, name: &str) -> Option<ActorAddress>; fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension>;
/// Register a name → address mapping. Returns `Err` if the name is taken.
fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error>;
/// Subscribe to death notifications for `target`. Returns a MonitorRef for cancellation.
fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> MonitorRef;
/// Cancel a monitor subscription.
fn demonitor(&self, mref: MonitorRef);
/// Add actor to a named group.
fn join_group(&self, actor: ActorAddress, group: String);
/// Remove actor from a named group.
fn leave_group(&self, actor: ActorAddress, group: &str);
/// Return all members of a named group.
fn group_members(&self, group: &str) -> Vec<ActorAddress>;
} }
/// Actor syscall interface — passed to `ActorInterface::handle()`. /// Actor syscall interface — passed to `ActorInterface::handle()`.
/// ///
/// Wraps a `&dyn ContextInner` to solve the object-safety problem while /// Wraps a `&dyn ContextInner` to solve the object-safety problem while
/// providing a typed public API. /// providing a typed public API. Registry methods (naming, monitoring, groups)
/// are provided by extension traits in `swactor-std`.
pub struct Ctx<'a> { pub struct Ctx<'a> {
inner: &'a dyn ContextInner, inner: &'a dyn ContextInner,
self_addr: ActorAddress, self_addr: ActorAddress,
@ -275,6 +231,11 @@ impl<'a> Ctx<'a> {
self.self_addr self.self_addr
} }
/// Access the runtime extension (if installed).
pub fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> {
self.inner.extension()
}
/// Send a typed message to an actor address. /// Send a typed message to an actor address.
pub fn send<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { pub fn send<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
self.inner.send_any(addr, Box::new(msg)) self.inner.send_any(addr, Box::new(msg))
@ -329,560 +290,4 @@ impl<'a> Ctx<'a> {
period, period,
}); });
} }
/// Look up an actor address by its registered name.
///
/// Returns `None` if no actor is registered under that name.
pub fn where_is(&self, name: &str) -> Option<ActorAddress> {
self.inner.where_is(name)
}
/// Spawn a new actor with a registered name.
///
/// The name is reserved immediately (before the actor starts processing).
/// Returns `Err` if the name is already taken.
pub fn spawn_named<A: ActorInterface>(
&self,
name: impl Into<String>,
actor: A,
) -> Result<ActorAddress, Error> {
let addr = ActorAddress::new_random();
self.inner.register_name(name.into(), addr)?;
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
self.inner.spawn_any(addr, boxed);
Ok(addr)
}
/// Subscribe to death notifications for `target`.
///
/// When `target` dies (stop or panic), a [`Down`] message is delivered to
/// this actor's mailbox as a normal message. Multiple monitors of the same
/// target create independent subscriptions.
pub fn monitor(&self, target: ActorAddress) -> MonitorRef {
self.inner.monitor(self.self_addr, target)
}
/// Cancel a previously created monitor subscription.
pub fn demonitor(&self, mref: MonitorRef) {
self.inner.demonitor(mref);
}
/// Join a named group. The group is created if it doesn't exist.
///
/// An actor can be a member of multiple groups simultaneously.
pub fn join_group(&self, group: impl Into<String>) {
self.inner.join_group(self.self_addr, group.into());
}
/// Leave a named group. Empty groups are automatically deleted.
pub fn leave_group(&self, group: &str) {
self.inner.leave_group(self.self_addr, group);
}
/// Broadcast a message to all members of a named group.
///
/// The message is cloned for each recipient. Returns the number of
/// messages successfully enqueued.
pub fn publish<M: Message>(&self, group: &str, msg: M) -> usize {
let members = self.inner.group_members(group);
let mut count = 0;
for member in &members {
if self.inner.send_any(*member, Box::new(msg.clone())).is_ok() {
count += 1;
}
}
count
}
/// Return all current members of a named group.
pub fn group_members(&self, group: &str) -> Vec<ActorAddress> {
self.inner.group_members(group)
}
/// Spawn a restartable actor. On panic, recreated via `factory` up to
/// `max_restarts` times before permanent poisoning.
pub fn spawn_restartable<A, F>(
&self,
actor: A,
factory: F,
max_restarts: u32,
) -> Result<ActorAddress, Error>
where
A: ActorInterface,
F: Fn() -> A + Send + Sync + 'static,
{
let addr = ActorAddress::new_random();
let boxed: Box<dyn AnyActor> = Box::new(Actor::new_restartable(
actor,
Arc::new(factory),
max_restarts,
));
self.inner.spawn_any(addr, boxed);
Ok(addr)
}
}
// ─── Supervision ────────────────────────────────────────────────────────────
/// How a child should be restarted when it dies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestartPolicy {
/// Always restart, regardless of stop reason.
Permanent,
/// Restart only on abnormal exit (Panicked). Normal stops are final.
Transient,
/// Never restart. The child is removed on any exit.
Temporary,
}
/// Strategy for handling child failures.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SupervisorStrategy {
/// Only restart the failed child. Other children are unaffected.
OneForOne,
/// Terminate all children and restart them all in spec order.
OneForAll,
/// Terminate children started after the failed child, then restart
/// the failed child and all terminated children in spec order.
RestForOne,
}
/// Specification for a supervised child actor.
///
/// The `start` closure is called with `&Ctx` and should spawn the child actor
/// (typically via `ctx.spawn()`). The supervisor monitors the returned address
/// and applies the restart policy when the child dies.
///
/// Children should be spawned with `ctx.spawn()`, not `ctx.spawn_restartable()`,
/// since the supervisor itself manages restarts.
pub struct ChildSpec {
/// Unique identifier for this child.
pub id: String,
/// How to restart this child.
pub restart: RestartPolicy,
/// Factory to spawn the child. Called with `&Ctx`, returns the child's address.
pub start: Arc<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync>,
}
impl ChildSpec {
pub fn new(
id: impl Into<String>,
restart: RestartPolicy,
start: impl Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync + 'static,
) -> Self {
Self {
id: id.into(),
restart,
start: Arc::new(start),
}
}
}
/// Tracked state for an active child within a supervisor.
struct ActiveChild {
addr: ActorAddress,
_monitor_ref: MonitorRef,
}
/// Internal phase for coordinating multi-child restart (OneForAll, RestForOne).
///
/// In `Normal` phase, the supervisor processes Down messages and applies the strategy.
/// When a coordinated restart is needed, it transitions to `Stopping` (sends stop
/// signals, waits for Down confirmations) then restarts all affected children.
enum SupervisorPhase {
/// Normal operation — process Down messages and apply strategy.
Normal,
/// Waiting for children to confirm death before restarting.
Stopping {
/// Children we're still waiting for Down confirmation.
awaiting: Vec<ActorAddress>,
/// Spec indices to restart once all confirmations received.
restart_set: Vec<usize>,
},
}
/// A supervisor actor that manages child actors according to a restart strategy.
///
/// Children are spawned during `on_start`. When a child dies, the supervisor
/// receives a [`Down`] notification via [`ActorInterface::handle_down`] and
/// applies the configured strategy and restart policy.
///
/// # Strategies
///
/// - **OneForOne**: Only the failed child is restarted.
/// - **OneForAll**: All children are stopped, then all restarted in spec order.
/// - **RestForOne**: The failed child and all children started after it are
/// stopped, then restarted in spec order.
///
/// # Restart Intensity
///
/// The supervisor tracks total restarts. When `total_restarts > max_restarts`,
/// the supervisor stops itself (meltdown protection), escalating the failure
/// to its own supervisor if one exists.
///
/// # Example
///
/// ```ignore
/// let sup = Supervisor::new(
/// SupervisorStrategy::OneForOne,
/// 5, // max 5 restarts before meltdown
/// vec![
/// ChildSpec::new("worker", RestartPolicy::Permanent, |ctx| {
/// ctx.spawn(MyWorker::new())
/// }),
/// ],
/// );
/// let sup_addr = rt.spawn(sup)?;
/// ```
pub struct Supervisor {
strategy: SupervisorStrategy,
max_restarts: u32,
specs: Vec<ChildSpec>,
children: Vec<Option<ActiveChild>>,
total_restarts: u32,
phase: SupervisorPhase,
}
impl Supervisor {
pub fn new(
strategy: SupervisorStrategy,
max_restarts: u32,
specs: Vec<ChildSpec>,
) -> Self {
let children = (0..specs.len()).map(|_| None).collect();
Self {
strategy,
max_restarts,
specs,
children,
total_restarts: 0,
phase: SupervisorPhase::Normal,
}
}
fn start_child(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> {
let addr = (self.specs[idx].start)(ctx)?;
let mref = ctx.monitor(addr);
self.children[idx] = Some(ActiveChild {
addr,
_monitor_ref: mref,
});
Ok(())
}
fn find_child_idx(&self, addr: ActorAddress) -> Option<usize> {
self.children
.iter()
.position(|c| c.as_ref().map_or(false, |ac| ac.addr == addr))
}
/// Check meltdown intensity — returns true if we should stop.
fn check_intensity(&mut self) -> bool {
self.total_restarts += 1;
self.total_restarts > self.max_restarts
}
/// Try to finish the coordinated restart: restart all children in `restart_set`.
fn finish_restart(&mut self, ctx: &Ctx) {
let restart_set = match &mut self.phase {
SupervisorPhase::Stopping { restart_set, .. } => {
std::mem::take(restart_set)
}
_ => return,
};
self.phase = SupervisorPhase::Normal;
for idx in restart_set {
if let Err(e) = self.start_child(ctx, idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[idx].id, e
);
}
}
}
/// Begin a coordinated restart for the given spec indices.
/// Stops any living children in the set, then waits for their Down messages.
fn begin_coordinated_restart(&mut self, ctx: &Ctx, restart_indices: Vec<usize>) {
let mut awaiting = Vec::new();
for &idx in &restart_indices {
if let Some(child) = self.children[idx].take() {
let _ = ctx.stop_actor(child.addr);
awaiting.push(child.addr);
}
}
if awaiting.is_empty() {
// All children already dead — restart immediately.
for idx in &restart_indices {
if let Err(e) = self.start_child(ctx, *idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[*idx].id, e
);
}
}
} else {
self.phase = SupervisorPhase::Stopping {
awaiting,
restart_set: restart_indices,
};
}
}
}
impl ActorInterface for Supervisor {
type Incoming = ();
type Response = ();
fn handle(&mut self, _ctx: &Ctx, _msg: ()) {}
fn on_start(&mut self, ctx: &Ctx) {
for idx in 0..self.specs.len() {
if let Err(e) = self.start_child(ctx, idx) {
eprintln!(
"swactor: supervisor failed to start child '{}': {}",
self.specs[idx].id, e
);
}
}
}
fn on_stop(&mut self, ctx: &Ctx) {
for child in self.children.iter().flatten() {
let _ = ctx.stop_actor(child.addr);
}
}
fn handle_down(&mut self, ctx: &Ctx, down: Down) {
// During coordinated restart: track Down confirmations.
if matches!(self.phase, SupervisorPhase::Stopping { .. }) {
// Clear from children tracking
if let Some(idx) = self.find_child_idx(down.addr) {
self.children[idx] = None;
}
// Remove from awaiting list
if let SupervisorPhase::Stopping { awaiting, .. } = &mut self.phase {
awaiting.retain(|a| *a != down.addr);
}
let done = matches!(&self.phase,
SupervisorPhase::Stopping { awaiting, .. } if awaiting.is_empty());
if done {
self.finish_restart(ctx);
}
return;
}
// Normal phase: handle child death.
let Some(idx) = self.find_child_idx(down.addr) else {
return;
};
self.children[idx] = None;
let should_restart = match self.specs[idx].restart {
RestartPolicy::Permanent => true,
RestartPolicy::Transient => down.reason == StopReason::Panicked,
RestartPolicy::Temporary => false,
};
if !should_restart {
return;
}
if self.check_intensity() {
eprintln!(
"swactor: supervisor reached max restarts ({}), shutting down",
self.max_restarts
);
ctx.stop_self();
return;
}
match self.strategy {
SupervisorStrategy::OneForOne => {
if let Err(e) = self.start_child(ctx, idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[idx].id, e
);
}
}
SupervisorStrategy::OneForAll => {
// Stop all other living children, then restart all in order.
let restart_indices: Vec<usize> = (0..self.specs.len()).collect();
self.begin_coordinated_restart(ctx, restart_indices);
}
SupervisorStrategy::RestForOne => {
// Stop children after the failed one, then restart failed + rest.
let restart_indices: Vec<usize> = (idx..self.specs.len()).collect();
self.begin_coordinated_restart(ctx, restart_indices);
}
}
}
}
// ---------------------------------------------------------------------------
// Router — pool of identical workers with configurable routing strategy
// ---------------------------------------------------------------------------
/// Strategy for distributing messages across pool workers.
#[derive(Debug, Clone)]
pub enum RoutingStrategy {
/// Sequential round-robin distribution.
RoundRobin,
/// Random worker selection.
Random,
/// Send to all workers (message is cloned to each).
Broadcast,
}
/// A router actor that manages a pool of identical workers and distributes
/// incoming messages across them according to a [`RoutingStrategy`].
///
/// Workers are spawned during `on_start`, monitored for failures, and
/// automatically replaced to maintain the target pool size. Meltdown
/// protection stops the router when total restarts exceed `max_restarts`.
///
/// # Example
///
/// ```ignore
/// let router = Router::new(
/// RoutingStrategy::RoundRobin,
/// 5,
/// |ctx| ctx.spawn(MyWorker::new()),
/// 10,
/// );
/// let router_addr = rt.spawn(router)?;
/// rt.send_to(router_addr, WorkerMessage::DoWork(42))?;
/// ```
pub struct Router<M: Message> {
strategy: RoutingStrategy,
pool_size: usize,
factory: Arc<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync>,
workers: Vec<Option<ActiveChild>>,
rr_index: usize,
total_restarts: u32,
max_restarts: u32,
_marker: PhantomData<M>,
}
impl<M: Message> Router<M> {
pub fn new(
strategy: RoutingStrategy,
pool_size: usize,
factory: impl Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync + 'static,
max_restarts: u32,
) -> Self {
Self {
strategy,
pool_size,
factory: Arc::new(factory),
workers: (0..pool_size).map(|_| None).collect(),
rr_index: 0,
total_restarts: 0,
max_restarts,
_marker: PhantomData,
}
}
fn start_worker(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> {
let addr = (self.factory)(ctx)?;
let mref = ctx.monitor(addr);
self.workers[idx] = Some(ActiveChild {
addr,
_monitor_ref: mref,
});
Ok(())
}
fn find_worker_idx(&self, addr: ActorAddress) -> Option<usize> {
self.workers
.iter()
.position(|w| w.as_ref().map_or(false, |ac| ac.addr == addr))
}
fn live_workers(&self) -> Vec<ActorAddress> {
self.workers
.iter()
.filter_map(|w| w.as_ref().map(|ac| ac.addr))
.collect()
}
fn select_one(&mut self) -> Option<ActorAddress> {
let live = self.live_workers();
if live.is_empty() {
return None;
}
match self.strategy {
RoutingStrategy::RoundRobin => {
let idx = self.rr_index % live.len();
self.rr_index = self.rr_index.wrapping_add(1);
Some(live[idx])
}
RoutingStrategy::Random => {
let mut buf = [0u8; 8];
crate::get_random(&mut buf);
let r = u64::from_ne_bytes(buf) as usize;
Some(live[r % live.len()])
}
RoutingStrategy::Broadcast => None, // handled separately
}
}
}
impl<M: Message> ActorInterface for Router<M> {
type Incoming = M;
type Response = ();
fn handle(&mut self, ctx: &Ctx, msg: M) {
match self.strategy {
RoutingStrategy::Broadcast => {
let live = self.live_workers();
for addr in live {
let _ = ctx.send(addr, msg.clone());
}
}
_ => {
if let Some(addr) = self.select_one() {
let _ = ctx.send(addr, msg);
}
}
}
}
fn on_start(&mut self, ctx: &Ctx) {
for idx in 0..self.pool_size {
if let Err(e) = self.start_worker(ctx, idx) {
eprintln!("swactor: router failed to start worker {idx}: {e}");
}
}
}
fn on_stop(&mut self, ctx: &Ctx) {
for child in self.workers.iter().flatten() {
let _ = ctx.stop_actor(child.addr);
}
}
fn handle_down(&mut self, ctx: &Ctx, down: Down) {
let Some(idx) = self.find_worker_idx(down.addr) else {
return;
};
self.workers[idx] = None;
self.total_restarts += 1;
if self.total_restarts > self.max_restarts {
eprintln!(
"swactor: router reached max restarts ({}), shutting down",
self.max_restarts
);
ctx.stop_self();
return;
}
if let Err(e) = self.start_worker(ctx, idx) {
eprintln!("swactor: router failed to restart worker {idx}: {e}");
}
}
} }

View file

@ -1,4 +1,3 @@
use std::sync::Arc; use std::sync::Arc;
use crossbeam_queue::{ArrayQueue, SegQueue}; use crossbeam_queue::{ArrayQueue, SegQueue};

View file

@ -1,11 +1,11 @@
use std::any::Any; use std::any::Any;
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::hash::{BuildHasher, Hasher}; use std::hash::{BuildHasher, Hasher};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock, RwLock}; use std::sync::{Arc, OnceLock, RwLock};
use std::thread::Thread; use std::thread::Thread;
use crate::actor::{ActorAddress, AnyActor, Message, MonitorRef}; use crate::actor::{ActorAddress, AnyActor, Message};
use crate::channel::Sender; use crate::channel::Sender;
use crate::config::RuntimeConfig; use crate::config::RuntimeConfig;
use crate::stats::WorkerStats; use crate::stats::WorkerStats;
@ -21,7 +21,7 @@ use crate::Error;
/// ///
/// This is safe because the input is already random (uniform distribution), /// This is safe because the input is already random (uniform distribution),
/// so additional mixing would be redundant. /// so additional mixing would be redundant.
pub(crate) struct AddrHasher(u64); pub struct AddrHasher(u64);
impl Hasher for AddrHasher { impl Hasher for AddrHasher {
#[inline] #[inline]
@ -42,7 +42,7 @@ impl Hasher for AddrHasher {
/// BuildHasher for creating AddrHasher instances. /// BuildHasher for creating AddrHasher instances.
#[derive(Default, Clone)] #[derive(Default, Clone)]
pub(crate) struct AddrBuildHasher; pub struct AddrBuildHasher;
impl BuildHasher for AddrBuildHasher { impl BuildHasher for AddrBuildHasher {
type Hasher = AddrHasher; type Hasher = AddrHasher;
@ -55,10 +55,10 @@ impl BuildHasher for AddrBuildHasher {
/// HashMap optimized for ActorAddress keys. /// HashMap optimized for ActorAddress keys.
/// Uses identity hashing since ActorAddress bytes are already random. /// Uses identity hashing since ActorAddress bytes are already random.
pub(crate) type AddrMap<V> = HashMap<ActorAddress, V, AddrBuildHasher>; pub type AddrMap<V> = HashMap<ActorAddress, V, AddrBuildHasher>;
/// HashSet optimized for ActorAddress keys. /// HashSet optimized for ActorAddress keys.
pub(crate) type AddrSet = HashSet<ActorAddress, AddrBuildHasher>; pub type AddrSet = HashSet<ActorAddress, AddrBuildHasher>;
// ─── Address Map Types ─────────────────────────────────────────────────────── // ─── Address Map Types ───────────────────────────────────────────────────────
@ -239,9 +239,7 @@ pub(crate) struct TickContext<'a> {
pub(crate) placement: &'a Placement, pub(crate) placement: &'a Placement,
pub(crate) inbox_registry: &'a InboxRegistry, pub(crate) inbox_registry: &'a InboxRegistry,
pub(crate) config: &'a RuntimeConfig, pub(crate) config: &'a RuntimeConfig,
pub(crate) name_registry: &'a NameRegistry, pub(crate) extension: Option<&'a dyn crate::extension::RuntimeExtension>,
pub(crate) monitor_registry: &'a MonitorRegistry,
pub(crate) group_registry: &'a GroupRegistry,
pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>, pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>,
/// Thread handles for waking parked workers on cross-worker sends. /// Thread handles for waking parked workers on cross-worker sends.
pub(crate) worker_threads: &'a [OnceLock<Thread>], pub(crate) worker_threads: &'a [OnceLock<Thread>],
@ -251,216 +249,6 @@ pub(crate) struct TickContext<'a> {
pub(crate) transport_router: Option<&'a crate::transport::TransportRouter>, pub(crate) transport_router: Option<&'a crate::transport::TransportRouter>,
} }
// ─── Name Registry ──────────────────────────────────────────────────────────
/// Named actor registry — maps human-readable names to actor addresses.
///
/// `RwLock<HashMap>` — same pattern as `AddressMap`. Write-rare (spawn/death),
/// read-often (lookup). A reverse map enables O(1) cleanup on actor death.
pub(crate) struct NameRegistry {
names: RwLock<HashMap<String, ActorAddress>>,
reverse: RwLock<AddrMap<String>>,
}
impl NameRegistry {
pub fn new() -> Self {
Self {
names: RwLock::new(HashMap::new()),
reverse: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
}
}
/// Register a name → address mapping. Returns `Err` if the name is already taken.
pub fn register(&self, name: String, addr: ActorAddress) -> Result<(), crate::Error> {
let mut names = self.names.write().unwrap();
if names.contains_key(&name) {
return Err(crate::Error::from("Name already registered"));
}
names.insert(name.clone(), addr);
drop(names);
self.reverse.write().unwrap().insert(addr, name);
Ok(())
}
/// Look up an actor address by name.
pub fn lookup(&self, name: &str) -> Option<ActorAddress> {
self.names.read().unwrap().get(name).copied()
}
/// Unregister a name, returning the address it was bound to.
pub fn unregister(&self, name: &str) -> Option<ActorAddress> {
let addr = self.names.write().unwrap().remove(name)?;
self.reverse.write().unwrap().remove(&addr);
Some(addr)
}
/// Remove a name by address (called on actor death for auto-cleanup).
pub fn unregister_by_addr(&self, addr: &ActorAddress) {
if let Some(name) = self.reverse.write().unwrap().remove(addr) {
self.names.write().unwrap().remove(&name);
}
}
/// Return all registered names.
pub fn registered_names(&self) -> Vec<String> {
self.names.read().unwrap().keys().cloned().collect()
}
}
// ─── Monitor Registry ────────────────────────────────────────────────────────
/// Tracks monitor subscriptions: watched actor → list of (MonitorRef, watcher address).
///
/// Write-rare (monitor/demonitor/death), read at cleanup time.
pub(crate) struct MonitorRegistry {
/// watched_addr → [(mref, watcher_addr)]
monitors: RwLock<AddrMap<Vec<(MonitorRef, ActorAddress)>>>,
/// mref → watched_addr (for O(1) demonitor)
ref_to_target: RwLock<HashMap<MonitorRef, ActorAddress>>,
next_ref: AtomicU64,
}
impl MonitorRegistry {
pub fn new() -> Self {
Self {
monitors: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
ref_to_target: RwLock::new(HashMap::new()),
next_ref: AtomicU64::new(1),
}
}
/// Register a monitor: `watcher` wants to know when `target` dies.
pub fn register(&self, watcher: ActorAddress, target: ActorAddress) -> MonitorRef {
let id = self.next_ref.fetch_add(1, Ordering::Relaxed);
let mref = MonitorRef(id);
self.monitors.write().unwrap()
.entry(target)
.or_default()
.push((mref, watcher));
self.ref_to_target.write().unwrap().insert(mref, target);
mref
}
/// Cancel a monitor by its ref.
pub fn deregister(&self, mref: MonitorRef) {
if let Some(target) = self.ref_to_target.write().unwrap().remove(&mref) {
let mut monitors = self.monitors.write().unwrap();
if let Some(watchers) = monitors.get_mut(&target) {
watchers.retain(|(r, _)| *r != mref);
if watchers.is_empty() {
monitors.remove(&target);
}
}
}
}
/// Remove and return all monitors for a dead actor.
pub fn take_monitors(&self, target: &ActorAddress) -> Vec<(MonitorRef, ActorAddress)> {
let watchers = self.monitors.write().unwrap().remove(target).unwrap_or_default();
let mut ref_map = self.ref_to_target.write().unwrap();
for (mref, _) in &watchers {
ref_map.remove(mref);
}
watchers
}
/// Remove all monitor subscriptions where `addr` is the watcher (dead watcher cleanup).
pub fn remove_watcher(&self, addr: &ActorAddress) {
let mut monitors = self.monitors.write().unwrap();
let mut ref_map = self.ref_to_target.write().unwrap();
// Iterate all targets and remove entries where this addr is the watcher
monitors.retain(|_target, watchers| {
watchers.retain(|(mref, watcher)| {
if watcher == addr {
ref_map.remove(mref);
false
} else {
true
}
});
!watchers.is_empty()
});
}
}
// ─── Group Registry ─────────────────────────────────────────────────────────
/// Actor groups (pub-sub). Actors join/leave named groups; messages can be
/// broadcast to all members of a group.
///
/// 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, AddrSet>>,
/// actor_addr → set of group names (reverse map for O(G) cleanup on death)
memberships: RwLock<AddrMap<HashSet<String>>>,
}
impl GroupRegistry {
pub fn new() -> Self {
Self {
groups: RwLock::new(HashMap::new()),
memberships: RwLock::new(HashMap::with_hasher(AddrBuildHasher)),
}
}
/// Add an actor to a named group. Group is created if it doesn't exist.
pub fn join(&self, group: String, addr: ActorAddress) {
self.groups.write().unwrap()
.entry(group.clone())
.or_insert_with(|| HashSet::with_hasher(AddrBuildHasher))
.insert(addr);
self.memberships.write().unwrap()
.entry(addr)
.or_default()
.insert(group);
}
/// Remove an actor from a named group. Empty groups are auto-deleted.
pub fn leave(&self, group: &str, addr: &ActorAddress) {
let mut groups = self.groups.write().unwrap();
if let Some(members) = groups.get_mut(group) {
members.remove(addr);
if members.is_empty() {
groups.remove(group);
}
}
drop(groups);
if let Some(membership) = self.memberships.write().unwrap().get_mut(addr) {
membership.remove(group);
}
}
/// Return all members of a group.
pub fn members(&self, group: &str) -> Vec<ActorAddress> {
self.groups.read().unwrap()
.get(group)
.map(|s| s.iter().copied().collect())
.unwrap_or_default()
}
/// Remove a dead actor from all its groups.
pub fn cleanup(&self, addr: &ActorAddress) {
let group_names = self.memberships.write().unwrap().remove(addr);
if let Some(names) = group_names {
let mut groups = self.groups.write().unwrap();
for name in names {
if let Some(members) = groups.get_mut(&name) {
members.remove(addr);
if members.is_empty() {
groups.remove(&name);
}
}
}
}
}
/// Return all active group names.
pub fn group_names(&self) -> Vec<String> {
self.groups.read().unwrap().keys().cloned().collect()
}
}
impl<'a> TickContext<'a> { impl<'a> TickContext<'a> {
/// Route a message whose destination is not in the local address map. /// Route a message whose destination is not in the local address map.
/// Tries inbox registry, then remote transport, then falls back to inbox error. /// Tries inbox registry, then remote transport, then falls back to inbox error.

28
src/extension.rs Normal file
View file

@ -0,0 +1,28 @@
use std::any::Any;
use crate::actor::{ActorAddress, StopReason};
/// Extension hook for runtime lifecycle events.
///
/// Stored as `Arc<dyn RuntimeExtension>` in the Runtime. Workers access it
/// via TickContext. Ctx methods that need registry access downcast to the
/// concrete type via `as_any()`.
///
/// Core calls these methods at appropriate tick phases:
/// - `on_actor_death`: called during phase 7 (cleanup_dead) with newly dead actors
/// - `cleanup_dead`: called during phase 7 to clean up extension state
pub trait RuntimeExtension: Send + Sync {
/// Called during phase 7 (cleanup_dead) for each dead actor.
/// Returns (destination, message) pairs for death notifications.
/// The core delivers these through normal routing (pending_local or transfer queue).
fn on_actor_death(
&self,
dead: &[(ActorAddress, StopReason)],
) -> Vec<(ActorAddress, Box<dyn Any + Send>)>;
/// Clean up extension state for dead actors (names, groups, monitors).
fn cleanup_dead(&self, dead: &[ActorAddress]);
/// Downcast support for Ctx extension traits.
fn as_any(&self) -> &dyn Any;
}

View file

@ -1,10 +1,13 @@
pub mod actor; pub mod actor;
pub mod extension;
pub mod worker; pub mod worker;
pub(crate) mod channel; pub(crate) mod channel;
pub(crate) mod error; pub(crate) mod error;
pub use error::Error; pub use error::Error;
// Re-export identity hashing types for ActorAddress-keyed collections.
pub use delivery::{AddrBuildHasher, AddrMap, AddrSet};
pub mod config; pub mod config;
pub(crate) mod delivery; pub(crate) mod delivery;

View file

@ -9,7 +9,8 @@ use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopS
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, MailboxOverflow, RuntimeConfig}; pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
use crate::delivery::{AddressMap, Envelope, GroupRegistry, InboxRegistry, MonitorRegistry, NameRegistry, Placement, TickContext, WorkerId}; use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId};
use crate::extension::RuntimeExtension;
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
pub use crate::stats::{RuntimeStats, WorkerInfo}; pub use crate::stats::{RuntimeStats, WorkerInfo};
@ -95,9 +96,7 @@ pub struct Runtime {
config: RuntimeConfig, config: RuntimeConfig,
address_map: Arc<AddressMap>, address_map: Arc<AddressMap>,
inbox_registry: Arc<InboxRegistry>, inbox_registry: Arc<InboxRegistry>,
name_registry: Arc<NameRegistry>, extension: Option<Arc<dyn RuntimeExtension>>,
monitor_registry: Arc<MonitorRegistry>,
group_registry: Arc<GroupRegistry>,
transfer_txs: Vec<Sender<Envelope>>, transfer_txs: Vec<Sender<Envelope>>,
spawn_txs: Vec<Sender<(ActorAddress, Box<dyn AnyActor>)>>, spawn_txs: Vec<Sender<(ActorAddress, Box<dyn AnyActor>)>>,
placement: Placement, placement: Placement,
@ -155,9 +154,6 @@ impl Runtime {
let address_map = Arc::new(AddressMap::with_capacity(config.max_actors)); let address_map = Arc::new(AddressMap::with_capacity(config.max_actors));
let inbox_registry = Arc::new(InboxRegistry::new()); let inbox_registry = Arc::new(InboxRegistry::new());
let name_registry = Arc::new(NameRegistry::new());
let monitor_registry = Arc::new(MonitorRegistry::new());
let group_registry = Arc::new(GroupRegistry::new());
let mut transfer_txs = Vec::with_capacity(num_workers); let mut transfer_txs = Vec::with_capacity(num_workers);
let mut spawn_txs = Vec::with_capacity(num_workers); let mut spawn_txs = Vec::with_capacity(num_workers);
@ -195,9 +191,7 @@ impl Runtime {
config, config,
address_map, address_map,
inbox_registry, inbox_registry,
name_registry, extension: None,
monitor_registry,
group_registry,
transfer_txs, transfer_txs,
spawn_txs, spawn_txs,
placement, placement,
@ -242,96 +236,18 @@ impl Runtime {
Ok(addr) Ok(addr)
} }
/// Spawn a restartable actor. On panic, the actor is recreated using `factory` /// Install a runtime extension. Extensions provide higher-level features
/// up to `max_restarts` times before being permanently poisoned. /// (naming, monitoring, groups) via lifecycle hooks.
/// The mailbox is cleared on each restart — the new instance starts fresh.
pub fn spawn_restartable<A, F>(
&self,
actor: A,
factory: F,
max_restarts: u32,
) -> Result<ActorAddress, Error>
where
A: ActorInterface,
F: Fn() -> A + Send + Sync + 'static,
{
let addr = ActorAddress::new_random();
let worker_id = self.placement.next_worker();
self.address_map.insert(addr, worker_id);
let boxed: Box<dyn AnyActor> = Box::new(Actor::new_restartable(
actor,
std::sync::Arc::new(factory),
max_restarts,
));
self.spawn_txs[worker_id.as_usize()]
.send((addr, boxed));
Ok(addr)
}
/// Spawn an actor with a registered name, returns its address.
/// ///
/// The name is reserved immediately. Returns `Err` if the name is already taken. /// Must be called before `run()` or `tick()`.
pub fn spawn_named<A: ActorInterface>( pub fn with_extension(mut self, ext: Arc<dyn RuntimeExtension>) -> Self {
&self, self.extension = Some(ext);
name: impl Into<String>, self
actor: A,
) -> Result<ActorAddress, Error> {
let addr = ActorAddress::new_random();
self.name_registry.register(name.into(), addr)?;
let worker_id = self.placement.next_worker();
self.address_map.insert(addr, worker_id);
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
self.spawn_txs[worker_id.as_usize()].send((addr, boxed));
Ok(addr)
} }
/// Look up an actor address by its registered name. /// Access the installed runtime extension (if any).
pub fn where_is(&self, name: &str) -> Option<ActorAddress> { pub fn extension(&self) -> Option<&dyn RuntimeExtension> {
self.name_registry.lookup(name) self.extension.as_deref()
}
/// Unregister a name. Returns the address it was bound to, or `None`.
pub fn unregister(&self, name: &str) -> Option<ActorAddress> {
self.name_registry.unregister(name)
}
/// Return all currently registered actor names.
pub fn registered_names(&self) -> Vec<String> {
self.name_registry.registered_names()
}
/// Add an actor to a named group. The group is created if it doesn't exist.
pub fn join_group(&self, addr: ActorAddress, group: impl Into<String>) {
self.group_registry.join(group.into(), addr);
}
/// Remove an actor from a named group. Empty groups are auto-deleted.
pub fn leave_group(&self, addr: ActorAddress, group: &str) {
self.group_registry.leave(group, &addr);
}
/// Broadcast a message to all members of a named group.
///
/// Returns the number of messages successfully enqueued.
pub fn publish_to<M: Message>(&self, group: &str, msg: M) -> usize {
let members = self.group_registry.members(group);
let mut count = 0;
for member in &members {
if self.send_to(*member, msg.clone()).is_ok() {
count += 1;
}
}
count
}
/// Return all current members of a named group.
pub fn group_members(&self, group: &str) -> Vec<ActorAddress> {
self.group_registry.members(group)
}
/// Return all active group names.
pub fn groups(&self) -> Vec<String> {
self.group_registry.group_names()
} }
/// Send a request and get a handle for the response. /// Send a request and get a handle for the response.
@ -385,9 +301,7 @@ impl Runtime {
placement: &self.placement, placement: &self.placement,
inbox_registry: &self.inbox_registry, inbox_registry: &self.inbox_registry,
config: &self.config, config: &self.config,
name_registry: &self.name_registry, extension: self.extension.as_deref(),
monitor_registry: &self.monitor_registry,
group_registry: &self.group_registry,
stats_hook: self.stats_hook.as_deref(), stats_hook: self.stats_hook.as_deref(),
worker_threads: &self.worker_threads, worker_threads: &self.worker_threads,
#[cfg(feature = "transport")] #[cfg(feature = "transport")]
@ -589,31 +503,7 @@ impl ContextInner for Runtime {
eprintln!("swactor: schedule_timer called outside worker context — ignored"); eprintln!("swactor: schedule_timer called outside worker context — ignored");
} }
fn where_is(&self, name: &str) -> Option<ActorAddress> { fn extension(&self) -> Option<&dyn RuntimeExtension> {
self.name_registry.lookup(name) self.extension.as_deref()
}
fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error> {
self.name_registry.register(name, addr)
}
fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> crate::actor::MonitorRef {
self.monitor_registry.register(watcher, target)
}
fn demonitor(&self, mref: crate::actor::MonitorRef) {
self.monitor_registry.deregister(mref);
}
fn join_group(&self, actor: ActorAddress, group: String) {
self.group_registry.join(group, actor);
}
fn leave_group(&self, actor: ActorAddress, group: &str) {
self.group_registry.leave(group, &actor);
}
fn group_members(&self, group: &str) -> Vec<ActorAddress> {
self.group_registry.members(group)
} }
} }

View file

@ -321,9 +321,35 @@ impl Worker {
if !dead.is_empty() { if !dead.is_empty() {
for &(addr, _) in &dead { for &(addr, _) in &dead {
tc.address_map.remove(&addr); tc.address_map.remove(&addr);
tc.name_registry.unregister_by_addr(&addr);
tc.group_registry.cleanup(&addr);
} }
if let Some(ext) = tc.extension {
// Get death notifications (monitors) before cleaning up state
let notifications = ext.on_actor_death(&dead);
// Clean up extension state (names, groups, dead watcher monitors)
let dead_addrs: Vec<_> = dead.iter().map(|(a, _)| *a).collect();
ext.cleanup_dead(&dead_addrs);
// Deliver Down notifications through normal routing
for (dest, msg) in notifications {
if self.pool.contains(&dest) {
self.pool.deliver(&dest, msg);
} else {
match tc.address_map.lookup(&dest) {
Some(wid) => {
tc.transfer_txs[wid.as_usize()]
.send(Envelope::new(dest, msg));
crate::runtime::notify_worker(tc.worker_threads, wid.as_usize());
}
None => {
let _ = tc.inbox_registry.try_deliver(dest, msg);
}
}
}
}
}
// Re-publish num_actors after cleanup so stats reflect removal // Re-publish num_actors after cleanup so stats reflect removal
self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed);
did_work = true; did_work = true;
@ -335,31 +361,6 @@ impl Worker {
self.pool.deliver(&addr, msg); self.pool.deliver(&addr, msg);
} }
// Emit Down notifications for monitored dead actors
for &(addr, reason) in &dead {
let watchers = tc.monitor_registry.take_monitors(&addr);
for (_mref, watcher) in watchers {
let down = crate::actor::Down { addr, reason };
// Route through normal delivery path
if self.pool.contains(&watcher) {
self.pool.deliver(&watcher, Box::new(down));
} else {
match tc.address_map.lookup(&watcher) {
Some(wid) => {
tc.transfer_txs[wid.as_usize()]
.send(Envelope::new(watcher, Box::new(down)));
crate::runtime::notify_worker(tc.worker_threads, wid.as_usize());
}
None => {
let _ = tc.inbox_registry.try_deliver(watcher, Box::new(down));
}
}
}
}
// Clean up any monitors the dead actor had placed on others
tc.monitor_registry.remove_watcher(&addr);
}
// GC orphaned interval timers for actors that were just removed // GC orphaned interval timers for actors that were just removed
let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect(); let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect();
self.timers.gc_dead_intervals(&dead_addrs); self.timers.gc_dead_intervals(&dead_addrs);
@ -447,32 +448,8 @@ impl ContextInner for WorkerContext<'_> {
self.timer_requests.borrow_mut().push(request); self.timer_requests.borrow_mut().push(request);
} }
fn where_is(&self, name: &str) -> Option<ActorAddress> { fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> {
self.tc.name_registry.lookup(name) self.tc.extension
}
fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), crate::Error> {
self.tc.name_registry.register(name, addr)
}
fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> crate::actor::MonitorRef {
self.tc.monitor_registry.register(watcher, target)
}
fn demonitor(&self, mref: crate::actor::MonitorRef) {
self.tc.monitor_registry.deregister(mref);
}
fn join_group(&self, actor: ActorAddress, group: String) {
self.tc.group_registry.join(group, actor);
}
fn leave_group(&self, actor: ActorAddress, group: &str) {
self.tc.group_registry.leave(group, &actor);
}
fn group_members(&self, group: &str) -> Vec<ActorAddress> {
self.tc.group_registry.members(group)
} }
} }
@ -625,20 +602,10 @@ impl ActorPool {
Err(_) => { Err(_) => {
stats.panics.fetch_add(1, Ordering::Relaxed); stats.panics.fetch_add(1, Ordering::Relaxed);
slot.mailbox.clear(); slot.mailbox.clear();
// Try restart before poisoning
if let Some(fresh_actor) = slot.actor.try_restart() {
slot.actor = fresh_actor;
slot.started = false; // on_start will be called on next tick
stats.restarts.fetch_add(1, Ordering::Relaxed);
eprintln!("swactor: actor {addr} panicked — restarted");
#[cfg(feature = "tracing")]
tracing::warn!(actor_addr = %addr, "actor.restarted");
} else {
eprintln!("swactor: actor {addr} panicked — poisoned, future messages will be discarded"); eprintln!("swactor: actor {addr} panicked — poisoned, future messages will be discarded");
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::error!(actor_addr = %addr, "actor.panicked"); tracing::error!(actor_addr = %addr, "actor.panicked");
slot.poisoned = true; slot.poisoned = true;
}
break; break;
} }
Ok(Some(type_name)) => { Ok(Some(type_name)) => {

File diff suppressed because it is too large Load diff