zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: zacheryasc:6cb5044d064bfcb5f38f7524ea2a265ea4b4be0e
Branches Tags
zacheryasc:master
zacheryasc:development
zacheryasc:backup/pre-rewrite-msg
..
compare: zacheryasc:73f667227a078b8207bc3c4494cb05e29ded0b24
Branches Tags
zacheryasc:development
zacheryasc:master
zacheryasc:backup/pre-rewrite-msg

4 commits
6cb5044d06 ... 73f667227a

Author SHA1 Message Date
Zachery Aaron Shores-Chmielewski
73f667227a feat: worker thread info added to runtime stats 2026-02-06 21:20:49 +07:00
Zachery Aaron Shores-Chmielewski
8db066f646 feat: worker thread architecture diagram 2026-02-06 21:20:49 +07:00
zacheryasc
be0eb8bc71 feat: jupyter example (#10) 2026-02-06 21:20:49 +07:00
zacheryasc
75dbd0b540 feat: runtime information display (#9) 
Show basic information from the runtime, such as number of actors, worker threads, etc.
 2026-02-06 21:20:49 +07:00
13 changed files with 3590 additions and 27 deletions
Show stats Expand all files Collapse all files

550
docs/worker-thread.md Normal file
View 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 │
│ │
└───────────────────┘
```

98
examples/python/.ipynb_checkpoints/getting_started-checkpoint.ipynb Normal file
View file

@ -0,0 +1,98 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Swactor — Getting Started\n",
"\n",
"This notebook walks through the basics of the swactor actor runtime from Python."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from swactor import Runtime"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Single-threaded: spawn, send, tick\n",
"\n",
"The simplest way to use swactor is with `tick()` — manually stepping the runtime one tick at a time."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"def echo(ctx, msg):\n",
" ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
"\n",
"rt = Runtime()\n",
"addr = rt.spawn(echo)\n",
"inbox = rt.inbox()\n",
"\n",
"rt.send(addr, {\"name\": \"world\", \"reply_to\": inbox.addr})\n",
"rt.tick()\n",
"\n",
"print(inbox.try_recv())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Multi-threaded: background runtime\n",
"\n",
"For real workloads you can run the runtime on background threads with `RuntimeConfig`."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import time\n",
"from swactor import Runtime, RuntimeConfig\n",
"\n",
"def echo(ctx, msg):\n",
" ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
"\n",
"rt = Runtime(RuntimeConfig(num_threads=2))\n",
"addr = rt.spawn(echo)\n",
"inbox = rt.inbox()\n",
"handle = rt.run()\n",
"\n",
"for name in [\"alice\", \"bob\", \"charlie\"]:\n",
" handle.send(addr, {\"name\": name, \"reply_to\": inbox.addr})\n",
" time.sleep(0.05) # give the runtime a moment\n",
" print(inbox.try_recv())\n",
"\n",
"handle.shutdown()\n",
"handle.join()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"name": "python",
"version": "3.9.0"
}
},
"nbformat": 4,
"nbformat_minor": 4
}

98
examples/python/getting_started.ipynb Normal file
View file

@ -0,0 +1,98 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Swactor — Getting Started\n",
"\n",
"This notebook walks through the basics of the swactor actor runtime from Python."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from swactor import Runtime"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Single-threaded: spawn, send, tick\n",
"\n",
"The simplest way to use swactor is with `tick()` — manually stepping the runtime one tick at a time."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"def echo(ctx, msg):\n",
" ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
"\n",
"rt = Runtime()\n",
"addr = rt.spawn(echo)\n",
"inbox = rt.inbox()\n",
"\n",
"rt.send(addr, {\"name\": \"world\", \"reply_to\": inbox.addr})\n",
"rt.tick()\n",
"\n",
"print(inbox.try_recv())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Multi-threaded: background runtime\n",
"\n",
"For real workloads you can run the runtime on background threads with `RuntimeConfig`."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import time\n",
"from swactor import Runtime, RuntimeConfig\n",
"\n",
"def echo(ctx, msg):\n",
" ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
"\n",
"rt = Runtime(RuntimeConfig(num_threads=2))\n",
"addr = rt.spawn(echo)\n",
"inbox = rt.inbox()\n",
"handle = rt.run()\n",
"\n",
"for name in [\"alice\", \"bob\", \"charlie\"]:\n",
" handle.send(addr, {\"name\": name, \"reply_to\": inbox.addr})\n",
" time.sleep(0.05) # give the runtime a moment\n",
" print(inbox.try_recv())\n",
"\n",
"handle.shutdown()\n",
"handle.join()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"name": "python",
"version": "3.9.0"
}
},
"nbformat": 4,
"nbformat_minor": 4
}

2
examples/hello_async.py → examples/python/hello_async.py
View file

@ -30,6 +30,8 @@ async def main():
reply = await recv(inbox)
print(reply)
# show us our actors!
print(handle.stats())
handle.shutdown()
handle.join()

0
examples/hello_single_thread.py → examples/python/hello_single_thread.py
View file

3
pyproject.toml
View file

@ -7,5 +7,8 @@ name = "swactor"
version = "0.1.0"
requires-python = ">=3.9"
[dependency-groups]
dev = ["jupyter", "ipykernel"]
[tool.maturin]
features = ["python"]

10
src/address_map.rs
View file

@ -49,6 +49,16 @@ impl AddressMap {
pub fn len(&self) -> usize {
self.inner.read().unwrap().len()
}
/// Returns a snapshot of all (address, worker) pairs.
pub fn snapshot(&self) -> Vec<(ActorAddress, WorkerId)> {
self.inner
.read()
.unwrap()
.iter()
.map(|(addr, wid)| (*addr, *wid))
.collect()
}
}
/// Round-robin actor placement strategy.

139
src/python.rs
View file

@ -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 ───────────────────────────────────────────────────────────
@ -176,16 +175,14 @@ impl ActorInterface for PyActor {
for effect in effects {
match effect {
Effect::Send { addr, msg } => {
let _ = ctx.raw_inner().send_via_queue(
let _ = ctx.raw_inner().send_any(
addr,
Box::new(PyMsg(msg)) as Box<dyn Any + Send>,
);
}
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);
}
}
@ -364,6 +361,14 @@ impl PyRuntime {
})
}
fn stats(&self) -> PyResult<PyRuntimeStats> {
let rt = self
.inner
.as_ref()
.ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?;
Ok(build_stats(rt))
}
fn shutdown(&self) -> PyResult<()> {
let rt = self
.inner
@ -419,6 +424,16 @@ impl PyRuntimeHandle {
Ok(PyInbox { inner: inbox })
}
fn stats(&self) -> PyResult<PyRuntimeStats> {
let handle = self
.inner
.as_ref()
.ok_or_else(|| {
pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()")
})?;
Ok(build_stats(&handle.runtime))
}
fn shutdown(&self) -> PyResult<()> {
let handle = self
.inner
@ -442,6 +457,117 @@ impl PyRuntimeHandle {
}
}
// ─── ActorInfo / RuntimeStats ────────────────────────────────────────────────
#[pyclass(name = "ActorInfo")]
#[derive(Clone)]
pub struct PyActorInfo {
#[pyo3(get)]
address: PyActorAddress,
#[pyo3(get)]
worker_id: usize,
}
#[pymethods]
impl PyActorInfo {
fn __repr__(&self) -> String {
let hex = self.address.hex();
format!("ActorInfo(address={hex}, worker={})", self.worker_id)
}
}
#[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 {
#[pyo3(get)]
num_actors: usize,
#[pyo3(get)]
num_workers: usize,
#[pyo3(get)]
actors: Vec<PyActorInfo>,
#[pyo3(get)]
workers: Vec<PyWorkerInfo>,
}
#[pymethods]
impl PyRuntimeStats {
fn __repr__(&self) -> String {
let mut out = format!(
"RuntimeStats(actors={}, workers={})",
self.num_actors, self.num_workers
);
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()));
}
}
}
out
}
fn __str__(&self) -> String {
self.__repr__()
}
}
fn build_stats(runtime: &Runtime) -> PyRuntimeStats {
let stats = runtime.stats();
let actors: Vec<PyActorInfo> = stats
.actors
.into_iter()
.map(|(addr, wid)| PyActorInfo {
address: PyActorAddress::from(addr),
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: stats.num_workers,
actors,
workers,
}
}
// ─── Module registration ─────────────────────────────────────────────────────
pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
@ -451,5 +577,8 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<PyRuntimeConfig>()?;
m.add_class::<PyRuntime>()?;
m.add_class::<PyRuntimeHandle>()?;
m.add_class::<PyActorInfo>()?;
m.add_class::<PyWorkerInfo>()?;
m.add_class::<PyRuntimeStats>()?;
Ok(())
}

57
src/runtime.rs
View file

@ -10,10 +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::{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,
@ -112,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()
@ -139,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 {
@ -151,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 {
@ -165,6 +185,7 @@ impl Runtime {
spawn_txs,
placement,
is_running: AtomicBool::new(false),
worker_stats,
single_worker: Some(RefCell::new(worker)),
pending_workers: None,
}
@ -178,6 +199,7 @@ impl Runtime {
spawn_txs,
placement,
is_running: AtomicBool::new(false),
worker_stats,
single_worker: None,
pending_workers: Some(workers),
}
@ -276,6 +298,37 @@ impl Runtime {
})
}
/// 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
};
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
pub fn shutdown(&self) {
self.is_running.store(false, Ordering::Release);

46
src/worker/mod.rs
View file

@ -1,7 +1,8 @@
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};
@ -11,6 +12,23 @@ use crate::config::{BackoffPolicy, RuntimeConfig};
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(crate) address_map: &'a AddressMap,
@ -27,6 +45,7 @@ pub(crate) struct Worker {
pool: ActorPool,
transfer_rx: Receiver<Envelope>,
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
stats: Arc<WorkerStats>,
}
impl Worker {
@ -34,12 +53,14 @@ impl Worker {
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,
}
}
@ -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
}
@ -216,9 +244,9 @@ impl ActorPool {
}
}
/// 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;
/// 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());
@ -227,17 +255,21 @@ impl ActorPool {
for _ in 0..n {
if let Some(msg) = slot.mailbox.pop_front() {
slot.actor.handle_any(&ctx, msg);
count += 1;
}
}
did_work = true;
}
}
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()
}
}

26
src/worker/tests.rs
View file

@ -8,7 +8,7 @@ use crate::address_map::{AddressMap, Placement, WorkerId};
use crate::channel::Receiver;
use crate::config::{BackoffPolicy, RuntimeConfig};
use crate::runtime::{ContextInner, Ctx, Envelope, InboxRegistry};
use super::{ActorPool, Mailbox, TickContext, Worker};
use super::{ActorPool, Mailbox, TickContext, Worker, WorkerStats};
use crate::Error;
// ── Test helpers ────────────────────────────────────────────────────
@ -328,13 +328,13 @@ fn pool_tick_all_processes_messages() {
pool.deliver(&addr, Box::new(3u64));
let stub = StubContextInner { waterlevel: 100 };
let did_work = pool.tick_all(&stub);
assert!(did_work);
let processed = pool.tick_all(&stub);
assert_eq!(processed, 3);
assert_eq!(counter.load(Ordering::Relaxed), 3);
}
#[test]
fn pool_tick_all_empty_returns_false() {
fn pool_tick_all_empty_returns_zero() {
let mut pool = ActorPool::new();
let addr = make_addr(1);
let (actor, _) = make_test_actor(addr);
@ -342,8 +342,8 @@ fn pool_tick_all_empty_returns_false() {
// No messages delivered
let stub = StubContextInner { waterlevel: 100 };
let did_work = pool.tick_all(&stub);
assert!(!did_work);
let processed = pool.tick_all(&stub);
assert_eq!(processed, 0);
}
// ── Worker tests ────────────────────────────────────────────────────
@ -356,7 +356,7 @@ fn worker_tick_once_no_work() {
let transfer_tx = transfer_rx.new_sender();
let spawn_tx = spawn_rx.new_sender();
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new()));
let address_map = AddressMap::new();
let placement = Placement::new(1);
@ -387,7 +387,7 @@ fn worker_tick_once_drains_spawns() {
let (actor, _) = make_test_actor(addr);
spawn_tx.try_send((addr, actor)).ok().unwrap();
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new()));
let address_map = AddressMap::new();
let placement = Placement::new(1);
@ -420,7 +420,7 @@ fn worker_tick_once_drains_transfers() {
let addr = make_addr(1);
let (actor, counter) = make_test_actor(addr);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new()));
// Spawn the actor first
spawn_tx.try_send((addr, actor)).ok().unwrap();
@ -464,7 +464,7 @@ fn worker_tick_once_processes_messages() {
let addr = make_addr(1);
let (actor, counter) = make_test_actor(addr);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new()));
spawn_tx.try_send((addr, actor)).ok().unwrap();
@ -510,7 +510,7 @@ fn worker_tick_once_multiple_spawns() {
spawn_tx.try_send((addr, actor)).ok().unwrap();
}
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new()));
let address_map = AddressMap::new();
let placement = Placement::new(1);
@ -543,7 +543,7 @@ fn worker_tick_once_wrong_type_no_panic() {
let addr = make_addr(1);
let (actor, counter) = make_test_actor(addr);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new()));
spawn_tx.try_send((addr, actor)).ok().unwrap();
@ -584,7 +584,7 @@ fn worker_run_stops_on_signal() {
let transfer_tx = transfer_rx.new_sender();
let spawn_tx = spawn_rx.new_sender();
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx);
let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, Arc::new(WorkerStats::new()));
let is_running = AtomicBool::new(false); // start as false → should exit immediately
let backoff = BackoffPolicy::default();

136
tests/stats_demo.rs Normal file
View 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]);
}
}
}
}

2452
uv.lock
View file

File diff suppressed because it is too large Load diff