diff --git a/Cargo.toml b/Cargo.toml index c223be1..353f364 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,3 +25,7 @@ criterion = { version = "0.5", features = ["html_reports"] } [[bench]] name = "runtime_benchmarks" harness = false + +[[bench]] +name = "worker_benchmarks" +harness = false diff --git a/benches/worker_benchmarks.rs b/benches/worker_benchmarks.rs new file mode 100644 index 0000000..7db7129 --- /dev/null +++ b/benches/worker_benchmarks.rs @@ -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 = 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 = 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 = 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 = 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 = 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 = 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 = 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); diff --git a/docs/worker-thread.md b/docs/worker-thread.md new file mode 100644 index 0000000..bb909da --- /dev/null +++ b/docs/worker-thread.md @@ -0,0 +1,550 @@ +# Worker Thread Architecture + +## Structure + +``` +┌─ Worker ───────────────────────────────────────────────────────────────┐ +│ │ +│ id: WorkerId │ +│ │ +│ ┌─ spawn_rx ─────────────────────┐ ┌─ transfer_rx ──────────────────┐│ +│ │ Receiver<(Addr, Box)>│ │ Receiver ││ +│ │ │ │ ││ +│ │ from: Runtime.spawn() │ │ from: other workers, Runtime ││ +│ │ ctx.spawn() │ │ ctx.send() ││ +│ └────────────────────────────────┘ └────────────────────────────────┘│ +│ │ +│ ┌─ ActorPool ──────────────────────────────────────────────────────┐ │ +│ │ │ │ +│ │ actors: HashMap │ │ +│ │ │ │ +│ │ ┌─ ActorSlot [addr_0] ──────────────────────────────────────┐ │ │ +│ │ │ │ │ │ +│ │ │ ┌─ mailbox ──────────────────────────────────────────┐ │ │ │ +│ │ │ │ VecDeque> │ │ │ │ +│ │ │ │ │ │ │ │ +│ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │ +│ │ │ │ │ msg │ │ msg │ │ msg │ │ ... │ <- push_back │ │ │ │ +│ │ │ │ └─────┘ └─────┘ └─────┘ └─────┘ │ │ │ │ +│ │ │ │ pop_front -> untyped; Box │ │ │ │ +│ │ │ └────────────────────────────────────────────────────┘ │ │ │ +│ │ │ │ │ │ +│ │ │ ┌─ actor ────────────────────────────────────────────┐ │ │ │ +│ │ │ │ Box │ │ │ │ +│ │ │ │ │ │ │ │ +│ │ │ │ wraps Actor(A) where A: ActorInterface │ │ │ │ +│ │ │ │ │ │ │ │ +│ │ │ │ handle_any(ctx, msg): │ │ │ │ +│ │ │ │ downcast Box -> A::Incoming │ │ │ │ +│ │ │ │ ok -> A.handle(ctx, typed_msg) │ │ │ │ +│ │ │ │ err -> silently drop │ │ │ │ +│ │ │ └────────────────────────────────────────────────────┘ │ │ │ +│ │ │ │ │ │ +│ │ └───────────────────────────────────────────────────────────┘ │ │ +│ │ │ │ +│ │ ┌─ ActorSlot [addr_1] ──────────────────────────────────────┐ │ │ +│ │ │ ... │ │ │ +│ │ └───────────────────────────────────────────────────────────┘ │ │ +│ │ │ │ +│ └──────────────────────────────────────────────────────────────────┘ │ +│ │ +└────────────────────────────────────────────────────────────────────────┘ +``` + +## Shared State (borrowed via TickContext) + +Lives on `Arc`, 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 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) ││ +│ │ │ ││ +│ │ v ││ +│ │ pool.insert(addr, actor) ││ +│ │ │ ││ +│ │ v ││ +│ │ ActorSlot { ││ +│ │ mailbox: VecDeque::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)>> │ ││ +│ │ └────────────────────────────────────────────────────────────┘ ││ +│ │ ││ +│ │ 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 │ │ │ +│ │ │ │ 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 ─────────────────────────────────────────────────────┐ +│ │ +│ ┌─ ring: ArrayQueue ──────────────────────────────────────┐ │ +│ │ fixed capacity, lock-free, bounded │ │ +│ │ ┌───┬───┬───┬───┬───┬───┬───┬───┐ │ │ +│ │ │ │ │ │ │ │ │ │ │ (pre-allocated) │ │ +│ │ └───┴───┴───┴───┴───┴───┴───┴───┘ │ │ +│ └────────────────────────────────────────────────────────────┘ │ +│ │ +│ ┌─ overflow: SegQueue ────────────────────────────────────┐ │ +│ │ 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 ─────────┐ ┌─ Receiver ─────────┐ │ +│ │ Arc>│ │ Arc> │ │ +│ │ .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 > │ +│ │ +│ ┌──────────────────────────────────────────────────────────────────┐ │ +│ │ 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> > │ +│ │ +│ For addresses belonging to external Inbox, not actors. │ +│ │ +│ ┌─ Registration ─────────────────────────────────────────────────┐ │ +│ │ │ │ +│ │ Runtime.new_inbox::() │ │ +│ │ addr = ActorAddress::new_random() │ │ +│ │ receiver = Receiver::::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 -> M, push into Receiver │ │ +│ │ │ │ +│ └────────────────────────────────────────────────────────────────┘ │ +│ │ +│ ┌─ Consumption (user code) ──────────────────────────────────────┐ │ +│ │ │ │ +│ │ let inbox = rt.new_inbox::()?; │ │ +│ │ // 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 ────────────────────────────────────────────────────────────┐ +│ │ +│ ┌───────────┐ ┌────────────┐ ┌─────────────┐ ┌────────────┐ │ +│ │AddressMap │ │ Placement │ │InboxRegistry│ │ is_running │ │ +│ │ addr->wid │ │ round-robin│ │ addr->Sender│ │ AtomicBool │ │ +│ └─────┬─────┘ └──────┬─────┘ └──────┬──────┘ └──────┬─────┘ │ +│ │ │ │ │ │ +│ ┌─────┴───────────────┴──────────────┴───────────────┴────────────────┐ │ +│ │ TickContext (borrows all above) │ │ +│ └──────────────────────────┬──────────────────────────────────────────┘ │ +│ │ │ +│ ┌─ transfer_txs[] ────┐ │ ┌─ spawn_txs[] ──────┐ │ +│ │ [0]: Sender│ │ +│ │ [1]: Sender│ │ +│ │ [2]: Sender│ │ +│ └──┬──────┬──────┬────┘ │ └──┬──────┬──────┬────┘ │ +│ │ │ │ │ │ │ │ │ +└────┼──────┼──────┼────────┼──────┼──────┼──────┼──────────────────────────┘ + │ │ │ │ │ │ │ + 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 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 │ + │ │ + └───────────────────┘ +``` + diff --git a/src/actor.rs b/src/actor.rs index 7a7a9b8..8f5bc11 100644 --- a/src/actor.rs +++ b/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 -where - A: ActorInterface, -{ - addr: ActorAddress, - mailbox: Mailbox, - inner: A, -} +/// The actor process as represented in the Runtime — thin wrapper around user state. +pub(crate) struct Actor(A); impl Actor { - pub(crate) fn new(addr: ActorAddress, mailbox: Mailbox, 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) -> bool; + fn handle_any(&mut self, ctx: &Ctx, msg: Box); } impl AnyActor for Actor 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) -> bool { + fn handle_any(&mut self, ctx: &Ctx, msg: Box) { if let Ok(typed) = msg.downcast::() { - self.mailbox.push(*typed); - true - } else { - false + self.0.handle(ctx, *typed); } } } diff --git a/src/python.rs b/src/python.rs index 57b5a5e..0c15d09 100644 --- a/src/python.rs +++ b/src/python.rs @@ -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 = Box::new(actor); + let boxed: Box = 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, + #[pyo3(get)] + workers: Vec, } #[pymethods] @@ -498,19 +520,13 @@ impl PyRuntimeStats { self.num_actors, self.num_workers ); - // Group actors by worker - let mut by_worker: std::collections::BTreeMap> = - 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 = snapshot + let stats = runtime.stats(); + let actors: Vec = stats + .actors .into_iter() .map(|(addr, wid)| PyActorInfo { address: PyActorAddress::from(addr), - worker_id: wid.as_usize(), + worker_id: wid, + }) + .collect(); + let workers: Vec = 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::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; m.add_class::()?; Ok(()) } diff --git a/src/runtime.rs b/src/runtime.rs index 7680899..b797164 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -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, +} + /// Generic message inbox for receiving messages outside of the runtime. pub struct Inbox { addr: ActorAddress, @@ -82,9 +97,7 @@ impl<'a> Ctx<'a> { /// Spawn a new actor, returning its address. pub fn spawn(&self, actor: A) -> Result { let addr = ActorAddress::new_random(); - let waterlevel = self.inner.mailbox_waterlevel(); - let actor = Actor::new(addr, Mailbox::new(waterlevel), actor); - let boxed: Box = Box::new(actor); + let boxed: Box = Box::new(Actor::new(actor)); self.inner.spawn_any(addr, boxed)?; Ok(addr) } @@ -115,6 +128,7 @@ pub struct Runtime { spawn_txs: Vec)>>, placement: Placement, is_running: AtomicBool, + worker_stats: Vec>, /// Single-threaded mode: worker stored inline single_worker: Option>, /// 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 = Box::new(actor); + let boxed: Box = 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 diff --git a/src/worker.rs b/src/worker/mod.rs similarity index 67% rename from src/worker.rs rename to src/worker/mod.rs index 1ca7c0f..493b9c4 100644 --- a/src/worker.rs +++ b/src/worker/mod.rs @@ -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], - pub spawn_txs: &'a [Sender<(ActorAddress, Box)>], - 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], + pub(crate) spawn_txs: &'a [Sender<(ActorAddress, Box)>], + 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, spawn_rx: Receiver<(ActorAddress, Box)>, + stats: Arc, } impl Worker { - pub fn new( + pub(crate) fn new( id: WorkerId, transfer_rx: Receiver, spawn_rx: Receiver<(ActorAddress, Box)>, + stats: Arc, ) -> 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)>> = 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>, + actor: Box, +} + +/// Per-worker actor storage. Owns per-actor mailboxes. pub(crate) struct ActorPool { - actors: HashMap>, + actors: HashMap, } impl ActorPool { @@ -179,42 +223,58 @@ impl ActorPool { } pub fn insert(&mut self, addr: ActorAddress, actor: Box) { - self.actors.insert(addr, actor); + self.actors.insert(addr, ActorSlot { + mailbox: VecDeque::new(), + actor, + }); } pub fn remove(&mut self, addr: &ActorAddress) -> Option> { - 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) -> 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 { +pub(crate) struct Mailbox { queue: VecDeque, waterlevel: usize, } @@ -247,12 +307,9 @@ impl Mailbox { /// - `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; diff --git a/src/worker/tests.rs b/src/worker/tests.rs new file mode 100644 index 0000000..720d968 --- /dev/null +++ b/src/worker/tests.rs @@ -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); + +impl AnyActor for CounterActor { + fn handle_any(&mut self, _ctx: &Ctx, msg: Box) { + if msg.downcast::().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, + feed_transfer: crate::channel::Sender, + feed_spawn: crate::channel::Sender<(ActorAddress, Box)>, + address_map: AddressMap, + placement: Placement, + inbox_registry: InboxRegistry, + config: RuntimeConfig, + tc_transfer_txs: Vec>, + tc_spawn_txs: Vec)>>, + counters: HashMap>, +} + +impl Env { + fn new() -> Self { + Self::with_config(RuntimeConfig::default()) + } + + fn with_config(config: RuntimeConfig) -> Self { + let transfer_rx = Receiver::::new(256); + let spawn_rx = Receiver::<(ActorAddress, Box)>::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 = 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::::new(64); + let spawn_rx = Receiver::<(ActorAddress, Box)>::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)); + }); +} diff --git a/tests/runtime_api_tests.rs b/tests/runtime_api.rs similarity index 100% rename from tests/runtime_api_tests.rs rename to tests/runtime_api.rs diff --git a/tests/stats_demo.rs b/tests/stats_demo.rs new file mode 100644 index 0000000..49901a0 --- /dev/null +++ b/tests/stats_demo.rs @@ -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]); + } + } + } +}