Compare commits
4 commits
6cb5044d06
...
73f667227a
| Author | SHA1 | Date | |
|---|---|---|---|
|
73f667227a | ||
|
|
8db066f646 | ||
| be0eb8bc71 | |||
| 75dbd0b540 |
13 changed files with 3590 additions and 27 deletions
550
docs/worker-thread.md
Normal file
550
docs/worker-thread.md
Normal file
|
|
@ -0,0 +1,550 @@
|
|||
# Worker Thread Architecture
|
||||
|
||||
## Structure
|
||||
|
||||
```
|
||||
┌─ Worker ───────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ id: WorkerId │
|
||||
│ │
|
||||
│ ┌─ spawn_rx ─────────────────────┐ ┌─ transfer_rx ──────────────────┐│
|
||||
│ │ Receiver<(Addr, Box<AnyActor>)>│ │ Receiver<Envelope> ││
|
||||
│ │ │ │ ││
|
||||
│ │ from: Runtime.spawn() │ │ from: other workers, Runtime ││
|
||||
│ │ ctx.spawn() │ │ ctx.send() ││
|
||||
│ └────────────────────────────────┘ └────────────────────────────────┘│
|
||||
│ │
|
||||
│ ┌─ ActorPool ──────────────────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ actors: HashMap<ActorAddress, ActorSlot> │ │
|
||||
│ │ │ │
|
||||
│ │ ┌─ ActorSlot [addr_0] ──────────────────────────────────────┐ │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ ┌─ mailbox ──────────────────────────────────────────┐ │ │ │
|
||||
│ │ │ │ VecDeque<Box<dyn Any + Send>> │ │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │
|
||||
│ │ │ │ │ msg │ │ msg │ │ msg │ │ ... │ <- push_back │ │ │ │
|
||||
│ │ │ │ └─────┘ └─────┘ └─────┘ └─────┘ │ │ │ │
|
||||
│ │ │ │ pop_front -> untyped; Box<Any> │ │ │ │
|
||||
│ │ │ └────────────────────────────────────────────────────┘ │ │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ ┌─ actor ────────────────────────────────────────────┐ │ │ │
|
||||
│ │ │ │ Box<dyn AnyActor> │ │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ │ wraps Actor<A>(A) where A: ActorInterface │ │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ │ handle_any(ctx, msg): │ │ │ │
|
||||
│ │ │ │ downcast Box<Any> -> A::Incoming │ │ │ │
|
||||
│ │ │ │ ok -> A.handle(ctx, typed_msg) │ │ │ │
|
||||
│ │ │ │ err -> silently drop │ │ │ │
|
||||
│ │ │ └────────────────────────────────────────────────────┘ │ │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ └───────────────────────────────────────────────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ │ ┌─ ActorSlot [addr_1] ──────────────────────────────────────┐ │ │
|
||||
│ │ │ ... │ │ │
|
||||
│ │ └───────────────────────────────────────────────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Shared State (borrowed via TickContext)
|
||||
|
||||
Lives on `Arc<Runtime>`, shared read-only across all worker threads.
|
||||
|
||||
```
|
||||
┌─ TickContext<'a> ──────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ address_map: &AddressMap -- ActorAddress -> WorkerId lookup │
|
||||
│ transfer_txs: &[Sender] -- one Sender per worker (cross-send) │
|
||||
│ spawn_txs: &[Sender] -- one Sender per worker (spawn reqs) │
|
||||
│ placement: &Placement -- round-robin next-worker picker │
|
||||
│ inbox_registry: &InboxRegistry -- external Inbox<M> receivers │
|
||||
│ config: &RuntimeConfig -- waterlevel, backoff params, etc. │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Run Loop
|
||||
|
||||
```
|
||||
┌─ Worker::run ──────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ ┌────────────────────────────────────┐ │
|
||||
│ │ is_running.load() ? │ │
|
||||
│ └──────────┬─────────────────────────┘ │
|
||||
│ yes │ │
|
||||
│ v │
|
||||
│ ┌────────────────────────────────────┐ │
|
||||
│ │ tick_once(&tc) │─────────┐ │
|
||||
│ └──────────┬─────────────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ ┌────┴────┐ │ │
|
||||
│ v v │ │
|
||||
│ did work no work │ │
|
||||
│ │ │ │ │
|
||||
│ v v │ │
|
||||
│ idle = 0 idle++ │ │
|
||||
│ │ │ │ │
|
||||
│ │ ┌────┴────────────────────────┐ │ │
|
||||
│ │ │ idle < spin_thr: spin │ │ │
|
||||
│ │ │ idle < yield_thr: yield │ │ │
|
||||
│ │ │ else: sleep(incr, capped) │ │ │
|
||||
│ │ └─────────────────────────┬───┘ │ │
|
||||
│ │ │ │ │
|
||||
│ └──────────┬───────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ └──── loop back ───────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Tick Once (four phases)
|
||||
|
||||
```
|
||||
┌─ tick_once ────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ PHASE 1 --- Drain Spawn Queue │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ spawn_rx --try_recv()--> (addr, Box<dyn AnyActor>) ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ pool.insert(addr, actor) ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ ActorSlot { ││
|
||||
│ │ mailbox: VecDeque::new() ││
|
||||
│ │ actor: <the new actor> ││
|
||||
│ │ } ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │ │
|
||||
│ v │
|
||||
│ PHASE 2 --- Drain Transfer Queue │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ transfer_rx --try_recv()--> Envelope { dest, payload } ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ pool.deliver(&dest, payload) ││
|
||||
│ │ │ ││
|
||||
│ │ v ││
|
||||
│ │ slot.mailbox.push_back(msg) ││
|
||||
│ │ (untyped; type check at handle time) ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │ │
|
||||
│ v │
|
||||
│ PHASE 3 --- Tick All Actors │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ ┌─ WorkerContext (on stack) ─────────────────────────────────┐ ││
|
||||
│ │ │ implements ContextInner │ ││
|
||||
│ │ │ owns pending_local: RefCell<Vec<(Addr, Box<Any>)>> │ ││
|
||||
│ │ └────────────────────────────────────────────────────────────┘ ││
|
||||
│ │ ││
|
||||
│ │ for each (addr, slot) in pool: ││
|
||||
│ │ ││
|
||||
│ │ ┌─ drain_count ──────────────────────────────────────────┐ ││
|
||||
│ │ │ len = slot.mailbox.len() │ ││
|
||||
│ │ │ len < waterlevel --> n = len (drain all) │ ││
|
||||
│ │ │ len >= waterlevel --> n = len / 2 (backpressure) │ ││
|
||||
│ │ └────────────────────────────────────────────────────────┘ ││
|
||||
│ │ ││
|
||||
│ │ ctx = Ctx { inner: &worker_ctx, self_addr: addr } ││
|
||||
│ │ ││
|
||||
│ │ repeat n times: ││
|
||||
│ │ msg = slot.mailbox.pop_front() ││
|
||||
│ │ slot.actor.handle_any(&ctx, msg) ││
|
||||
│ │ │ ││
|
||||
│ │ │ actor calls ctx.send() or ctx.spawn() ││
|
||||
│ │ v ││
|
||||
│ │ ││
|
||||
│ │ ┌─ WorkerContext routes ─────────────────────────────────────┐ ││
|
||||
│ │ │ │ ││
|
||||
│ │ │ send_any(addr, msg): │ ││
|
||||
│ │ │ ┌──────────────┬──────────────┬─────────────────┐ │ ││
|
||||
│ │ │ │ same worker │ other worker │ unknown addr │ │ ││
|
||||
│ │ │ │ │ │ │ │ ││
|
||||
│ │ │ │ pending_ │ transfer_tx │ inbox_registry │ │ ││
|
||||
│ │ │ │ local.push()│ [wid].send()│ .try_deliver() │ │ ││
|
||||
│ │ │ └──────────────┴──────────────┴─────────────────┘ │ ││
|
||||
│ │ │ │ ││
|
||||
│ │ │ spawn_any(addr, actor): │ ││
|
||||
│ │ │ wid = placement.next_worker() │ ││
|
||||
│ │ │ address_map.insert(addr, wid) │ ││
|
||||
│ │ │ spawn_txs[wid].send((addr, actor)) │ ││
|
||||
│ │ │ │ ││
|
||||
│ │ └────────────────────────────────────────────────────────────┘ ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │ │
|
||||
│ v │
|
||||
│ PHASE 4 --- Drain Pending Local │
|
||||
│ ┌────────────────────────────────────────────────────────────────────┐│
|
||||
│ │ ││
|
||||
│ │ for (addr, msg) in pending_local.into_inner(): ││
|
||||
│ │ pool.deliver(&addr, msg) ││
|
||||
│ │ --> slot.mailbox.push_back(msg) ││
|
||||
│ │ ││
|
||||
│ │ these sit in the mailbox until NEXT tick ││
|
||||
│ │ ││
|
||||
│ └────────────────────────────────────────────────────────────────────┘│
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## External Interactions
|
||||
|
||||
Everything the worker talks to, and everything that talks to it.
|
||||
|
||||
### Who writes into the worker's queues
|
||||
|
||||
```
|
||||
┌─ User Code ────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ let rt = Runtime::new(config); │
|
||||
│ let addr = rt.spawn(my_actor)?; // --┐ │
|
||||
│ rt.send_to(addr, MyMsg(42))?; // --┤ │
|
||||
│ // │ │
|
||||
└──────────────────────────────────┼──┼───────────────────────────────────┘
|
||||
│ │
|
||||
┌───────────────────────────┘ │
|
||||
│ │
|
||||
v v
|
||||
┌─ Runtime ──────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ spawn(): │
|
||||
│ addr = ActorAddress::new_random() │
|
||||
│ wid = placement.next_worker() -- round-robin pick │
|
||||
│ address_map.insert(addr, wid) -- register globally │
|
||||
│ spawn_txs[wid].try_send((addr, boxed)) -- push to worker queue │
|
||||
│ │ │
|
||||
│ │ ┌────────────────────────────────────────────┐ │
|
||||
│ └──────>│ Worker.spawn_rx (Receiver side) │ │
|
||||
│ └────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ send_to(): │
|
||||
│ wid = address_map.lookup(&addr) │
|
||||
│ transfer_txs[wid].try_send(Envelope::new(addr, msg)) │
|
||||
│ │ │
|
||||
│ │ ┌────────────────────────────────────────────┐ │
|
||||
│ └──────>│ Worker.transfer_rx (Receiver side) │ │
|
||||
│ └────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### Who the worker talks to during a tick
|
||||
|
||||
```
|
||||
┌─ Worker (during phase 3: tick_all) ────────────────────────────────────┐
|
||||
│ │
|
||||
│ An actor calls ctx.send(addr, msg) or ctx.spawn(new_actor). │
|
||||
│ These go through WorkerContext, which implements ContextInner. │
|
||||
│ │
|
||||
│ ctx.send(addr, msg) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ ┌─ WorkerContext.send_any ─────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ address_map.lookup(addr) --> which worker owns this actor? │ │
|
||||
│ │ │ │ │
|
||||
│ │ ┌────┴──────────────┬──────────────────┬──────────────────┐ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ v v v │ │ │
|
||||
│ │ SAME WORKER OTHER WORKER NOT FOUND │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ pending_local │ transfer_txs │ inbox_registry │ │ │
|
||||
│ │ │ .push(addr,msg) │ [wid].send() │ .try_deliver() │ │ │
|
||||
│ │ │ │ │ │ │ │
|
||||
│ │ │ stays in this │ crosses to │ goes to an │ │ │
|
||||
│ │ │ worker; delivered │ another worker │ external │ │ │
|
||||
│ │ │ in phase 4 │ thread's │ Inbox<M> │ │ │
|
||||
│ │ │ │ transfer_rx │ receiver │ │ │
|
||||
│ │ └───────────────────┴──────────────────┴──────────────────┘ │ │
|
||||
│ │ │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ctx.spawn(new_actor) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ ┌─ WorkerContext.spawn_any ────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ wid = placement.next_worker() -- round-robin target │ │
|
||||
│ │ address_map.insert(addr, wid) -- register in global map │ │
|
||||
│ │ spawn_txs[wid].try_send(...) -- enqueue for target worker │ │
|
||||
│ │ │ │
|
||||
│ │ may land on THIS worker or a DIFFERENT worker │ │
|
||||
│ │ target picks it up in phase 1 of its next tick │ │
|
||||
│ │ │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### The channel connecting everything
|
||||
|
||||
Each worker has two inbound channels. The channels are lock-free MPSC queues
|
||||
backed by `crossbeam::ArrayQueue` with a `SegQueue` overflow.
|
||||
|
||||
```
|
||||
┌─ HybridChannel<T> ─────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ ┌─ ring: ArrayQueue<T> ──────────────────────────────────────┐ │
|
||||
│ │ fixed capacity, lock-free, bounded │ │
|
||||
│ │ ┌───┬───┬───┬───┬───┬───┬───┬───┐ │ │
|
||||
│ │ │ │ │ │ │ │ │ │ │ (pre-allocated) │ │
|
||||
│ │ └───┴───┴───┴───┴───┴───┴───┴───┘ │ │
|
||||
│ └────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ┌─ overflow: SegQueue<T> ────────────────────────────────────┐ │
|
||||
│ │ unbounded, lock-free, linked nodes │ │
|
||||
│ │ used only when ring is full │ │
|
||||
│ └────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ push(v): try ring first, spill to overflow │
|
||||
│ pop(): drain ring first, then overflow │
|
||||
│ │
|
||||
│ ┌─ Sender<T> ─────────┐ ┌─ Receiver<T> ─────────┐ │
|
||||
│ │ Arc<HybridChannel<T>>│ │ Arc<HybridChannel<T>> │ │
|
||||
│ │ .try_send(v) │──────>│ .try_recv() -> Option │ │
|
||||
│ │ clonable (new_sender)│ same │ single consumer │ │
|
||||
│ └──────────────────────┘ Arc └────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
|
||||
Who holds what:
|
||||
|
||||
transfer channel:
|
||||
Sender held by: Runtime.transfer_txs[i], workers via TickContext
|
||||
Receiver held by: Worker[i].transfer_rx
|
||||
|
||||
spawn channel:
|
||||
Sender held by: Runtime.spawn_txs[i], workers via TickContext
|
||||
Receiver held by: Worker[i].spawn_rx
|
||||
```
|
||||
|
||||
### The AddressMap: global actor directory
|
||||
|
||||
```
|
||||
┌─ AddressMap ───────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ RwLock< HashMap<ActorAddress, WorkerId> > │
|
||||
│ │
|
||||
│ ┌──────────────────────────────────────────────────────────────────┐ │
|
||||
│ │ addr_0 -> WorkerId(0) │ │
|
||||
│ │ addr_1 -> WorkerId(2) │ │
|
||||
│ │ addr_2 -> WorkerId(0) │ │
|
||||
│ │ addr_3 -> WorkerId(1) │ │
|
||||
│ │ ... │ │
|
||||
│ └──────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ READERS (concurrent, RwLock read): │
|
||||
│ WorkerContext.send_any() -- every message send does a lookup │
|
||||
│ Runtime.send_to() -- external sends do a lookup │
|
||||
│ │
|
||||
│ WRITERS (rare, exclusive lock): │
|
||||
│ Runtime.spawn() -- registers new actor at spawn time │
|
||||
│ WorkerContext.spawn_any() -- actor spawns another actor │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### The InboxRegistry: escape hatch to user code
|
||||
|
||||
```
|
||||
┌─ InboxRegistry ────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ RwLock< HashMap<ActorAddress, Arc<dyn SenderT>> > │
|
||||
│ │
|
||||
│ For addresses belonging to external Inbox<M>, not actors. │
|
||||
│ │
|
||||
│ ┌─ Registration ─────────────────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ Runtime.new_inbox::<M>() │ │
|
||||
│ │ addr = ActorAddress::new_random() │ │
|
||||
│ │ receiver = Receiver::<M>::new(capacity) │ │
|
||||
│ │ sender = receiver.new_sender() │ │
|
||||
│ │ inbox_registry.register(addr, Arc::new(sender)) │ │
|
||||
│ │ returns Inbox { addr, inner: receiver } │ │
|
||||
│ │ │ │
|
||||
│ └────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ┌─ Delivery (when address_map lookup fails) ─────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ WorkerContext.send_any(addr, msg) │ │
|
||||
│ │ address_map.lookup(addr) -> None │ │
|
||||
│ │ inbox_registry.try_deliver(addr, msg) │ │
|
||||
│ │ senders.read().get(addr).try_send_any(msg) │ │
|
||||
│ │ downcast Box<Any> -> M, push into Receiver<M> │ │
|
||||
│ │ │ │
|
||||
│ └────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ┌─ Consumption (user code) ──────────────────────────────────────┐ │
|
||||
│ │ │ │
|
||||
│ │ let inbox = rt.new_inbox::<MyMsg>()?; │ │
|
||||
│ │ // later, from any thread: │ │
|
||||
│ │ if let Some(msg) = inbox.try_recv() { ... } │ │
|
||||
│ │ │ │
|
||||
│ └────────────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### Full system topology
|
||||
|
||||
```
|
||||
┌─ User Code ────────────────────────────────────────────────────────────┐
|
||||
│ rt.spawn() rt.send_to() inbox.try_recv() rt.shutdown() │
|
||||
└────┬──────────────────┬──────────────────┬──────────────────┬──────────┘
|
||||
│ │ ^ │
|
||||
v v │ v
|
||||
┌─ Arc<Runtime> ────────────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ ┌───────────┐ ┌────────────┐ ┌─────────────┐ ┌────────────┐ │
|
||||
│ │AddressMap │ │ Placement │ │InboxRegistry│ │ is_running │ │
|
||||
│ │ addr->wid │ │ round-robin│ │ addr->Sender│ │ AtomicBool │ │
|
||||
│ └─────┬─────┘ └──────┬─────┘ └──────┬──────┘ └──────┬─────┘ │
|
||||
│ │ │ │ │ │
|
||||
│ ┌─────┴───────────────┴──────────────┴───────────────┴────────────────┐ │
|
||||
│ │ TickContext (borrows all above) │ │
|
||||
│ └──────────────────────────┬──────────────────────────────────────────┘ │
|
||||
│ │ │
|
||||
│ ┌─ transfer_txs[] ────┐ │ ┌─ spawn_txs[] ──────┐ │
|
||||
│ │ [0]: Sender<Envelope│ │ │ [0]: Sender<(A,Box)>│ │
|
||||
│ │ [1]: Sender<Envelope│ │ │ [1]: Sender<(A,Box)>│ │
|
||||
│ │ [2]: Sender<Envelope│ │ │ [2]: Sender<(A,Box)>│ │
|
||||
│ └──┬──────┬──────┬────┘ │ └──┬──────┬──────┬────┘ │
|
||||
│ │ │ │ │ │ │ │ │
|
||||
└────┼──────┼──────┼────────┼──────┼──────┼──────┼──────────────────────────┘
|
||||
│ │ │ │ │ │ │
|
||||
v v v │ v v v
|
||||
┌────────┐┌────────┐┌───────┴┐┌────────┐┌────────┐┌────────┐
|
||||
│xfer ││xfer ││xfer ││ spawn ││ spawn ││ spawn │
|
||||
│_rx[0] ││_rx[1] ││_rx[2] ││ _rx[0] ││ _rx[1] ││ _rx[2] │
|
||||
└───┬────┘└───┬────┘└───┬────┘└───┬────┘└───┬────┘└───┬────┘
|
||||
│ │ │ │ │ │
|
||||
v v v v v v
|
||||
┌─ Worker 0 ──────┐ ┌─ Worker 1 ──────┐ ┌─ Worker 2 ──────┐
|
||||
│ │ │ │ │ │
|
||||
│ ┌─ pool ──────┐ │ │ ┌─ pool ──────┐ │ │ ┌─ pool ──────┐ │
|
||||
│ │ ┌──────────┐│ │ │ │ ┌──────────┐│ │ │ │ ┌──────────┐│ │
|
||||
│ │ │ slot: ││ │ │ │ │ slot: ││ │ │ │ │ slot: ││ │
|
||||
│ │ │ mailbox ││ │ │ │ │ mailbox ││ │ │ │ │ mailbox ││ │
|
||||
│ │ │ actor ││ │ │ │ │ actor ││ │ │ │ │ actor ││ │
|
||||
│ │ └──────────┘│ │ │ │ └──────────┘│ │ │ │ └──────────┘│ │
|
||||
│ │ ┌──────────┐│ │ │ │ ┌──────────┐│ │ │ │ │ │
|
||||
│ │ │ slot: ││ │ │ │ │ slot: ││ │ │ └─────────────┘ │
|
||||
│ │ │ mailbox ││ │ │ │ │ mailbox ││ │ │ │
|
||||
│ │ │ actor ││ │ │ │ │ actor ││ │ │ thread 2 │
|
||||
│ │ └──────────┘│ │ │ │ └──────────┘│ │ └──────────────────┘
|
||||
│ └─────────────┘ │ │ └─────────────┘ │
|
||||
│ │ │ │
|
||||
│ thread 0 │ │ thread 1 │
|
||||
└──────────────────┘ └──────────────────┘
|
||||
|
||||
Workers also send to EACH OTHER during phase 3:
|
||||
WorkerContext.send_any() -> transfer_txs[other_wid].try_send()
|
||||
WorkerContext.spawn_any() -> spawn_txs[target_wid].try_send()
|
||||
```
|
||||
|
||||
## Message Lifecycle
|
||||
|
||||
```
|
||||
PRODUCERS
|
||||
┌──────────────────┬──────────────────────┐
|
||||
│ │ │
|
||||
v v v
|
||||
┌─────────────┐ ┌──────────────┐ ┌──────────────────────┐
|
||||
│ Runtime │ │ ctx.send() │ │ ctx.send() │
|
||||
│ .send_to() │ │ same worker │ │ other worker │
|
||||
└──────┬──────┘ └──────┬───────┘ └──────────┬───────────┘
|
||||
│ │ │
|
||||
v v v
|
||||
┌─────────────┐ ┌─────────────┐ ┌──────────────────────┐
|
||||
│ transfer_tx │ │ pending_ │ │ transfer_tx │
|
||||
│ [wid].send()│ │ local.push()│ │ [wid].send() │
|
||||
└──────┬──────┘ └──────┬──────┘ └──────────┬───────────┘
|
||||
│ │ │
|
||||
│ (end of phase 3) │
|
||||
│ │ │
|
||||
│ phase 4: │
|
||||
│ │ │
|
||||
v v v
|
||||
┌────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ pool.deliver(&addr, msg) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ slot.mailbox.push_back(msg) │
|
||||
│ │
|
||||
└───────────────────────┬────────────────────────────┘
|
||||
│
|
||||
next tick_once
|
||||
phase 3
|
||||
│
|
||||
v
|
||||
┌────────────────────────────────────────────────────┐
|
||||
│ │
|
||||
│ msg = slot.mailbox.pop_front() │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ slot.actor.handle_any(&ctx, msg) │
|
||||
│ │ │
|
||||
│ v │
|
||||
│ ┌──────────────────────────────────────────┐ │
|
||||
│ │ downcast Box<dyn Any> to A::Incoming │ │
|
||||
│ │ │ │
|
||||
│ │ ok: A.handle(ctx, typed_msg) │ │
|
||||
│ │ err: silently dropped │ │
|
||||
│ └──────────────────────────────────────────┘ │
|
||||
│ │
|
||||
└────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Shutdown Flow
|
||||
|
||||
```
|
||||
┌─ User Code ──────┐
|
||||
│ │
|
||||
│ rt.shutdown() │
|
||||
│ │ │
|
||||
└───────┼───────────┘
|
||||
│
|
||||
v
|
||||
┌─ Runtime ──────────────────────────────────────┐
|
||||
│ │
|
||||
│ is_running.store(false, Release) │
|
||||
│ │
|
||||
└────────────────────────┬────────────────────────┘
|
||||
│
|
||||
┌─────────────────┼─────────────────┐
|
||||
│ │ │
|
||||
v v v
|
||||
┌─ Worker 0 ────┐ ┌─ Worker 1 ────┐ ┌─ Worker 2 ────┐
|
||||
│ │ │ │ │ │
|
||||
│ is_running │ │ is_running │ │ is_running │
|
||||
│ .load(Acquire)│ │ .load(Acquire)│ │ .load(Acquire)│
|
||||
│ -> false │ │ -> false │ │ -> false │
|
||||
│ │ │ │ │ │
|
||||
│ run() returns │ │ run() returns │ │ run() returns │
|
||||
│ thread exits │ │ thread exits │ │ thread exits │
|
||||
└────────────────┘ └────────────────┘ └────────────────┘
|
||||
│ │ │
|
||||
└─────────────────┼─────────────────┘
|
||||
│
|
||||
v
|
||||
┌─ RuntimeHandle ──┐
|
||||
│ │
|
||||
│ .join() │
|
||||
│ waits for all │
|
||||
│ JoinHandles │
|
||||
│ │
|
||||
└───────────────────┘
|
||||
```
|
||||
|
||||
|
|
@ -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
98
examples/python/getting_started.ipynb
Normal 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
|
||||
}
|
||||
|
|
@ -30,6 +30,8 @@ async def main():
|
|||
reply = await recv(inbox)
|
||||
print(reply)
|
||||
|
||||
# show us our actors!
|
||||
print(handle.stats())
|
||||
handle.shutdown()
|
||||
handle.join()
|
||||
|
||||
|
|
@ -7,5 +7,8 @@ name = "swactor"
|
|||
version = "0.1.0"
|
||||
requires-python = ">=3.9"
|
||||
|
||||
[dependency-groups]
|
||||
dev = ["jupyter", "ipykernel"]
|
||||
|
||||
[tool.maturin]
|
||||
features = ["python"]
|
||||
|
|
|
|||
|
|
@ -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
139
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 ───────────────────────────────────────────────────────────
|
||||
|
|
@ -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(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
136
tests/stats_demo.rs
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
use swactor::{
|
||||
actor::{ActorAddress, ActorInterface},
|
||||
runtime::{Ctx, Runtime, RuntimeConfig},
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Ping {
|
||||
reply_to: ActorAddress,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Pong;
|
||||
|
||||
struct PingActor;
|
||||
|
||||
impl ActorInterface for PingActor {
|
||||
type Incoming = Ping;
|
||||
type Response = Pong;
|
||||
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Ping) {
|
||||
let _ = ctx.send(msg.reply_to, Pong);
|
||||
}
|
||||
}
|
||||
|
||||
/// Counter that just counts messages.
|
||||
struct Counter(u64);
|
||||
|
||||
impl ActorInterface for Counter {
|
||||
type Incoming = u64;
|
||||
type Response = ();
|
||||
|
||||
fn handle(&mut self, _ctx: &Ctx, _msg: u64) {
|
||||
self.0 += 1;
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stats_demo_single_thread() {
|
||||
let rt = Runtime::new(RuntimeConfig::default());
|
||||
|
||||
// Spawn a few actors
|
||||
let ping1 = rt.spawn(PingActor).unwrap();
|
||||
let ping2 = rt.spawn(PingActor).unwrap();
|
||||
let counter = rt.spawn(Counter(0)).unwrap();
|
||||
|
||||
// Send some messages (they queue up before we tick)
|
||||
for i in 0..20u64 {
|
||||
rt.send_to(counter, i).unwrap();
|
||||
}
|
||||
|
||||
// Stats BEFORE ticking — messages are in the transfer queue, not yet in mailboxes
|
||||
let s = rt.stats();
|
||||
println!("=== Before any ticks ===");
|
||||
print_stats(&s);
|
||||
|
||||
// Tick once — drains transfer queue into mailboxes, then processes messages
|
||||
rt.tick();
|
||||
|
||||
let s = rt.stats();
|
||||
println!("\n=== After 1 tick ===");
|
||||
print_stats(&s);
|
||||
|
||||
// Tick a few more times to drain remaining messages
|
||||
for _ in 0..5 {
|
||||
rt.tick();
|
||||
}
|
||||
|
||||
let s = rt.stats();
|
||||
println!("\n=== After 6 ticks total ===");
|
||||
print_stats(&s);
|
||||
|
||||
assert_eq!(s.num_workers, 1);
|
||||
assert_eq!(s.actors.len(), 3);
|
||||
assert_eq!(s.workers[0].num_actors, 3);
|
||||
// All 20 messages should be processed by now
|
||||
assert_eq!(s.workers[0].mailbox_depth, 0);
|
||||
assert!(s.workers[0].messages_processed >= 20);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stats_demo_multi_thread() {
|
||||
let config = RuntimeConfig {
|
||||
num_threads: 3,
|
||||
..Default::default()
|
||||
};
|
||||
let rt = Runtime::new(config);
|
||||
|
||||
// Spawn actors — round-robin will spread them across 3 workers
|
||||
let mut addrs = Vec::new();
|
||||
for _ in 0..6 {
|
||||
addrs.push(rt.spawn(Counter(0)).unwrap());
|
||||
}
|
||||
|
||||
// Send messages to each actor
|
||||
for &addr in &addrs {
|
||||
for i in 0..10u64 {
|
||||
rt.send_to(addr, i).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
let handle = rt.run().unwrap();
|
||||
|
||||
// Let it process
|
||||
std::thread::sleep(std::time::Duration::from_millis(50));
|
||||
|
||||
let s = handle.runtime.stats();
|
||||
println!("\n=== Multi-threaded (3 workers, 6 actors, 60 messages) ===");
|
||||
print_stats(&s);
|
||||
|
||||
handle.shutdown();
|
||||
handle.join();
|
||||
|
||||
assert_eq!(s.num_workers, 3);
|
||||
assert_eq!(s.actors.len(), 6);
|
||||
let total_processed: u64 = s.workers.iter().map(|w| w.messages_processed).sum();
|
||||
assert_eq!(total_processed, 60);
|
||||
}
|
||||
|
||||
fn print_stats(s: &swactor::runtime::RuntimeStats) {
|
||||
println!(
|
||||
"RuntimeStats(actors={}, workers={})",
|
||||
s.actors.len(),
|
||||
s.num_workers
|
||||
);
|
||||
for w in &s.workers {
|
||||
println!(
|
||||
" Worker {}: {} actors, {} queued, {} processed",
|
||||
w.id, w.num_actors, w.mailbox_depth, w.messages_processed
|
||||
);
|
||||
for (addr, wid) in &s.actors {
|
||||
if *wid == w.id {
|
||||
println!(" - {:x?}...", &addr.0[..4]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue