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

fix: gossip dashboard (#24)

Browse source 
Better, but still imperfect, performance in replay for large simulation sizes.
This commit is contained in:
zacheryasc 2026-02-09 09:07:38 +00:00
parent 1a64f421bf
commit d61ea50414
17 changed files with 1685 additions and 1241 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -216,7 +216,6 @@ version = "0.1.0"
dependencies = [
"serde",
"serde_json",
"swactor",
"swactor-gossip",
"tiny_http",
"toml",

1
crates/gossip-dashboard/Cargo.toml
View file

@ -4,7 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
swactor = { path = "../..", features = ["serde"] }
swactor-gossip = { path = "../swactor-gossip" }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

126
crates/gossip-dashboard/README.md Normal file
View file

@ -0,0 +1,126 @@
# gossip-dashboard
Interactive web dashboard for visualizing gossip protocol simulations.
The workflow has two steps:
1. **Generate traces** -- run simulations against TOML config files, producing `.trace.json` files.
2. **Replay traces** -- start the dashboard server, point it at a directory of traces, and explore them in the browser.
## Quick start
```bash
# 1. Generate traces from the bundled configs (outputs to traces/)
cargo run -p gossip-dashboard --example generate_traces
# 2. Launch the dashboard
cargo run -p gossip-dashboard --example replay -- traces
# => open http://localhost:8080
```
## Commands
### `generate_traces`
Runs gossip simulations and writes `.trace.json` files.
```
generate_traces # all bundled configs -> traces/
generate_traces <out-dir> # all bundled configs -> <out-dir>/
generate_traces <out-dir> <config.toml> [more.toml …] # specific configs -> <out-dir>/
```
Bundled configs live in `examples/configs/`. When no config paths are given, every `.toml` in that directory is run.
Output filenames are derived from the simulation name (lowercased, spaces to underscores). Example output:
```
traces/
ring_10_nodes.trace.json
star_7_nodes.trace.json
chain_8_nodes.trace.json
full_mesh_6_nodes.trace.json
partition_&_heal_8_nodes.trace.json
```
### `replay`
Starts an HTTP server that serves the dashboard UI and the trace data.
```
replay <trace-dir> [port]
```
| Argument | Required | Default | Description |
|-------------|----------|---------|------------------------------------------|
| `trace-dir` | yes | -- | Directory containing `.trace.json` files |
| `port` | no | 8080 | Port to bind on |
The server exposes three endpoints:
| Route | Description |
|--------------------------|------------------------------------|
| `GET /` | Dashboard HTML |
| `GET /traces` | JSON list of available trace files |
| `GET /trace.json?file=…` | Fetch a specific trace |
## Configuration (TOML)
Each simulation is defined by a TOML file. Example (`ring_10.toml`):
```toml
name = "Ring (10 nodes)"
topology = "ring"
num_nodes = 10
num_rounds = 15
ticks_per_round = 5
num_threads = 1
[initial_data]
color = "blue"
version = "1"
status = "active"
```
### Fields
| Field | Type | Required | Description |
|--------------------|-------------------|----------|-------------------------------------------------------------------|
| `name` | string | yes | Display name for the simulation |
| `topology` | string | yes | Network topology (see below) |
| `num_nodes` | integer | yes | Number of gossip nodes |
| `num_rounds` | integer | yes | Number of gossip rounds to run |
| `ticks_per_round` | integer | yes | Simulation ticks per round |
| `num_threads` | integer | yes | Worker threads (`1` = deterministic single-threaded) |
| `heal_after_round` | integer | no | Round after which partitioned halves are bridged |
| `initial_data` | table of strings | no | Key-value pairs seeded on node 0 before gossip begins |
### Topologies
| Value | Shape |
|---------------|--------------------------------------------------------------------------|
| `ring` | Each node connects to the next, forming a circle |
| `star` | Node 0 is a hub with bidirectional links to every other node |
| `full_mesh` | Every node connects bidirectionally to every other node |
| `chain` | Unidirectional chain: node 0 -> 1 -> 2 -> ... -> N-1 |
| `partitioned` | Two isolated full-mesh halves; use `heal_after_round` to bridge them |
## Bundled configs
| File | Topology | Nodes | Rounds | Notes |
|---------------------------|-------------|-------|--------|----------------------------|
| `ring_10.toml` | ring | 10 | 15 | |
| `star_7.toml` | star | 7 | 10 | |
| `full_mesh_6.toml` | full_mesh | 6 | 8 | |
| `chain_8.toml` | chain | 8 | 20 | |
| `partitioned_8_heal.toml` | partitioned | 8 | 20 | Heals after round 10 |
## Dashboard UI
Once a trace is loaded in the browser:
- **Graph canvas** -- nodes arranged in a circle then refined with force-directed layout. Nodes and edges flash as events are replayed.
- **Stats panel** -- total nodes, edges, messages, current round.
- **Worker logs** -- per-thread activity feed.
- **Event table** -- full event log with columns: Seq, Round, Thread, Node, Event, Details.
- **Playback controls** -- First / Prev / Play / Pause / Next / Last, timeline slider, speed adjustment (10 ms -- 2000 ms per event).

11
crates/gossip-dashboard/examples/configs/chain_8.toml Normal file
View file

@ -0,0 +1,11 @@
name = "Chain (8 nodes)"
topology = "chain"
num_nodes = 8
num_rounds = 20
ticks_per_round = 4
num_threads = 1
[initial_data]
color = "yellow"
version = "1"
status = "pending"

12
crates/gossip-dashboard/examples/configs/full_mesh_6.toml Normal file
View file

@ -0,0 +1,12 @@
name = "Full Mesh (6 nodes)"
topology = "full_mesh"
num_nodes = 6
num_rounds = 8
ticks_per_round = 4
num_threads = 1
[initial_data]
color = "green"
version = "3"
status = "ok"
region = "us-east"

11
crates/gossip-dashboard/examples/configs/partitioned_800_heal.toml Normal file
View file

@ -0,0 +1,11 @@
name = "Partition & Heal (100 nodes)"
topology = "partitioned"
num_nodes = 100
num_rounds = 10
ticks_per_round = 5
num_threads = 4
heal_after_round = 5
[initial_data]
color = "blue"
version = "1"

5
crates/gossip-dashboard/examples/sim.toml → crates/gossip-dashboard/examples/configs/partitioned_8_heal.toml
View file

@ -1,11 +1,10 @@
name = "Partitioned-8 Heal"
name = "Partition & Heal (8 nodes)"
topology = "partitioned"
num_nodes = 8
num_rounds = 20
ticks_per_round = 5
num_threads = 2
num_threads = 1
heal_after_round = 10
port = 8080
[initial_data]
color = "blue"

11
crates/gossip-dashboard/examples/configs/ring_10.toml Normal file
View file

@ -0,0 +1,11 @@
name = "Ring (10 nodes)"
topology = "ring"
num_nodes = 10
num_rounds = 15
ticks_per_round = 5
num_threads = 1
[initial_data]
color = "blue"
version = "1"
status = "active"

11
crates/gossip-dashboard/examples/configs/star_7.toml Normal file
View file

@ -0,0 +1,11 @@
name = "Star (7 nodes)"
topology = "star"
num_nodes = 7
num_rounds = 10
ticks_per_round = 4
num_threads = 1
[initial_data]
color = "red"
version = "2"
status = "ready"

40
crates/gossip-dashboard/examples/dashboard.rs
View file

@ -1,40 +0,0 @@
use gossip_dashboard::{DashboardConfig, config::SimFileConfig, run_with_dashboard, save_trace};
use swactor_gossip::sim::{SimConfig, Topology};
fn main() {
let args: Vec<String> = std::env::args().collect();
let (config, dash) = if let Some(path) = args.get(1) {
let file_config = SimFileConfig::load(path).expect("failed to load config file");
file_config.into_sim_config()
} else {
let config = SimConfig {
name: "Ring-10 Demo".to_string(),
topology: Topology::Ring,
num_nodes: 10,
initial_data: vec![
("color".into(), b"blue".to_vec()),
("version".into(), b"1".to_vec()),
("status".into(), b"active".to_vec()),
],
num_rounds: 15,
ticks_per_round: 5,
heal_after_round: None,
num_threads: 2,
};
let dash = DashboardConfig { port: 8080 };
(config, dash)
};
eprintln!("Starting gossip dashboard at http://localhost:{}", dash.port);
eprintln!("Open in your browser to see the simulation live.");
let trace = run_with_dashboard(config, dash);
let path = "demo.trace.json";
save_trace(&trace, path).expect("failed to save trace");
eprintln!("Trace saved to {path}");
eprintln!(
"Replay with: cargo run -p gossip-dashboard --example replay -- {path}"
);
}

69
crates/gossip-dashboard/examples/generate_traces.rs Normal file
View file

@ -0,0 +1,69 @@
use std::path::PathBuf;
use gossip_dashboard::config::SimFileConfig;
use gossip_dashboard::save_trace;
use swactor_gossip::sim::run_simulation;
const CONFIGS_DIR: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/examples/configs");
fn main() {
let args: Vec<String> = std::env::args().collect();
let (out_dir, configs) = match args.len() {
// generate_traces → traces/ + all bundled configs
1 => ("traces".to_string(), collect_configs(CONFIGS_DIR)),
// generate_traces <out-dir> → custom dir + all bundled configs
2 if !args[1].ends_with(".toml") => (args[1].clone(), collect_configs(CONFIGS_DIR)),
// generate_traces <out-dir> <config.toml ...>
n if n >= 3 => (args[1].clone(), args[2..].iter().map(PathBuf::from).collect()),
_ => {
eprintln!("Usage:");
eprintln!(" generate_traces # all configs -> traces/");
eprintln!(" generate_traces <out-dir> # all configs -> out-dir/");
eprintln!(" generate_traces <out-dir> <config.toml> [more.toml ...]");
std::process::exit(1);
}
};
if configs.is_empty() {
eprintln!("No .toml configs found in {CONFIGS_DIR}");
std::process::exit(1);
}
std::fs::create_dir_all(&out_dir).expect("failed to create output directory");
for path in &configs {
let path_str = path.to_string_lossy();
let file_config = SimFileConfig::load(&path_str)
.unwrap_or_else(|e| panic!("failed to load {path_str}: {e}"));
let config = file_config.into_sim_config();
eprintln!("Running: {} ...", config.name);
let trace = run_simulation(config);
let filename = format!(
"{}/{}.trace.json",
out_dir,
trace.name.to_lowercase().replace(' ', "_").replace(['(', ')'], "")
);
save_trace(&trace, &filename).expect("failed to save trace");
eprintln!(
" -> {} ({} nodes, {} events)",
filename,
trace.node_names.len(),
trace.events.len()
);
}
eprintln!("Done. View with: cargo run -p gossip-dashboard --example replay -- {out_dir}");
}
fn collect_configs(dir: &str) -> Vec<PathBuf> {
let mut paths: Vec<PathBuf> = std::fs::read_dir(dir)
.unwrap_or_else(|e| panic!("cannot read configs dir {dir}: {e}"))
.flatten()
.map(|e| e.path())
.filter(|p| p.extension().is_some_and(|ext| ext == "toml"))
.collect();
paths.sort();
paths
}

20
crates/gossip-dashboard/examples/replay.rs
View file

@ -1,19 +1,15 @@
use gossip_dashboard::{load_trace, serve_replay};
use gossip_dashboard::serve_dashboard;
fn main() {
let args: Vec<String> = std::env::args().collect();
let path = args
let trace_dir = args
.get(1)
.expect("Usage: replay <trace.json>");
.expect("Usage: replay <trace-dir> [port]");
let trace = load_trace(path).expect("failed to load trace");
let port: u16 = args
.get(2)
.and_then(|s| s.parse().ok())
.unwrap_or(8080);
eprintln!(
"Loaded trace '{}': {} nodes, {} events",
trace.name,
trace.node_names.len(),
trace.events.len()
);
serve_replay(&trace, 8081);
serve_dashboard(trace_dir, port);
}

15
crates/gossip-dashboard/src/config.rs
View file

@ -5,8 +5,6 @@ use std::io;
use serde::Deserialize;
use swactor_gossip::sim::{SimConfig, Topology};
use crate::server::DashboardConfig;
#[derive(Deserialize)]
pub struct SimFileConfig {
pub name: String,
@ -16,7 +14,6 @@ pub struct SimFileConfig {
pub ticks_per_round: usize,
pub num_threads: usize,
pub heal_after_round: Option<usize>,
pub port: Option<u16>,
pub initial_data: Option<BTreeMap<String, String>>,
}
@ -26,7 +23,7 @@ impl SimFileConfig {
toml::from_str(&contents).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
}
pub fn into_sim_config(self) -> (SimConfig, DashboardConfig) {
pub fn into_sim_config(self) -> SimConfig {
let topology = match self.topology.to_lowercase().as_str() {
"ring" => Topology::Ring,
"star" => Topology::Star,
@ -43,7 +40,7 @@ impl SimFileConfig {
.map(|(k, v)| (k, v.into_bytes()))
.collect();
let sim = SimConfig {
SimConfig {
name: self.name,
topology,
num_nodes: self.num_nodes,
@ -52,12 +49,6 @@ impl SimFileConfig {
ticks_per_round: self.ticks_per_round,
heal_after_round: self.heal_after_round,
num_threads: self.num_threads,
};
let dash = DashboardConfig {
port: self.port.unwrap_or(8080),
};
(sim, dash)
}
}
}

1749
crates/gossip-dashboard/src/dashboard_html.rs
View file

File diff suppressed because it is too large Load diff

2
crates/gossip-dashboard/src/lib.rs
View file

@ -2,7 +2,7 @@ pub mod config;
mod dashboard_html;
mod server;
pub use server::{DashboardConfig, run_with_dashboard, serve_replay};
pub use server::serve_dashboard;
use std::fs;
use std::io;

822
crates/gossip-dashboard/src/server.rs
View file

@ -1,720 +1,78 @@
use std::collections::HashMap;
use std::io::{self, Read as IoRead};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
use std::time::Duration;
use std::fs;
use std::path::{Path, PathBuf};
use serde::Serialize;
use swactor::actor::ActorAddress;
use swactor::config::RuntimeConfig;
use swactor::runtime::Runtime;
use swactor_gossip::protocol::{GossipActor, GossipMessage};
use swactor_gossip::sim::{heal_partition_via_handle, wire_topology, SimConfig};
use swactor_gossip::trace::{
EventLog, GossipEvent, GossipEventKind, NameRegistry, NodeSnapshot, SimulationTrace,
TickCounter,
};
use crate::dashboard_html::DASHBOARD_HTML;
// ── Configuration ──────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub struct DashboardConfig {
pub port: u16,
}
impl Default for DashboardConfig {
fn default() -> Self {
Self { port: 8080 }
}
}
// ── Shared dashboard state ─────────────────────────────────────────────
struct DashboardState {
event_log: EventLog,
name_registry: NameRegistry,
init_data: Mutex<Option<InitData>>,
stats: Mutex<StatsSnapshot>,
done: AtomicBool,
}
// ── Trace directory scanning ──────────────────────────────────────────
#[derive(Debug, Clone, Serialize)]
struct InitData {
struct TraceEntry {
file: String,
name: String,
nodes: Vec<NodeInfo>,
edges: Vec<[String; 2]>,
num_threads: usize,
nodes: usize,
events: usize,
}
#[derive(Debug, Clone, Serialize)]
struct NodeInfo {
name: String,
addr: String,
}
#[derive(Debug, Clone, Default, Serialize)]
struct StatsSnapshot {
total_nodes: usize,
total_edges: usize,
total_messages: usize,
current_round: u64,
total_rounds: usize,
}
// ── SSE channel adapter ────────────────────────────────────────────────
/// Adapts an `mpsc::Receiver<Vec<u8>>` to `std::io::Read` for tiny_http streaming.
struct ChannelReader {
rx: mpsc::Receiver<Vec<u8>>,
buf: Vec<u8>,
pos: usize,
}
impl ChannelReader {
fn new(rx: mpsc::Receiver<Vec<u8>>) -> Self {
Self {
rx,
buf: Vec::new(),
pos: 0,
}
}
}
impl IoRead for ChannelReader {
fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
// 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
}
}
}
// ── SSE formatting helpers ─────────────────────────────────────────────
fn format_sse(event: &str, data: &str) -> Vec<u8> {
format!("event: {event}\ndata: {data}\n\n").into_bytes()
}
fn event_kind_name(kind: &GossipEventKind) -> &'static str {
match kind {
GossipEventKind::LocalSet { .. } => "LocalSet",
GossipEventKind::GossipRoundStarted { .. } => "GossipRoundStarted",
GossipEventKind::GossipRoundNoPeers => "GossipRoundNoPeers",
GossipEventKind::PushReceived { .. } => "PushReceived",
GossipEventKind::QueryReceived { .. } => "QueryReceived",
GossipEventKind::PeerAdded { .. } => "PeerAdded",
GossipEventKind::PeerRemoved { .. } => "PeerRemoved",
GossipEventKind::StateSnapshot { .. } => "StateSnapshot",
}
}
#[derive(Serialize)]
struct SseGossipEvent {
seq: usize,
tick: u64,
node: String,
thread: Option<String>,
kind: String,
detail: serde_json::Value,
}
fn gossip_event_to_sse(seq: usize, ev: &GossipEvent) -> SseGossipEvent {
let detail = match &ev.kind {
GossipEventKind::LocalSet { key } => {
serde_json::json!({ "key": key })
}
GossipEventKind::GossipRoundStarted { target_name } => {
serde_json::json!({ "target": target_name })
}
GossipEventKind::GossipRoundNoPeers => serde_json::json!({}),
GossipEventKind::PushReceived {
from_name,
keys_updated,
} => {
serde_json::json!({ "from": from_name, "keys_updated": keys_updated })
}
GossipEventKind::QueryReceived { key } => {
serde_json::json!({ "key": key })
}
GossipEventKind::PeerAdded { peer_name } => {
serde_json::json!({ "peer": peer_name })
}
GossipEventKind::PeerRemoved { peer_name } => {
serde_json::json!({ "peer": peer_name })
}
GossipEventKind::StateSnapshot { snapshot } => {
serde_json::json!({
"entries": snapshot.entries.len(),
"peer_count": snapshot.peer_count,
})
}
fn scan_traces(dir: &Path) -> Vec<TraceEntry> {
let mut entries = Vec::new();
let Ok(read_dir) = fs::read_dir(dir) else {
return entries;
};
SseGossipEvent {
seq,
tick: ev.tick,
node: ev.node_name.clone(),
thread: ev.thread_name.clone(),
kind: event_kind_name(&ev.kind).to_string(),
detail,
}
}
// ── HTTP server ────────────────────────────────────────────────────────
fn spawn_http_server(state: Arc<DashboardState>, port: u16, mode: &str) {
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);
let mode = mode.to_string();
// Spawn a pool of handler threads.
for _ in 0..4 {
let server = Arc::clone(&server);
let state = Arc::clone(&state);
let mode = mode.clone();
thread::spawn(move || {
loop {
let request = match server.recv() {
Ok(r) => r,
Err(_) => break,
};
let url = request.url().to_string();
match url.as_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::<tiny_http::Header>()
.unwrap(),
);
let _ = request.respond(response);
}
"/events" => {
handle_sse(request, Arc::clone(&state));
}
"/trace.json" => {
handle_trace_json(request, Arc::clone(&state));
}
_ => {
let response =
tiny_http::Response::from_string("Not Found").with_status_code(404);
let _ = request.respond(response);
}
}
}
for entry in read_dir.flatten() {
let path = entry.path();
let fname = path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
if !fname.ends_with(".trace.json") {
continue;
}
let Ok(data) = fs::read_to_string(&path) else {
continue;
};
// Parse as generic JSON to extract metadata without full deserialization.
let Ok(val) = serde_json::from_str::<serde_json::Value>(&data) else {
continue;
};
let name = val
.get("name")
.and_then(|v| v.as_str())
.unwrap_or(&fname)
.to_string();
let nodes = val
.get("node_names")
.and_then(|v| v.as_array())
.map(|a| a.len())
.unwrap_or(0);
let events = val
.get("events")
.and_then(|v| v.as_array())
.map(|a| a.len())
.unwrap_or(0);
entries.push(TraceEntry {
file: fname,
name,
nodes,
events,
});
}
entries.sort_by(|a, b| a.file.cmp(&b.file));
entries
}
fn handle_sse(request: tiny_http::Request, state: Arc<DashboardState>) {
let (tx, rx) = mpsc::channel::<Vec<u8>>();
let reader = ChannelReader::new(rx);
// ── HTTP server ───────────────────────────────────────────────────────
// Send SSE headers via a streaming response.
let response = tiny_http::Response::new(
tiny_http::StatusCode(200),
vec![
"Content-Type: text/event-stream"
.parse::<tiny_http::Header>()
.unwrap(),
"Cache-Control: no-cache"
.parse::<tiny_http::Header>()
.unwrap(),
"Connection: keep-alive"
.parse::<tiny_http::Header>()
.unwrap(),
],
Box::new(reader) as Box<dyn IoRead + Send>,
None,
None,
);
// Spawn producer thread that polls for new events.
thread::spawn(move || {
let mut cursor: usize = 0;
// Wait for init data.
loop {
if let Some(init) = state.init_data.lock().unwrap().as_ref() {
let json = serde_json::to_string(init).unwrap();
if tx.send(format_sse("init", &json)).is_err() {
return;
}
break;
}
thread::sleep(Duration::from_millis(50));
}
// Poll for events.
loop {
{
let log = state.event_log.lock().unwrap();
while cursor < log.len() {
let ev = &log[cursor];
// Filter out StateSnapshot events from SSE stream.
if !matches!(ev.kind, GossipEventKind::StateSnapshot { .. }) {
let sse_ev = gossip_event_to_sse(cursor, ev);
let json = serde_json::to_string(&sse_ev).unwrap();
if tx.send(format_sse("gossip", &json)).is_err() {
return;
}
}
cursor += 1;
}
}
// Send stats update.
{
let stats = state.stats.lock().unwrap().clone();
let json = serde_json::to_string(&stats).unwrap();
if tx.send(format_sse("stats", &json)).is_err() {
return;
}
}
if state.done.load(Ordering::Relaxed) {
let _ = tx.send(format_sse("done", "{}"));
let _ = tx.send(Vec::new()); // EOF
return;
}
thread::sleep(Duration::from_millis(50));
}
});
// This blocks until the reader is consumed / connection closes.
let _ = request.respond(response);
}
fn handle_trace_json(request: tiny_http::Request, state: Arc<DashboardState>) {
// Build a partial trace from current state.
let events = state.event_log.lock().unwrap().clone();
let names_map = state.name_registry.lock().unwrap().clone();
let init = state.init_data.lock().unwrap().clone();
let trace = SimulationTrace {
name: init.as_ref().map(|i| i.name.clone()).unwrap_or_default(),
node_names: init
.as_ref()
.map(|i| i.nodes.iter().map(|n| n.name.clone()).collect())
.unwrap_or_default(),
node_addrs: {
let mut addrs: Vec<ActorAddress> = Vec::new();
if let Some(init) = &init {
// Reconstruct addrs from name_registry in node order.
let inv: HashMap<String, ActorAddress> =
names_map.into_iter().map(|(a, n)| (n, a)).collect();
for node in &init.nodes {
if let Some(&addr) = inv.get(&node.name) {
addrs.push(addr);
}
}
}
addrs
},
topology_edges: init
.as_ref()
.map(|i| {
i.edges
.iter()
.map(|e| (e[0].clone(), e[1].clone()))
.collect()
})
.unwrap_or_default(),
events,
snapshots_per_round: Vec::new(),
num_rounds: init
.as_ref()
.map(|_| {
state
.stats
.lock()
.unwrap()
.total_rounds
})
.unwrap_or(0),
total_keys: 0,
};
let json = serde_json::to_string(&trace).unwrap();
let response = tiny_http::Response::from_string(json).with_header(
"Content-Type: application/json"
.parse::<tiny_http::Header>()
.unwrap(),
);
let _ = request.respond(response);
}
// ── Public API: run_with_dashboard ─────────────────────────────────────
pub fn run_with_dashboard(config: SimConfig, dash: DashboardConfig) -> SimulationTrace {
let num_threads = config.num_threads.max(1);
let event_log: EventLog = Arc::new(Mutex::new(Vec::new()));
let tick_counter: TickCounter = Arc::new(AtomicU64::new(0));
let name_registry: NameRegistry = Arc::new(Mutex::new(HashMap::new()));
let state = Arc::new(DashboardState {
event_log: Arc::clone(&event_log),
name_registry: Arc::clone(&name_registry),
init_data: Mutex::new(None),
stats: Mutex::new(StatsSnapshot::default()),
done: AtomicBool::new(false),
});
// Start HTTP server.
spawn_http_server(Arc::clone(&state), dash.port, "live");
if num_threads < 2 {
run_dashboard_single_threaded(config, state, event_log, tick_counter, name_registry)
} else {
run_dashboard_multi_threaded(config, state, event_log, tick_counter, name_registry)
}
}
fn run_dashboard_single_threaded(
config: SimConfig,
state: Arc<DashboardState>,
event_log: EventLog,
tick_counter: TickCounter,
name_registry: NameRegistry,
) -> SimulationTrace {
let rt = Runtime::new(RuntimeConfig {
num_threads: 1,
max_actors: (config.num_nodes + 64).next_power_of_two(),
actor_max_messages: (config.num_nodes * 4).max(1_000),
..Default::default()
});
let mut addrs = Vec::with_capacity(config.num_nodes);
let mut names = Vec::with_capacity(config.num_nodes);
for i in 0..config.num_nodes {
let name = format!("node-{i}");
let actor = GossipActor::traced(
Arc::clone(&event_log),
Arc::clone(&tick_counter),
Arc::clone(&name_registry),
);
let addr = rt.spawn(actor).unwrap();
name_registry.lock().unwrap().insert(addr, name.clone());
addrs.push(addr);
names.push(name);
}
let edges = wire_topology(&rt, &config.topology, &addrs, &names);
for _ in 0..3 {
rt.tick();
}
// Publish init data.
publish_init(&state, &config, &addrs, &names, &edges);
let total_keys = config.initial_data.len();
for (key, value) in &config.initial_data {
rt.send_to(
addrs[0],
GossipMessage::Set {
key: key.clone(),
value: value.clone(),
},
)
.unwrap();
}
rt.tick();
let mut snapshots_per_round: Vec<Vec<(String, NodeSnapshot)>> = Vec::new();
for round in 0..config.num_rounds {
if config.heal_after_round == Some(round) {
swactor_gossip::sim::heal_partition(&rt, &config.topology, &addrs, &names);
for _ in 0..3 {
rt.tick();
}
}
tick_counter.store((round + 1) as u64, Ordering::Relaxed);
update_stats(&state, &config, round, &event_log);
for &addr in &addrs {
rt.send_to(addr, GossipMessage::DoGossipRound).unwrap();
}
for _ in 0..config.ticks_per_round {
rt.tick();
}
for &addr in &addrs {
rt.send_to(addr, GossipMessage::TakeSnapshot).unwrap();
}
for _ in 0..3 {
rt.tick();
}
let current_round_tick = (round + 1) as u64;
let log = event_log.lock().unwrap();
let mut round_snapshots: Vec<(String, NodeSnapshot)> = Vec::new();
for event in log.iter().rev() {
if event.tick != current_round_tick {
break;
}
if let GossipEventKind::StateSnapshot { ref snapshot } = event.kind {
round_snapshots.push((event.node_name.clone(), snapshot.clone()));
}
}
round_snapshots.reverse();
snapshots_per_round.push(round_snapshots);
}
state.done.store(true, Ordering::Relaxed);
let events = event_log.lock().unwrap().clone();
SimulationTrace {
name: config.name,
node_names: names,
node_addrs: addrs,
topology_edges: edges,
events,
snapshots_per_round,
num_rounds: config.num_rounds,
total_keys,
}
}
fn run_dashboard_multi_threaded(
config: SimConfig,
state: Arc<DashboardState>,
event_log: EventLog,
tick_counter: TickCounter,
name_registry: NameRegistry,
) -> SimulationTrace {
let ticks_per_round = config.ticks_per_round;
let settle_ms = (ticks_per_round as u64 * 2).max(10);
let rt = Runtime::new(RuntimeConfig {
num_threads: config.num_threads,
max_actors: (config.num_nodes + 64).next_power_of_two(),
actor_max_messages: (config.num_nodes * 4).max(1_000),
..Default::default()
});
let mut addrs = Vec::with_capacity(config.num_nodes);
let mut names = Vec::with_capacity(config.num_nodes);
for i in 0..config.num_nodes {
let name = format!("node-{i}");
let actor = GossipActor::traced(
Arc::clone(&event_log),
Arc::clone(&tick_counter),
Arc::clone(&name_registry),
);
let addr = rt.spawn(actor).unwrap();
name_registry.lock().unwrap().insert(addr, name.clone());
addrs.push(addr);
names.push(name);
}
let edges = wire_topology(&rt, &config.topology, &addrs, &names);
// Publish init data.
publish_init(&state, &config, &addrs, &names, &edges);
let total_keys = config.initial_data.len();
for (key, value) in &config.initial_data {
rt.send_to(
addrs[0],
GossipMessage::Set {
key: key.clone(),
value: value.clone(),
},
)
.unwrap();
}
let handle = rt.run().expect("failed to start multi-threaded runtime");
thread::sleep(Duration::from_millis(settle_ms * 2));
let mut snapshots_per_round: Vec<Vec<(String, NodeSnapshot)>> = Vec::new();
for round in 0..config.num_rounds {
if config.heal_after_round == Some(round) {
heal_partition_via_handle(&handle, &config.topology, &addrs, &names);
thread::sleep(Duration::from_millis(settle_ms));
}
tick_counter.store((round + 1) as u64, Ordering::Relaxed);
update_stats(&state, &config, round, &event_log);
for &addr in &addrs {
handle
.runtime
.send_to(addr, GossipMessage::DoGossipRound)
.unwrap();
}
thread::sleep(Duration::from_millis(settle_ms));
for &addr in &addrs {
handle
.runtime
.send_to(addr, GossipMessage::TakeSnapshot)
.unwrap();
}
thread::sleep(Duration::from_millis(settle_ms / 2));
let current_round_tick = (round + 1) as u64;
let log = event_log.lock().unwrap();
let mut round_snapshots: Vec<(String, NodeSnapshot)> = Vec::new();
for event in log.iter().rev() {
if event.tick != current_round_tick {
break;
}
if let GossipEventKind::StateSnapshot { ref snapshot } = event.kind {
round_snapshots.push((event.node_name.clone(), snapshot.clone()));
}
}
round_snapshots.reverse();
snapshots_per_round.push(round_snapshots);
}
handle.shutdown();
handle.join();
state.done.store(true, Ordering::Relaxed);
let events = event_log.lock().unwrap().clone();
SimulationTrace {
name: config.name,
node_names: names,
node_addrs: addrs,
topology_edges: edges,
events,
snapshots_per_round,
num_rounds: config.num_rounds,
total_keys,
}
}
// ── Helpers ────────────────────────────────────────────────────────────
fn publish_init(
state: &DashboardState,
config: &SimConfig,
addrs: &[ActorAddress],
names: &[String],
edges: &[(String, String)],
) {
let nodes: Vec<NodeInfo> = names
.iter()
.zip(addrs.iter())
.map(|(name, addr)| NodeInfo {
name: name.clone(),
addr: format!("{addr}"),
})
.collect();
let edge_pairs: Vec<[String; 2]> = edges
.iter()
.map(|(a, b)| [a.clone(), b.clone()])
.collect();
*state.init_data.lock().unwrap() = Some(InitData {
name: config.name.clone(),
nodes,
edges: edge_pairs,
num_threads: config.num_threads,
});
*state.stats.lock().unwrap() = StatsSnapshot {
total_nodes: config.num_nodes,
total_edges: edges.len(),
total_messages: 0,
current_round: 0,
total_rounds: config.num_rounds,
};
}
fn update_stats(state: &DashboardState, config: &SimConfig, round: usize, event_log: &EventLog) {
let msg_count = event_log.lock().unwrap().len();
let mut stats = state.stats.lock().unwrap();
stats.current_round = (round + 1) as u64;
stats.total_messages = msg_count;
stats.total_rounds = config.num_rounds;
}
// ── Replay mode ────────────────────────────────────────────────────────
pub fn serve_replay(trace: &SimulationTrace, port: u16) {
let event_log: EventLog = Arc::new(Mutex::new(trace.events.clone()));
let name_registry: NameRegistry = Arc::new(Mutex::new(
trace
.node_names
.iter()
.zip(trace.node_addrs.iter())
.map(|(n, a)| (*a, n.clone()))
.collect(),
));
let edges: Vec<[String; 2]> = trace
.topology_edges
.iter()
.map(|(a, b)| [a.clone(), b.clone()])
.collect();
let nodes: Vec<NodeInfo> = trace
.node_names
.iter()
.zip(trace.node_addrs.iter())
.map(|(name, addr)| NodeInfo {
name: name.clone(),
addr: format!("{addr}"),
})
.collect();
// Keep state alive for potential future SSE support in replay mode.
let _state = Arc::new(DashboardState {
event_log,
name_registry,
init_data: Mutex::new(Some(InitData {
name: trace.name.clone(),
nodes,
edges,
num_threads: 1,
})),
stats: Mutex::new(StatsSnapshot {
total_nodes: trace.node_names.len(),
total_edges: trace.topology_edges.len(),
total_messages: trace.events.len(),
current_round: trace.num_rounds as u64,
total_rounds: trace.num_rounds,
}),
done: AtomicBool::new(true),
});
pub fn serve_dashboard(trace_dir: &str, port: u16) {
let dir = PathBuf::from(trace_dir);
assert!(dir.is_dir(), "trace directory does not exist: {trace_dir}");
let addr = format!("0.0.0.0:{port}");
let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server");
eprintln!("Replay dashboard at http://localhost:{port}");
eprintln!("Dashboard at http://localhost:{port}");
eprintln!("Serving traces from: {trace_dir}");
loop {
let request = match server.recv() {
@ -725,16 +83,16 @@ pub fn serve_replay(trace: &SimulationTrace, port: u16) {
let url = request.url().to_string();
match url.as_str() {
"/" => {
let html = DASHBOARD_HTML.replace("__DASHBOARD_MODE__", "replay");
let response = tiny_http::Response::from_string(html).with_header(
let response = tiny_http::Response::from_string(DASHBOARD_HTML).with_header(
"Content-Type: text/html; charset=utf-8"
.parse::<tiny_http::Header>()
.unwrap(),
);
let _ = request.respond(response);
}
"/trace.json" => {
let json = serde_json::to_string(trace).unwrap();
"/traces" => {
let entries = scan_traces(&dir);
let json = serde_json::to_string(&entries).unwrap();
let response = tiny_http::Response::from_string(json).with_header(
"Content-Type: application/json"
.parse::<tiny_http::Header>()
@ -742,6 +100,39 @@ pub fn serve_replay(trace: &SimulationTrace, port: u16) {
);
let _ = request.respond(response);
}
_ if url.starts_with("/trace.json?file=") => {
let raw = url.strip_prefix("/trace.json?file=").unwrap();
let file = percent_decode(raw);
// Reject path traversal attempts.
if file.contains('/')
|| file.contains('\\')
|| file.contains("..")
|| file.is_empty()
{
let response =
tiny_http::Response::from_string("Bad Request").with_status_code(400);
let _ = request.respond(response);
continue;
}
let path = dir.join(&file);
match fs::read_to_string(&path) {
Ok(data) => {
let response = tiny_http::Response::from_string(data).with_header(
"Content-Type: application/json"
.parse::<tiny_http::Header>()
.unwrap(),
);
let _ = request.respond(response);
}
Err(_) => {
let response =
tiny_http::Response::from_string("Not Found").with_status_code(404);
let _ = request.respond(response);
}
}
}
_ => {
let response =
tiny_http::Response::from_string("Not Found").with_status_code(404);
@ -750,3 +141,32 @@ pub fn serve_replay(trace: &SimulationTrace, port: u16) {
}
}
}
// ── Helpers ───────────────────────────────────────────────────────────
fn percent_decode(s: &str) -> String {
let mut result = Vec::new();
let bytes = s.as_bytes();
let mut i = 0;
while i < bytes.len() {
if bytes[i] == b'%' && i + 2 < bytes.len() {
if let (Some(h), Some(l)) = (hex_val(bytes[i + 1]), hex_val(bytes[i + 2])) {
result.push(h << 4 | l);
i += 3;
continue;
}
}
result.push(bytes[i]);
i += 1;
}
String::from_utf8_lossy(&result).to_string()
}
fn hex_val(b: u8) -> Option<u8> {
match b {
b'0'..=b'9' => Some(b - b'0'),
b'a'..=b'f' => Some(b - b'a' + 10),
b'A'..=b'F' => Some(b - b'A' + 10),
_ => None,
}
}

20
crates/swactor-gossip/src/sim.rs
View file

@ -289,13 +289,14 @@ pub fn heal_partition_via_handle(
handle: &swactor::runtime::RuntimeHandle,
topology: &Topology,
addrs: &[ActorAddress],
_names: &[String],
) {
names: &[String],
) -> Vec<(String, String)> {
if !matches!(topology, Topology::Partitioned) {
return;
return Vec::new();
}
let n = addrs.len();
let half = n / 2;
let mut new_edges = Vec::new();
if half > 0 && half < n {
handle
.runtime
@ -305,7 +306,10 @@ pub fn heal_partition_via_handle(
.runtime
.send_to(addrs[half], GossipMessage::AddPeer(addrs[half - 1]))
.unwrap();
new_edges.push((names[half - 1].clone(), names[half].clone()));
new_edges.push((names[half].clone(), names[half - 1].clone()));
}
new_edges
}
// ── Topology wiring ──────────────────────────────────────────────────────
@ -377,18 +381,22 @@ pub fn heal_partition(
rt: &Runtime,
topology: &Topology,
addrs: &[ActorAddress],
_names: &[String],
) {
names: &[String],
) -> Vec<(String, String)> {
if !matches!(topology, Topology::Partitioned) {
return;
return Vec::new();
}
let n = addrs.len();
let half = n / 2;
let mut new_edges = Vec::new();
// Add bidirectional links between the two halves (bridge nodes).
if half > 0 && half < n {
rt.send_to(addrs[half - 1], GossipMessage::AddPeer(addrs[half]))
.unwrap();
rt.send_to(addrs[half], GossipMessage::AddPeer(addrs[half - 1]))
.unwrap();
new_edges.push((names[half - 1].clone(), names[half].clone()));
new_edges.push((names[half].clone(), names[half - 1].clone()));
}
new_edges
}