fix: gossip simulation dashboard
This commit is contained in:
Zachery Aaron Shores-Chmielewski
parent
9beca6c5dc
commit
72df65a155
17 changed files with 615 additions and 991 deletions
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -216,7 +216,6 @@ version = "0.1.0"
|
|||
dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"swactor",
|
||||
"swactor-gossip",
|
||||
"tiny_http",
|
||||
"toml",
|
||||
|
|
|
|||
|
|
@ -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
126
crates/gossip-dashboard/README.md
Normal 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
11
crates/gossip-dashboard/examples/configs/chain_8.toml
Normal 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
12
crates/gossip-dashboard/examples/configs/full_mesh_6.toml
Normal 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"
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
name = "Partition & Heal"
|
||||
topology = "partitioned"
|
||||
num_nodes = 800
|
||||
num_rounds = 10
|
||||
ticks_per_round = 5
|
||||
num_threads = 4
|
||||
heal_after_round = 5
|
||||
|
||||
[initial_data]
|
||||
color = "blue"
|
||||
version = "1"
|
||||
|
|
@ -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
11
crates/gossip-dashboard/examples/configs/ring_10.toml
Normal 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
11
crates/gossip-dashboard/examples/configs/star_7.toml
Normal 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"
|
||||
|
|
@ -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
69
crates/gossip-dashboard/examples/generate_traces.rs
Normal 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
|
||||
}
|
||||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,12 +13,24 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
padding: 12px 20px; background: #161822; border-bottom: 1px solid #2a2d3a;
|
||||
}
|
||||
.header h1 { font-size: 18px; font-weight: 600; color: #c0c6d4; }
|
||||
|
||||
.trace-select {
|
||||
background: #1a1d2e; color: #e0e0e0; border: 1px solid #2a2d3a;
|
||||
border-radius: 4px; padding: 6px 12px; font-size: 13px;
|
||||
cursor: pointer; min-width: 240px; max-width: 420px;
|
||||
}
|
||||
.trace-select:hover { border-color: #4f46e5; }
|
||||
.trace-select:focus { outline: none; border-color: #6366f1; }
|
||||
.trace-select:disabled { cursor: default; opacity: 0.5; }
|
||||
|
||||
.status-badge {
|
||||
display: flex; align-items: center; gap: 6px; font-size: 13px; color: #9ca3af;
|
||||
}
|
||||
.status-dot { width: 8px; height: 8px; border-radius: 50%; background: #3b82f6; }
|
||||
.status-dot.done { background: #22c55e; }
|
||||
.status-dot.replay { background: #f59e0b; }
|
||||
.status-dot { width: 8px; height: 8px; border-radius: 50%; background: #f59e0b; }
|
||||
.status-dot.ready { background: #22c55e; }
|
||||
.status-dot.loading { background: #3b82f6; animation: pulse 1s infinite; }
|
||||
.status-dot.error { background: #ef4444; }
|
||||
@keyframes pulse { 0%,100% { opacity: 1; } 50% { opacity: 0.4; } }
|
||||
|
||||
.main { display: grid; grid-template-columns: 1fr 1fr; grid-template-rows: auto 1fr; height: calc(100vh - 48px); }
|
||||
|
||||
|
|
@ -82,18 +94,18 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
border-radius: 4px; padding: 2px 6px; font-size: 12px; text-align: right;
|
||||
}
|
||||
.replay-controls .speed-group label { font-size: 11px; color: #6b7280; white-space: nowrap; }
|
||||
|
||||
.node-flash { animation: flash 0.4s ease-out; }
|
||||
@keyframes flash { 0% { filter: brightness(2); } 100% { filter: brightness(1); } }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
|
||||
<div class="header">
|
||||
<h1>Gossip Simulation Dashboard</h1>
|
||||
<h1>Gossip Dashboard</h1>
|
||||
<select id="traceSelect" class="trace-select" disabled>
|
||||
<option value="">Loading traces...</option>
|
||||
</select>
|
||||
<div class="status-badge">
|
||||
<div class="status-dot" id="statusDot"></div>
|
||||
<span id="statusText">Connecting...</span>
|
||||
<div class="status-dot loading" id="statusDot"></div>
|
||||
<span id="statusText">Loading traces...</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
|
@ -141,23 +153,23 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
</div>
|
||||
|
||||
<script>
|
||||
const MODE = "__DASHBOARD_MODE__"; // replaced by server: "live" or "replay"
|
||||
const MAX_TABLE_ROWS = 2000;
|
||||
|
||||
// ── State ──────────────────────────────────────────────────────────
|
||||
let nodes = [];
|
||||
let edges = [];
|
||||
let initialEdgeSet = new Set(); // "a->b" keys from topology_edges
|
||||
let edgeAppearance = {}; // "a->b" -> event index (-1 = initial)
|
||||
let nodePositions = {};
|
||||
let allEvents = [];
|
||||
let replayCursor = 0;
|
||||
let workerLogs = {};
|
||||
let isDone = false;
|
||||
|
||||
// ── Canvas / Graph ─────────────────────────────────────────────────
|
||||
const canvas = document.getElementById('graphCanvas');
|
||||
const ctx2d = canvas.getContext('2d');
|
||||
let nodeFlash = {}; // nodeName -> timestamp
|
||||
let edgeFlash = {}; // "a->b" -> timestamp
|
||||
let nodeFlash = {};
|
||||
let edgeFlash = {};
|
||||
|
||||
function resizeCanvas() {
|
||||
const rect = canvas.parentElement.getBoundingClientRect();
|
||||
|
|
@ -178,7 +190,6 @@ function initPositions() {
|
|||
const angle = (2 * Math.PI * i) / n - Math.PI / 2;
|
||||
nodePositions[node.name] = { x: cx + r * Math.cos(angle), y: cy + r * Math.sin(angle) };
|
||||
});
|
||||
// Run force simulation
|
||||
runForceLayout(rect.width, rect.height);
|
||||
}
|
||||
|
||||
|
|
@ -217,10 +228,8 @@ function runForceLayout(w, h) {
|
|||
|
||||
// Gravity toward center
|
||||
nodes.forEach(n => {
|
||||
let dx = pos[n.name].x - cx, dy = pos[n.name].y - cy;
|
||||
let dist = Math.sqrt(dx * dx + dy * dy) || 0.01;
|
||||
disp[n.name].x -= dx * 0.01;
|
||||
disp[n.name].y -= dy * 0.01;
|
||||
disp[n.name].x -= (pos[n.name].x - cx) * 0.01;
|
||||
disp[n.name].y -= (pos[n.name].y - cy) * 0.01;
|
||||
});
|
||||
|
||||
// Apply with damping
|
||||
|
|
@ -231,7 +240,6 @@ function runForceLayout(w, h) {
|
|||
let cap = Math.min(dist, 10 * temp);
|
||||
pos[n.name].x += (d.x / dist) * cap;
|
||||
pos[n.name].y += (d.y / dist) * cap;
|
||||
// Keep within bounds
|
||||
pos[n.name].x = Math.max(40, Math.min(w - 40, pos[n.name].x));
|
||||
pos[n.name].y = Math.max(40, Math.min(h - 40, pos[n.name].y));
|
||||
});
|
||||
|
|
@ -312,42 +320,20 @@ function makeRow(ev) {
|
|||
return tr;
|
||||
}
|
||||
|
||||
function addEventToTable(ev) {
|
||||
const tbody = document.getElementById('eventTableBody');
|
||||
if (tbody.children.length >= MAX_TABLE_ROWS) {
|
||||
tbody.removeChild(tbody.firstChild);
|
||||
}
|
||||
tbody.appendChild(makeRow(ev));
|
||||
|
||||
const wrap = document.getElementById('eventTableWrap');
|
||||
wrap.scrollTop = wrap.scrollHeight;
|
||||
}
|
||||
|
||||
function formatDetail(kind, detail) {
|
||||
if (!detail) return '';
|
||||
switch (kind) {
|
||||
case 'GossipRoundStarted': return '→ ' + detail.target;
|
||||
case 'PushReceived': return '← ' + detail.from + ' (' + detail.keys_updated + ' keys)';
|
||||
case 'GossipRoundStarted': return '→ ' + (detail.target_name || detail.target);
|
||||
case 'PushReceived': return '← ' + (detail.from_name || detail.from) + ' (' + detail.keys_updated + ' keys)';
|
||||
case 'LocalSet': return 'key=' + detail.key;
|
||||
case 'PeerAdded': return '+ ' + detail.peer;
|
||||
case 'PeerRemoved': return '- ' + detail.peer;
|
||||
case 'PeerAdded': return '+ ' + (detail.peer_name || detail.peer);
|
||||
case 'PeerRemoved': return '- ' + (detail.peer_name || detail.peer);
|
||||
case 'QueryReceived': return 'key=' + detail.key;
|
||||
case 'StateSnapshot': return detail.entries + ' entries, ' + detail.peer_count + ' peers';
|
||||
default: return JSON.stringify(detail);
|
||||
}
|
||||
}
|
||||
|
||||
function addWorkerEntry(thread, text) {
|
||||
if (!thread) return;
|
||||
if (!workerLogs[thread]) {
|
||||
workerLogs[thread] = [];
|
||||
rebuildWorkerColumns();
|
||||
}
|
||||
workerLogs[thread].push(text);
|
||||
if (workerLogs[thread].length > 50) workerLogs[thread].shift();
|
||||
updateWorkerColumn(thread);
|
||||
}
|
||||
|
||||
function rebuildWorkerColumns() {
|
||||
const container = document.getElementById('workerColumns');
|
||||
container.innerHTML = '';
|
||||
|
|
@ -364,7 +350,6 @@ function updateWorkerColumn(thread) {
|
|||
const col = document.getElementById('worker-' + thread);
|
||||
if (!col) return;
|
||||
const entries = workerLogs[thread];
|
||||
// Keep header + entries
|
||||
let html = '<div class="worker-col-header">' + thread + '</div>';
|
||||
for (const e of entries) {
|
||||
html += '<div class="worker-entry">' + e + '</div>';
|
||||
|
|
@ -373,82 +358,136 @@ function updateWorkerColumn(thread) {
|
|||
col.scrollTop = col.scrollHeight;
|
||||
}
|
||||
|
||||
function processEvent(ev) {
|
||||
addEventToTable(ev);
|
||||
// ── Replay engine ──────────────────────────────────────────────────
|
||||
let replayRafId = 0;
|
||||
let playTimer = null;
|
||||
|
||||
// Flash node
|
||||
nodeFlash[ev.node] = performance.now();
|
||||
|
||||
// Flash edges on Push
|
||||
if (ev.kind === 'GossipRoundStarted' && ev.detail && ev.detail.target) {
|
||||
edgeFlash[ev.node + '->' + ev.detail.target] = performance.now();
|
||||
edgeFlash[ev.detail.target + '->' + ev.node] = performance.now();
|
||||
}
|
||||
if (ev.kind === 'PushReceived' && ev.detail && ev.detail.from) {
|
||||
edgeFlash[ev.detail.from + '->' + ev.node] = performance.now();
|
||||
edgeFlash[ev.node + '->' + ev.detail.from] = performance.now();
|
||||
function stopPlayback() {
|
||||
if (playTimer !== null) {
|
||||
clearInterval(playTimer);
|
||||
playTimer = null;
|
||||
document.getElementById('btnPlay').innerHTML = '▶';
|
||||
document.getElementById('btnPlay').title = 'Play';
|
||||
}
|
||||
}
|
||||
|
||||
// Worker log
|
||||
function startPlayback() {
|
||||
stopPlayback();
|
||||
if (replayCursor >= allEvents.length) return;
|
||||
const speed = parseInt(document.getElementById('speedInput').value) || 100;
|
||||
document.getElementById('btnPlay').innerHTML = '⏸';
|
||||
document.getElementById('btnPlay').title = 'Pause';
|
||||
playTimer = setInterval(() => {
|
||||
if (replayCursor >= allEvents.length) {
|
||||
stopPlayback();
|
||||
return;
|
||||
}
|
||||
replayTo(replayCursor + 1);
|
||||
}, speed);
|
||||
}
|
||||
|
||||
function replayTo(pos) {
|
||||
replayCursor = pos;
|
||||
if (replayRafId) return;
|
||||
replayRafId = requestAnimationFrame(() => {
|
||||
replayRafId = 0;
|
||||
replayToImpl(replayCursor);
|
||||
});
|
||||
}
|
||||
|
||||
function replayToImpl(pos) {
|
||||
nodeFlash = {};
|
||||
edgeFlash = {};
|
||||
|
||||
// Recompute visible edges based on replay position
|
||||
edges = Object.entries(edgeAppearance)
|
||||
.filter(([_, idx]) => idx < pos)
|
||||
.map(([key]) => key.split('->'));
|
||||
|
||||
const frag = document.createDocumentFragment();
|
||||
const start = Math.max(0, pos - MAX_TABLE_ROWS);
|
||||
for (let i = start; i < pos && i < allEvents.length; i++) {
|
||||
frag.appendChild(makeRow(allEvents[i]));
|
||||
}
|
||||
const tbody = document.getElementById('eventTableBody');
|
||||
tbody.textContent = '';
|
||||
tbody.appendChild(frag);
|
||||
|
||||
const wrap = document.getElementById('eventTableWrap');
|
||||
wrap.scrollTop = wrap.scrollHeight;
|
||||
|
||||
// Rebuild worker logs
|
||||
workerLogs = {};
|
||||
const workerStart = Math.max(0, pos - 200);
|
||||
for (let i = workerStart; i < pos && i < allEvents.length; i++) {
|
||||
const ev = allEvents[i];
|
||||
if (!ev.thread) continue;
|
||||
if (!workerLogs[ev.thread]) workerLogs[ev.thread] = [];
|
||||
let text = ev.node + ': ' + ev.kind;
|
||||
if (ev.kind === 'GossipRoundStarted') text += ' -> ' + ev.detail.target;
|
||||
if (ev.kind === 'PushReceived') text += ' <- ' + ev.detail.from;
|
||||
addWorkerEntry(ev.thread, text);
|
||||
|
||||
requestAnimationFrame(drawGraph);
|
||||
}
|
||||
|
||||
// ── Live mode (SSE) ────────────────────────────────────────────────
|
||||
function startLive() {
|
||||
const dot = document.getElementById('statusDot');
|
||||
const statusText = document.getElementById('statusText');
|
||||
|
||||
const es = new EventSource('/events');
|
||||
|
||||
es.addEventListener('init', (e) => {
|
||||
const data = JSON.parse(e.data);
|
||||
nodes = data.nodes;
|
||||
edges = data.edges;
|
||||
statusText.textContent = 'Live: ' + data.name;
|
||||
dot.className = 'status-dot';
|
||||
resizeCanvas();
|
||||
initPositions();
|
||||
drawGraph();
|
||||
});
|
||||
|
||||
es.addEventListener('gossip', (e) => {
|
||||
const ev = JSON.parse(e.data);
|
||||
allEvents.push(ev);
|
||||
processEvent(ev);
|
||||
});
|
||||
|
||||
es.addEventListener('stats', (e) => {
|
||||
const data = JSON.parse(e.data);
|
||||
updateStats(data);
|
||||
});
|
||||
|
||||
es.addEventListener('done', (e) => {
|
||||
isDone = true;
|
||||
dot.className = 'status-dot done';
|
||||
statusText.textContent += ' (done)';
|
||||
es.close();
|
||||
});
|
||||
|
||||
es.onerror = () => {
|
||||
if (!isDone) {
|
||||
statusText.textContent = 'Disconnected';
|
||||
dot.style.background = '#ef4444';
|
||||
if (ev.kind === 'GossipRoundStarted' && ev.detail) text += ' -> ' + (ev.detail.target_name || ev.detail.target);
|
||||
if (ev.kind === 'PushReceived' && ev.detail) text += ' <- ' + (ev.detail.from_name || ev.detail.from);
|
||||
workerLogs[ev.thread].push(text);
|
||||
if (workerLogs[ev.thread].length > 50) workerLogs[ev.thread].shift();
|
||||
}
|
||||
};
|
||||
rebuildWorkerColumns();
|
||||
for (const thread of Object.keys(workerLogs)) {
|
||||
updateWorkerColumn(thread);
|
||||
}
|
||||
|
||||
// Flash the last event
|
||||
if (pos > 0 && pos <= allEvents.length) {
|
||||
const ev = allEvents[pos - 1];
|
||||
nodeFlash[ev.node] = performance.now();
|
||||
if (ev.kind === 'GossipRoundStarted' && ev.detail && (ev.detail.target_name || ev.detail.target)) {
|
||||
const t = ev.detail.target_name || ev.detail.target;
|
||||
edgeFlash[ev.node + '->' + t] = performance.now();
|
||||
}
|
||||
if (ev.kind === 'PushReceived' && ev.detail && (ev.detail.from_name || ev.detail.from)) {
|
||||
const f = ev.detail.from_name || ev.detail.from;
|
||||
edgeFlash[f + '->' + ev.node] = performance.now();
|
||||
}
|
||||
}
|
||||
|
||||
document.getElementById('replaySlider').value = pos;
|
||||
updateReplayPos();
|
||||
drawGraph();
|
||||
}
|
||||
|
||||
// ── Replay mode ────────────────────────────────────────────────────
|
||||
async function startReplay() {
|
||||
function updateReplayPos() {
|
||||
document.getElementById('replayPos').textContent = replayCursor + '/' + allEvents.length;
|
||||
}
|
||||
|
||||
// ── Trace loading ──────────────────────────────────────────────────
|
||||
function resetState() {
|
||||
stopPlayback();
|
||||
nodes = [];
|
||||
edges = [];
|
||||
initialEdgeSet = new Set();
|
||||
edgeAppearance = {};
|
||||
nodePositions = {};
|
||||
allEvents = [];
|
||||
replayCursor = 0;
|
||||
workerLogs = {};
|
||||
nodeFlash = {};
|
||||
edgeFlash = {};
|
||||
document.getElementById('eventTableBody').textContent = '';
|
||||
document.getElementById('workerColumns').innerHTML = '';
|
||||
}
|
||||
|
||||
async function loadTrace(file) {
|
||||
const dot = document.getElementById('statusDot');
|
||||
const statusText = document.getElementById('statusText');
|
||||
dot.className = 'status-dot replay';
|
||||
dot.className = 'status-dot loading';
|
||||
statusText.textContent = 'Loading...';
|
||||
|
||||
const resp = await fetch('/trace.json');
|
||||
resetState();
|
||||
|
||||
const resp = await fetch('/trace.json?file=' + encodeURIComponent(file));
|
||||
if (!resp.ok) {
|
||||
dot.className = 'status-dot error';
|
||||
statusText.textContent = 'Failed to load trace';
|
||||
return;
|
||||
}
|
||||
const trace = await resp.json();
|
||||
|
||||
nodes = trace.node_names.map((name, i) => ({
|
||||
|
|
@ -457,10 +496,16 @@ async function startReplay() {
|
|||
}));
|
||||
edges = trace.topology_edges.map(e => [e[0], e[1]]);
|
||||
|
||||
// Precompute edge timeline: initial edges are always visible (-1),
|
||||
// dynamically added edges (from PeerAdded) appear at their event index.
|
||||
initialEdgeSet = new Set(edges.map(([a, b]) => a + '->' + b));
|
||||
edgeAppearance = {};
|
||||
edges.forEach(([a, b]) => { edgeAppearance[a + '->' + b] = -1; });
|
||||
|
||||
// Build events list (filter out StateSnapshot for display)
|
||||
let seq = 0;
|
||||
allEvents = trace.events
|
||||
.filter(ev => ev.kind !== 'StateSnapshot')
|
||||
.filter(ev => typeof ev.kind === 'object' ? !('StateSnapshot' in ev.kind) : ev.kind !== 'StateSnapshot')
|
||||
.map(ev => {
|
||||
const kind = typeof ev.kind === 'string' ? ev.kind : Object.keys(ev.kind)[0];
|
||||
const detail = typeof ev.kind === 'string' ? {} : ev.kind[kind] || {};
|
||||
|
|
@ -474,10 +519,24 @@ async function startReplay() {
|
|||
};
|
||||
});
|
||||
|
||||
statusText.textContent = 'Replay: ' + trace.name + ' (' + allEvents.length + ' events)';
|
||||
// Index PeerAdded events that introduce new edges (e.g. partition heal)
|
||||
allEvents.forEach((ev, i) => {
|
||||
if (ev.kind === 'PeerAdded' && ev.detail) {
|
||||
const peer = ev.detail.peer_name || ev.detail.peer;
|
||||
if (peer) {
|
||||
const key = ev.node + '->' + peer;
|
||||
if (!(key in edgeAppearance)) {
|
||||
edgeAppearance[key] = i;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
dot.className = 'status-dot ready';
|
||||
statusText.textContent = trace.name + ' (' + allEvents.length + ' events)';
|
||||
updateStats({
|
||||
total_nodes: trace.node_names.length,
|
||||
total_edges: trace.topology_edges.length,
|
||||
total_edges: Object.keys(edgeAppearance).length,
|
||||
total_messages: allEvents.length,
|
||||
current_round: trace.num_rounds,
|
||||
total_rounds: trace.num_rounds
|
||||
|
|
@ -487,7 +546,7 @@ async function startReplay() {
|
|||
initPositions();
|
||||
drawGraph();
|
||||
|
||||
// Show replay controls
|
||||
// Set up replay controls
|
||||
const controls = document.getElementById('replayControls');
|
||||
controls.style.display = 'flex';
|
||||
const slider = document.getElementById('replaySlider');
|
||||
|
|
@ -510,106 +569,48 @@ async function startReplay() {
|
|||
};
|
||||
}
|
||||
|
||||
let replayRafId = 0;
|
||||
let playTimer = null;
|
||||
|
||||
function stopPlayback() {
|
||||
if (playTimer !== null) {
|
||||
clearInterval(playTimer);
|
||||
playTimer = null;
|
||||
document.getElementById('btnPlay').innerHTML = '▶';
|
||||
document.getElementById('btnPlay').title = 'Play';
|
||||
}
|
||||
}
|
||||
|
||||
function startPlayback() {
|
||||
stopPlayback();
|
||||
if (replayCursor >= allEvents.length) return; // already at end
|
||||
const speed = parseInt(document.getElementById('speedInput').value) || 100;
|
||||
document.getElementById('btnPlay').innerHTML = '⏸';
|
||||
document.getElementById('btnPlay').title = 'Pause';
|
||||
playTimer = setInterval(() => {
|
||||
if (replayCursor >= allEvents.length) {
|
||||
stopPlayback();
|
||||
return;
|
||||
}
|
||||
replayTo(replayCursor + 1);
|
||||
}, speed);
|
||||
}
|
||||
|
||||
function replayTo(pos) {
|
||||
// Debounce: only run once per animation frame
|
||||
replayCursor = pos;
|
||||
if (replayRafId) return;
|
||||
replayRafId = requestAnimationFrame(() => {
|
||||
replayRafId = 0;
|
||||
replayToImpl(replayCursor);
|
||||
});
|
||||
}
|
||||
|
||||
function replayToImpl(pos) {
|
||||
nodeFlash = {};
|
||||
edgeFlash = {};
|
||||
|
||||
// ── Build table rows into a fragment (no reflows) ──
|
||||
const frag = document.createDocumentFragment();
|
||||
const start = Math.max(0, pos - MAX_TABLE_ROWS);
|
||||
for (let i = start; i < pos && i < allEvents.length; i++) {
|
||||
frag.appendChild(makeRow(allEvents[i]));
|
||||
}
|
||||
const tbody = document.getElementById('eventTableBody');
|
||||
tbody.textContent = ''; // fast clear
|
||||
tbody.appendChild(frag);
|
||||
|
||||
const wrap = document.getElementById('eventTableWrap');
|
||||
wrap.scrollTop = wrap.scrollHeight;
|
||||
|
||||
// ── Rebuild worker logs in one pass ──
|
||||
workerLogs = {};
|
||||
const workerStart = Math.max(0, pos - 200); // only last ~200 events for worker logs
|
||||
for (let i = workerStart; i < pos && i < allEvents.length; i++) {
|
||||
const ev = allEvents[i];
|
||||
if (!ev.thread) continue;
|
||||
if (!workerLogs[ev.thread]) workerLogs[ev.thread] = [];
|
||||
let text = ev.node + ': ' + ev.kind;
|
||||
if (ev.kind === 'GossipRoundStarted' && ev.detail) text += ' -> ' + ev.detail.target;
|
||||
if (ev.kind === 'PushReceived' && ev.detail) text += ' <- ' + ev.detail.from;
|
||||
workerLogs[ev.thread].push(text);
|
||||
if (workerLogs[ev.thread].length > 50) workerLogs[ev.thread].shift();
|
||||
}
|
||||
rebuildWorkerColumns();
|
||||
for (const thread of Object.keys(workerLogs)) {
|
||||
updateWorkerColumn(thread);
|
||||
}
|
||||
|
||||
// Flash the last event
|
||||
if (pos > 0 && pos <= allEvents.length) {
|
||||
const ev = allEvents[pos - 1];
|
||||
nodeFlash[ev.node] = performance.now();
|
||||
if (ev.kind === 'GossipRoundStarted' && ev.detail && ev.detail.target) {
|
||||
edgeFlash[ev.node + '->' + ev.detail.target] = performance.now();
|
||||
}
|
||||
if (ev.kind === 'PushReceived' && ev.detail && ev.detail.from) {
|
||||
edgeFlash[ev.detail.from + '->' + ev.node] = performance.now();
|
||||
}
|
||||
}
|
||||
|
||||
document.getElementById('replaySlider').value = pos;
|
||||
updateReplayPos();
|
||||
drawGraph();
|
||||
}
|
||||
|
||||
function updateReplayPos() {
|
||||
document.getElementById('replayPos').textContent = replayCursor + '/' + allEvents.length;
|
||||
}
|
||||
|
||||
// ── Boot ───────────────────────────────────────────────────────────
|
||||
resizeCanvas();
|
||||
if (MODE === 'replay') {
|
||||
startReplay();
|
||||
} else {
|
||||
startLive();
|
||||
}
|
||||
|
||||
(async function boot() {
|
||||
const select = document.getElementById('traceSelect');
|
||||
const dot = document.getElementById('statusDot');
|
||||
const statusText = document.getElementById('statusText');
|
||||
|
||||
let traces;
|
||||
try {
|
||||
const resp = await fetch('/traces');
|
||||
traces = await resp.json();
|
||||
} catch (e) {
|
||||
dot.className = 'status-dot error';
|
||||
statusText.textContent = 'Failed to fetch trace list';
|
||||
select.innerHTML = '<option value="">Error</option>';
|
||||
return;
|
||||
}
|
||||
|
||||
if (traces.length === 0) {
|
||||
dot.className = 'status-dot error';
|
||||
statusText.textContent = 'No traces found';
|
||||
select.innerHTML = '<option value="">No traces found</option>';
|
||||
return;
|
||||
}
|
||||
|
||||
select.innerHTML = '';
|
||||
traces.forEach(t => {
|
||||
const opt = document.createElement('option');
|
||||
opt.value = t.file;
|
||||
opt.textContent = t.name + ' (' + t.nodes + ' nodes, ' + t.events + ' events)';
|
||||
select.appendChild(opt);
|
||||
});
|
||||
select.disabled = false;
|
||||
|
||||
select.onchange = () => {
|
||||
if (select.value) loadTrace(select.value);
|
||||
};
|
||||
|
||||
// Auto-load first trace
|
||||
loadTrace(traces[0].file);
|
||||
})();
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -1,747 +1,138 @@
|
|||
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,
|
||||
fn scan_traces(dir: &Path) -> Vec<TraceEntry> {
|
||||
let mut entries = Vec::new();
|
||||
let Ok(read_dir) = fs::read_dir(dir) else {
|
||||
return entries;
|
||||
};
|
||||
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
|
||||
}
|
||||
|
||||
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);
|
||||
// ── HTTP server ───────────────────────────────────────────────────────
|
||||
|
||||
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!("Dashboard at http://localhost:{port}");
|
||||
eprintln!("Serving traces from: {trace_dir}");
|
||||
|
||||
loop {
|
||||
let request = match server.recv() {
|
||||
Ok(r) => r,
|
||||
Err(_) => break,
|
||||
};
|
||||
|
||||
let url = request.url().to_string();
|
||||
match url.as_str() {
|
||||
"/" => {
|
||||
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);
|
||||
}
|
||||
"/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>()
|
||||
.unwrap(),
|
||||
);
|
||||
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;
|
||||
}
|
||||
|
||||
// Wait for next chunk.
|
||||
match self.rx.recv() {
|
||||
let path = dir.join(&file);
|
||||
match fs::read_to_string(&path) {
|
||||
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,
|
||||
})
|
||||
}
|
||||
};
|
||||
|
||||
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"
|
||||
let response = tiny_http::Response::from_string(data).with_header(
|
||||
"Content-Type: application/json"
|
||||
.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));
|
||||
}
|
||||
_ => {
|
||||
Err(_) => {
|
||||
let response =
|
||||
tiny_http::Response::from_string("Not Found").with_status_code(404);
|
||||
let _ = request.respond(response);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_sse(request: tiny_http::Request, state: Arc<DashboardState>) {
|
||||
let (tx, rx) = mpsc::channel::<Vec<u8>>();
|
||||
let reader = ChannelReader::new(rx);
|
||||
|
||||
// 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),
|
||||
});
|
||||
|
||||
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}");
|
||||
|
||||
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__", "replay");
|
||||
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);
|
||||
}
|
||||
"/trace.json" => {
|
||||
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);
|
||||
}
|
||||
_ => {
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue