feat: worker thread api (#6)
Make the worker thread api clearly seperated and ready for test harness
This commit is contained in:
Zachery Aaron Shores-Chmielewski
parent
29bb51004f
commit
d3fdf9e550
10 changed files with 1462 additions and 106 deletions
|
|
@ -25,3 +25,7 @@ criterion = { version = "0.5", features = ["html_reports"] }
|
|||
[[bench]]
|
||||
name = "runtime_benchmarks"
|
||||
harness = false
|
||||
|
||||
[[bench]]
|
||||
name = "worker_benchmarks"
|
||||
harness = false
|
||||
|
|
|
|||
153
benches/worker_benchmarks.rs
Normal file
153
benches/worker_benchmarks.rs
Normal file
|
|
@ -0,0 +1,153 @@
|
|||
use criterion::{
|
||||
criterion_group, criterion_main, BenchmarkId, Criterion, Throughput,
|
||||
};
|
||||
use swactor::worker::Mailbox;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Push throughput
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn mailbox_push(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("mailbox_push");
|
||||
for n in [100, 1_000, 10_000] {
|
||||
group.throughput(Throughput::Elements(n as u64));
|
||||
group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
|
||||
b.iter(|| {
|
||||
let mut mb: Mailbox<u64> = Mailbox::new(n);
|
||||
for i in 0..n {
|
||||
mb.push(i as u64);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Pop throughput
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn mailbox_pop(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("mailbox_pop");
|
||||
for n in [100, 1_000, 10_000] {
|
||||
group.throughput(Throughput::Elements(n as u64));
|
||||
group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
|
||||
b.iter_batched(
|
||||
|| {
|
||||
let mut mb: Mailbox<u64> = Mailbox::new(n);
|
||||
for i in 0..n {
|
||||
mb.push(i as u64);
|
||||
}
|
||||
mb
|
||||
},
|
||||
|mut mb| {
|
||||
for _ in 0..n {
|
||||
std::hint::black_box(mb.pop());
|
||||
}
|
||||
},
|
||||
criterion::BatchSize::SmallInput,
|
||||
);
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Interleaved push+pop
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn mailbox_interleaved(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("mailbox_interleaved");
|
||||
for n in [100, 1_000, 10_000] {
|
||||
group.throughput(Throughput::Elements(n as u64 * 2));
|
||||
group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
|
||||
b.iter(|| {
|
||||
let mut mb: Mailbox<u64> = Mailbox::new(n);
|
||||
for i in 0..n {
|
||||
mb.push(i as u64);
|
||||
std::hint::black_box(mb.pop());
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// drain_count O(1) verification
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn mailbox_drain_count(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("mailbox_drain_count");
|
||||
|
||||
// Below waterlevel
|
||||
group.bench_function("below", |b| {
|
||||
let mut mb: Mailbox<u64> = Mailbox::new(100);
|
||||
for i in 0..50 {
|
||||
mb.push(i);
|
||||
}
|
||||
b.iter(|| std::hint::black_box(mb.drain_count()));
|
||||
});
|
||||
|
||||
// At waterlevel
|
||||
group.bench_function("at", |b| {
|
||||
let mut mb: Mailbox<u64> = Mailbox::new(100);
|
||||
for i in 0..100 {
|
||||
mb.push(i);
|
||||
}
|
||||
b.iter(|| std::hint::black_box(mb.drain_count()));
|
||||
});
|
||||
|
||||
// Above waterlevel
|
||||
group.bench_function("above", |b| {
|
||||
let mut mb: Mailbox<u64> = Mailbox::new(100);
|
||||
for i in 0..500 {
|
||||
mb.push(i);
|
||||
}
|
||||
b.iter(|| std::hint::black_box(mb.drain_count()));
|
||||
});
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Simulated actor tick: drain_count + pop N
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn mailbox_actor_tick(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("mailbox_actor_tick");
|
||||
|
||||
for (wl, fill) in [(10, 5), (10, 10), (10, 50), (100, 200)] {
|
||||
let param = format!("wl={wl},fill={fill}");
|
||||
group.bench_function(BenchmarkId::from_parameter(¶m), |b| {
|
||||
b.iter_batched(
|
||||
|| {
|
||||
let mut mb: Mailbox<u64> = Mailbox::new(wl);
|
||||
for i in 0..fill {
|
||||
mb.push(i as u64);
|
||||
}
|
||||
mb
|
||||
},
|
||||
|mut mb| {
|
||||
let n = mb.drain_count();
|
||||
for _ in 0..n {
|
||||
std::hint::black_box(mb.pop());
|
||||
}
|
||||
},
|
||||
criterion::BatchSize::SmallInput,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
criterion_group!(
|
||||
benches,
|
||||
mailbox_push,
|
||||
mailbox_pop,
|
||||
mailbox_interleaved,
|
||||
mailbox_drain_count,
|
||||
mailbox_actor_tick,
|
||||
);
|
||||
criterion_main!(benches);
|
||||
550
docs/worker-thread.md
Normal file
550
docs/worker-thread.md
Normal file
|
|
@ -0,0 +1,550 @@
|
|||
# Worker Thread Architecture
|
||||
|
||||
## Structure
|
||||
|
||||
```
|
||||
┌─ Worker ───────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ id: WorkerId │
|
||||
│ │
|
||||
│ ┌─ spawn_rx ─────────────────────┐ ┌─ transfer_rx ──────────────────┐│
|
||||
│ │ Receiver<(Addr, Box<AnyActor>)>│ │ Receiver<Envelope> ││
|
||||
│ │ │ │ ││
|
||||
│ │ from: Runtime.spawn() │ │ from: other workers, Runtime ││
|
||||
│ │ ctx.spawn() │ │ ctx.send() ││
|
||||
│ └────────────────────────────────┘ └────────────────────────────────┘│
|
||||
│ │
|
||||
│ ┌─ ActorPool ──────────────────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ actors: HashMap<ActorAddress, ActorSlot> │ │
|
||||
│ │ │ │
|
||||
│ │ ┌─ ActorSlot [addr_0] ──────────────────────────────────────┐ │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ ┌─ mailbox ──────────────────────────────────────────┐ │ │ │
|
||||
│ │ │ │ VecDeque<Box<dyn Any + Send>> │ │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │
|
||||
│ │ │ │ │ msg │ │ msg │ │ msg │ │ ... │ <- push_back │ │ │ │
|
||||
│ │ │ │ └─────┘ └─────┘ └─────┘ └─────┘ │ │ │ │
|
||||
│ │ │ │ pop_front -> untyped; Box<Any> │ │ │ │
|
||||
│ │ │ └────────────────────────────────────────────────────┘ │ │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ ┌─ actor ────────────────────────────────────────────┐ │ │ │
|
||||
│ │ │ │ Box<dyn AnyActor> │ │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ │ wraps Actor<A>(A) where A: ActorInterface │ │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ │ handle_any(ctx, msg): │ │ │ │
|
||||
│ │ │ │ downcast Box<Any> -> A::Incoming │ │ │ │
|
||||
│ │ │ │ ok -> A.handle(ctx, typed_msg) │ │ │ │
|
||||
│ │ │ │ err -> silently drop │ │ │ │
|
||||
│ │ │ └────────────────────────────────────────────────────┘ │ │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ └───────────────────────────────────────────────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ │ ┌─ ActorSlot [addr_1] ──────────────────────────────────────┐ │ │
|
||||
│ │ │ ... │ │ │
|
||||
│ │ └───────────────────────────────────────────────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Shared State (borrowed via TickContext)
|
||||
|
||||
Lives on `Arc<Runtime>`, shared read-only across all worker threads.
|
||||
|
||||
```
|
||||
┌─ TickContext<'a> ──────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ address_map: &AddressMap -- ActorAddress -> WorkerId lookup │
|
||||
│ transfer_txs: &[Sender] -- one Sender per worker (cross-send) │
|
||||
│ spawn_txs: &[Sender] -- one Sender per worker (spawn reqs) │
|
||||
│ placement: &Placement -- round-robin next-worker picker │
|
||||
│ inbox_registry: &InboxRegistry -- external Inbox<M> receivers │
|
||||
│ config: &RuntimeConfig -- waterlevel, backoff params, etc. │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Run Loop
|
||||
|
||||
```
|
||||
┌─ Worker::run ──────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ ┌────────────────────────────────────┐ │
|
||||
│ │ is_running.load() ? │ │
|
||||
│ └──────────┬─────────────────────────┘ │
|
||||
│ yes │ │
|
||||
│ v │
|
||||
│ ┌────────────────────────────────────┐ │
|
||||
│ │ tick_once(&tc) │─────────┐ │
|
||||
│ └──────────┬─────────────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ ┌────┴────┐ │ │
|
||||
│ v v │ │
|
||||
│ did work no work │ │
|
||||
│ │ │ │ │
|
||||
│ v v │ │
|
||||
│ idle = 0 idle++ │ │
|
||||
│ │ │ │ │
|
||||
│ │ ┌────┴────────────────────────┐ │ │
|
||||
│ │ │ idle < spin_thr: spin │ │ │
|
||||
│ │ │ idle < yield_thr: yield │ │ │
|
||||
│ │ │ else: sleep(incr, capped) │ │ │
|
||||
│ │ └─────────────────────────┬───┘ │ │
|
||||
│ │ │ │ │
|
||||
│ └──────────┬───────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ └──── loop back ───────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Tick Once (four phases)
|
||||
|
||||
```
|
||||
┌─ tick_once ────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ PHASE 1 --- Drain Spawn Queue │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ spawn_rx --try_recv()--> (addr, Box<dyn AnyActor>) ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ pool.insert(addr, actor) ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ ActorSlot { ││
|
||||
│ │ mailbox: VecDeque::new() ││
|
||||
│ │ actor: <the new actor> ││
|
||||
│ │ } ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │ │
|
||||
│ v │
|
||||
│ PHASE 2 --- Drain Transfer Queue │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ transfer_rx --try_recv()--> Envelope { dest, payload } ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ pool.deliver(&dest, payload) ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ slot.mailbox.push_back(msg) ││
|
||||
│ │ (untyped; type check at handle time) ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │ │
|
||||
│ v │
|
||||
│ PHASE 3 --- Tick All Actors │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ ┌─ WorkerContext (on stack) ─────────────────────────────────┐ ││
|
||||
│ │ │ implements ContextInner │ ││
|
||||
│ │ │ owns pending_local: RefCell<Vec<(Addr, Box<Any>)>> │ ││
|
||||
│ │ └────────────────────────────────────────────────────────────┘ ││
|
||||
│ │ ││
|
||||
│ │ for each (addr, slot) in pool: ││
|
||||
│ │ ││
|
||||
│ │ ┌─ drain_count ──────────────────────────────────────────┐ ││
|
||||
│ │ │ len = slot.mailbox.len() │ ││
|
||||
│ │ │ len < waterlevel --> n = len (drain all) │ ││
|
||||
│ │ │ len >= waterlevel --> n = len / 2 (backpressure) │ ││
|
||||
│ │ └────────────────────────────────────────────────────────┘ ││
|
||||
│ │ ││
|
||||
│ │ ctx = Ctx { inner: &worker_ctx, self_addr: addr } ││
|
||||
│ │ ││
|
||||
│ │ repeat n times: ││
|
||||
│ │ msg = slot.mailbox.pop_front() ││
|
||||
│ │ slot.actor.handle_any(&ctx, msg) ││
|
||||
│ │ │ ││
|
||||
│ │ │ actor calls ctx.send() or ctx.spawn() ││
|
||||
│ │ v ││
|
||||
│ │ ││
|
||||
│ │ ┌─ WorkerContext routes ─────────────────────────────────────┐ ││
|
||||
│ │ │ │ ││
|
||||
│ │ │ send_any(addr, msg): │ ││
|
||||
│ │ │ ┌──────────────┬──────────────┬─────────────────┐ │ ││
|
||||
│ │ │ │ same worker │ other worker │ unknown addr │ │ ││
|
||||
│ │ │ │ │ │ │ │ ││
|
||||
│ │ │ │ pending_ │ transfer_tx │ inbox_registry │ │ ││
|
||||
│ │ │ │ local.push()│ [wid].send()│ .try_deliver() │ │ ││
|
||||
│ │ │ └──────────────┴──────────────┴─────────────────┘ │ ││
|
||||
│ │ │ │ ││
|
||||
│ │ │ spawn_any(addr, actor): │ ││
|
||||
│ │ │ wid = placement.next_worker() │ ││
|
||||
│ │ │ address_map.insert(addr, wid) │ ││
|
||||
│ │ │ spawn_txs[wid].send((addr, actor)) │ ││
|
||||
│ │ │ │ ││
|
||||
│ │ └────────────────────────────────────────────────────────────┘ ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │ │
|
||||
│ v │
|
||||
│ PHASE 4 --- Drain Pending Local │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ for (addr, msg) in pending_local.into_inner(): ││
|
||||
│ │ pool.deliver(&addr, msg) ││
|
||||
│ │ --> slot.mailbox.push_back(msg) ││
|
||||
│ │ ││
|
||||
│ │ these sit in the mailbox until NEXT tick ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## External Interactions
|
||||
|
||||
Everything the worker talks to, and everything that talks to it.
|
||||
|
||||
### Who writes into the worker's queues
|
||||
|
||||
```
|
||||
┌─ User Code ────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ let rt = Runtime::new(config); │
|
||||
│ let addr = rt.spawn(my_actor)?; // --┐ │
|
||||
│ rt.send_to(addr, MyMsg(42))?; // --┤ │
|
||||
│ // │ │
|
||||
└──────────────────────────────────┼──┼───────────────────────────────────┘
|
||||
│ │
|
||||
┌───────────────────────────┘ │
|
||||
│ │
|
||||
v v
|
||||
┌─ Runtime ──────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ spawn(): │
|
||||
│ addr = ActorAddress::new_random() │
|
||||
│ wid = placement.next_worker() -- round-robin pick │
|
||||
│ address_map.insert(addr, wid) -- register globally │
|
||||
│ spawn_txs[wid].try_send((addr, boxed)) -- push to worker queue │
|
||||
│ │ │
|
||||
│ │ ┌────────────────────────────────────────────┐ │
|
||||
│ └──────>│ Worker.spawn_rx (Receiver side) │ │
|
||||
│ └────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ send_to(): │
|
||||
│ wid = address_map.lookup(&addr) │
|
||||
│ transfer_txs[wid].try_send(Envelope::new(addr, msg)) │
|
||||
│ │ │
|
||||
│ │ ┌────────────────────────────────────────────┐ │
|
||||
│ └──────>│ Worker.transfer_rx (Receiver side) │ │
|
||||
│ └────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### Who the worker talks to during a tick
|
||||
|
||||
```
|
||||
┌─ Worker (during phase 3: tick_all) ────────────────────────────────────┐
|
||||
│ │
|
||||
│ An actor calls ctx.send(addr, msg) or ctx.spawn(new_actor). │
|
||||
│ These go through WorkerContext, which implements ContextInner. │
|
||||
│ │
|
||||
│ ctx.send(addr, msg) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ ┌─ WorkerContext.send_any ─────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ address_map.lookup(addr) --> which worker owns this actor? │ │
|
||||
│ │ │ │ │
|
||||
│ │ ┌────┴──────────────┬──────────────────┬──────────────────┐ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ v v v │ │ │
|
||||
│ │ SAME WORKER OTHER WORKER NOT FOUND │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ pending_local │ transfer_txs │ inbox_registry │ │ │
|
||||
│ │ │ .push(addr,msg) │ [wid].send() │ .try_deliver() │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ stays in this │ crosses to │ goes to an │ │ │
|
||||
│ │ │ worker; delivered │ another worker │ external │ │ │
|
||||
│ │ │ in phase 4 │ thread's │ Inbox<M> │ │ │
|
||||
│ │ │ │ transfer_rx │ receiver │ │ │
|
||||
│ │ └───────────────────┴──────────────────┴──────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ctx.spawn(new_actor) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ ┌─ WorkerContext.spawn_any ────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ wid = placement.next_worker() -- round-robin target │ │
|
||||
│ │ address_map.insert(addr, wid) -- register in global map │ │
|
||||
│ │ spawn_txs[wid].try_send(...) -- enqueue for target worker │ │
|
||||
│ │ │ │
|
||||
│ │ may land on THIS worker or a DIFFERENT worker │ │
|
||||
│ │ target picks it up in phase 1 of its next tick │ │
|
||||
│ │ │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### The channel connecting everything
|
||||
|
||||
Each worker has two inbound channels. The channels are lock-free MPSC queues
|
||||
backed by `crossbeam::ArrayQueue` with a `SegQueue` overflow.
|
||||
|
||||
```
|
||||
┌─ HybridChannel<T> ─────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ ┌─ ring: ArrayQueue<T> ──────────────────────────────────────┐ │
|
||||
│ │ fixed capacity, lock-free, bounded │ │
|
||||
│ │ ┌───┬───┬───┬───┬───┬───┬───┬───┐ │ │
|
||||
│ │ │ │ │ │ │ │ │ │ │ (pre-allocated) │ │
|
||||
│ │ └───┴───┴───┴───┴───┴───┴───┴───┘ │ │
|
||||
│ └────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ┌─ overflow: SegQueue<T> ────────────────────────────────────┐ │
|
||||
│ │ unbounded, lock-free, linked nodes │ │
|
||||
│ │ used only when ring is full │ │
|
||||
│ └────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ push(v): try ring first, spill to overflow │
|
||||
│ pop(): drain ring first, then overflow │
|
||||
│ │
|
||||
│ ┌─ Sender<T> ─────────┐ ┌─ Receiver<T> ─────────┐ │
|
||||
│ │ Arc<HybridChannel<T>>│ │ Arc<HybridChannel<T>> │ │
|
||||
│ │ .try_send(v) │──────>│ .try_recv() -> Option │ │
|
||||
│ │ clonable (new_sender)│ same │ single consumer │ │
|
||||
│ └──────────────────────┘ Arc └────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
|
||||
Who holds what:
|
||||
|
||||
transfer channel:
|
||||
Sender held by: Runtime.transfer_txs[i], workers via TickContext
|
||||
Receiver held by: Worker[i].transfer_rx
|
||||
|
||||
spawn channel:
|
||||
Sender held by: Runtime.spawn_txs[i], workers via TickContext
|
||||
Receiver held by: Worker[i].spawn_rx
|
||||
```
|
||||
|
||||
### The AddressMap: global actor directory
|
||||
|
||||
```
|
||||
┌─ AddressMap ───────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ RwLock< HashMap<ActorAddress, WorkerId> > │
|
||||
│ │
|
||||
│ ┌──────────────────────────────────────────────────────────────────┐ │
|
||||
│ │ addr_0 -> WorkerId(0) │ │
|
||||
│ │ addr_1 -> WorkerId(2) │ │
|
||||
│ │ addr_2 -> WorkerId(0) │ │
|
||||
│ │ addr_3 -> WorkerId(1) │ │
|
||||
│ │ ... │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ READERS (concurrent, RwLock read): │
|
||||
│ WorkerContext.send_any() -- every message send does a lookup │
|
||||
│ Runtime.send_to() -- external sends do a lookup │
|
||||
│ │
|
||||
│ WRITERS (rare, exclusive lock): │
|
||||
│ Runtime.spawn() -- registers new actor at spawn time │
|
||||
│ WorkerContext.spawn_any() -- actor spawns another actor │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### The InboxRegistry: escape hatch to user code
|
||||
|
||||
```
|
||||
┌─ InboxRegistry ────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ RwLock< HashMap<ActorAddress, Arc<dyn SenderT>> > │
|
||||
│ │
|
||||
│ For addresses belonging to external Inbox<M>, not actors. │
|
||||
│ │
|
||||
│ ┌─ Registration ─────────────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ Runtime.new_inbox::<M>() │ │
|
||||
│ │ addr = ActorAddress::new_random() │ │
|
||||
│ │ receiver = Receiver::<M>::new(capacity) │ │
|
||||
│ │ sender = receiver.new_sender() │ │
|
||||
│ │ inbox_registry.register(addr, Arc::new(sender)) │ │
|
||||
│ │ returns Inbox { addr, inner: receiver } │ │
|
||||
│ │ │ │
|
||||
│ └────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ┌─ Delivery (when address_map lookup fails) ─────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ WorkerContext.send_any(addr, msg) │ │
|
||||
│ │ address_map.lookup(addr) -> None │ │
|
||||
│ │ inbox_registry.try_deliver(addr, msg) │ │
|
||||
│ │ senders.read().get(addr).try_send_any(msg) │ │
|
||||
│ │ downcast Box<Any> -> M, push into Receiver<M> │ │
|
||||
│ │ │ │
|
||||
│ └────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ┌─ Consumption (user code) ──────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ let inbox = rt.new_inbox::<MyMsg>()?; │ │
|
||||
│ │ // later, from any thread: │ │
|
||||
│ │ if let Some(msg) = inbox.try_recv() { ... } │ │
|
||||
│ │ │ │
|
||||
│ └────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### Full system topology
|
||||
|
||||
```
|
||||
┌─ User Code ────────────────────────────────────────────────────────────┐
|
||||
│ rt.spawn() rt.send_to() inbox.try_recv() rt.shutdown() │
|
||||
└────┬──────────────────┬──────────────────┬──────────────────┬──────────┘
|
||||
│ │ ^ │
|
||||
v v │ v
|
||||
┌─ Arc<Runtime> ────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ ┌───────────┐ ┌────────────┐ ┌─────────────┐ ┌────────────┐ │
|
||||
│ │AddressMap │ │ Placement │ │InboxRegistry│ │ is_running │ │
|
||||
│ │ addr->wid │ │ round-robin│ │ addr->Sender│ │ AtomicBool │ │
|
||||
│ └─────┬─────┘ └──────┬─────┘ └──────┬──────┘ └──────┬─────┘ │
|
||||
│ │ │ │ │ │
|
||||
│ ┌─────┴───────────────┴──────────────┴───────────────┴────────────────┐ │
|
||||
│ │ TickContext (borrows all above) │ │
|
||||
│ └──────────────────────────┬──────────────────────────────────────────┘ │
|
||||
│ │ │
|
||||
│ ┌─ transfer_txs[] ────┐ │ ┌─ spawn_txs[] ──────┐ │
|
||||
│ │ [0]: Sender<Envelope│ │ │ [0]: Sender<(A,Box)>│ │
|
||||
│ │ [1]: Sender<Envelope│ │ │ [1]: Sender<(A,Box)>│ │
|
||||
│ │ [2]: Sender<Envelope│ │ │ [2]: Sender<(A,Box)>│ │
|
||||
│ └──┬──────┬──────┬────┘ │ └──┬──────┬──────┬────┘ │
|
||||
│ │ │ │ │ │ │ │ │
|
||||
└────┼──────┼──────┼────────┼──────┼──────┼──────┼──────────────────────────┘
|
||||
│ │ │ │ │ │ │
|
||||
v v v │ v v v
|
||||
┌────────┐┌────────┐┌───────┴┐┌────────┐┌────────┐┌────────┐
|
||||
│xfer ││xfer ││xfer ││ spawn ││ spawn ││ spawn │
|
||||
│_rx[0] ││_rx[1] ││_rx[2] ││ _rx[0] ││ _rx[1] ││ _rx[2] │
|
||||
└───┬────┘└───┬────┘└───┬────┘└───┬────┘└───┬────┘└───┬────┘
|
||||
│ │ │ │ │ │
|
||||
v v v v v v
|
||||
┌─ Worker 0 ──────┐ ┌─ Worker 1 ──────┐ ┌─ Worker 2 ──────┐
|
||||
│ │ │ │ │ │
|
||||
│ ┌─ pool ──────┐ │ │ ┌─ pool ──────┐ │ │ ┌─ pool ──────┐ │
|
||||
│ │ ┌──────────┐│ │ │ │ ┌──────────┐│ │ │ │ ┌──────────┐│ │
|
||||
│ │ │ slot: ││ │ │ │ │ slot: ││ │ │ │ │ slot: ││ │
|
||||
│ │ │ mailbox ││ │ │ │ │ mailbox ││ │ │ │ │ mailbox ││ │
|
||||
│ │ │ actor ││ │ │ │ │ actor ││ │ │ │ │ actor ││ │
|
||||
│ │ └──────────┘│ │ │ │ └──────────┘│ │ │ │ └──────────┘│ │
|
||||
│ │ ┌──────────┐│ │ │ │ ┌──────────┐│ │ │ │ │ │
|
||||
│ │ │ slot: ││ │ │ │ │ slot: ││ │ │ └─────────────┘ │
|
||||
│ │ │ mailbox ││ │ │ │ │ mailbox ││ │ │ │
|
||||
│ │ │ actor ││ │ │ │ │ actor ││ │ │ thread 2 │
|
||||
│ │ └──────────┘│ │ │ │ └──────────┘│ │ └──────────────────┘
|
||||
│ └─────────────┘ │ │ └─────────────┘ │
|
||||
│ │ │ │
|
||||
│ thread 0 │ │ thread 1 │
|
||||
└──────────────────┘ └──────────────────┘
|
||||
|
||||
Workers also send to EACH OTHER during phase 3:
|
||||
WorkerContext.send_any() -> transfer_txs[other_wid].try_send()
|
||||
WorkerContext.spawn_any() -> spawn_txs[target_wid].try_send()
|
||||
```
|
||||
|
||||
## Message Lifecycle
|
||||
|
||||
```
|
||||
PRODUCERS
|
||||
┌──────────────────┬──────────────────────┐
|
||||
│ │ │
|
||||
v v v
|
||||
┌─────────────┐ ┌──────────────┐ ┌──────────────────────┐
|
||||
│ Runtime │ │ ctx.send() │ │ ctx.send() │
|
||||
│ .send_to() │ │ same worker │ │ other worker │
|
||||
└──────┬──────┘ └──────┬───────┘ └──────────┬───────────┘
|
||||
│ │ │
|
||||
v v v
|
||||
┌─────────────┐ ┌─────────────┐ ┌──────────────────────┐
|
||||
│ transfer_tx │ │ pending_ │ │ transfer_tx │
|
||||
│ [wid].send()│ │ local.push()│ │ [wid].send() │
|
||||
└──────┬──────┘ └──────┬──────┘ └──────────┬───────────┘
|
||||
│ │ │
|
||||
│ (end of phase 3) │
|
||||
│ │ │
|
||||
│ phase 4: │
|
||||
│ │ │
|
||||
v v v
|
||||
┌────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ pool.deliver(&addr, msg) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ slot.mailbox.push_back(msg) │
|
||||
│ │
|
||||
└───────────────────────┬────────────────────────────┘
|
||||
│
|
||||
next tick_once
|
||||
phase 3
|
||||
│
|
||||
v
|
||||
┌────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ msg = slot.mailbox.pop_front() │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ slot.actor.handle_any(&ctx, msg) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ ┌──────────────────────────────────────────┐ │
|
||||
│ │ downcast Box<dyn Any> to A::Incoming │ │
|
||||
│ │ │ │
|
||||
│ │ ok: A.handle(ctx, typed_msg) │ │
|
||||
│ │ err: silently dropped │ │
|
||||
│ └──────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Shutdown Flow
|
||||
|
||||
```
|
||||
┌─ User Code ──────┐
|
||||
│ │
|
||||
│ rt.shutdown() │
|
||||
│ │ │
|
||||
└───────┼───────────┘
|
||||
│
|
||||
v
|
||||
┌─ Runtime ──────────────────────────────────────┐
|
||||
│ │
|
||||
│ is_running.store(false, Release) │
|
||||
│ │
|
||||
└────────────────────────┬────────────────────────┘
|
||||
│
|
||||
┌─────────────────┼─────────────────┐
|
||||
│ │ │
|
||||
v v v
|
||||
┌─ Worker 0 ────┐ ┌─ Worker 1 ────┐ ┌─ Worker 2 ────┐
|
||||
│ │ │ │ │ │
|
||||
│ is_running │ │ is_running │ │ is_running │
|
||||
│ .load(Acquire)│ │ .load(Acquire)│ │ .load(Acquire)│
|
||||
│ -> false │ │ -> false │ │ -> false │
|
||||
│ │ │ │ │ │
|
||||
│ run() returns │ │ run() returns │ │ run() returns │
|
||||
│ thread exits │ │ thread exits │ │ thread exits │
|
||||
└────────────────┘ └────────────────┘ └────────────────┘
|
||||
│ │ │
|
||||
└─────────────────┼─────────────────┘
|
||||
│
|
||||
v
|
||||
┌─ RuntimeHandle ──┐
|
||||
│ │
|
||||
│ .join() │
|
||||
│ waits for all │
|
||||
│ JoinHandles │
|
||||
│ │
|
||||
└───────────────────┘
|
||||
```
|
||||
|
||||
51
src/actor.rs
51
src/actor.rs
|
|
@ -1,6 +1,6 @@
|
|||
use std::any::Any;
|
||||
|
||||
use crate::{get_random, runtime::{ContextInner, Ctx}, worker::Mailbox};
|
||||
use crate::runtime::Ctx;
|
||||
|
||||
/// The primary trait defining data that can be passed to and from actor processes
|
||||
pub trait Message: 'static + Sized + Clone + Send + Sync {}
|
||||
|
|
@ -20,63 +20,32 @@ pub struct ActorAddress(pub [u8; 32]);
|
|||
impl ActorAddress {
|
||||
pub fn new_random() -> Self {
|
||||
let mut bytes = [0u8; 32];
|
||||
get_random(&mut bytes);
|
||||
crate::get_random(&mut bytes);
|
||||
Self(bytes)
|
||||
}
|
||||
}
|
||||
|
||||
/// The actor process as represented in the Runtime, with the actor state stored with its mailbox.
|
||||
pub(crate) struct Actor<A>
|
||||
where
|
||||
A: ActorInterface,
|
||||
{
|
||||
addr: ActorAddress,
|
||||
mailbox: Mailbox<A::Incoming>,
|
||||
inner: A,
|
||||
}
|
||||
/// The actor process as represented in the Runtime — thin wrapper around user state.
|
||||
pub(crate) struct Actor<A: ActorInterface>(A);
|
||||
|
||||
impl<A: ActorInterface> Actor<A> {
|
||||
pub(crate) fn new(addr: ActorAddress, mailbox: Mailbox<A::Incoming>, inner: A) -> Self {
|
||||
Self {
|
||||
addr,
|
||||
mailbox,
|
||||
inner,
|
||||
}
|
||||
pub(crate) fn new(inner: A) -> Self {
|
||||
Self(inner)
|
||||
}
|
||||
}
|
||||
|
||||
/// Trait for type-erased actors
|
||||
/// Trait for type-erased actors — single-message handler.
|
||||
pub(crate) trait AnyActor: Send {
|
||||
/// Tick the actor, processing pending messages. Returns `true` if any work was done.
|
||||
fn tick(&mut self, inner: &dyn ContextInner) -> bool;
|
||||
/// Deliver a type-erased message into this actor's mailbox.
|
||||
/// Returns `true` if the downcast succeeded.
|
||||
fn deliver(&mut self, msg: Box<dyn Any + Send>) -> bool;
|
||||
fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>);
|
||||
}
|
||||
|
||||
impl<A> AnyActor for Actor<A>
|
||||
where
|
||||
A: ActorInterface,
|
||||
{
|
||||
fn tick(&mut self, inner: &dyn ContextInner) -> bool {
|
||||
let n = self.mailbox.drain_count();
|
||||
if n > 0 {
|
||||
let ctx = Ctx::new(inner, self.addr);
|
||||
for _ in 0..n {
|
||||
if let Some(msg) = self.mailbox.pop() {
|
||||
self.inner.handle(&ctx, msg);
|
||||
}
|
||||
}
|
||||
}
|
||||
n > 0
|
||||
}
|
||||
|
||||
fn deliver(&mut self, msg: Box<dyn Any + Send>) -> bool {
|
||||
fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) {
|
||||
if let Ok(typed) = msg.downcast::<A::Incoming>() {
|
||||
self.mailbox.push(*typed);
|
||||
true
|
||||
} else {
|
||||
false
|
||||
self.0.handle(ctx, *typed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ use pyo3::types::PyModule;
|
|||
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor};
|
||||
use crate::config::{BackoffPolicy, RuntimeConfig};
|
||||
use crate::runtime::{Ctx, Inbox, Runtime, RuntimeHandle};
|
||||
use crate::worker::Mailbox;
|
||||
use crate::Error;
|
||||
|
||||
// ─── PyMsg newtype ───────────────────────────────────────────────────────────
|
||||
|
|
@ -182,10 +181,8 @@ impl ActorInterface for PyActor {
|
|||
);
|
||||
}
|
||||
Effect::Spawn { addr, handler } => {
|
||||
let waterlevel = ctx.raw_inner().mailbox_waterlevel();
|
||||
let actor = PyActor::new(handler);
|
||||
let actor = Actor::new(addr, Mailbox::new(waterlevel), actor);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(actor);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
|
||||
let _ = ctx.raw_inner().spawn_any(addr, boxed);
|
||||
}
|
||||
}
|
||||
|
|
@ -479,6 +476,29 @@ impl PyActorInfo {
|
|||
}
|
||||
}
|
||||
|
||||
#[pyclass(name = "WorkerInfo")]
|
||||
#[derive(Clone)]
|
||||
pub struct PyWorkerInfo {
|
||||
#[pyo3(get)]
|
||||
id: usize,
|
||||
#[pyo3(get)]
|
||||
num_actors: usize,
|
||||
#[pyo3(get)]
|
||||
mailbox_depth: usize,
|
||||
#[pyo3(get)]
|
||||
messages_processed: u64,
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
impl PyWorkerInfo {
|
||||
fn __repr__(&self) -> String {
|
||||
format!(
|
||||
"WorkerInfo(id={}, actors={}, queued={}, processed={})",
|
||||
self.id, self.num_actors, self.mailbox_depth, self.messages_processed
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[pyclass(name = "RuntimeStats")]
|
||||
#[derive(Clone)]
|
||||
pub struct PyRuntimeStats {
|
||||
|
|
@ -488,6 +508,8 @@ pub struct PyRuntimeStats {
|
|||
num_workers: usize,
|
||||
#[pyo3(get)]
|
||||
actors: Vec<PyActorInfo>,
|
||||
#[pyo3(get)]
|
||||
workers: Vec<PyWorkerInfo>,
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
|
|
@ -498,19 +520,13 @@ impl PyRuntimeStats {
|
|||
self.num_actors, self.num_workers
|
||||
);
|
||||
|
||||
// Group actors by worker
|
||||
let mut by_worker: std::collections::BTreeMap<usize, Vec<&PyActorInfo>> =
|
||||
std::collections::BTreeMap::new();
|
||||
for info in &self.actors {
|
||||
by_worker.entry(info.worker_id).or_default().push(info);
|
||||
}
|
||||
|
||||
for wid in 0..self.num_workers {
|
||||
let actors = by_worker.get(&wid);
|
||||
let count = actors.map_or(0, |v| v.len());
|
||||
out.push_str(&format!("\n Worker {wid}: {count} actors"));
|
||||
if let Some(actors) = actors {
|
||||
for info in actors {
|
||||
for w in &self.workers {
|
||||
out.push_str(&format!(
|
||||
"\n Worker {}: {} actors, {} queued, {} processed",
|
||||
w.id, w.num_actors, w.mailbox_depth, w.messages_processed
|
||||
));
|
||||
for info in &self.actors {
|
||||
if info.worker_id == w.id {
|
||||
out.push_str(&format!("\n - {}", info.address.hex()));
|
||||
}
|
||||
}
|
||||
|
|
@ -525,18 +541,30 @@ impl PyRuntimeStats {
|
|||
}
|
||||
|
||||
fn build_stats(runtime: &Runtime) -> PyRuntimeStats {
|
||||
let (num_workers, snapshot) = runtime.stats();
|
||||
let actors: Vec<PyActorInfo> = snapshot
|
||||
let stats = runtime.stats();
|
||||
let actors: Vec<PyActorInfo> = stats
|
||||
.actors
|
||||
.into_iter()
|
||||
.map(|(addr, wid)| PyActorInfo {
|
||||
address: PyActorAddress::from(addr),
|
||||
worker_id: wid.as_usize(),
|
||||
worker_id: wid,
|
||||
})
|
||||
.collect();
|
||||
let workers: Vec<PyWorkerInfo> = stats
|
||||
.workers
|
||||
.into_iter()
|
||||
.map(|w| PyWorkerInfo {
|
||||
id: w.id,
|
||||
num_actors: w.num_actors,
|
||||
mailbox_depth: w.mailbox_depth,
|
||||
messages_processed: w.messages_processed,
|
||||
})
|
||||
.collect();
|
||||
PyRuntimeStats {
|
||||
num_actors: actors.len(),
|
||||
num_workers,
|
||||
num_workers: stats.num_workers,
|
||||
actors,
|
||||
workers,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -550,6 +578,7 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
|
|||
m.add_class::<PyRuntime>()?;
|
||||
m.add_class::<PyRuntimeHandle>()?;
|
||||
m.add_class::<PyActorInfo>()?;
|
||||
m.add_class::<PyWorkerInfo>()?;
|
||||
m.add_class::<PyRuntimeStats>()?;
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,11 +10,26 @@ use crate::address_map::{AddressMap, Placement, WorkerId};
|
|||
use crate::channel::{Receiver, Sender};
|
||||
// Re-export config types so existing code using `runtime::RuntimeConfig` still works
|
||||
pub use crate::config::{BackoffPolicy, RuntimeConfig};
|
||||
use crate::worker::Mailbox;
|
||||
use crate::worker::{TickContext, Worker};
|
||||
use crate::worker::{TickContext, Worker, WorkerStats};
|
||||
use crate::Error;
|
||||
|
||||
|
||||
/// Snapshot of per-worker state.
|
||||
pub struct WorkerInfo {
|
||||
pub id: usize,
|
||||
pub num_actors: usize,
|
||||
pub mailbox_depth: usize,
|
||||
pub messages_processed: u64,
|
||||
}
|
||||
|
||||
/// Snapshot of overall runtime state.
|
||||
pub struct RuntimeStats {
|
||||
pub num_workers: usize,
|
||||
/// Each entry is (address, worker_id).
|
||||
pub actors: Vec<(ActorAddress, usize)>,
|
||||
pub workers: Vec<WorkerInfo>,
|
||||
}
|
||||
|
||||
/// Generic message inbox for receiving messages outside of the runtime.
|
||||
pub struct Inbox<M: Message> {
|
||||
addr: ActorAddress,
|
||||
|
|
@ -82,9 +97,7 @@ impl<'a> Ctx<'a> {
|
|||
/// Spawn a new actor, returning its address.
|
||||
pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
|
||||
let addr = ActorAddress::new_random();
|
||||
let waterlevel = self.inner.mailbox_waterlevel();
|
||||
let actor = Actor::new(addr, Mailbox::new(waterlevel), actor);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(actor);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
|
||||
self.inner.spawn_any(addr, boxed)?;
|
||||
Ok(addr)
|
||||
}
|
||||
|
|
@ -115,6 +128,7 @@ pub struct Runtime {
|
|||
spawn_txs: Vec<Sender<(ActorAddress, Box<dyn AnyActor>)>>,
|
||||
placement: Placement,
|
||||
is_running: AtomicBool,
|
||||
worker_stats: Vec<Arc<WorkerStats>>,
|
||||
/// Single-threaded mode: worker stored inline
|
||||
single_worker: Option<RefCell<Worker>>,
|
||||
/// Multi-threaded mode: workers waiting to be assigned to threads by run()
|
||||
|
|
@ -142,6 +156,7 @@ impl Runtime {
|
|||
|
||||
let mut transfer_txs = Vec::with_capacity(num_workers);
|
||||
let mut spawn_txs = Vec::with_capacity(num_workers);
|
||||
let mut worker_stats = Vec::with_capacity(num_workers);
|
||||
let mut workers = Vec::with_capacity(num_workers);
|
||||
|
||||
for i in 0..num_workers {
|
||||
|
|
@ -154,7 +169,9 @@ impl Runtime {
|
|||
let spawn_tx = spawn_rx.new_sender();
|
||||
spawn_txs.push(spawn_tx);
|
||||
|
||||
workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx));
|
||||
let stats = Arc::new(WorkerStats::new());
|
||||
worker_stats.push(stats.clone());
|
||||
workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats));
|
||||
}
|
||||
|
||||
if config.num_threads < 2 {
|
||||
|
|
@ -168,6 +185,7 @@ impl Runtime {
|
|||
spawn_txs,
|
||||
placement,
|
||||
is_running: AtomicBool::new(false),
|
||||
worker_stats,
|
||||
single_worker: Some(RefCell::new(worker)),
|
||||
pending_workers: None,
|
||||
}
|
||||
|
|
@ -181,6 +199,7 @@ impl Runtime {
|
|||
spawn_txs,
|
||||
placement,
|
||||
is_running: AtomicBool::new(false),
|
||||
worker_stats,
|
||||
single_worker: None,
|
||||
pending_workers: Some(workers),
|
||||
}
|
||||
|
|
@ -192,8 +211,7 @@ impl Runtime {
|
|||
let addr = ActorAddress::new_random();
|
||||
let worker_id = self.placement.next_worker();
|
||||
self.address_map.insert(addr, worker_id);
|
||||
let actor = Actor::new(addr, Mailbox::new(self.config.mailbox_waterlevel), actor);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(actor);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
|
||||
self.spawn_txs[worker_id.as_usize()]
|
||||
.try_send((addr, boxed))
|
||||
.map_err(|_| Error::from("Runtime error: spawn queue full"))?;
|
||||
|
|
@ -280,15 +298,35 @@ impl Runtime {
|
|||
})
|
||||
}
|
||||
|
||||
/// Returns a snapshot of runtime stats: all actor addresses with their worker assignments,
|
||||
/// plus the number of workers.
|
||||
pub(crate) fn stats(&self) -> (usize, Vec<(ActorAddress, WorkerId)>) {
|
||||
/// Returns a snapshot of runtime stats: actor placements and per-worker info.
|
||||
pub fn stats(&self) -> RuntimeStats {
|
||||
let num_workers = if self.config.num_threads < 2 {
|
||||
1
|
||||
} else {
|
||||
self.config.num_threads
|
||||
};
|
||||
(num_workers, self.address_map.snapshot())
|
||||
let workers = self
|
||||
.worker_stats
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, ws)| WorkerInfo {
|
||||
id: i,
|
||||
num_actors: ws.num_actors.load(Ordering::Relaxed),
|
||||
mailbox_depth: ws.total_mailbox_depth.load(Ordering::Relaxed),
|
||||
messages_processed: ws.messages_processed.load(Ordering::Relaxed),
|
||||
})
|
||||
.collect();
|
||||
let actors = self
|
||||
.address_map
|
||||
.snapshot()
|
||||
.into_iter()
|
||||
.map(|(addr, wid)| (addr, wid.as_usize()))
|
||||
.collect();
|
||||
RuntimeStats {
|
||||
num_workers,
|
||||
actors,
|
||||
workers,
|
||||
}
|
||||
}
|
||||
|
||||
/// Signal all workers to stop
|
||||
|
|
|
|||
|
|
@ -1,24 +1,42 @@
|
|||
use std::any::Any;
|
||||
use std::cell::RefCell;
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
|
||||
use crate::actor::{ActorAddress, AnyActor, Message};
|
||||
use crate::address_map::{AddressMap, Placement, WorkerId};
|
||||
use crate::channel::{Receiver, Sender};
|
||||
use crate::config::{BackoffPolicy, RuntimeConfig};
|
||||
use crate::runtime::{ContextInner, Envelope, InboxRegistry};
|
||||
use crate::runtime::{ContextInner, Ctx, Envelope, InboxRegistry};
|
||||
use crate::Error;
|
||||
|
||||
/// Per-worker stats published via atomics. Readable from any thread.
|
||||
pub(crate) struct WorkerStats {
|
||||
pub num_actors: AtomicUsize,
|
||||
pub total_mailbox_depth: AtomicUsize,
|
||||
pub messages_processed: AtomicU64,
|
||||
}
|
||||
|
||||
impl WorkerStats {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
num_actors: AtomicUsize::new(0),
|
||||
total_mailbox_depth: AtomicUsize::new(0),
|
||||
messages_processed: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared state passed to tick_once — single thin pointer avoids register spill.
|
||||
pub(crate) struct TickContext<'a> {
|
||||
pub address_map: &'a AddressMap,
|
||||
pub transfer_txs: &'a [Sender<Envelope>],
|
||||
pub spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
|
||||
pub placement: &'a Placement,
|
||||
pub inbox_registry: &'a InboxRegistry,
|
||||
pub config: &'a RuntimeConfig,
|
||||
pub(crate) address_map: &'a AddressMap,
|
||||
pub(crate) transfer_txs: &'a [Sender<Envelope>],
|
||||
pub(crate) spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
|
||||
pub(crate) placement: &'a Placement,
|
||||
pub(crate) inbox_registry: &'a InboxRegistry,
|
||||
pub(crate) config: &'a RuntimeConfig,
|
||||
}
|
||||
|
||||
/// A worker owns a set of actors and runs them in a loop.
|
||||
|
|
@ -27,24 +45,27 @@ pub(crate) struct Worker {
|
|||
pool: ActorPool,
|
||||
transfer_rx: Receiver<Envelope>,
|
||||
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
|
||||
stats: Arc<WorkerStats>,
|
||||
}
|
||||
|
||||
impl Worker {
|
||||
pub fn new(
|
||||
pub(crate) fn new(
|
||||
id: WorkerId,
|
||||
transfer_rx: Receiver<Envelope>,
|
||||
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
|
||||
stats: Arc<WorkerStats>,
|
||||
) -> Self {
|
||||
Self {
|
||||
id,
|
||||
pool: ActorPool::new(),
|
||||
transfer_rx,
|
||||
spawn_rx,
|
||||
stats,
|
||||
}
|
||||
}
|
||||
|
||||
/// Run one iteration of the worker loop. Returns `true` if any work was done.
|
||||
pub fn tick_once(&mut self, tc: &TickContext) -> bool {
|
||||
pub(crate) fn tick_once(&mut self, tc: &TickContext) -> bool {
|
||||
let mut did_work = false;
|
||||
|
||||
// 1. Drain spawn queue → add actors to pool
|
||||
|
|
@ -65,6 +86,7 @@ impl Worker {
|
|||
let pending_local: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
|
||||
RefCell::new(Vec::new());
|
||||
|
||||
let processed;
|
||||
{
|
||||
let worker_ctx = WorkerContext {
|
||||
worker_id: self.id,
|
||||
|
|
@ -76,7 +98,8 @@ impl Worker {
|
|||
config: tc.config,
|
||||
pending_local: &pending_local,
|
||||
};
|
||||
if self.pool.tick_all(&worker_ctx) {
|
||||
processed = self.pool.tick_all(&worker_ctx);
|
||||
if processed > 0 {
|
||||
did_work = true;
|
||||
}
|
||||
}
|
||||
|
|
@ -90,6 +113,11 @@ impl Worker {
|
|||
self.pool.deliver(&addr, msg);
|
||||
}
|
||||
|
||||
// 5. Publish stats
|
||||
self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed);
|
||||
self.stats.total_mailbox_depth.store(self.pool.total_mailbox_depth(), Ordering::Relaxed);
|
||||
self.stats.messages_processed.fetch_add(processed as u64, Ordering::Relaxed);
|
||||
|
||||
did_work
|
||||
}
|
||||
|
||||
|
|
@ -166,9 +194,25 @@ impl ContextInner for WorkerContext<'_> {
|
|||
}
|
||||
}
|
||||
|
||||
/// Per-worker actor storage.
|
||||
/// How many messages to process this tick:
|
||||
/// - `len < waterlevel` → process all (`len`)
|
||||
/// - `len >= waterlevel` → process half (`len >> 1`)
|
||||
pub(crate) fn drain_count(len: usize, waterlevel: usize) -> usize {
|
||||
if len < waterlevel {
|
||||
len
|
||||
} else {
|
||||
len >> 1
|
||||
}
|
||||
}
|
||||
|
||||
struct ActorSlot {
|
||||
mailbox: VecDeque<Box<dyn Any + Send>>,
|
||||
actor: Box<dyn AnyActor>,
|
||||
}
|
||||
|
||||
/// Per-worker actor storage. Owns per-actor mailboxes.
|
||||
pub(crate) struct ActorPool {
|
||||
actors: HashMap<ActorAddress, Box<dyn AnyActor>>,
|
||||
actors: HashMap<ActorAddress, ActorSlot>,
|
||||
}
|
||||
|
||||
impl ActorPool {
|
||||
|
|
@ -179,42 +223,58 @@ impl ActorPool {
|
|||
}
|
||||
|
||||
pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
|
||||
self.actors.insert(addr, actor);
|
||||
self.actors.insert(addr, ActorSlot {
|
||||
mailbox: VecDeque::new(),
|
||||
actor,
|
||||
});
|
||||
}
|
||||
|
||||
pub fn remove(&mut self, addr: &ActorAddress) -> Option<Box<dyn AnyActor>> {
|
||||
self.actors.remove(addr)
|
||||
self.actors.remove(addr).map(|slot| slot.actor)
|
||||
}
|
||||
|
||||
/// Deliver a type-erased message to the actor at `addr`.
|
||||
/// Returns `true` if the actor was found and the message type matched.
|
||||
/// Returns `true` if the actor exists (message is queued; type check deferred to tick).
|
||||
pub fn deliver(&mut self, addr: &ActorAddress, msg: Box<dyn Any + Send>) -> bool {
|
||||
if let Some(actor) = self.actors.get_mut(addr) {
|
||||
actor.deliver(msg)
|
||||
if let Some(slot) = self.actors.get_mut(addr) {
|
||||
slot.mailbox.push_back(msg);
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
/// Tick all actors in the pool. Returns `true` if any actor processed messages.
|
||||
pub fn tick_all(&mut self, inner: &dyn ContextInner) -> bool {
|
||||
let mut did_work = false;
|
||||
for actor in self.actors.values_mut() {
|
||||
if actor.tick(inner) {
|
||||
did_work = true;
|
||||
/// Tick all actors in the pool. Returns the number of messages processed.
|
||||
pub fn tick_all(&mut self, inner: &dyn ContextInner) -> usize {
|
||||
let mut count = 0;
|
||||
for (&addr, slot) in self.actors.iter_mut() {
|
||||
let len = slot.mailbox.len();
|
||||
let n = drain_count(len, inner.mailbox_waterlevel());
|
||||
if n > 0 {
|
||||
let ctx = Ctx::new(inner, addr);
|
||||
for _ in 0..n {
|
||||
if let Some(msg) = slot.mailbox.pop_front() {
|
||||
slot.actor.handle_any(&ctx, msg);
|
||||
count += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
did_work
|
||||
count
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
self.actors.len()
|
||||
}
|
||||
|
||||
pub fn total_mailbox_depth(&self) -> usize {
|
||||
self.actors.values().map(|slot| slot.mailbox.len()).sum()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
pub struct Mailbox<M: Message> {
|
||||
pub(crate) struct Mailbox<M: Message> {
|
||||
queue: VecDeque<M>,
|
||||
waterlevel: usize,
|
||||
}
|
||||
|
|
@ -247,12 +307,9 @@ impl<M: Message> Mailbox<M> {
|
|||
/// - `len < waterlevel` → process all (`len`)
|
||||
/// - `len >= waterlevel` → process half (`len >> 1`)
|
||||
pub fn drain_count(&self) -> usize {
|
||||
let len = self.queue.len();
|
||||
if len < self.waterlevel {
|
||||
len
|
||||
} else {
|
||||
len >> 1
|
||||
}
|
||||
drain_count(self.queue.len(), self.waterlevel)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
420
src/worker/tests.rs
Normal file
420
src/worker/tests.rs
Normal file
|
|
@ -0,0 +1,420 @@
|
|||
use std::any::Any;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
|
||||
use crate::actor::{ActorAddress, AnyActor};
|
||||
use crate::address_map::{AddressMap, Placement, WorkerId};
|
||||
use crate::channel::Receiver;
|
||||
use crate::config::{BackoffPolicy, RuntimeConfig};
|
||||
use crate::runtime::{Ctx, Envelope, InboxRegistry};
|
||||
|
||||
use super::{TickContext, Worker, WorkerStats};
|
||||
|
||||
// ── Actors ─────────────────────────────────────────────────────────
|
||||
|
||||
/// Counts how many u64 messages it successfully handled.
|
||||
struct CounterActor(Arc<AtomicUsize>);
|
||||
|
||||
impl AnyActor for CounterActor {
|
||||
fn handle_any(&mut self, _ctx: &Ctx, msg: Box<dyn Any + Send>) {
|
||||
if msg.downcast::<u64>().is_ok() {
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Harness ────────────────────────────────────────────────────────
|
||||
|
||||
fn addr(id: u8) -> ActorAddress {
|
||||
let mut bytes = [0u8; 32];
|
||||
bytes[0] = id;
|
||||
ActorAddress(bytes)
|
||||
}
|
||||
|
||||
/// Self-contained single-worker test environment.
|
||||
///
|
||||
/// Holds the worker, its channels, shared state for TickContext, and
|
||||
/// per-actor handle counters — everything needed to drive scenarios.
|
||||
struct Env {
|
||||
worker: Worker,
|
||||
stats: Arc<WorkerStats>,
|
||||
feed_transfer: crate::channel::Sender<Envelope>,
|
||||
feed_spawn: crate::channel::Sender<(ActorAddress, Box<dyn AnyActor>)>,
|
||||
address_map: AddressMap,
|
||||
placement: Placement,
|
||||
inbox_registry: InboxRegistry,
|
||||
config: RuntimeConfig,
|
||||
tc_transfer_txs: Vec<crate::channel::Sender<Envelope>>,
|
||||
tc_spawn_txs: Vec<crate::channel::Sender<(ActorAddress, Box<dyn AnyActor>)>>,
|
||||
counters: HashMap<u8, Arc<AtomicUsize>>,
|
||||
}
|
||||
|
||||
impl Env {
|
||||
fn new() -> Self {
|
||||
Self::with_config(RuntimeConfig::default())
|
||||
}
|
||||
|
||||
fn with_config(config: RuntimeConfig) -> Self {
|
||||
let transfer_rx = Receiver::<Envelope>::new(256);
|
||||
let spawn_rx = Receiver::<(ActorAddress, Box<dyn AnyActor>)>::new(256);
|
||||
|
||||
let feed_transfer = transfer_rx.new_sender();
|
||||
let feed_spawn = spawn_rx.new_sender();
|
||||
let tc_transfer = transfer_rx.new_sender();
|
||||
let tc_spawn = spawn_rx.new_sender();
|
||||
|
||||
let stats = Arc::new(WorkerStats::new());
|
||||
let worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats.clone());
|
||||
|
||||
Self {
|
||||
worker,
|
||||
stats,
|
||||
feed_transfer,
|
||||
feed_spawn,
|
||||
address_map: AddressMap::new(),
|
||||
placement: Placement::new(1),
|
||||
inbox_registry: InboxRegistry::new(),
|
||||
config,
|
||||
tc_transfer_txs: vec![tc_transfer],
|
||||
tc_spawn_txs: vec![tc_spawn],
|
||||
counters: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Enqueue an actor spawn (drained on next tick, phase 1).
|
||||
fn spawn(&mut self, id: u8) {
|
||||
let counter = Arc::new(AtomicUsize::new(0));
|
||||
let actor: Box<dyn AnyActor> = Box::new(CounterActor(counter.clone()));
|
||||
self.feed_spawn.try_send((addr(id), actor)).ok().unwrap();
|
||||
self.counters.insert(id, counter);
|
||||
}
|
||||
|
||||
/// Enqueue a u64 message (drained on next tick, phase 2).
|
||||
fn send(&self, id: u8, val: u64) {
|
||||
self.feed_transfer
|
||||
.try_send(Envelope::new(addr(id), Box::new(val)))
|
||||
.ok()
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// Enqueue a wrong-typed message (String instead of u64).
|
||||
fn send_bad(&self, id: u8) {
|
||||
self.feed_transfer
|
||||
.try_send(Envelope::new(addr(id), Box::new("bad".to_string())))
|
||||
.ok()
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// Remove an actor from the pool (immediate, no tick needed).
|
||||
fn remove(&mut self, id: u8) {
|
||||
self.worker.pool.remove(&addr(id));
|
||||
}
|
||||
|
||||
/// Run one tick of the worker loop.
|
||||
fn tick(&mut self) {
|
||||
let tc = TickContext {
|
||||
address_map: &self.address_map,
|
||||
transfer_txs: &self.tc_transfer_txs,
|
||||
spawn_txs: &self.tc_spawn_txs,
|
||||
placement: &self.placement,
|
||||
inbox_registry: &self.inbox_registry,
|
||||
config: &self.config,
|
||||
};
|
||||
self.worker.tick_once(&tc);
|
||||
}
|
||||
|
||||
// ── Readouts ───────────────────────────────────────────────────
|
||||
|
||||
fn handled(&self, id: u8) -> usize {
|
||||
self.counters[&id].load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn pool_len(&self) -> usize {
|
||||
self.worker.pool.len()
|
||||
}
|
||||
|
||||
fn depth(&self) -> usize {
|
||||
self.stats.total_mailbox_depth.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn processed(&self) -> u64 {
|
||||
self.stats.messages_processed.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn num_actors_stat(&self) -> usize {
|
||||
self.stats.num_actors.load(Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
// ── Step-driven runner ─────────────────────────────────────────────
|
||||
|
||||
enum Step {
|
||||
Spawn(u8),
|
||||
Send(u8, u64),
|
||||
SendBad(u8),
|
||||
Remove(u8),
|
||||
Tick,
|
||||
Expect { pool_len: usize, depth: usize, processed: u64 },
|
||||
ExpectHandled(u8, usize),
|
||||
}
|
||||
|
||||
fn run(steps: &[Step]) {
|
||||
run_with(RuntimeConfig::default(), steps);
|
||||
}
|
||||
|
||||
fn run_with(config: RuntimeConfig, steps: &[Step]) {
|
||||
let mut env = Env::with_config(config);
|
||||
for (i, step) in steps.iter().enumerate() {
|
||||
match step {
|
||||
Step::Spawn(id) => env.spawn(*id),
|
||||
Step::Send(id, val) => env.send(*id, *val),
|
||||
Step::SendBad(id) => env.send_bad(*id),
|
||||
Step::Remove(id) => env.remove(*id),
|
||||
Step::Tick => env.tick(),
|
||||
Step::Expect { pool_len, depth, processed } => {
|
||||
assert_eq!(env.pool_len(), *pool_len, "step {i}: pool_len");
|
||||
assert_eq!(env.depth(), *depth, "step {i}: depth");
|
||||
assert_eq!(env.processed(), *processed, "step {i}: processed");
|
||||
}
|
||||
Step::ExpectHandled(id, n) => {
|
||||
assert_eq!(env.handled(*id), *n, "step {i}: handled({})", id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Tests ──────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn spawn_send_process() {
|
||||
run(&[
|
||||
Step::Spawn(1),
|
||||
Step::Spawn(2),
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 2, depth: 0, processed: 0 },
|
||||
|
||||
Step::Send(1, 10),
|
||||
Step::Send(1, 20),
|
||||
Step::Send(2, 30),
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 2, depth: 0, processed: 3 },
|
||||
Step::ExpectHandled(1, 2),
|
||||
Step::ExpectHandled(2, 1),
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remove_drops_future_messages() {
|
||||
run(&[
|
||||
Step::Spawn(1),
|
||||
Step::Spawn(2),
|
||||
Step::Tick,
|
||||
|
||||
// Remove actor 1 directly from pool
|
||||
Step::Remove(1),
|
||||
Step::Expect { pool_len: 1, depth: 0, processed: 0 },
|
||||
|
||||
// Messages to actor 1 are drained from the transfer queue
|
||||
// but pool.deliver finds no slot — silently dropped
|
||||
Step::Send(1, 42),
|
||||
Step::Send(2, 99),
|
||||
Step::Tick,
|
||||
|
||||
Step::Expect { pool_len: 1, depth: 0, processed: 1 },
|
||||
Step::ExpectHandled(1, 0),
|
||||
Step::ExpectHandled(2, 1),
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wrong_type_silently_dropped() {
|
||||
run(&[
|
||||
Step::Spawn(1),
|
||||
Step::Tick,
|
||||
|
||||
// Mix correct (u64) and incorrect (String) types
|
||||
Step::Send(1, 1),
|
||||
Step::SendBad(1),
|
||||
Step::Send(1, 2),
|
||||
Step::SendBad(1),
|
||||
Step::SendBad(1),
|
||||
Step::Send(1, 3),
|
||||
Step::Tick,
|
||||
|
||||
// All 6 popped from mailbox ("processed" by the pool),
|
||||
// but only the 3 u64 messages were handled by the actor
|
||||
Step::Expect { pool_len: 1, depth: 0, processed: 6 },
|
||||
Step::ExpectHandled(1, 3),
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn backpressure_drains_half() {
|
||||
let config = RuntimeConfig {
|
||||
mailbox_waterlevel: 4,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
run_with(config, &[
|
||||
Step::Spawn(1),
|
||||
Step::Tick,
|
||||
|
||||
// Send 10 messages
|
||||
Step::Send(1, 0), Step::Send(1, 1), Step::Send(1, 2), Step::Send(1, 3),
|
||||
Step::Send(1, 4), Step::Send(1, 5), Step::Send(1, 6), Step::Send(1, 7),
|
||||
Step::Send(1, 8), Step::Send(1, 9),
|
||||
|
||||
// Tick 1: 10 in mailbox, drain_count(10, 4) = 5
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 1, depth: 5, processed: 5 },
|
||||
|
||||
// Tick 2: 5 remaining, drain_count(5, 4) = 2
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 1, depth: 3, processed: 7 },
|
||||
|
||||
// Tick 3: 3 remaining, drain_count(3, 4) = 3 (below waterlevel → all)
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 1, depth: 0, processed: 10 },
|
||||
Step::ExpectHandled(1, 10),
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spawn_and_send_same_tick() {
|
||||
// Spawn is phase 1, transfer is phase 2, processing is phase 3.
|
||||
// All three happen within a single tick_once call.
|
||||
run(&[
|
||||
Step::Spawn(1),
|
||||
Step::Send(1, 42),
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 1, depth: 0, processed: 1 },
|
||||
Step::ExpectHandled(1, 1),
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stats_track_pool_mutations() {
|
||||
let mut env = Env::new();
|
||||
|
||||
// Before any tick, stats are zeroed
|
||||
assert_eq!(env.num_actors_stat(), 0);
|
||||
assert_eq!(env.depth(), 0);
|
||||
assert_eq!(env.processed(), 0);
|
||||
|
||||
// Spawn 3 + tick → stats reflect 3 actors
|
||||
env.spawn(1);
|
||||
env.spawn(2);
|
||||
env.spawn(3);
|
||||
env.tick();
|
||||
assert_eq!(env.num_actors_stat(), 3);
|
||||
|
||||
// Send 5 to actor 1 + tick → processed increases
|
||||
for i in 0..5 {
|
||||
env.send(1, i);
|
||||
}
|
||||
env.tick();
|
||||
assert_eq!(env.processed(), 5);
|
||||
assert_eq!(env.depth(), 0);
|
||||
assert_eq!(env.num_actors_stat(), 3);
|
||||
|
||||
// Remove actor 2 + tick → stats update
|
||||
env.remove(2);
|
||||
env.tick();
|
||||
assert_eq!(env.num_actors_stat(), 2);
|
||||
assert_eq!(env.pool_len(), 2);
|
||||
}
|
||||
|
||||
/// A long mixed-action sequence: spawns, sends, removes, wrong types,
|
||||
/// and backpressure — all in one run.
|
||||
#[test]
|
||||
fn interleaved_lifecycle() {
|
||||
run(&[
|
||||
// ── Phase 1: build the pool ────────────────────────────────
|
||||
Step::Spawn(1),
|
||||
Step::Spawn(2),
|
||||
Step::Spawn(3),
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 3, depth: 0, processed: 0 },
|
||||
|
||||
// ── Phase 2: normal message flow ───────────────────────────
|
||||
Step::Send(1, 100),
|
||||
Step::Send(2, 200),
|
||||
Step::Send(3, 300),
|
||||
Step::Tick,
|
||||
Step::ExpectHandled(1, 1),
|
||||
Step::ExpectHandled(2, 1),
|
||||
Step::ExpectHandled(3, 1),
|
||||
Step::Expect { pool_len: 3, depth: 0, processed: 3 },
|
||||
|
||||
// ── Phase 3: remove actor 2, send to all 3 ────────────────
|
||||
Step::Remove(2),
|
||||
Step::Send(1, 101),
|
||||
Step::Send(2, 201), // actor 2 gone — dropped at deliver
|
||||
Step::Send(3, 301),
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 2, depth: 0, processed: 5 },
|
||||
Step::ExpectHandled(1, 2),
|
||||
Step::ExpectHandled(2, 1), // unchanged since removal
|
||||
Step::ExpectHandled(3, 2),
|
||||
|
||||
// ── Phase 4: late spawn + immediate send ───────────────────
|
||||
Step::Spawn(4),
|
||||
Step::Send(4, 400),
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 3, depth: 0, processed: 6 },
|
||||
Step::ExpectHandled(4, 1),
|
||||
|
||||
// ── Phase 5: bad types mixed with good ─────────────────────
|
||||
Step::SendBad(1),
|
||||
Step::SendBad(1),
|
||||
Step::SendBad(1),
|
||||
Step::Send(1, 999),
|
||||
Step::Tick,
|
||||
// 4 popped (3 bad + 1 good), only 1 handled by actor
|
||||
Step::Expect { pool_len: 3, depth: 0, processed: 10 },
|
||||
Step::ExpectHandled(1, 3), // 2 from prior phases + 1 good
|
||||
|
||||
// ── Phase 6: remove all, send to ghosts ────────────────────
|
||||
Step::Remove(1),
|
||||
Step::Remove(3),
|
||||
Step::Remove(4),
|
||||
Step::Expect { pool_len: 0, depth: 0, processed: 10 },
|
||||
Step::Send(1, 0),
|
||||
Step::Send(3, 0),
|
||||
Step::Tick,
|
||||
Step::Expect { pool_len: 0, depth: 0, processed: 10 },
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_loop_stops_on_shutdown() {
|
||||
let transfer_rx = Receiver::<Envelope>::new(64);
|
||||
let spawn_rx = Receiver::<(ActorAddress, Box<dyn AnyActor>)>::new(64);
|
||||
let transfer_tx = transfer_rx.new_sender();
|
||||
let spawn_tx = spawn_rx.new_sender();
|
||||
|
||||
let stats = Arc::new(WorkerStats::new());
|
||||
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats);
|
||||
|
||||
let is_running = AtomicBool::new(false);
|
||||
let backoff = BackoffPolicy::default();
|
||||
let address_map = AddressMap::new();
|
||||
let placement = Placement::new(1);
|
||||
let inbox_registry = InboxRegistry::new();
|
||||
let config = RuntimeConfig::default();
|
||||
|
||||
let tc = TickContext {
|
||||
address_map: &address_map,
|
||||
transfer_txs: &[transfer_tx],
|
||||
spawn_txs: &[spawn_tx],
|
||||
placement: &placement,
|
||||
inbox_registry: &inbox_registry,
|
||||
config: &config,
|
||||
};
|
||||
|
||||
thread::scope(|s| {
|
||||
s.spawn(|| worker.run(&tc, &is_running, &backoff));
|
||||
});
|
||||
}
|
||||
136
tests/stats_demo.rs
Normal file
136
tests/stats_demo.rs
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
use swactor::{
|
||||
actor::{ActorAddress, ActorInterface},
|
||||
runtime::{Ctx, Runtime, RuntimeConfig},
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Ping {
|
||||
reply_to: ActorAddress,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Pong;
|
||||
|
||||
struct PingActor;
|
||||
|
||||
impl ActorInterface for PingActor {
|
||||
type Incoming = Ping;
|
||||
type Response = Pong;
|
||||
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Ping) {
|
||||
let _ = ctx.send(msg.reply_to, Pong);
|
||||
}
|
||||
}
|
||||
|
||||
/// Counter that just counts messages.
|
||||
struct Counter(u64);
|
||||
|
||||
impl ActorInterface for Counter {
|
||||
type Incoming = u64;
|
||||
type Response = ();
|
||||
|
||||
fn handle(&mut self, _ctx: &Ctx, _msg: u64) {
|
||||
self.0 += 1;
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stats_demo_single_thread() {
|
||||
let rt = Runtime::new(RuntimeConfig::default());
|
||||
|
||||
// Spawn a few actors
|
||||
let ping1 = rt.spawn(PingActor).unwrap();
|
||||
let ping2 = rt.spawn(PingActor).unwrap();
|
||||
let counter = rt.spawn(Counter(0)).unwrap();
|
||||
|
||||
// Send some messages (they queue up before we tick)
|
||||
for i in 0..20u64 {
|
||||
rt.send_to(counter, i).unwrap();
|
||||
}
|
||||
|
||||
// Stats BEFORE ticking — messages are in the transfer queue, not yet in mailboxes
|
||||
let s = rt.stats();
|
||||
println!("=== Before any ticks ===");
|
||||
print_stats(&s);
|
||||
|
||||
// Tick once — drains transfer queue into mailboxes, then processes messages
|
||||
rt.tick();
|
||||
|
||||
let s = rt.stats();
|
||||
println!("\n=== After 1 tick ===");
|
||||
print_stats(&s);
|
||||
|
||||
// Tick a few more times to drain remaining messages
|
||||
for _ in 0..5 {
|
||||
rt.tick();
|
||||
}
|
||||
|
||||
let s = rt.stats();
|
||||
println!("\n=== After 6 ticks total ===");
|
||||
print_stats(&s);
|
||||
|
||||
assert_eq!(s.num_workers, 1);
|
||||
assert_eq!(s.actors.len(), 3);
|
||||
assert_eq!(s.workers[0].num_actors, 3);
|
||||
// All 20 messages should be processed by now
|
||||
assert_eq!(s.workers[0].mailbox_depth, 0);
|
||||
assert!(s.workers[0].messages_processed >= 20);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stats_demo_multi_thread() {
|
||||
let config = RuntimeConfig {
|
||||
num_threads: 3,
|
||||
..Default::default()
|
||||
};
|
||||
let rt = Runtime::new(config);
|
||||
|
||||
// Spawn actors — round-robin will spread them across 3 workers
|
||||
let mut addrs = Vec::new();
|
||||
for _ in 0..6 {
|
||||
addrs.push(rt.spawn(Counter(0)).unwrap());
|
||||
}
|
||||
|
||||
// Send messages to each actor
|
||||
for &addr in &addrs {
|
||||
for i in 0..10u64 {
|
||||
rt.send_to(addr, i).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
let handle = rt.run().unwrap();
|
||||
|
||||
// Let it process
|
||||
std::thread::sleep(std::time::Duration::from_millis(50));
|
||||
|
||||
let s = handle.runtime.stats();
|
||||
println!("\n=== Multi-threaded (3 workers, 6 actors, 60 messages) ===");
|
||||
print_stats(&s);
|
||||
|
||||
handle.shutdown();
|
||||
handle.join();
|
||||
|
||||
assert_eq!(s.num_workers, 3);
|
||||
assert_eq!(s.actors.len(), 6);
|
||||
let total_processed: u64 = s.workers.iter().map(|w| w.messages_processed).sum();
|
||||
assert_eq!(total_processed, 60);
|
||||
}
|
||||
|
||||
fn print_stats(s: &swactor::runtime::RuntimeStats) {
|
||||
println!(
|
||||
"RuntimeStats(actors={}, workers={})",
|
||||
s.actors.len(),
|
||||
s.num_workers
|
||||
);
|
||||
for w in &s.workers {
|
||||
println!(
|
||||
" Worker {}: {} actors, {} queued, {} processed",
|
||||
w.id, w.num_actors, w.mailbox_depth, w.messages_processed
|
||||
);
|
||||
for (addr, wid) in &s.actors {
|
||||
if *wid == w.id {
|
||||
println!(" - {:x?}...", &addr.0[..4]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue