From 32d1fea2fa77a32189236e27d2d36f8eafdbc150 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 9 Feb 2026 15:59:14 +0700 Subject: [PATCH] feat: runtime dashboard and docs Live and replay demo for a runtime dashboard. Added docs with svg files. --- Cargo.lock | 121 +++ Cargo.toml | 4 +- crates/runtime-dashboard/.gitignore | 2 + crates/runtime-dashboard/Cargo.toml | 12 + crates/runtime-dashboard/README.md | 36 + .../docs/actor_lifecycle.svg | 145 ++++ .../runtime-dashboard/docs/architecture.svg | 692 +++++++++++++++ crates/runtime-dashboard/docs/dataflow.svg | 806 ++++++++++++++++++ .../docs/message_lifecycle.svg | 155 ++++ crates/runtime-dashboard/docs/render_docs.sh | 81 ++ .../docs/runtime_lifecycle.svg | 229 +++++ crates/runtime-dashboard/docs/tick_cycle.svg | 258 ++++++ .../runtime-dashboard/docs/type_erasure.svg | 222 +++++ .../examples/dashboard_demo.rs | 155 ++++ .../runtime-dashboard/examples/record_demo.rs | 125 +++ .../runtime-dashboard/examples/replay_demo.rs | 23 + .../runtime-dashboard/src/dashboard_html.rs | 369 ++++++++ crates/runtime-dashboard/src/layer.rs | 227 +++++ crates/runtime-dashboard/src/lib.rs | 198 +++++ crates/runtime-dashboard/src/server.rs | 337 ++++++++ crates/runtime-dashboard/src/trace.rs | 18 + src/crate_test/mod.rs | 8 +- src/runtime.rs | 69 +- src/stats.rs | 96 +++ src/worker.rs | 91 +- tools/package-lock.json | 2 +- 26 files changed, 4466 insertions(+), 15 deletions(-) create mode 100644 crates/runtime-dashboard/.gitignore create mode 100644 crates/runtime-dashboard/Cargo.toml create mode 100644 crates/runtime-dashboard/README.md create mode 100644 crates/runtime-dashboard/docs/actor_lifecycle.svg create mode 100644 crates/runtime-dashboard/docs/architecture.svg create mode 100644 crates/runtime-dashboard/docs/dataflow.svg create mode 100644 crates/runtime-dashboard/docs/message_lifecycle.svg create mode 100755 crates/runtime-dashboard/docs/render_docs.sh create mode 100644 crates/runtime-dashboard/docs/runtime_lifecycle.svg create mode 100644 crates/runtime-dashboard/docs/tick_cycle.svg create mode 100644 crates/runtime-dashboard/docs/type_erasure.svg create mode 100644 crates/runtime-dashboard/examples/dashboard_demo.rs create mode 100644 crates/runtime-dashboard/examples/record_demo.rs create mode 100644 crates/runtime-dashboard/examples/replay_demo.rs create mode 100644 crates/runtime-dashboard/src/dashboard_html.rs create mode 100644 crates/runtime-dashboard/src/layer.rs create mode 100644 crates/runtime-dashboard/src/lib.rs create mode 100644 crates/runtime-dashboard/src/server.rs create mode 100644 crates/runtime-dashboard/src/trace.rs diff --git a/Cargo.lock b/Cargo.lock index 3710e17..97b5b8c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -312,6 +312,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "libc" version = "0.2.180" @@ -339,6 +345,15 @@ dependencies = [ "autocfg", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -360,6 +375,12 @@ version = "11.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" +[[package]] +name = "pin-project-lite" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" + [[package]] name = "plotters" version = "0.3.7" @@ -524,6 +545,18 @@ version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a96887878f22d7bad8a3b6dc5b7440e0ada9a245242924394987b21cf2210a4c" +[[package]] +name = "runtime-dashboard" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "swactor", + "tiny_http", + "tracing", + "tracing-subscriber", +] + [[package]] name = "rustversion" version = "1.0.22" @@ -591,6 +624,21 @@ dependencies = [ "serde", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + +[[package]] +name = "smallvec" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" + [[package]] name = "swactor" version = "0.1.0" @@ -600,6 +648,7 @@ dependencies = [ "crossbeam-utils", "getrandom", "serde", + "tracing", ] [[package]] @@ -646,6 +695,15 @@ version = "0.12.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1" +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "tiny_http" version = "0.12.0" @@ -709,6 +767,63 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e" +dependencies = [ + "nu-ansi-term", + "sharded-slab", + "smallvec", + "thread_local", + "tracing-core", + "tracing-log", +] + [[package]] name = "unicode-ident" version = "1.0.22" @@ -721,6 +836,12 @@ version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "walkdir" version = "2.5.0" diff --git a/Cargo.toml b/Cargo.toml index 119d8f8..b659968 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [".", "crates/swactor-python", "crates/swactor-wasm", "crates/swactor-gossip", "crates/gossip-dashboard"] +members = [".", "crates/swactor-python", "crates/swactor-wasm", "crates/swactor-gossip", "crates/gossip-dashboard", "crates/runtime-dashboard"] exclude = ["tools/depgraph"] [package] @@ -15,11 +15,13 @@ crate-type = ["rlib"] default = ["getrandom"] getrandom = ["dep:getrandom"] serde = ["dep:serde"] +tracing = ["dep:tracing"] no_random = [] # compile without access to a source of randomness [dependencies] getrandom = { version = "0.2", optional = true } serde = { version = "1", features = ["derive"], optional = true } +tracing = { version = "0.1", optional = true } crossbeam-queue = "0.3.12" crossbeam-utils = "0.8.21" diff --git a/crates/runtime-dashboard/.gitignore b/crates/runtime-dashboard/.gitignore new file mode 100644 index 0000000..fc68aa3 --- /dev/null +++ b/crates/runtime-dashboard/.gitignore @@ -0,0 +1,2 @@ +docs/generated/ +docs/**dot \ No newline at end of file diff --git a/crates/runtime-dashboard/Cargo.toml b/crates/runtime-dashboard/Cargo.toml new file mode 100644 index 0000000..76b4102 --- /dev/null +++ b/crates/runtime-dashboard/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "runtime-dashboard" +version = "0.1.0" +edition = "2024" + +[dependencies] +swactor = { path = "../..", features = ["serde", "tracing"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["registry"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tiny_http = "0.12" diff --git a/crates/runtime-dashboard/README.md b/crates/runtime-dashboard/README.md new file mode 100644 index 0000000..cc45ba0 --- /dev/null +++ b/crates/runtime-dashboard/README.md @@ -0,0 +1,36 @@ +# runtime-dashboard + +Visual dashboard and architectural diagrams for the swactor runtime. + +## Generating Diagrams + +Render all `.dot` sources into SVGs: + +```bash +./render_docs.sh +``` + +**Prerequisites** (one of): +- [Graphviz](https://graphviz.org/) — `apt install graphviz` / `brew install graphviz` +- [Node.js](https://nodejs.org/) — the script auto-installs `@viz-js/viz` into `tools/` + +Generated SVGs are written to `docs/generated/` (gitignored). + +## Diagram Index + +### DOT sources (`docs/*.dot` → `docs/generated/*.svg`) + +| Diagram | Description | +|---------|-------------| +| `architecture.dot` | Structural map of all structs/traits, grouped by module, with ownership/Arc/borrow/trait-impl edges | +| `dataflow.dot` | 5 behavioral flows: cross-worker send, same-worker send, actor spawn, external inbox, 6-phase tick cycle | +| `tick_cycle.dot` | Focused view of the `Worker::tick_once` pipeline and its 6 phases | +| `type_erasure.dot` | How generic message/actor types are erased via `Box` and `Box` | + +### Hand-authored SVGs (`docs/*.svg`, committed) + +| Diagram | Description | +|---------|-------------| +| `actor_lifecycle.svg` | Lifecycle states of an actor from spawn to shutdown | +| `message_lifecycle.svg` | Path of a message from send through inbox to handler | +| `runtime_lifecycle.svg` | Runtime startup, worker creation, and shutdown sequence | diff --git a/crates/runtime-dashboard/docs/actor_lifecycle.svg b/crates/runtime-dashboard/docs/actor_lifecycle.svg new file mode 100644 index 0000000..5e17b8d --- /dev/null +++ b/crates/runtime-dashboard/docs/actor_lifecycle.svg @@ -0,0 +1,145 @@ + + + + + + + + + + + + + + + + + + + + + + + Actor Lifecycle + src/runtime.rs · src/worker.rs + + + + + ▶ + + + + spawn() + + + + + SPAWNED + in spawn queue + + + + Phase 1/4 + drain → + pool.insert() + + + + + READY + in pool, mailbox empty + + + + send_to / + deliver pushes + to VecDeque + + + + + MAILBOX + PENDING + + + + Phase 3 + pop_front() + + + + + PROCESSING + handler executing + + + + handler OK + + + + panic! + catch_unwind + + + + + PANIC + caught, actor survives + + + + next msg, + next tick + + + + + DEAD + removed from pool + + + + shutdown / + pool.remove() + + + + shutdown + + + + + + + + Legend + + + + Transition + + + + Success + + + + Error / Panic + + + + Shutdown path + + + runtime.rs:125-134 · worker.rs:204-219 + diff --git a/crates/runtime-dashboard/docs/architecture.svg b/crates/runtime-dashboard/docs/architecture.svg new file mode 100644 index 0000000..4d216f8 --- /dev/null +++ b/crates/runtime-dashboard/docs/architecture.svg @@ -0,0 +1,692 @@ + + + + + + +architecture + + +cluster_legend + +Legend + + +cluster_runtime + +runtime.rs + + +cluster_actor + +actor.rs + + +cluster_delivery + +delivery.rs + + +cluster_worker + +worker.rs + + +cluster_channel + +channel.rs + + +cluster_stats + +stats.rs + + +cluster_config + +config.rs + + + +legend + +─────── +Ownership (solid) +━━━━━━━ +Arc-shared (bold orange) +- - - - - - +Borrow / &'a (dashed purple) +· · · · · ·▷ +Trait impl (dotted green) + + + +Runtime + + + +Runtime + +config: RuntimeConfig + +address_map: Arc<AddressMap> + +inbox_registry: Arc<InboxRegistry> + +transfer_txs: Vec<Sender<Envelope>> + +spawn_txs: Vec<Sender<(Addr, Box<dyn AnyActor>)>> + +placement: Placement + +is_running: AtomicBool + +worker_stats: Vec<Arc<WorkerStats>> + +tick_workers: RefCell<Vec<Worker>> + + + +ContextInner + +«trait» ContextInner +fn send_any(addr, msg) +fn spawn_any(addr, actor) + + + +Runtime->ContextInner + + +impl + + + +AddressMap + + + +AddressMap + +inner: RwLock<HashMap<ActorAddress, WorkerId>> + + + +Runtime->AddressMap + + +Arc + + + +Placement + + + +Placement + +next: AtomicUsize + +num_workers: usize + + + +Runtime->Placement + + +owns + + + +InboxRegistry + + + +InboxRegistry + +senders: RwLock<HashMap<Addr, Arc<dyn SenderT>>> + + + +Runtime->InboxRegistry + + +Arc + + + +Worker + + + +Worker + +id: WorkerId + +pool: ActorPool + +transfer_rx: Receiver<Envelope> + +spawn_rx: Receiver<(Addr, Box<dyn AnyActor>)> + +stats: Arc<WorkerStats> + + + +Runtime->Worker + + +RefCell<Vec<>> + + + +SenderCh + + + +Sender<T> + +queue: Arc<HybridChannel<T>> + + + +Runtime->SenderCh + + +transfer_txs + +spawn_txs + + + +WorkerStats + + + +WorkerStats + +num_actors: AtomicUsize + +total_mailbox_depth: AtomicUsize + +messages_processed: AtomicU64 + + + +Runtime->WorkerStats + + +Vec<Arc> + + + +RuntimeConfig + + + +RuntimeConfig + +max_actors: usize + +actor_max_messages: usize + +num_threads: usize + +backoff_policy: BackoffPolicy + + + +Runtime->RuntimeConfig + + +owns + + + +RuntimeHandle + + + +RuntimeHandle + +runtime: Arc<Runtime> + +threads: Vec<JoinHandle<()>> + + + +RuntimeHandle->Runtime + + +Arc + + + +Inbox + + + +Inbox<M> + +addr: ActorAddress + +inner: Receiver<M> + + + +ActorAddress + + + +ActorAddress + +0: [u8; 32] + +Copy, Eq, Hash + + + +Inbox->ActorAddress + + +addr + + + +ReceiverCh + + + +Receiver<T> + +queue: Arc<HybridChannel<T>> + + + +Inbox->ReceiverCh + + +owns + + + +Actor + + + +Actor<A> + +0: A   +(where A: ActorInterface) + + + +ActorInterface + +«trait» ActorInterface +type Incoming: Message +type Response: Message +fn handle(&mut self, ctx, msg) + + + +Actor->ActorInterface + + +wraps A: impl + + + +AnyActor + +«trait» AnyActor +fn handle_any(&mut self, ctx, msg) + + + +Actor->AnyActor + + +impl + + + +Ctx + + + +Ctx<'a> + +inner: &'a dyn ContextInner + +self_addr: ActorAddress + + + +Ctx->ActorAddress + + +self_addr + + + +Ctx->ContextInner + + +&'a dyn + + + +MessageTrait + +«trait» Message +'static + Clone + Send + Sync + + + +Envelope + + + +Envelope + +dest: ActorAddress + +payload: Box<dyn Any + Send> + + + +Envelope->ActorAddress + + +dest + + + +WorkerId + + + +WorkerId + +0: usize + + + +AddressMap->ActorAddress + + +HashMap key + + + +AddressMap->WorkerId + + +HashMap val + + + +SenderT + +«trait» SenderT +fn try_send_any(msg: Box<dyn Any>) + + + +InboxRegistry->SenderT + + +Arc<dyn> + + + +TickContext + + + +TickContext<'a> + +address_map: &'a AddressMap + +transfer_txs: &'a [Sender<Envelope>] + +spawn_txs: &'a [Sender<...>] + +placement: &'a Placement + +inbox_registry: &'a InboxRegistry + +config: &'a RuntimeConfig + + + +TickContext->AddressMap + + +&'a + + + +TickContext->Placement + + +&'a + + + +TickContext->InboxRegistry + + +&'a + + + +TickContext->SenderCh + + +&'a [] + + + +TickContext->RuntimeConfig + + +&'a + + + +Worker->WorkerId + + +id + + + +ActorPool + + + +ActorPool + +actors: HashMap<ActorAddress, ActorSlot> + + + +Worker->ActorPool + + +owns + + + +Worker->ReceiverCh + + +transfer_rx + +spawn_rx + + + +Worker->WorkerStats + + +Arc + + + +WorkerContext + + + +WorkerContext<'a> + +worker_id: WorkerId + +tc: &'a TickContext<'a> + +pending_local: &'a RefCell<Vec<...>> + + + +WorkerContext->ContextInner + + +impl + + + +WorkerContext->TickContext + + +&'a + + + +ActorSlot + + + +ActorSlot + +mailbox: VecDeque<Box<dyn Any + Send>> + +actor: Box<dyn AnyActor> + + + +ActorPool->ActorSlot + + +HashMap<Addr, _> + + + +ActorSlot->AnyActor + + +Box<dyn> + + + +HybridChannel + + + +HybridChannel<T> + +ring: ArrayQueue<T> + +overflow: SegQueue<T> + + + +HybridChannel->HybridChannel + + + + +SenderCh->SenderT + + +impl for +Sender<M> + + + +SenderCh->HybridChannel + + +Arc + + + +ReceiverCh->HybridChannel + + +Arc + + + +WorkerInfo + + + +WorkerInfo + +id: usize + +num_actors: usize + +mailbox_depth: usize + +messages_processed: u64 + + + +RuntimeStats + + + +RuntimeStats + +num_workers: usize + +actors: Vec<(ActorAddress, usize)> + +workers: Vec<WorkerInfo> + + + +RuntimeStats->WorkerInfo + + +Vec<> + + + +BackoffPolicy + + + +BackoffPolicy + +spin_threshold: u32 + +yield_threshold: u32 + +sleep_increment_us: u64 + +sleep_max_us: u64 + + + +RuntimeConfig->BackoffPolicy + + +owns + + + diff --git a/crates/runtime-dashboard/docs/dataflow.svg b/crates/runtime-dashboard/docs/dataflow.svg new file mode 100644 index 0000000..45b38a1 --- /dev/null +++ b/crates/runtime-dashboard/docs/dataflow.svg @@ -0,0 +1,806 @@ + + + + + + +dataflow + + +cluster_legend + +Data on Arrows + + +cluster_flow_a + +FLOW A — Cross-Worker Message Send + + +cluster_flow_b + +FLOW B — Same-Worker Message Send (during tick) + + +cluster_flow_c + +FLOW C — Actor Spawn + + +cluster_flow_d + +FLOW D — External Inbox + + +cluster_flow_e + +FLOW E — The 6-Phase Tick Cycle (Worker::tick_once) + + + +dlegend +━━━ +Envelope (dest + payload) +━━━ +Spawn tuple (addr, Box<dyn AnyActor>) +━━━ +Pending local (addr, Box<dyn Any>) +━━━ +Box<dyn Any + Send> (raw message) +━━━ +Stats (atomics) +━━━ +WorkerId (lookup result) + + + +a_caller + +Caller + + + +a_send + +Runtime +::send_to(addr, msg) + + + +a_caller->a_send + + + + + +a_lookup + +AddressMap +::lookup(addr) + + + +a_send->a_lookup + + + + + +a_box + +Box::new(msg) +→ Envelope + + + +a_lookup->a_box + + +WorkerId + + + +a_tx + +transfer_txs +[worker_id] +.try_send() + + + +a_box->a_tx + + +Envelope + + + +a_hybrid + + +HybridChannel +ring → overflow + + + +a_tx->a_hybrid + + +Envelope + + + +a_rx + +Worker +transfer_rx +.try_recv() + + + +a_hybrid->a_rx + + +Envelope + + + +a_deliver + +ActorPool +::deliver(addr) + + + +a_rx->a_deliver + + +Envelope + + + +a_mailbox + +ActorSlot +mailbox +.push_back() + + + +a_deliver->a_mailbox + + +payload + + + +b_actor + +Actor::handle() +calls ctx.send() + + + +b_wctx + +WorkerContext +::send_any() + + + +b_actor->b_wctx + + + + + +b_lookup + +AddressMap +::lookup(addr) + + + +b_wctx->b_lookup + + + + + +b_pending + +pending_local +.borrow_mut() +.push((addr, msg)) + + + +b_lookup->b_pending + + +same WorkerId + + + +b_phase5 + +Phase 5: +DELIVER +PENDING +LOCAL + + + +b_pending->b_phase5 + + +buffered + + + +b_deliver + +ActorPool +::deliver(addr) + + + +b_phase5->b_deliver + + +(addr, payload) + + + +b_mailbox + +ActorSlot +mailbox +.push_back() + + + +b_deliver->b_mailbox + + +payload + + + +c_caller + +Caller / +ctx.spawn() + + + +c_spawn + +Runtime +::spawn(actor) + + + +c_caller->c_spawn + + + + + +c_addr + +ActorAddress +::new_random() + + + +c_spawn->c_addr + + + + + +c_place + +Placement +::next_worker() +round-robin + + + +c_addr->c_place + + + + + +c_addrmap + +AddressMap +::insert(addr, wid) + + + +c_place->c_addrmap + + +WorkerId + + + +c_box + +Box::new( +Actor::new(a)) + + + +c_addrmap->c_box + + + + + +c_tx + +spawn_txs +[worker_id] +.try_send() + + + +c_box->c_tx + + +(addr, actor_box) + + + +c_hybrid + + +HybridChannel + + + +c_tx->c_hybrid + + +spawn tuple + + + +c_rx + +Worker +spawn_rx +.try_recv() + + + +c_hybrid->c_rx + + +spawn tuple + + + +c_pool + +ActorPool +::insert(addr, +actor_box) + + + +c_rx->c_pool + + +(addr, actor_box) + + + +d_caller + +Caller + + + +d_new + +Runtime +::new_inbox::<M>() + + + +d_caller->d_new + + + + + +d_addr + +ActorAddress +::new_random() + + + +d_new->d_addr + + + + + +d_recv + +Receiver<M> +::new(capacity) + + + +d_addr->d_recv + + + + + +d_sender + +Sender<M> += recv.new_sender() + + + +d_recv->d_sender + + +clone Arc + + + +d_inbox + +return Inbox { +addr, inner: recv +} + + + +d_recv->d_inbox + + +Receiver + + + +d_reg + +InboxRegistry +::register( +addr, Arc<sender>) + + + +d_sender->d_reg + + +Arc<dyn SenderT> + + + +d_send + +Runtime +::send_to( +inbox_addr, msg) + + + +d_miss + +AddressMap +::lookup → miss + + + +d_send->d_miss + + + + + +d_try + +InboxRegistry +::try_deliver( +addr, Box(msg)) + + + +d_miss->d_try + + +not a worker +actor + + + +d_dyn + +Arc<dyn SenderT> +::try_send_any() +downcast → Sender<M> + + + +d_try->d_dyn + + +Box<dyn Any> + + + +d_hybrid + + +HybridChannel + + + +d_dyn->d_hybrid + + +M (typed) + + + +d_tryrecv + +Inbox +::try_recv() + + + +d_hybrid->d_tryrecv + + +M + + + +e_phase1 + + +Phase 1 + +DRAIN SPAWN QUEUE +spawn_rx.try_recv() +→ ActorPool::insert(addr, actor) +repeat until empty + + + +e_phase2 + + +Phase 2 + +DRAIN TRANSFER QUEUE +transfer_rx.try_recv() → Envelope +→ ActorPool::deliver(dest, payload) +repeat up to max(len>>1, 1) + + + +e_phase1->e_phase2 + + + + + +e_pool + + + + +ActorPool +(HashMap<Addr, +ActorSlot>) + + + +e_phase1->e_pool + + +insert + + + +e_phase3 + + +Phase 3 + +TICK ALL ACTORS +for (addr, slot) in pool: +  while msg = mailbox.pop_front(): +    catch_unwind(actor.handle_any(ctx, msg)) +actors may call ctx.send / ctx.spawn + + + +e_phase2->e_phase3 + + + + + +e_phase2->e_pool + + +deliver + + + +e_phase4 + + +Phase 4 + +DRAIN SPAWN QUEUE AGAIN +(picks up spawns from phase 3) +spawn_rx.try_recv() +→ ActorPool::insert() + + + +e_phase3->e_phase4 + + + + + +e_spawn_rx + + +spawn_rx +(HybridChannel) + + + +e_phase3->e_spawn_rx + + +ctx.spawn → +spawn_txs + + + +e_phase3->e_pool + + + +pop mailbox + +handle_any + + + +e_pending + +pending_local +(RefCell<Vec>) + + + +e_phase3->e_pending + + +ctx.send → +buffer local + + + +e_phase5 + + +Phase 5 + +DELIVER PENDING LOCAL +for (addr, payload) in pending_local.take(): +  ActorPool::deliver(addr, payload) +same-worker messages from phase 3 + + + +e_phase4->e_phase5 + + + + + +e_phase4->e_pool + + +insert + + + +e_phase6 + + +Phase 6 + +UPDATE STATS +stats.num_actors.store(pool.len()) +stats.total_mailbox_depth.store(...) +stats.messages_processed.fetch_add(n) + + + +e_phase5->e_phase6 + + + + + +e_phase5->e_pool + + +deliver + + + +e_stats + +WorkerStats +(Arc, atomics) + + + +e_phase6->e_stats + + +store/fetch_add + + + +e_spawn_rx->e_phase1 + + +(addr, actor_box) + + + +e_spawn_rx->e_phase4 + + +(addr, actor_box) + + + +e_transfer_rx + + +transfer_rx +(HybridChannel) + + + +e_transfer_rx->e_phase2 + + +Envelope + + + +e_pending->e_phase5 + + +(addr, payload) + + + diff --git a/crates/runtime-dashboard/docs/message_lifecycle.svg b/crates/runtime-dashboard/docs/message_lifecycle.svg new file mode 100644 index 0000000..7224312 --- /dev/null +++ b/crates/runtime-dashboard/docs/message_lifecycle.svg @@ -0,0 +1,155 @@ + + + + + + + + + + + + + + + + + + + + + + + + Message Lifecycle + src/worker.rs · src/delivery.rs · src/channel.rs · src/actor.rs + + + + + + ORIGIN + + + + runtime.send_to( + addr, msg) + + + + ctx.send( + addr, msg) + + external caller + from actor handler + + + + + + + + + + + + + + ROUTING + + + + AddressMap + ::lookup() + + + + not found + + + InboxRegistry + ::try_deliver() + + + + + Receiver<M> + external consumer + + + + same + worker + + + pending_local + RefCell<Vec> buffer + + delivered in Phase 5 + + + + different + worker + + + Envelope::new() + → HybridChannel + ArrayQueue + SegQueue + + drained in Phase 2 + + + + + + + + + + DELIVERY + + + + pool.deliver(addr, payload) + push to VecDeque mailbox + + + + Phase 3 + + + + mailbox.pop_front() + → AnyActor::handle_any() + + + + + + + downcast + + + + + User Handler + called + + + + + Dropped + type mismatch + + + OK + fail + + + worker.rs:134-151 · delivery.rs:92-116 · channel.rs · actor.rs:53-59 + diff --git a/crates/runtime-dashboard/docs/render_docs.sh b/crates/runtime-dashboard/docs/render_docs.sh new file mode 100755 index 0000000..7516be0 --- /dev/null +++ b/crates/runtime-dashboard/docs/render_docs.sh @@ -0,0 +1,81 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +DOCS_DIR="$SCRIPT_DIR" +OUT_DIR="$DOCS_DIR" +TOOLS_DIR="$(cd "$SCRIPT_DIR/../../../tools" && pwd)" + + +mkdir -p "$OUT_DIR" + +# Collect .dot sources +dots=("$DOCS_DIR"/*.dot) +if [ ${#dots[@]} -eq 0 ]; then + echo "No .dot files found in $DOCS_DIR" + exit 0 +fi + +# Pick a renderer: prefer graphviz `dot`, fall back to @viz-js/viz via Node +render_with_dot() { + for src in "${dots[@]}"; do + name="$(basename "$src" .dot)" + echo " dot: $name.dot -> generated/$name.svg" + dot -Tsvg "$src" -o "$OUT_DIR/$name.svg" + done +} + +render_with_vizjs() { + local tmpfile + tmpfile="$(mktemp "${TMPDIR:-/tmp}/render_docs.XXXXXX.mjs")" + trap 'rm -f "$tmpfile"' RETURN + + cat > "$tmpfile" < f.endsWith(".dot")); + +for (const file of dots) { + const src = readFileSync(join(docsDir, file), "utf-8"); + const name = basename(file, ".dot"); + const svg = viz.renderString(src, { format: "svg" }); + writeFileSync(join(outDir, \`\${name}.svg\`), svg); + console.log(\` viz-js: \${file} -> generated/\${name}.svg\`); +} +NODEJS + + node "$tmpfile" +} + +echo "Rendering DOT diagrams..." + +if command -v dot &>/dev/null; then + render_with_dot +elif command -v node &>/dev/null; then + # Ensure @viz-js/viz is available + if [ -f "$TOOLS_DIR/package.json" ]; then + if ! [ -d "$TOOLS_DIR/node_modules/@viz-js/viz" ]; then + echo "Installing @viz-js/viz..." + (cd "$TOOLS_DIR" && npm install --silent) + fi + else + echo "Error: tools/package.json not found at $TOOLS_DIR" >&2 + exit 1 + fi + render_with_vizjs +else + echo "Error: No renderer available." >&2 + echo "Install graphviz (apt install graphviz) or Node.js." >&2 + exit 1 +fi + +echo "Done. Output in ${OUT_DIR}" diff --git a/crates/runtime-dashboard/docs/runtime_lifecycle.svg b/crates/runtime-dashboard/docs/runtime_lifecycle.svg new file mode 100644 index 0000000..229763d --- /dev/null +++ b/crates/runtime-dashboard/docs/runtime_lifecycle.svg @@ -0,0 +1,229 @@ + + + + + + + + + + + + + + + + + + + + + + + + Runtime Lifecycle + src/runtime.rs · src/worker.rs · src/config.rs + + + + + + CONSTRUCTION + runtime.rs:80-122 + + + + Runtime::new(config) + parse RuntimeConfig + + + + + + Create Workers + N per config.workers + + + + + + Allocate Channels + transfer_tx/rx + spawn_tx/rx + per worker + + + + + + Init Shared State + AddressMap, InboxRegistry, + Placement + + + + ready + + + + + + TICK CYCLE + worker.rs:40-96 + + + + + + 1 + Drain Spawns + spawn_rx → + pool.insert() + new actors join + + + + + + + 2 + Drain Transfers + transfer_rx → + pool.deliver() + cross-worker msgs + + + + + + + 3 + tick_all() + pop_front → handle + + catch_unwind + process messages + + + + + + + 4 + Drain Spawns + again: actors + spawned in Phase 3 + handler-created + + + + + + + 5 + pending_local + drain buffer → + pool.deliver() + same-worker msgs + + + + + + + 6 + Publish Stats + update atomics + WorkerStats + + + + next tick iteration + + + one call to tick_once() executes phases 1–6 + + + + + + EXECUTION MODES + runtime.rs:185-217 + + + + + + tick() mode — single-threaded + + + + user calls + runtime.tick() + + + + + iterate + tick_workers + + + + + w.tick_once() + for each worker + + + + next + + RefCell<Vec<Worker>> — stays on caller's thread + + + + + + run() mode — multi-threaded + + + + workers → threads + std::thread::spawn + + + + + while is_running + { tick_once(); backoff() } + + + + + Handle + + + + loop + + + returns RuntimeHandle to caller + + + + Shutdown + is_running.store(false) → worker loops exit → thread.join() + + + + + + + + runtime.rs:80-122 (new) · worker.rs:40-96 (tick_once) · worker.rs:98-120 (run loop) · runtime.rs:185-217 (run) · config.rs (BackoffPolicy) + diff --git a/crates/runtime-dashboard/docs/tick_cycle.svg b/crates/runtime-dashboard/docs/tick_cycle.svg new file mode 100644 index 0000000..c509c77 --- /dev/null +++ b/crates/runtime-dashboard/docs/tick_cycle.svg @@ -0,0 +1,258 @@ + + + + + + +tick_cycle + + +cluster_backoff + +Worker::run() backoff + + + +start + + + + +phase1 + +Phase 1: Drain Spawn Queue +spawn_rx → pool.insert(addr, actor) + + + +start->phase1 + + + + + +end + + + + + +idle_check + +did_work? + + + +end->idle_check + + +run() loop + + + +phase2 + +Phase 2: Drain Transfer Queue +transfer_rx → pool.deliver(dest, payload) + + + +phase1->phase2 + + +(ActorAddress, Box<dyn AnyActor>) + + + +phase3 + +Phase 3: Tick All Actors +for each actor: pop mailbox → handle_any(ctx, msg) +catch_unwind wraps each handle call + + + +phase2->phase3 + + +Envelope → Box<dyn Any + Send> + + + +phase4 + +Phase 4: Drain Spawn Queue (again) +actors spawned during Phase 3 handlers + + + +phase3->phase4 + + + + + +route_decision + +send_any routing +(inside Phase 3 handlers) + + + +phase3->route_decision + + +ctx.send() + + + +phase5 + +Phase 5: Drain Pending Local +same-worker sends buffered in Phase 3 +pending_local → pool.deliver(addr, msg) + + + +phase4->phase5 + + +new actors ready for local delivery + + + +phase6 + +Phase 6: Publish Stats +AtomicStore: num_actors, mailbox_depth +AtomicFetchAdd: messages_processed + + + +phase5->phase6 + + + + + +phase6->end + + +returns did_work: bool + + + +local_buf + + + +pending_local buffer +(same worker) + + + +route_decision->local_buf + + +same worker + + + +cross_q + + + +transfer_tx[wid] +(cross worker) + + + +route_decision->cross_q + + +different worker + + + +inbox_reg + + + +inbox_registry +(external inbox) + + + +route_decision->inbox_reg + + +not in address_map + + + +local_buf->phase5 + + +delivered +next phase + + + +cross_q->phase2 + + +delivered +next tick +(other worker) + + + +idle_check->start + + +yes / idle_count=0 + + + +spinning + +Spin (hot loop) + + + +idle_check->spinning + + +idle < spin_threshold + + + +yielding + +thread::yield_now() + + + +idle_check->yielding + + +idle < yield_threshold + + + +sleeping + +thread::sleep(μs) + + + +idle_check->sleeping + + +idle ≥ yield_threshold + + + diff --git a/crates/runtime-dashboard/docs/type_erasure.svg b/crates/runtime-dashboard/docs/type_erasure.svg new file mode 100644 index 0000000..0be2bd1 --- /dev/null +++ b/crates/runtime-dashboard/docs/type_erasure.svg @@ -0,0 +1,222 @@ + + + + + + +type_erasure + + +cluster_typed + +TYPED ZONE + + +cluster_erased + +ERASED ZONE (Box<dyn Any + Send>) + + +cluster_recovery + +RECOVERY ZONE + + + +user_send + +ctx.send::<M>(addr, msg) +actor.rs — Ctx::send() +M: 'static + Clone + Send + Sync + + + +box_erase + +Box::new(msg) → Box<dyn Any + Send> +actor.rs — Ctx::send() line 94 +Type information erased here + + + +user_send->box_erase + + +M + + + +send_any + +ContextInner::send_any(addr, msg) +worker.rs — WorkerContext::send_any() + + + +box_erase->send_any + + +Box<dyn Any + Send> + + + +route + +Route lookup +address_map.lookup(addr) + + + +send_any->route + + + + + +pending + + + +pending_local +Vec<(Addr, Box<dyn Any>)> + + + +route->pending + + +same worker +(wid == self) + + + +envelope + +Envelope::new(addr, payload) +delivery.rs +dest: ActorAddress +payload: Box<dyn Any + Send> + + + +route->envelope + + +cross worker +(wid != self) + + + +inbox_send + +InboxRegistry::try_deliver() +SenderT::try_send_any(msg) +downcast inside SenderT impl + + + +route->inbox_send + + +not found +(external inbox) + + + +mailbox + + +ActorSlot::mailbox +VecDeque<Box<dyn Any>> + + + +pending->mailbox + + +Phase 5 +local delivery + + + +channel + + +HybridChannel +(ArrayQueue + SegQueue) + + + +envelope->channel + + +try_send + + + +channel->mailbox + + +transfer_rx.pop() +→ pool.deliver() + + + +handle_any + +Actor<A>::handle_any(ctx, msg) +actor.rs line 56-61 + + + +mailbox->handle_any + + +pop_front() + + + +downcast + +msg.downcast::<A::Incoming>() + + + +handle_any->downcast + + + + + +ok_path + +A::handle(ctx, *typed) +Fully typed handler invocation + + + +downcast->ok_path + + +Ok(typed) + + + +err_path + +Silent drop +Wrong type — message discarded +type_mismatches counter++ + + + +downcast->err_path + + +Err(_) + + + diff --git a/crates/runtime-dashboard/examples/dashboard_demo.rs b/crates/runtime-dashboard/examples/dashboard_demo.rs new file mode 100644 index 0000000..3e35a60 --- /dev/null +++ b/crates/runtime-dashboard/examples/dashboard_demo.rs @@ -0,0 +1,155 @@ +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; + +use runtime_dashboard::{start_dashboard, DashboardConfig}; + +// ── Demo actors ───────────────────────────────────────────────────────── + +#[derive(Clone)] +struct Ping(ActorAddress); + +struct PingActor { + count: u32, +} + +impl PingActor { + fn new() -> Self { + Self { count: 0 } + } +} + +impl ActorInterface for PingActor { + type Incoming = Ping; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: Ping) { + self.count += 1; + // Forward to the target — creates cross-worker traffic + if self.count < 200 { + let _ = ctx.send(msg.0, Ping(ctx.self_addr())); + } + } +} + +#[derive(Clone)] +struct Tick; + +struct CounterActor { + ticks: u64, +} + +impl CounterActor { + fn new() -> Self { + Self { ticks: 0 } + } +} + +impl ActorInterface for CounterActor { + type Incoming = Tick; + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: Tick) { + self.ticks += 1; + } +} + +// ── Main ──────────────────────────────────────────────────────────────── + +fn main() { + let stop = Arc::new(AtomicBool::new(false)); + + // Handle Ctrl+C gracefully + { + let stop = Arc::clone(&stop); + let _ = std::panic::catch_unwind(|| { + // Try to register a signal handler; fall back to running until killed + unsafe { + libc_signal(2, move || stop.store(true, Ordering::Relaxed)); + } + }); + } + + let dash = start_dashboard(DashboardConfig { + port: 9090, + ..Default::default() + }); + dash.install_tracing(); + + let rt = Runtime::new(RuntimeConfig { + num_threads: 4, + max_actors: 1024, + actor_max_messages: 2000, + ..Default::default() + }); + + // Spawn ping actors for cross-worker traffic + let mut ping_addrs = Vec::new(); + for _ in 0..16 { + let addr = rt.spawn(PingActor::new()).unwrap(); + ping_addrs.push(addr); + } + + // Spawn counter actors for sustained traffic + let mut counter_addrs = Vec::new(); + for _ in 0..20 { + let addr = rt.spawn(CounterActor::new()).unwrap(); + counter_addrs.push(addr); + } + + let handle = rt.run().expect("failed to start runtime"); + dash.set_runtime(handle.runtime.clone()); + + eprintln!("Dashboard at http://localhost:9090 — press Ctrl+C to stop"); + + // Kick off ping-pong chains + for i in 0..ping_addrs.len() { + let target = ping_addrs[(i + 1) % ping_addrs.len()]; + let _ = handle.runtime.send_to(ping_addrs[i], Ping(target)); + } + + let mut round: u64 = 0; + while !stop.load(Ordering::Relaxed) { + // Send ticks to all counter actors + for addr in &counter_addrs { + let _ = handle.runtime.send_to(*addr, Tick); + } + + // Periodically spawn more actors + if round % 150 == 75 && counter_addrs.len() < 200 { + for _ in 0..8 { + match handle.runtime.spawn(CounterActor::new()) { + Ok(addr) => counter_addrs.push(addr), + Err(_) => break, + } + } + } + + // Periodically re-kick ping chains + if round % 80 == 0 && round > 0 { + for i in 0..ping_addrs.len() { + let target = ping_addrs[(i + 1) % ping_addrs.len()]; + let _ = handle.runtime.send_to(ping_addrs[i], Ping(target)); + } + } + + round += 1; + thread::sleep(Duration::from_millis(200)); + } + + eprintln!("\nShutting down..."); + handle.shutdown(); + dash.shutdown(); + handle.join(); +} + +// Minimal signal handling without external deps +unsafe fn libc_signal(_sig: i32, _handler: impl FnOnce()) { + // This is a no-op fallback; the loop checks the AtomicBool + // In practice, Ctrl+C will terminate the process +} diff --git a/crates/runtime-dashboard/examples/record_demo.rs b/crates/runtime-dashboard/examples/record_demo.rs new file mode 100644 index 0000000..3d424fb --- /dev/null +++ b/crates/runtime-dashboard/examples/record_demo.rs @@ -0,0 +1,125 @@ +use std::thread; +use std::time::Duration; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; + +use runtime_dashboard::{start_dashboard, DashboardConfig}; + +// ── Demo actors ───────────────────────────────────────────────────────── + +#[derive(Clone)] +struct Ping(ActorAddress); + +struct PingActor { + count: u32, +} + +impl PingActor { + fn new() -> Self { + Self { count: 0 } + } +} + +impl ActorInterface for PingActor { + type Incoming = Ping; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: Ping) { + self.count += 1; + if self.count < 100 { + let _ = ctx.send(msg.0, Ping(ctx.self_addr())); + } + } +} + +#[derive(Clone)] +struct Tick; + +struct CounterActor; + +impl ActorInterface for CounterActor { + type Incoming = Tick; + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: Tick) {} +} + +// ── Main ──────────────────────────────────────────────────────────────── + +fn main() { + let dash = start_dashboard(DashboardConfig { + port: 9090, + record: true, // Enable trace recording + ..Default::default() + }); + dash.install_tracing(); + + let rt = Runtime::new(RuntimeConfig { + num_threads: 4, + max_actors: 512, + actor_max_messages: 1000, + ..Default::default() + }); + + let mut ping_addrs = Vec::new(); + for _ in 0..12 { + let addr = rt.spawn(PingActor::new()).unwrap(); + ping_addrs.push(addr); + } + + let mut counter_addrs = Vec::new(); + for _ in 0..16 { + let addr = rt.spawn(CounterActor).unwrap(); + counter_addrs.push(addr); + } + + let handle = rt.run().expect("failed to start runtime"); + dash.set_runtime(handle.runtime.clone()); + + eprintln!("Recording trace for 10 seconds..."); + eprintln!("Dashboard at http://localhost:9090"); + + // Kick off ping-pong chains + for i in 0..ping_addrs.len() { + let target = ping_addrs[(i + 1) % ping_addrs.len()]; + let _ = handle.runtime.send_to(ping_addrs[i], Ping(target)); + } + + for round in 0..50 { + for addr in &counter_addrs { + let _ = handle.runtime.send_to(*addr, Tick); + } + + // Spawn more actors mid-recording + if round == 20 { + for _ in 0..8 { + let addr = handle.runtime.spawn(CounterActor).unwrap(); + counter_addrs.push(addr); + } + eprintln!(" Spawned 8 more actors"); + } + + // Re-kick pings + if round == 25 { + for i in 0..ping_addrs.len() { + let target = ping_addrs[(i + 1) % ping_addrs.len()]; + let _ = handle.runtime.send_to(ping_addrs[i], Ping(target)); + } + } + + thread::sleep(Duration::from_millis(200)); + } + + handle.shutdown(); + dash.shutdown(); + handle.join(); + + // Save trace + let path = "runtime_trace.json"; + match dash.save_trace(path) { + Ok(()) => eprintln!("Trace saved to {path}"), + Err(e) => eprintln!("Failed to save trace: {e}"), + } +} diff --git a/crates/runtime-dashboard/examples/replay_demo.rs b/crates/runtime-dashboard/examples/replay_demo.rs new file mode 100644 index 0000000..704b4fe --- /dev/null +++ b/crates/runtime-dashboard/examples/replay_demo.rs @@ -0,0 +1,23 @@ +use runtime_dashboard::{serve_replay, ReplayConfig}; + +fn main() { + let args: Vec = std::env::args().collect(); + if args.len() < 2 { + eprintln!("Usage: replay_demo [speed]"); + eprintln!(" speed: playback multiplier (default 1.0, e.g. 2.0 = 2x speed)"); + std::process::exit(1); + } + + let path = &args[1]; + let speed = args + .get(2) + .and_then(|s| s.parse::().ok()) + .unwrap_or(1.0); + + eprintln!("Replaying {path} at {speed}x speed"); + + if let Err(e) = serve_replay(path, ReplayConfig { port: 9090, speed }) { + eprintln!("Error: {e}"); + std::process::exit(1); + } +} diff --git a/crates/runtime-dashboard/src/dashboard_html.rs b/crates/runtime-dashboard/src/dashboard_html.rs new file mode 100644 index 0000000..d3b0519 --- /dev/null +++ b/crates/runtime-dashboard/src/dashboard_html.rs @@ -0,0 +1,369 @@ +pub const DASHBOARD_HTML: &str = r##" + + + + +Swactor Runtime Dashboard + + + +
+

+ Swactor Runtime Dashboard + + REPLAY +

+
+ + + +
+
+
+
+
+ +
+
+

Worker Distribution

+ +
+
+ +
+

Stats

+
+
0
Actors
+
0
Messages
+
0
Workers
+
0
Mailbox
+
+
+ +
+

Actors

+
+ + + +
AddressWorker
+
+
+ +
+

Activity Log

+
+ + + +
SeqTimeLevelWorkerMessageFields
+
+
+
+ + + + +"##; diff --git a/crates/runtime-dashboard/src/layer.rs b/crates/runtime-dashboard/src/layer.rs new file mode 100644 index 0000000..8633a0d --- /dev/null +++ b/crates/runtime-dashboard/src/layer.rs @@ -0,0 +1,227 @@ +use std::collections::VecDeque; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; +use tracing::field::{Field, Visit}; +use tracing::span; +use tracing::Subscriber; +use tracing_subscriber::layer::Context; +use tracing_subscriber::registry::LookupSpan; +use tracing_subscriber::Layer; + +/// A single captured tracing event. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DashboardEvent { + pub seq: u64, + pub timestamp_ms: u64, + pub level: String, + pub message: String, + pub worker_id: Option, + pub fields: serde_json::Map, +} + +/// Thread-safe ring buffer for dashboard events, with optional full-log recording. +pub struct EventStore { + events: Mutex>, + capacity: usize, + next_seq: AtomicU64, + /// When recording is enabled, all events are also appended here (unbounded). + full_log: Option>>, +} + +impl EventStore { + pub fn new(capacity: usize, record: bool) -> Self { + Self { + events: Mutex::new(VecDeque::with_capacity(capacity)), + capacity, + next_seq: AtomicU64::new(0), + full_log: if record { + Some(Mutex::new(Vec::new())) + } else { + None + }, + } + } + + pub fn push(&self, mut event: DashboardEvent) { + event.seq = self.next_seq.fetch_add(1, Ordering::Relaxed); + if let Some(ref log) = self.full_log { + log.lock().unwrap().push(event.clone()); + } + let mut events = self.events.lock().unwrap(); + if events.len() >= self.capacity { + events.pop_front(); + } + events.push_back(event); + } + + /// Read events starting from `cursor`. Returns the new events and the updated cursor. + pub fn read_from(&self, cursor: u64) -> (Vec, u64) { + let events = self.events.lock().unwrap(); + if events.is_empty() { + return (Vec::new(), cursor); + } + + let first_seq = events.front().unwrap().seq; + let last_seq = events.back().unwrap().seq; + + if cursor > last_seq { + return (Vec::new(), cursor); + } + + let start = if cursor <= first_seq { + 0 + } else { + (cursor - first_seq) as usize + }; + + let batch: Vec = events.iter().skip(start).cloned().collect(); + let new_cursor = last_seq + 1; + (batch, new_cursor) + } + + /// Returns a clone of the full event log. Only available when recording is enabled. + pub fn all_events(&self) -> Option> { + self.full_log + .as_ref() + .map(|log| log.lock().unwrap().clone()) + } +} + +pub(crate) fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} + +/// Visitor that extracts the message string and collects other fields. +struct FieldVisitor { + message: String, + fields: serde_json::Map, +} + +impl FieldVisitor { + fn new() -> Self { + Self { + message: String::new(), + fields: serde_json::Map::new(), + } + } +} + +impl Visit for FieldVisitor { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + self.message = format!("{:?}", value); + } else { + self.fields.insert( + field.name().to_string(), + serde_json::Value::String(format!("{:?}", value)), + ); + } + } + + fn record_u64(&mut self, field: &Field, value: u64) { + if field.name() == "message" { + self.message = value.to_string(); + } else { + self.fields.insert( + field.name().to_string(), + serde_json::Value::Number(value.into()), + ); + } + } + + fn record_i64(&mut self, field: &Field, value: i64) { + if field.name() == "message" { + self.message = value.to_string(); + } else { + self.fields.insert( + field.name().to_string(), + serde_json::Value::Number(value.into()), + ); + } + } + + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "message" { + self.message = value.to_string(); + } else { + self.fields.insert( + field.name().to_string(), + serde_json::Value::String(value.to_string()), + ); + } + } +} + +/// A tracing Layer that captures events into an EventStore. +pub struct DashboardLayer { + store: std::sync::Arc, +} + +impl DashboardLayer { + pub fn new(store: std::sync::Arc) -> Self { + Self { store } + } +} + +impl Layer for DashboardLayer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_event(&self, event: &tracing::Event<'_>, ctx: Context<'_, S>) { + let mut visitor = FieldVisitor::new(); + event.record(&mut visitor); + + // Walk span context to find worker_id + let mut worker_id = None; + if let Some(scope) = ctx.event_scope(event) { + for span in scope { + let exts = span.extensions(); + if let Some(wid) = exts.get::() { + worker_id = Some(wid.0); + break; + } + } + } + + // Also check if worker_id was a field on the event itself + if worker_id.is_none() { + if let Some(serde_json::Value::Number(n)) = visitor.fields.get("worker_id") { + worker_id = n.as_u64().map(|v| v as usize); + } + } + + let dashboard_event = DashboardEvent { + seq: 0, // filled by push() + timestamp_ms: now_ms(), + level: event.metadata().level().to_string(), + message: visitor.message, + worker_id, + fields: visitor.fields, + }; + + self.store.push(dashboard_event); + } + + fn on_new_span(&self, attrs: &span::Attributes<'_>, id: &span::Id, ctx: Context<'_, S>) { + // Extract worker_id from span fields and store in extensions + let mut visitor = FieldVisitor::new(); + attrs.record(&mut visitor); + + if let Some(serde_json::Value::Number(n)) = visitor.fields.get("worker_id") { + if let Some(wid) = n.as_u64() { + if let Some(span) = ctx.span(id) { + span.extensions_mut().insert(WorkerIdField(wid as usize)); + } + } + } + } +} + +/// Stored in span extensions to propagate worker_id to child events. +struct WorkerIdField(usize); diff --git a/crates/runtime-dashboard/src/lib.rs b/crates/runtime-dashboard/src/lib.rs new file mode 100644 index 0000000..26e519a --- /dev/null +++ b/crates/runtime-dashboard/src/lib.rs @@ -0,0 +1,198 @@ +pub mod layer; +pub mod trace; +mod dashboard_html; +mod server; + +use std::io; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +use swactor::config::RuntimeConfig; +use swactor::runtime::{Runtime, RuntimeHandle}; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::util::SubscriberInitExt; + +use crate::layer::{now_ms, DashboardLayer, EventStore}; +use crate::trace::{RuntimeTrace, TimestampedStats}; + +/// Configuration for the runtime dashboard. +#[derive(Debug, Clone)] +pub struct DashboardConfig { + pub port: u16, + pub event_capacity: usize, + /// Enable trace recording for `save_trace()`. When true, all events + /// are kept in an unbounded log and stats are periodically sampled. + pub record: bool, +} + +impl Default for DashboardConfig { + fn default() -> Self { + Self { + port: 9090, + event_capacity: 10_000, + record: false, + } + } +} + +/// Configuration for replaying a recorded trace. +#[derive(Debug, Clone)] +pub struct ReplayConfig { + pub port: u16, + /// Playback speed multiplier (1.0 = real-time, 2.0 = double speed). + pub speed: f64, +} + +impl Default for ReplayConfig { + fn default() -> Self { + Self { + port: 9090, + speed: 1.0, + } + } +} + +/// Handle to a running dashboard. Allows attaching a runtime after creation. +pub struct DashboardHandle { + store: Arc, + runtime: Arc>>>, + shutdown: Arc, + stats_timeline: Arc>>, + recording: bool, +} + +impl DashboardHandle { + /// Install a global tracing subscriber with the dashboard layer. + pub fn install_tracing(&self) { + tracing_subscriber::registry() + .with(DashboardLayer::new(Arc::clone(&self.store))) + .init(); + } + + /// Return the raw `DashboardLayer` for users who want to compose their own subscriber. + pub fn layer(&self) -> DashboardLayer { + DashboardLayer::new(Arc::clone(&self.store)) + } + + /// Attach a runtime to the dashboard, enabling stats polling. + pub fn set_runtime(&self, runtime: Arc) { + *self.runtime.lock().unwrap() = Some(runtime); + } + + /// Whether trace recording is enabled. + pub fn is_recording(&self) -> bool { + self.recording + } + + /// Signal the dashboard to shut down (SSE clients receive "done"). + pub fn shutdown(&self) { + self.shutdown.store(true, Ordering::Release); + } + + /// Save the recorded trace to a JSON file. + /// + /// Only works when `DashboardConfig::record` was set to `true`. + pub fn save_trace(&self, path: &str) -> io::Result<()> { + let events = self.store.all_events().ok_or_else(|| { + io::Error::new( + io::ErrorKind::Other, + "recording not enabled (set DashboardConfig::record = true)", + ) + })?; + let stats_timeline = self.stats_timeline.lock().unwrap().clone(); + let trace = RuntimeTrace { + events, + stats_timeline, + }; + let json = serde_json::to_string(&trace) + .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?; + std::fs::write(path, json) + } +} + +/// Start a dashboard server and return a handle. +/// +/// The dashboard starts serving immediately. Call `install_tracing()` to set up +/// the global subscriber, and `set_runtime()` to enable stats polling. +pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle { + let store = Arc::new(EventStore::new(config.event_capacity, config.record)); + let runtime: Arc>>> = Arc::new(Mutex::new(None)); + let shutdown = Arc::new(AtomicBool::new(false)); + let stats_timeline = Arc::new(Mutex::new(Vec::new())); + + server::spawn_http_server( + Arc::clone(&store), + Arc::clone(&runtime), + Arc::clone(&shutdown), + config.port, + ); + + // Start stats recorder thread when recording is enabled + if config.record { + let rt_ref = Arc::clone(&runtime); + let timeline = Arc::clone(&stats_timeline); + let stop = Arc::clone(&shutdown); + thread::spawn(move || { + loop { + if stop.load(Ordering::Relaxed) { + return; + } + let maybe_rt = rt_ref.lock().unwrap().clone(); + if let Some(rt) = maybe_rt { + let stats = rt.stats(); + let ts = TimestampedStats { + timestamp_ms: now_ms(), + stats, + }; + timeline.lock().unwrap().push(ts); + } + thread::sleep(Duration::from_millis(200)); + } + }); + } + + eprintln!("Runtime dashboard at http://localhost:{}", config.port); + + DashboardHandle { + store, + runtime, + shutdown, + stats_timeline, + recording: config.record, + } +} + +/// Convenience: create a runtime, start a dashboard, install tracing, and run. +/// +/// Returns the runtime handle and dashboard handle. +pub fn run_with_dashboard( + rt_config: RuntimeConfig, + dash_config: DashboardConfig, +) -> (RuntimeHandle, DashboardHandle) { + let dash = start_dashboard(dash_config); + dash.install_tracing(); + + let rt = Runtime::new(rt_config); + let handle = rt.run().expect("failed to start runtime"); + dash.set_runtime(Arc::clone(&handle.runtime)); + + (handle, dash) +} + +/// Load a trace file and serve a replay dashboard. Blocks indefinitely. +pub fn serve_replay(path: &str, config: ReplayConfig) -> io::Result<()> { + let data = std::fs::read_to_string(path)?; + let trace: RuntimeTrace = serde_json::from_str(&data) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + + server::spawn_replay_server(Arc::new(trace), config.port, config.speed); + + eprintln!("Replay dashboard at http://localhost:{}", config.port); + eprintln!("Press Ctrl+C to stop"); + + loop { + thread::sleep(Duration::from_secs(3600)); + } +} diff --git a/crates/runtime-dashboard/src/server.rs b/crates/runtime-dashboard/src/server.rs new file mode 100644 index 0000000..15fabac --- /dev/null +++ b/crates/runtime-dashboard/src/server.rs @@ -0,0 +1,337 @@ +use std::io::{self, Read as IoRead}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{mpsc, Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use swactor::runtime::Runtime; + +use crate::dashboard_html::DASHBOARD_HTML; +use crate::layer::EventStore; +use crate::trace::RuntimeTrace; + +/// Format a server-sent event. +fn format_sse(event: &str, data: &str) -> Vec { + format!("event: {event}\ndata: {data}\n\n").into_bytes() +} + +/// Adapts an `mpsc::Receiver>` to `std::io::Read` for tiny_http streaming. +struct ChannelReader { + rx: mpsc::Receiver>, + buf: Vec, + pos: usize, +} + +impl ChannelReader { + fn new(rx: mpsc::Receiver>) -> Self { + Self { + rx, + buf: Vec::new(), + pos: 0, + } + } +} + +impl IoRead for ChannelReader { + fn read(&mut self, out: &mut [u8]) -> io::Result { + // Drain current buffer first. + if self.pos < self.buf.len() { + let n = std::cmp::min(out.len(), self.buf.len() - self.pos); + out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]); + self.pos += n; + return Ok(n); + } + + // Wait for next chunk. + match self.rx.recv() { + Ok(data) => { + if data.is_empty() { + return Ok(0); // EOF signal + } + let n = std::cmp::min(out.len(), data.len()); + out[..n].copy_from_slice(&data[..n]); + if n < data.len() { + self.buf = data; + self.pos = n; + } else { + self.buf.clear(); + self.pos = 0; + } + Ok(n) + } + Err(_) => Ok(0), // channel closed + } + } +} + +fn make_sse_response( + rx: mpsc::Receiver>, +) -> tiny_http::Response> { + let reader = ChannelReader::new(rx); + tiny_http::Response::new( + tiny_http::StatusCode(200), + vec![ + "Content-Type: text/event-stream" + .parse::() + .unwrap(), + "Cache-Control: no-cache" + .parse::() + .unwrap(), + "Connection: keep-alive" + .parse::() + .unwrap(), + ], + Box::new(reader) as Box, + None, + None, + ) +} + +fn respond_html(request: tiny_http::Request, mode: &str) { + let html = DASHBOARD_HTML.replace("__DASHBOARD_MODE__", mode); + let response = tiny_http::Response::from_string(html).with_header( + "Content-Type: text/html; charset=utf-8" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + +fn respond_404(request: tiny_http::Request) { + let response = tiny_http::Response::from_string("Not Found").with_status_code(404); + let _ = request.respond(response); +} + +// ── Live server ───────────────────────────────────────────────────────── + +/// Start the live HTTP server with a pool of handler threads. +pub(crate) fn spawn_http_server( + store: Arc, + runtime: Arc>>>, + shutdown: Arc, + port: u16, +) { + let addr = format!("0.0.0.0:{port}"); + let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server"); + let server = Arc::new(server); + + for _ in 0..4 { + let server = Arc::clone(&server); + let store = Arc::clone(&store); + let runtime = Arc::clone(&runtime); + let shutdown = Arc::clone(&shutdown); + thread::spawn(move || { + loop { + let request = match server.recv() { + Ok(r) => r, + Err(_) => break, + }; + + let url = request.url().to_string(); + match url.as_str() { + "/" => respond_html(request, "live"), + "/events" => { + handle_live_sse( + request, + Arc::clone(&store), + Arc::clone(&runtime), + Arc::clone(&shutdown), + ); + } + "/api/stats" => { + handle_stats_api(request, Arc::clone(&runtime)); + } + _ => respond_404(request), + } + } + }); + } +} + +fn handle_live_sse( + request: tiny_http::Request, + store: Arc, + runtime: Arc>>>, + shutdown: Arc, +) { + let (tx, rx) = mpsc::channel::>(); + let response = make_sse_response(rx); + + // Spawn producer thread + thread::spawn(move || { + let mut cursor: u64 = 0; + + loop { + // Send stats if runtime is available + { + let maybe_rt = runtime.lock().unwrap().clone(); + if let Some(rt) = maybe_rt { + let stats = rt.stats(); + let json = serde_json::to_string(&stats).unwrap(); + if tx.send(format_sse("stats", &json)).is_err() { + return; + } + } + } + + // Send new activity events + let (batch, new_cursor) = store.read_from(cursor); + if !batch.is_empty() { + let json = serde_json::to_string(&batch).unwrap(); + if tx.send(format_sse("activity", &json)).is_err() { + return; + } + cursor = new_cursor; + } + + if shutdown.load(Ordering::Relaxed) { + let _ = tx.send(format_sse("done", "{}")); + let _ = tx.send(Vec::new()); // EOF + return; + } + + thread::sleep(Duration::from_millis(200)); + } + }); + + // Blocks until connection closes + let _ = request.respond(response); +} + +fn handle_stats_api(request: tiny_http::Request, runtime: Arc>>>) { + let maybe_rt = runtime.lock().unwrap().clone(); + let json = match maybe_rt { + Some(rt) => serde_json::to_string(&rt.stats()).unwrap(), + None => "{}".to_string(), + }; + let response = tiny_http::Response::from_string(json).with_header( + "Content-Type: application/json" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + +// ── Replay server ─────────────────────────────────────────────────────── + +/// Start a replay HTTP server that serves a pre-recorded trace. +pub(crate) fn spawn_replay_server(trace: Arc, port: u16, speed: f64) { + let addr = format!("0.0.0.0:{port}"); + let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server"); + let server = Arc::new(server); + + for _ in 0..4 { + let server = Arc::clone(&server); + let trace = Arc::clone(&trace); + thread::spawn(move || { + loop { + let request = match server.recv() { + Ok(r) => r, + Err(_) => break, + }; + + let url = request.url().to_string(); + match url.as_str() { + "/" => respond_html(request, "replay"), + "/events" => { + handle_replay_sse(request, Arc::clone(&trace), speed); + } + _ => respond_404(request), + } + } + }); + } +} + +fn handle_replay_sse(request: tiny_http::Request, trace: Arc, speed: f64) { + let (tx, rx) = mpsc::channel::>(); + let response = make_sse_response(rx); + + thread::spawn(move || { + // Send replay metadata + let meta = serde_json::json!({ + "total_events": trace.events.len(), + "total_stats": trace.stats_timeline.len(), + "speed": speed, + }); + if tx.send(format_sse("replay_meta", &meta.to_string())).is_err() { + return; + } + + // Find the earliest timestamp across events and stats + let base_time = trace + .events + .first() + .map(|e| e.timestamp_ms) + .into_iter() + .chain(trace.stats_timeline.first().map(|s| s.timestamp_ms)) + .min() + .unwrap_or(0); + + let playback_start = Instant::now(); + let mut event_idx = 0; + let mut stats_idx = 0; + + loop { + let elapsed_ms = (playback_start.elapsed().as_millis() as f64 * speed) as u64; + let virtual_time = base_time + elapsed_ms; + + // Batch events up to virtual_time + let mut batch = Vec::new(); + while event_idx < trace.events.len() + && trace.events[event_idx].timestamp_ms <= virtual_time + { + batch.push(trace.events[event_idx].clone()); + event_idx += 1; + } + if !batch.is_empty() { + let json = serde_json::to_string(&batch).unwrap(); + if tx.send(format_sse("activity", &json)).is_err() { + return; + } + } + + // Send stats snapshots up to virtual_time + while stats_idx < trace.stats_timeline.len() + && trace.stats_timeline[stats_idx].timestamp_ms <= virtual_time + { + let json = + serde_json::to_string(&trace.stats_timeline[stats_idx].stats).unwrap(); + if tx.send(format_sse("stats", &json)).is_err() { + return; + } + stats_idx += 1; + } + + // Send progress + let total = trace.events.len() + trace.stats_timeline.len(); + let done_count = event_idx + stats_idx; + let progress = if total > 0 { + done_count as f64 / total as f64 + } else { + 1.0 + }; + let progress_json = serde_json::json!({ "progress": progress }); + if tx + .send(format_sse("replay_progress", &progress_json.to_string())) + .is_err() + { + return; + } + + // Check if replay is complete + if event_idx >= trace.events.len() + && stats_idx >= trace.stats_timeline.len() + { + let _ = tx.send(format_sse("done", "{}")); + let _ = tx.send(Vec::new()); // EOF + return; + } + + thread::sleep(Duration::from_millis(50)); + } + }); + + // Blocks until connection closes + let _ = request.respond(response); +} diff --git a/crates/runtime-dashboard/src/trace.rs b/crates/runtime-dashboard/src/trace.rs new file mode 100644 index 0000000..09505d5 --- /dev/null +++ b/crates/runtime-dashboard/src/trace.rs @@ -0,0 +1,18 @@ +use serde::{Deserialize, Serialize}; +use swactor::stats::RuntimeStats; + +use crate::layer::DashboardEvent; + +/// Complete trace of a runtime execution, suitable for saving/loading. +#[derive(Debug, Serialize, Deserialize)] +pub struct RuntimeTrace { + pub events: Vec, + pub stats_timeline: Vec, +} + +/// A stats snapshot with a wall-clock timestamp. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TimestampedStats { + pub timestamp_ms: u64, + pub stats: RuntimeStats, +} diff --git a/src/crate_test/mod.rs b/src/crate_test/mod.rs index c23af3c..1d0e99f 100644 --- a/src/crate_test/mod.rs +++ b/src/crate_test/mod.rs @@ -10,7 +10,7 @@ use crate::actor::{ActorAddress, AnyActor, Ctx}; use crate::channel::Receiver; use crate::config::RuntimeConfig; use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; -use crate::stats::WorkerStats; +use crate::stats::{MailboxSnapshot, WorkerStats}; use crate::worker::Worker; @@ -68,7 +68,8 @@ impl Env { 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()); + let mbox_snap = Arc::new(std::sync::Mutex::new(MailboxSnapshot::new())); + let worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats.clone(), mbox_snap); Self { worker, @@ -385,7 +386,8 @@ fn run_loop_stops_on_shutdown() { 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 mbox_snap = Arc::new(std::sync::Mutex::new(MailboxSnapshot::new())); + let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats, mbox_snap); let is_running = AtomicBool::new(false); let address_map = AddressMap::new(); diff --git a/src/runtime.rs b/src/runtime.rs index 01a8fc4..92e6561 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -9,7 +9,7 @@ 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::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; -use crate::stats::WorkerStats; +use crate::stats::{ActorInfo, MailboxSnapshot, WorkerStats}; // Re-export stats types so existing code using `runtime::*` still works pub use crate::stats::{RuntimeStats, WorkerInfo}; use crate::worker::Worker; @@ -66,6 +66,8 @@ pub struct Runtime { placement: Placement, is_running: AtomicBool, worker_stats: Vec>, + /// Per-worker mailbox snapshots, updated each tick by workers. + mailbox_snapshots: Vec>>, /// Workers available for tick(). run() drains this and moves workers to threads. tick_workers: RefCell>, } @@ -91,6 +93,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 mailbox_snapshots = Vec::with_capacity(num_workers); let mut workers = Vec::with_capacity(num_workers); for i in 0..num_workers { @@ -104,11 +107,13 @@ impl Runtime { spawn_txs.push(spawn_tx); let stats = Arc::new(WorkerStats::new()); + let mbox_snap = Arc::new(std::sync::Mutex::new(MailboxSnapshot::new())); worker_stats.push(stats.clone()); - workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats)); + mailbox_snapshots.push(mbox_snap.clone()); + workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats, mbox_snap)); } - Self { + let rt = Self { config, address_map, inbox_registry, @@ -117,8 +122,18 @@ impl Runtime { placement, is_running: AtomicBool::new(false), worker_stats, + mailbox_snapshots, tick_workers: RefCell::new(workers), - } + }; + + #[cfg(feature = "tracing")] + tracing::info!( + num_workers, + max_actors = rt.config.max_actors, + "runtime.created" + ); + + rt } /// Spawn an actor, returns its address @@ -130,12 +145,24 @@ impl Runtime { self.spawn_txs[worker_id.as_usize()] .try_send((addr, boxed)) .map_err(|_| Error::from("Runtime error: spawn queue full"))?; + + #[cfg(feature = "tracing")] + tracing::info!( + actor_addr = %addr, + worker_id = worker_id.as_usize(), + "actor.spawned" + ); + Ok(addr) } /// Send a message to an actor address pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { let msg_box: Box = Box::new(msg); + + #[cfg(feature = "tracing")] + tracing::trace!(dest = %addr, "message.sent"); + match self.address_map.lookup(&addr) { Some(wid) => self.transfer_txs[wid.as_usize()] .try_send(Envelope::new(addr, msg_box)) @@ -185,6 +212,9 @@ impl Runtime { pub fn run(self) -> Result { self.is_running.store(true, Ordering::Release); + #[cfg(feature = "tracing")] + tracing::info!(num_workers = self.config.num_threads.max(1), "runtime.started"); + let workers: Vec = self.tick_workers.replace(Vec::new()); let rt = Arc::new(self); @@ -232,6 +262,11 @@ impl Runtime { 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), + local_sends: ws.local_sends.load(Ordering::Relaxed), + cross_sends: ws.cross_sends.load(Ordering::Relaxed), + inbox_sends: ws.inbox_sends.load(Ordering::Relaxed), + type_mismatches: ws.type_mismatches.load(Ordering::Relaxed), + panics: ws.panics.load(Ordering::Relaxed), }) .collect(); let actors = self @@ -240,15 +275,41 @@ impl Runtime { .into_iter() .map(|(addr, wid)| (addr, wid.as_usize())) .collect(); + + // Collect per-actor mailbox depth details + let mut actor_details = Vec::new(); + for (wid, snap_lock) in self.mailbox_snapshots.iter().enumerate() { + let snap = snap_lock.lock().unwrap(); + for &(addr, depth) in &snap.depths { + actor_details.push(ActorInfo { + address: addr, + worker_id: wid, + mailbox_depth: depth, + }); + } + } + + // Drain tick timings from each worker + let tick_timings = self + .worker_stats + .iter() + .map(|ws| ws.drain_tick_timings()) + .collect(); + RuntimeStats { num_workers, actors, workers, + actor_details, + tick_timings, } } /// Signal all workers to stop pub fn shutdown(&self) { + #[cfg(feature = "tracing")] + tracing::info!("runtime.shutdown"); + self.is_running.store(false, Ordering::Release); } } diff --git a/src/stats.rs b/src/stats.rs index 67a6a2f..1f8b515 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -2,11 +2,32 @@ use std::sync::atomic::{AtomicU64, AtomicUsize}; use crate::actor::ActorAddress; +/// Timing data for one tick_once invocation. +#[derive(Debug, Clone, Default)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +pub struct TickTiming { + /// Microseconds spent in each of the 6 phases. + pub phase_us: [u64; 6], + /// Total messages processed this tick. + pub messages_processed: usize, + /// Whether any work was done this tick. + pub did_work: bool, +} + /// Per-worker stats published via atomics. Readable from any thread. pub struct WorkerStats { pub num_actors: AtomicUsize, pub total_mailbox_depth: AtomicUsize, pub messages_processed: AtomicU64, + // Message routing counters + pub local_sends: AtomicU64, + pub cross_sends: AtomicU64, + pub inbox_sends: AtomicU64, + // Error counters + pub type_mismatches: AtomicU64, + pub panics: AtomicU64, + // Tick timing ring buffer (last N ticks) + tick_timings: std::sync::Mutex>, } impl WorkerStats { @@ -15,22 +36,97 @@ impl WorkerStats { num_actors: AtomicUsize::new(0), total_mailbox_depth: AtomicUsize::new(0), messages_processed: AtomicU64::new(0), + local_sends: AtomicU64::new(0), + cross_sends: AtomicU64::new(0), + inbox_sends: AtomicU64::new(0), + type_mismatches: AtomicU64::new(0), + panics: AtomicU64::new(0), + tick_timings: std::sync::Mutex::new(RingBuffer::new(1024)), } } + + pub fn push_tick_timing(&self, timing: TickTiming) { + self.tick_timings.lock().unwrap().push(timing); + } + + /// Returns a snapshot of recent tick timings (drains the buffer). + pub fn drain_tick_timings(&self) -> Vec { + self.tick_timings.lock().unwrap().drain() + } } /// Snapshot of per-worker state. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] pub struct WorkerInfo { pub id: usize, pub num_actors: usize, pub mailbox_depth: usize, pub messages_processed: u64, + pub local_sends: u64, + pub cross_sends: u64, + pub inbox_sends: u64, + pub type_mismatches: u64, + pub panics: u64, +} + +/// Per-actor info for the dashboard. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +pub struct ActorInfo { + pub address: ActorAddress, + pub worker_id: usize, + pub mailbox_depth: usize, } /// Snapshot of overall runtime state. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] pub struct RuntimeStats { pub num_workers: usize, /// Each entry is (address, worker_id). pub actors: Vec<(ActorAddress, usize)>, pub workers: Vec, + /// Per-actor detail including mailbox depth. + pub actor_details: Vec, + /// Recent tick timings per worker (index = worker id). + pub tick_timings: Vec>, +} + +/// Simple ring buffer for storing recent values. +pub(crate) struct RingBuffer { + buf: Vec, + capacity: usize, +} + +impl RingBuffer { + pub fn new(capacity: usize) -> Self { + Self { + buf: Vec::with_capacity(capacity), + capacity, + } + } + + pub fn push(&mut self, value: T) { + if self.buf.len() >= self.capacity { + self.buf.remove(0); + } + self.buf.push(value); + } + + /// Drain all items, returning them and leaving the buffer empty. + pub fn drain(&mut self) -> Vec { + std::mem::take(&mut self.buf) + } +} + +/// Per-actor mailbox depth snapshot, collected by workers. +pub(crate) struct MailboxSnapshot { + pub depths: Vec<(ActorAddress, usize)>, +} + +impl MailboxSnapshot { + pub fn new() -> Self { + Self { depths: Vec::new() } + } } diff --git a/src/worker.rs b/src/worker.rs index 5e0b749..7e7b5da 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -4,11 +4,12 @@ use std::collections::{HashMap, VecDeque}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::thread; +use std::time::Instant; use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx}; use crate::channel::Receiver; use crate::delivery::{Envelope, TickContext, WorkerId}; -use crate::stats::WorkerStats; +use crate::stats::{MailboxSnapshot, TickTiming, WorkerStats}; use crate::Error; /// A worker owns a set of actors and runs them in a loop. @@ -18,6 +19,8 @@ pub(crate) struct Worker { transfer_rx: Receiver, spawn_rx: Receiver<(ActorAddress, Box)>, stats: Arc, + /// Shared snapshot of per-actor mailbox depths, readable by Runtime::stats(). + mailbox_snapshot: Arc>, } impl Worker { @@ -26,6 +29,7 @@ impl Worker { transfer_rx: Receiver, spawn_rx: Receiver<(ActorAddress, Box)>, stats: Arc, + mailbox_snapshot: Arc>, ) -> Self { Self { id, @@ -33,18 +37,32 @@ impl Worker { transfer_rx, spawn_rx, stats, + mailbox_snapshot, } } /// Run one iteration of the worker loop. Returns `true` if any work was done. pub(crate) fn tick_once(&mut self, tc: &TickContext) -> bool { + #[cfg(feature = "tracing")] + let _span = tracing::trace_span!("worker.tick", worker_id = self.id.0).entered(); + let mut did_work = false; + let t0 = Instant::now(); // 1. Drain spawn queue → add actors to pool + #[cfg(feature = "tracing")] + let mut spawn_count: usize = 0; while let Some((addr, actor)) = self.spawn_rx.try_recv() { self.pool.insert(addr, actor); + #[cfg(feature = "tracing")] + { spawn_count += 1; } did_work = true; } + #[cfg(feature = "tracing")] + if spawn_count > 0 { + tracing::debug!(worker_id = self.id.0, count = spawn_count, "worker.spawns_drained"); + } + let t1 = Instant::now(); // 2. Drain transfer queue → deliver envelopes to actors while let Some(envelope) = self.transfer_rx.try_recv() { @@ -53,6 +71,7 @@ impl Worker { self.pool.deliver(&dest, payload); did_work = true; } + let t2 = Instant::now(); // 3. Tick all actors with WorkerContext let pending_local: RefCell)>> = @@ -64,12 +83,23 @@ impl Worker { worker_id: self.id, tc, pending_local: &pending_local, + stats: &self.stats, }; - processed = self.pool.tick_all(&worker_ctx); + processed = self.pool.tick_all(&worker_ctx, &self.stats); if processed > 0 { did_work = true; } } + let t3 = Instant::now(); + + #[cfg(feature = "tracing")] + if processed > 0 { + tracing::debug!( + worker_id = self.id.0, + messages_processed = processed, + "worker.tick_all" + ); + } // 4. Drain spawn queue again — actors spawned during step 3 // must be in the pool before pending_local delivery. @@ -77,6 +107,7 @@ impl Worker { self.pool.insert(addr, actor); did_work = true; } + let t4 = Instant::now(); // 5. Drain pending_local buffer → deliver to local actors let pending = pending_local.into_inner(); @@ -86,16 +117,55 @@ impl Worker { for (addr, msg) in pending { self.pool.deliver(&addr, msg); } + let t5 = Instant::now(); // 6. 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); + // Publish per-actor mailbox depths + { + let depths: Vec<(ActorAddress, usize)> = self.pool.mailbox_depths(); + let mut snap = self.mailbox_snapshot.lock().unwrap(); + snap.depths = depths; + } + + let t6 = Instant::now(); + + // Record tick timing + let timing = TickTiming { + phase_us: [ + t1.duration_since(t0).as_micros() as u64, + t2.duration_since(t1).as_micros() as u64, + t3.duration_since(t2).as_micros() as u64, + t4.duration_since(t3).as_micros() as u64, + t5.duration_since(t4).as_micros() as u64, + t6.duration_since(t5).as_micros() as u64, + ], + messages_processed: processed, + did_work, + }; + self.stats.push_tick_timing(timing); + + #[cfg(feature = "tracing")] + if did_work { + tracing::debug!( + worker_id = self.id.0, + num_actors = self.pool.len(), + mailbox_depth = self.pool.total_mailbox_depth(), + messages_processed = processed, + "worker.stats" + ); + } + did_work } pub(crate) fn run(&mut self, tc: &TickContext, is_running: &AtomicBool) { + #[cfg(feature = "tracing")] + let _span = tracing::info_span!("worker.run", worker_id = self.id.0).entered(); + let backoff = &tc.config.backoff_policy; let mut idle_count: u32 = 0; while is_running.load(Ordering::Acquire) { @@ -128,6 +198,7 @@ struct WorkerContext<'a> { worker_id: WorkerId, tc: &'a TickContext<'a>, pending_local: &'a RefCell)>>, + stats: &'a WorkerStats, } impl ContextInner for WorkerContext<'_> { @@ -135,17 +206,20 @@ impl ContextInner for WorkerContext<'_> { match self.tc.address_map.lookup(&addr) { Some(wid) if wid == self.worker_id => { // Same worker: buffer for local delivery (after current tick round) + self.stats.local_sends.fetch_add(1, Ordering::Relaxed); self.pending_local.borrow_mut().push((addr, msg)); Ok(()) } Some(wid) => { // Cross worker: envelope through transfer queue + self.stats.cross_sends.fetch_add(1, Ordering::Relaxed); let envelope = Envelope::new(addr, msg); let _ = self.tc.transfer_txs[wid.as_usize()].try_send(envelope); Ok(()) } None => { // Try inbox registry (external inboxes) + self.stats.inbox_sends.fetch_add(1, Ordering::Relaxed); self.tc.inbox_registry.try_deliver(addr, msg) } } @@ -201,7 +275,7 @@ impl ActorPool { } /// Tick all actors in the pool. Returns the number of messages processed. - pub fn tick_all(&mut self, inner: &dyn ContextInner) -> usize { + pub fn tick_all(&mut self, inner: &dyn ContextInner, stats: &WorkerStats) -> usize { let mut count = 0; for (&addr, slot) in self.actors.iter_mut() { let ctx = Ctx::new(inner, addr); @@ -210,7 +284,10 @@ impl ActorPool { slot.actor.handle_any(&ctx, msg); })); if result.is_err() { + stats.panics.fetch_add(1, Ordering::Relaxed); eprintln!("swactor: actor {addr} panicked in handler"); + #[cfg(feature = "tracing")] + tracing::error!(actor_addr = %addr, "actor.panicked"); } count += 1; } @@ -225,7 +302,9 @@ impl ActorPool { pub fn total_mailbox_depth(&self) -> usize { self.actors.values().map(|slot| slot.mailbox.len()).sum() } + + /// Returns per-actor mailbox depths for dashboard reporting. + pub fn mailbox_depths(&self) -> Vec<(ActorAddress, usize)> { + self.actors.iter().map(|(&addr, slot)| (addr, slot.mailbox.len())).collect() + } } - - - diff --git a/tools/package-lock.json b/tools/package-lock.json index 1e3b1de..62ab53d 100644 --- a/tools/package-lock.json +++ b/tools/package-lock.json @@ -1,5 +1,5 @@ { - "name": "workspace", + "name": "tools", "lockfileVersion": 3, "requires": true, "packages": { -- 2.45.2