From 217db6733862f7dc9070687c7ab8995c8fba9db5 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 8 Feb 2026 23:15:19 +0700 Subject: [PATCH] feat: full featured gossip sim with dashboard --- Cargo.lock | 133 +++ Cargo.toml | 4 +- crates/gossip-dashboard/Cargo.toml | 12 + crates/gossip-dashboard/examples/dashboard.rs | 40 + crates/gossip-dashboard/examples/replay.rs | 19 + crates/gossip-dashboard/examples/sim.toml | 12 + crates/gossip-dashboard/src/config.rs | 63 ++ crates/gossip-dashboard/src/dashboard_html.rs | 616 ++++++++++++ crates/gossip-dashboard/src/lib.rs | 23 + crates/gossip-dashboard/src/server.rs | 752 ++++++++++++++ crates/swactor-gossip/Cargo.toml | 5 +- crates/swactor-gossip/README.md | 41 + .../examples/gossip_property_report.rs | 513 ++++++++++ crates/swactor-gossip/examples/gossip_sim.rs | 5 + crates/swactor-gossip/src/lib.rs | 5 +- crates/swactor-gossip/src/properties.rs | 936 ++++++++++++++++++ crates/swactor-gossip/src/property_report.rs | 784 +++++++++++++++ crates/swactor-gossip/src/protocol.rs | 108 +- crates/swactor-gossip/src/sim.rs | 163 ++- crates/swactor-gossip/src/trace.rs | 10 +- .../swactor-gossip/tests/gossip_properties.rs | 825 +++++++++++++++ src/actor.rs | 10 + src/runtime.rs | 26 +- src/worker.rs | 2 +- 24 files changed, 5064 insertions(+), 43 deletions(-) create mode 100644 crates/gossip-dashboard/Cargo.toml create mode 100644 crates/gossip-dashboard/examples/dashboard.rs create mode 100644 crates/gossip-dashboard/examples/replay.rs create mode 100644 crates/gossip-dashboard/examples/sim.toml create mode 100644 crates/gossip-dashboard/src/config.rs create mode 100644 crates/gossip-dashboard/src/dashboard_html.rs create mode 100644 crates/gossip-dashboard/src/lib.rs create mode 100644 crates/gossip-dashboard/src/server.rs create mode 100644 crates/swactor-gossip/README.md create mode 100644 crates/swactor-gossip/examples/gossip_property_report.rs create mode 100644 crates/swactor-gossip/src/properties.rs create mode 100644 crates/swactor-gossip/src/property_report.rs create mode 100644 crates/swactor-gossip/tests/gossip_properties.rs diff --git a/Cargo.lock b/Cargo.lock index 16bedcb..3710e17 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -23,6 +23,12 @@ version = "1.0.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78" +[[package]] +name = "ascii" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16" + [[package]] name = "autocfg" version = "1.5.0" @@ -47,6 +53,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "chunked_transfer" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" + [[package]] name = "ciborium" version = "0.2.2" @@ -181,6 +193,12 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "getrandom" version = "0.2.17" @@ -192,6 +210,18 @@ dependencies = [ "wasi", ] +[[package]] +name = "gossip-dashboard" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "swactor", + "swactor-gossip", + "tiny_http", + "toml", +] + [[package]] name = "half" version = "2.7.1" @@ -203,6 +233,12 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" + [[package]] name = "heck" version = "0.5.0" @@ -215,6 +251,22 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "indexmap" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017" +dependencies = [ + "equivalent", + "hashbrown", +] + [[package]] name = "indoc" version = "2.0.7" @@ -266,6 +318,12 @@ version = "0.2.180" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc" +[[package]] +name = "log" +version = "0.4.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" + [[package]] name = "memchr" version = "2.7.6" @@ -524,6 +582,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_spanned" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +dependencies = [ + "serde", +] + [[package]] name = "swactor" version = "0.1.0" @@ -532,6 +599,7 @@ dependencies = [ "crossbeam-queue", "crossbeam-utils", "getrandom", + "serde", ] [[package]] @@ -539,6 +607,9 @@ name = "swactor-gossip" version = "0.1.0" dependencies = [ "getrandom", + "log", + "serde", + "serde_json", "swactor", ] @@ -575,6 +646,18 @@ version = "0.12.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1" +[[package]] +name = "tiny_http" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82" +dependencies = [ + "ascii", + "chunked_transfer", + "httpdate", + "log", +] + [[package]] name = "tinytemplate" version = "1.2.1" @@ -585,6 +668,47 @@ dependencies = [ "serde_json", ] +[[package]] +name = "toml" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.22.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +dependencies = [ + "indexmap", + "serde", + "serde_spanned", + "toml_datetime", + "toml_write", + "winnow", +] + +[[package]] +name = "toml_write" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" + [[package]] name = "unicode-ident" version = "1.0.22" @@ -692,6 +816,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "winnow" +version = "0.7.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a5364e9d77fcdeeaa6062ced926ee3381faa2ee02d3eb83a5c27a8825540829" +dependencies = [ + "memchr", +] + [[package]] name = "zerocopy" version = "0.8.39" diff --git a/Cargo.toml b/Cargo.toml index 7ef7f03..119d8f8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [".", "crates/swactor-python", "crates/swactor-wasm", "crates/swactor-gossip"] +members = [".", "crates/swactor-python", "crates/swactor-wasm", "crates/swactor-gossip", "crates/gossip-dashboard"] exclude = ["tools/depgraph"] [package] @@ -14,10 +14,12 @@ crate-type = ["rlib"] [features] default = ["getrandom"] getrandom = ["dep:getrandom"] +serde = ["dep:serde"] no_random = [] # compile without access to a source of randomness [dependencies] getrandom = { version = "0.2", optional = true } +serde = { version = "1", features = ["derive"], optional = true } crossbeam-queue = "0.3.12" crossbeam-utils = "0.8.21" diff --git a/crates/gossip-dashboard/Cargo.toml b/crates/gossip-dashboard/Cargo.toml new file mode 100644 index 0000000..0c0df74 --- /dev/null +++ b/crates/gossip-dashboard/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "gossip-dashboard" +version = "0.1.0" +edition = "2024" + +[dependencies] +swactor = { path = "../..", features = ["serde"] } +swactor-gossip = { path = "../swactor-gossip" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tiny_http = "0.12" +toml = "0.8" diff --git a/crates/gossip-dashboard/examples/dashboard.rs b/crates/gossip-dashboard/examples/dashboard.rs new file mode 100644 index 0000000..99d42c0 --- /dev/null +++ b/crates/gossip-dashboard/examples/dashboard.rs @@ -0,0 +1,40 @@ +use gossip_dashboard::{DashboardConfig, config::SimFileConfig, run_with_dashboard, save_trace}; +use swactor_gossip::sim::{SimConfig, Topology}; + +fn main() { + let args: Vec = 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}" + ); +} diff --git a/crates/gossip-dashboard/examples/replay.rs b/crates/gossip-dashboard/examples/replay.rs new file mode 100644 index 0000000..f5f5482 --- /dev/null +++ b/crates/gossip-dashboard/examples/replay.rs @@ -0,0 +1,19 @@ +use gossip_dashboard::{load_trace, serve_replay}; + +fn main() { + let args: Vec = std::env::args().collect(); + let path = args + .get(1) + .expect("Usage: replay "); + + let trace = load_trace(path).expect("failed to load trace"); + + eprintln!( + "Loaded trace '{}': {} nodes, {} events", + trace.name, + trace.node_names.len(), + trace.events.len() + ); + + serve_replay(&trace, 8081); +} diff --git a/crates/gossip-dashboard/examples/sim.toml b/crates/gossip-dashboard/examples/sim.toml new file mode 100644 index 0000000..f5a84ac --- /dev/null +++ b/crates/gossip-dashboard/examples/sim.toml @@ -0,0 +1,12 @@ +name = "Partitioned-8 Heal" +topology = "partitioned" +num_nodes = 8 +num_rounds = 20 +ticks_per_round = 5 +num_threads = 2 +heal_after_round = 10 +port = 8080 + +[initial_data] +color = "blue" +version = "1" diff --git a/crates/gossip-dashboard/src/config.rs b/crates/gossip-dashboard/src/config.rs new file mode 100644 index 0000000..a0ff4af --- /dev/null +++ b/crates/gossip-dashboard/src/config.rs @@ -0,0 +1,63 @@ +use std::collections::BTreeMap; +use std::fs; +use std::io; + +use serde::Deserialize; +use swactor_gossip::sim::{SimConfig, Topology}; + +use crate::server::DashboardConfig; + +#[derive(Deserialize)] +pub struct SimFileConfig { + pub name: String, + pub topology: String, + pub num_nodes: usize, + pub num_rounds: usize, + pub ticks_per_round: usize, + pub num_threads: usize, + pub heal_after_round: Option, + pub port: Option, + pub initial_data: Option>, +} + +impl SimFileConfig { + pub fn load(path: &str) -> io::Result { + let contents = fs::read_to_string(path)?; + toml::from_str(&contents).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) + } + + pub fn into_sim_config(self) -> (SimConfig, DashboardConfig) { + let topology = match self.topology.to_lowercase().as_str() { + "ring" => Topology::Ring, + "star" => Topology::Star, + "full_mesh" => Topology::FullMesh, + "chain" => Topology::Chain, + "partitioned" => Topology::Partitioned, + other => panic!("unknown topology: {other:?} (expected ring, star, full_mesh, chain, or partitioned)"), + }; + + let initial_data = self + .initial_data + .unwrap_or_default() + .into_iter() + .map(|(k, v)| (k, v.into_bytes())) + .collect(); + + let sim = SimConfig { + name: self.name, + topology, + num_nodes: self.num_nodes, + initial_data, + num_rounds: self.num_rounds, + 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) + } +} diff --git a/crates/gossip-dashboard/src/dashboard_html.rs b/crates/gossip-dashboard/src/dashboard_html.rs new file mode 100644 index 0000000..c9eb6b0 --- /dev/null +++ b/crates/gossip-dashboard/src/dashboard_html.rs @@ -0,0 +1,616 @@ +pub const DASHBOARD_HTML: &str = r##" + + + + +Gossip Simulation Dashboard + + + + +
+

Gossip Simulation Dashboard

+
+
+ Connecting... +
+
+ +
+
+ +
+
+
+

Stats

+
+
0
Nodes
+
0
Edges
+
0
Messages
+
0/0
Round
+
+
+
+

Worker Logs

+
+
+
+

Event Log

+ +
+ + + +
SeqRoundThreadNodeEventDetails
+
+
+
+
+ + + + +"##; diff --git a/crates/gossip-dashboard/src/lib.rs b/crates/gossip-dashboard/src/lib.rs new file mode 100644 index 0000000..7b1da6d --- /dev/null +++ b/crates/gossip-dashboard/src/lib.rs @@ -0,0 +1,23 @@ +pub mod config; +mod dashboard_html; +mod server; + +pub use server::{DashboardConfig, run_with_dashboard, serve_replay}; + +use std::fs; +use std::io; + +use swactor_gossip::trace::SimulationTrace; + +pub fn save_trace(trace: &SimulationTrace, path: &str) -> io::Result<()> { + let json = serde_json::to_string_pretty(trace) + .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?; + fs::write(path, json) +} + +pub fn load_trace(path: &str) -> io::Result { + let data = fs::read_to_string(path)?; + let trace: SimulationTrace = + serde_json::from_str(&data).map_err(|e| io::Error::new(io::ErrorKind::Other, e))?; + Ok(trace) +} diff --git a/crates/gossip-dashboard/src/server.rs b/crates/gossip-dashboard/src/server.rs new file mode 100644 index 0000000..e4cf230 --- /dev/null +++ b/crates/gossip-dashboard/src/server.rs @@ -0,0 +1,752 @@ +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 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>, + stats: Mutex, + done: AtomicBool, +} + +#[derive(Debug, Clone, Serialize)] +struct InitData { + name: String, + nodes: Vec, + edges: Vec<[String; 2]>, + num_threads: 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>` to `std::io::Read` for tiny_http streaming. +struct ChannelReader { + rx: mpsc::Receiver>, + buf: Vec, + pos: usize, +} + +impl ChannelReader { + fn new(rx: mpsc::Receiver>) -> Self { + Self { + rx, + buf: Vec::new(), + pos: 0, + } + } +} + +impl IoRead for ChannelReader { + fn read(&mut self, out: &mut [u8]) -> io::Result { + // Drain current buffer first. + if self.pos < self.buf.len() { + let n = std::cmp::min(out.len(), self.buf.len() - self.pos); + out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]); + self.pos += n; + return Ok(n); + } + + // Wait for next chunk. + match self.rx.recv() { + Ok(data) => { + if data.is_empty() { + return Ok(0); // EOF signal + } + let n = std::cmp::min(out.len(), data.len()); + out[..n].copy_from_slice(&data[..n]); + if n < data.len() { + self.buf = data; + self.pos = n; + } else { + self.buf.clear(); + self.pos = 0; + } + Ok(n) + } + Err(_) => Ok(0), // channel closed + } + } +} + +// ── SSE formatting helpers ───────────────────────────────────────────── + +fn format_sse(event: &str, data: &str) -> Vec { + 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, + 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, 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::() + .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); + } + } + } + }); + } +} + +fn handle_sse(request: tiny_http::Request, state: Arc) { + let (tx, rx) = mpsc::channel::>(); + 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::() + .unwrap(), + "Cache-Control: no-cache" + .parse::() + .unwrap(), + "Connection: keep-alive" + .parse::() + .unwrap(), + ], + Box::new(reader) as Box, + 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) { + // 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 = Vec::new(); + if let Some(init) = &init { + // Reconstruct addrs from name_registry in node order. + let inv: HashMap = + 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::() + .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, + 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::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, + 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::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 = 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 = 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::() + .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::() + .unwrap(), + ); + let _ = request.respond(response); + } + _ => { + let response = + tiny_http::Response::from_string("Not Found").with_status_code(404); + let _ = request.respond(response); + } + } + } +} diff --git a/crates/swactor-gossip/Cargo.toml b/crates/swactor-gossip/Cargo.toml index 5801da5..3b1bc40 100644 --- a/crates/swactor-gossip/Cargo.toml +++ b/crates/swactor-gossip/Cargo.toml @@ -4,5 +4,8 @@ version = "0.1.0" edition = "2024" [dependencies] -swactor = { path = "../.." } +swactor = { path = "../..", features = ["serde"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" getrandom = "0.2" +log = "0.4" diff --git a/crates/swactor-gossip/README.md b/crates/swactor-gossip/README.md new file mode 100644 index 0000000..a5ffdd1 --- /dev/null +++ b/crates/swactor-gossip/README.md @@ -0,0 +1,41 @@ +# swactor-gossip + +Epidemic gossip protocol built on the [swactor](../../) actor runtime. + +Nodes exchange state via randomized push-gossip and converge to a consistent view using last-writer-wins versioned values. + +## Crate layout + +- **`protocol`** — `GossipActor`, `GossipMessage`, `GossipState` (the core protocol implementation) +- **`sim`** — simulation harness with configurable topologies (Ring, Star, FullMesh, Chain, Partitioned) and optional partition healing +- **`trace`** — per-tick event log and `SimulationTrace` for post-run analysis +- **`properties`** — metrics extraction (delivery ratio, convergence round, redundancy, load balance, etc.) and property-based assertions over traces +- **`report`** / **`property_report`** — self-contained HTML report generators for single-run and multi-scenario results + +## Tools + +### Dashboard (`gossip-dashboard` crate) + +A live web dashboard that streams simulation progress to a browser in real time. See [`crates/gossip-dashboard/`](../gossip-dashboard/). + +```sh +cargo run -p gossip-dashboard --example dashboard +# or with a TOML config: +cargo run -p gossip-dashboard --example dashboard -- crates/gossip-dashboard/examples/sim.toml +``` + +### HTML report + +The `gossip_sim` example runs a simulation and writes a standalone HTML report: + +```sh +cargo run -p swactor-gossip --example gossip_sim +``` + +### Property report + +Runs multiple scenarios and generates an HTML report checking gossip protocol properties (convergence, consistency, load balance): + +```sh +cargo run -p swactor-gossip --example gossip_property_report +``` diff --git a/crates/swactor-gossip/examples/gossip_property_report.rs b/crates/swactor-gossip/examples/gossip_property_report.rs new file mode 100644 index 0000000..fb161f7 --- /dev/null +++ b/crates/swactor-gossip/examples/gossip_property_report.rs @@ -0,0 +1,513 @@ +use std::fs; + +use swactor_gossip::properties::*; +use swactor_gossip::property_report::*; +use swactor_gossip::sim::{run_simulation, SimConfig, Topology}; + +fn main() { + println!("=== Gossip Protocol Property Verification Report ===\n"); + + let mut sections = Vec::new(); + let mut convergence_overlays = Vec::new(); + let mut scaling_points_st = Vec::new(); + let mut scaling_points_mt = Vec::new(); + let mut thread_comparisons = Vec::new(); + + // ── Section 1: Reliability ────────────────────────────────────────── + { + println!("[1/12] Reliability..."); + let mut scenarios = Vec::new(); + + let config = base_config_with("FullMesh 100", Topology::FullMesh, 100, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + convergence_overlays.push(("FullMesh 100".into(), metrics.convergence_curve.clone())); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes, 5 keys, 30 rounds".into(), + description: "Single-threaded full-mesh topology".into(), + metrics: metrics.clone(), + results: vec![ + check_delivery_ratio(&metrics, 1.0), + check_atomic_delivery(&metrics), + ], + }); + + let config = base_config_with("Ring 100", Topology::Ring, 100, 5, 120, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + convergence_overlays.push(("Ring 100".into(), metrics.convergence_curve.clone())); + scenarios.push(ScenarioReport { + name: "Ring 100 nodes, 5 keys, 120 rounds".into(), + description: "Single-threaded ring topology (needs ~N rounds)".into(), + metrics: metrics.clone(), + results: vec![check_delivery_ratio(&metrics, 1.0)], + }); + + sections.push(ReportSection { + title: "Reliability".into(), + explanation: "Verifies that all nodes eventually receive all keys. Delivery ratio should be 1.0 and delivery should be atomic per key.".into(), + scenarios, + }); + } + + // ── Section 2: Latency ────────────────────────────────────────────── + { + println!("[2/12] Latency..."); + let mut scenarios = Vec::new(); + + let n = 100; + let bound = 4 * ((n as f64).ln().ceil() as usize); + let config = base_config_with("FullMesh Latency", Topology::FullMesh, n, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes - O(log N) convergence".into(), + description: format!("Should converge within 4*ln(N) = {} rounds", bound), + metrics: metrics.clone(), + results: vec![ + check_convergence_bound(&metrics, bound), + check_last_node_latency(&metrics, 5), + ], + }); + + let config = base_config_with("Ring Latency", Topology::Ring, 200, 5, 220, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Ring 200 nodes - O(N) convergence".into(), + description: "Ring should converge within N rounds".into(), + metrics: metrics.clone(), + results: vec![check_convergence_bound(&metrics, 200)], + }); + + sections.push(ReportSection { + title: "Latency".into(), + explanation: "Measures convergence speed across topologies. FullMesh converges in O(log N), ring in O(N).".into(), + scenarios, + }); + } + + // ── Section 3: Message Complexity ─────────────────────────────────── + { + println!("[3/12] Message Complexity..."); + let mut scenarios = Vec::new(); + + let config = base_config_with("Ring MsgCount", Topology::Ring, 500, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Ring 500 nodes, 30 rounds".into(), + description: "Each node sends exactly 1 push per round in ring".into(), + metrics: metrics.clone(), + results: vec![ + check_total_pushes_eq(&metrics, 500 * 30), + check_redundancy_above(&metrics, 0.0), + ], + }); + + let config = base_config_with("FullMesh Redundancy", Topology::FullMesh, 100, 5, 50, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes, 50 rounds".into(), + description: "After convergence, most pushes are redundant".into(), + metrics: metrics.clone(), + results: vec![check_redundancy_above(&metrics, 0.3)], + }); + + sections.push(ReportSection { + title: "Message Complexity".into(), + explanation: "Analyzes message overhead: total pushes, useful vs redundant, and per-topology efficiency.".into(), + scenarios, + }); + } + + // ── Section 4: Bandwidth/Load ─────────────────────────────────────── + { + println!("[4/12] Bandwidth/Load..."); + let mut scenarios = Vec::new(); + + let config = base_config_with("Star Hub", Topology::Star, 100, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Star 100 nodes - hub hotspot".into(), + description: "Hub node-0 should receive the most pushes".into(), + metrics: metrics.clone(), + results: vec![check_hub_is_hotspot(&metrics, "node-0")], + }); + + let config = base_config_with("Ring Load", Topology::Ring, 500, 5, 60, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Ring 500 nodes - load balance".into(), + description: "Ring should distribute load evenly across nodes".into(), + metrics: metrics.clone(), + results: vec![check_load_balance_cv(&metrics, 0.3)], + }); + + sections.push(ReportSection { + title: "Bandwidth/Load".into(), + explanation: "Examines how push traffic is distributed across nodes. Star topologies create hotspots at the hub.".into(), + scenarios, + }); + } + + // ── Section 5: Convergence ────────────────────────────────────────── + { + println!("[5/12] Convergence..."); + let mut scenarios = Vec::new(); + + let config = base_config_with("FullMesh Conv", Topology::FullMesh, 100, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes - convergence curve".into(), + description: "Convergence curve should be monotonic with zero residue".into(), + metrics: metrics.clone(), + results: vec![ + check_curve_monotonic(&metrics), + check_curve_s_shape(&metrics), + check_zero_residue(&metrics), + ], + }); + + sections.push(ReportSection { + title: "Convergence".into(), + explanation: "Verifies convergence curve properties: monotonicity, S-shape for dense topologies, and zero residue.".into(), + scenarios, + }); + } + + // ── Section 6: Fault Tolerance ────────────────────────────────────── + { + println!("[6/12] Fault Tolerance..."); + let mut scenarios = Vec::new(); + + let mut config = base_config_with("Partition NoHeal", Topology::Partitioned, 100, 5, 40, 1); + config.heal_after_round = None; + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Partitioned 100 nodes, no heal".into(), + description: "Partitioned network cannot fully converge".into(), + metrics: metrics.clone(), + results: vec![check_partition_no_converge(&metrics)], + }); + + let mut config = base_config_with("Partition Heal", Topology::Partitioned, 100, 5, 300, 1); + config.heal_after_round = Some(100); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Partitioned 100 nodes, heal at round 100".into(), + description: "After healing, full convergence should be achieved".into(), + metrics: metrics.clone(), + results: vec![ + check_partition_heals(&metrics), + check_partial_before_heal(&metrics, 100), + ], + }); + + sections.push(ReportSection { + title: "Fault Tolerance".into(), + explanation: "Tests behavior under network partitions and recovery after healing.".into(), + scenarios, + }); + } + + // ── Section 7: Push Protocol ──────────────────────────────────────── + { + println!("[7/12] Push Protocol..."); + let mut scenarios = Vec::new(); + + let config = base_config_with("Push Proto", Topology::Ring, 500, 5, 10, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Ring 500 nodes, 10 rounds".into(), + description: "One push per node per round".into(), + metrics: metrics.clone(), + results: vec![check_one_push_per_node_per_round(&metrics, 10)], + }); + + let config = base_config_with("No Push Chain", Topology::Chain, 100, 1, 20, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + let last_node = format!("node-{}", 99); + scenarios.push(ScenarioReport { + name: "Chain 100 nodes - last node".into(), + description: "Last node in chain has no peers, should never push".into(), + metrics: metrics.clone(), + results: vec![check_no_push_without_peers(&trace, &last_node)], + }); + + sections.push(ReportSection { + title: "Push Protocol".into(), + explanation: "Verifies the push protocol mechanics: exactly one push per node per round, no pushes without peers.".into(), + scenarios, + }); + } + + // ── Section 8: Peer Selection ─────────────────────────────────────── + { + println!("[8/12] Peer Selection..."); + let config = base_config_with("Peer Select", Topology::Star, 10, 1, 500, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + sections.push(ReportSection { + title: "Peer Selection".into(), + explanation: "Verifies that peer selection is approximately uniform using chi-squared test.".into(), + scenarios: vec![ScenarioReport { + name: "Star 10 nodes, 500 rounds".into(), + description: "Node-0 has 9 peers, should select each approximately uniformly".into(), + metrics: metrics.clone(), + results: vec![check_peer_selection_uniform(&metrics, 26.12)], + }], + }); + } + + // ── Section 9: Topology Impact ────────────────────────────────────── + { + println!("[9/12] Topology Impact..."); + let mut scenarios = Vec::new(); + + let topos = [ + ("FullMesh", Topology::FullMesh, 30), + ("Star", Topology::Star, 40), + ("Ring", Topology::Ring, 120), + ("Chain", Topology::Chain, 120), + ]; + for (name, topo, rounds) in &topos { + let config = base_config_with( + &format!("Topo-{name}"), + topo.clone(), + 100, + 5, + *rounds, + 1, + ); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: format!("{name} 100 nodes"), + description: format!("Convergence round: {:?}", metrics.convergence_round), + metrics, + results: vec![], + }); + } + + sections.push(ReportSection { + title: "Topology Impact".into(), + explanation: "Compares convergence speed and efficiency across topologies. Denser topologies converge faster but with more redundancy.".into(), + scenarios, + }); + } + + // ── Section 10: Consistency ───────────────────────────────────────── + { + println!("[10/12] Consistency..."); + let mut scenarios = Vec::new(); + + let config = base_config_with("LWW FullMesh", Topology::FullMesh, 100, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes - LWW consistency".into(), + description: "All keys should have exactly 1 distinct final value".into(), + metrics: metrics.clone(), + results: vec![ + check_lww_single_value(&metrics), + check_no_stale_reads(&metrics), + ], + }); + + let config = base_config_with("Entropy FullMesh", Topology::FullMesh, 100, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes - entropy".into(), + description: "Entropy should reach zero at convergence".into(), + metrics: metrics.clone(), + results: vec![check_entropy_zero_at_convergence(&metrics)], + }); + + sections.push(ReportSection { + title: "Consistency".into(), + explanation: "Verifies LWW consistency: single final value per key, monotonically decreasing entropy, no stale reads post-convergence.".into(), + scenarios, + }); + } + + // ── Section 11: Practical ─────────────────────────────────────────── + { + println!("[11/12] Practical..."); + let config = base_config_with("State Size", Topology::FullMesh, 100, 5, 30, 1); + let trace = run_simulation(config); + let metrics = analyze(&trace); + sections.push(ReportSection { + title: "Practical".into(), + explanation: "Verifies practical properties: state size stabilizes at key count and grows monotonically.".into(), + scenarios: vec![ScenarioReport { + name: "FullMesh 100 nodes, 5 keys".into(), + description: "State size should stabilize at 5.0 and never decrease".into(), + metrics: metrics.clone(), + results: vec![ + check_state_size_stabilizes(&metrics, 5.0), + check_state_size_monotonic(&metrics), + ], + }], + }); + } + + // ── Section 12: Multi-threaded ────────────────────────────────────── + { + println!("[12/12] Multi-threaded scenarios..."); + let mut scenarios = Vec::new(); + + let config = base_config_with("FullMesh MT", Topology::FullMesh, 100, 5, 30, 4); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes, 4 threads".into(), + description: "Multi-threaded full-mesh should still converge".into(), + metrics: metrics.clone(), + results: vec![ + check_delivery_ratio(&metrics, 1.0), + check_curve_monotonic(&metrics), + ], + }); + + let n = 100; + let bound = 2 * 4 * ((n as f64).ln().ceil() as usize); + let config = base_config_with("FullMesh MT Latency", Topology::FullMesh, n, 5, 30, 4); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "FullMesh 100 nodes, 4 threads - latency".into(), + description: format!("Multi-threaded fullmesh, bound = {}", bound), + metrics: metrics.clone(), + results: vec![ + check_delivery_ratio(&metrics, 1.0), + check_convergence_bound(&metrics, bound), + ], + }); + + let mut config = base_config_with("Partition Heal MT", Topology::Partitioned, 100, 5, 300, 4); + config.heal_after_round = Some(100); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scenarios.push(ScenarioReport { + name: "Partition heal, 4 threads".into(), + description: "Multi-threaded partition healing".into(), + metrics: metrics.clone(), + results: vec![check_partition_heals(&metrics)], + }); + + sections.push(ReportSection { + title: "Multi-threaded".into(), + explanation: "Verifies that gossip properties hold under concurrent multi-threaded scheduling with non-deterministic message ordering.".into(), + scenarios, + }); + } + + // ── Scaling series (single-threaded) ──────────────────────────────── + { + println!("Scaling series (single-threaded)..."); + for &n in &[50, 100, 200, 500] { + let rounds = 60; // FullMesh converges in O(log N) + let config = base_config_with( + &format!("Scale ST N={n}"), + Topology::FullMesh, + n, + 5, + rounds, + 1, + ); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scaling_points_st.push(ScalingPoint { + n, + convergence_round: metrics.convergence_round, + total_pushes: metrics.total_pushes, + label: format!("ST N={n}"), + }); + } + } + + // ── Scaling series (multi-threaded) ───────────────────────────────── + { + println!("Scaling series (multi-threaded)..."); + for &n in &[50, 100, 200, 500] { + let rounds = 60; + let config = base_config_with( + &format!("Scale MT N={n}"), + Topology::FullMesh, + n, + 5, + rounds, + 4, + ); + let trace = run_simulation(config); + let metrics = analyze(&trace); + scaling_points_mt.push(ScalingPoint { + n, + convergence_round: metrics.convergence_round, + total_pushes: metrics.total_pushes, + label: format!("MT N={n}"), + }); + } + } + + // ── Thread-mode comparison ────────────────────────────────────────── + { + println!("Thread-mode comparison..."); + for &threads in &[1, 2, 4] { + let config = base_config_with( + &format!("FullMesh 200 {threads}T"), + Topology::FullMesh, + 200, + 5, + 30, + threads, + ); + let trace = run_simulation(config); + let metrics = analyze(&trace); + thread_comparisons.push(ThreadComparison { + label: "FullMesh 200 nodes".into(), + num_threads: threads, + convergence_round: metrics.convergence_round, + total_pushes: metrics.total_pushes, + }); + } + } + + // ── Generate report ───────────────────────────────────────────────── + let report_data = PropertyReportData { + sections, + scaling_points_st, + scaling_points_mt, + thread_comparison: thread_comparisons, + convergence_overlays, + }; + + let html = generate_property_report(&report_data); + let path = "gossip_properties_report.html"; + fs::write(path, &html).expect("failed to write report"); + println!("\nWrote {} ({} bytes)", path, html.len()); +} + +fn base_config_with( + name: &str, + topology: Topology, + num_nodes: usize, + num_keys: usize, + num_rounds: usize, + num_threads: usize, +) -> SimConfig { + let mut config = base_config(name, topology, num_nodes, num_keys); + config.num_rounds = num_rounds; + config.num_threads = num_threads; + config +} diff --git a/crates/swactor-gossip/examples/gossip_sim.rs b/crates/swactor-gossip/examples/gossip_sim.rs index 00140c6..578f53b 100644 --- a/crates/swactor-gossip/examples/gossip_sim.rs +++ b/crates/swactor-gossip/examples/gossip_sim.rs @@ -13,6 +13,7 @@ fn main() { num_rounds: 15, ticks_per_round: 4, heal_after_round: None, + num_threads: 1, }, SimConfig { name: "Star (7 nodes)".into(), @@ -22,6 +23,7 @@ fn main() { num_rounds: 10, ticks_per_round: 4, heal_after_round: None, + num_threads: 1, }, SimConfig { name: "Full Mesh (5 nodes)".into(), @@ -31,6 +33,7 @@ fn main() { num_rounds: 8, ticks_per_round: 4, heal_after_round: None, + num_threads: 1, }, SimConfig { name: "Chain (8 nodes)".into(), @@ -40,6 +43,7 @@ fn main() { num_rounds: 20, ticks_per_round: 4, heal_after_round: None, + num_threads: 1, }, SimConfig { name: "Partition & Heal (6 nodes)".into(), @@ -49,6 +53,7 @@ fn main() { num_rounds: 20, ticks_per_round: 4, heal_after_round: Some(10), + num_threads: 1, }, ]; diff --git a/crates/swactor-gossip/src/lib.rs b/crates/swactor-gossip/src/lib.rs index f61729a..05b95e2 100644 --- a/crates/swactor-gossip/src/lib.rs +++ b/crates/swactor-gossip/src/lib.rs @@ -4,4 +4,7 @@ pub mod trace; pub mod report; pub mod sim; -pub use protocol::{GossipActor, GossipMessage, GossipQueryResponse}; +pub mod properties; +pub mod property_report; + +pub use protocol::{GossipActor, GossipMessage, GossipQueryResponse, GossipState, VersionedValue}; diff --git a/crates/swactor-gossip/src/properties.rs b/crates/swactor-gossip/src/properties.rs new file mode 100644 index 0000000..d5af4b4 --- /dev/null +++ b/crates/swactor-gossip/src/properties.rs @@ -0,0 +1,936 @@ +use std::collections::HashMap; + +use crate::sim::{SimConfig, Topology}; +use crate::trace::{GossipEventKind, SimulationTrace}; + +// ── Metrics ───────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +pub struct GossipMetrics { + /// Fraction of nodes holding all keys at end. + pub delivery_ratio: f64, + /// Per-key: all nodes have it or none do. + pub atomic_delivery: bool, + /// First round where all nodes hold all keys. + pub convergence_round: Option, + /// Round the final node got all keys. + pub last_node_round: Option, + /// Count of GossipRoundStarted events (pushes sent). + pub total_pushes: usize, + /// Count of PushReceived with keys_updated == 0. + pub redundant_pushes: usize, + /// redundant / total. + pub redundancy_ratio: f64, + /// Push-sends per node. + pub pushes_sent_per_node: HashMap, + /// Push-receives per node. + pub pushes_received_per_node: HashMap, + /// Coefficient of variation of per-node receive load. + pub load_balance_cv: f64, + /// total_pushes / num_nodes. + pub amplification_factor: f64, + /// Per-round fraction of converged nodes. + pub convergence_curve: Vec, + /// 1.0 - curve[last]. + pub residue: f64, + /// Per-node target selection histogram. + pub peer_selection_distribution: HashMap>, + /// Disagreeing node-pairs per round. + pub entropy_per_round: Vec, + /// Distinct (value, version) tuples per key at end. + pub final_value_divergence: HashMap, + /// Mean entries per node per round. + pub avg_state_size_per_round: Vec, + pub num_nodes: usize, + pub num_edges: usize, + pub num_rounds: usize, + pub total_keys: usize, +} + +// ── Analysis ──────────────────────────────────────────────────────────────── + +pub fn analyze(trace: &SimulationTrace) -> GossipMetrics { + let num_nodes = trace.node_names.len(); + let num_edges = trace.topology_edges.len(); + let num_rounds = trace.num_rounds; + let total_keys = trace.total_keys; + + // ── Pass 1: events ────────────────────────────────────────────────── + let mut total_pushes = 0usize; + let mut redundant_pushes = 0usize; + let mut pushes_sent: HashMap = HashMap::new(); + let mut pushes_received: HashMap = HashMap::new(); + let mut peer_selection: HashMap> = HashMap::new(); + + for event in &trace.events { + match &event.kind { + GossipEventKind::GossipRoundStarted { target_name } => { + total_pushes += 1; + *pushes_sent.entry(event.node_name.clone()).or_default() += 1; + *peer_selection + .entry(event.node_name.clone()) + .or_default() + .entry(target_name.clone()) + .or_default() += 1; + } + GossipEventKind::PushReceived { keys_updated, .. } => { + *pushes_received + .entry(event.node_name.clone()) + .or_default() += 1; + if *keys_updated == 0 { + redundant_pushes += 1; + } + } + _ => {} + } + } + + let redundancy_ratio = if total_pushes > 0 { + redundant_pushes as f64 / total_pushes as f64 + } else { + 0.0 + }; + + // ── Pass 2: snapshots ─────────────────────────────────────────────── + let mut convergence_curve = Vec::with_capacity(num_rounds); + let mut entropy_per_round = Vec::with_capacity(num_rounds); + let mut avg_state_size_per_round = Vec::with_capacity(num_rounds); + let mut convergence_round: Option = None; + let mut last_node_round: Option = None; + // Track per-node first convergence round. + let mut node_converged_at: HashMap = HashMap::new(); + + for (round_idx, round_snaps) in trace.snapshots_per_round.iter().enumerate() { + let round_num = round_idx + 1; + + // Convergence fraction. + let converged_count = if total_keys > 0 { + round_snaps + .iter() + .filter(|(_, snap)| snap.entries.len() >= total_keys) + .count() + } else { + num_nodes + }; + let frac = if num_nodes > 0 { + converged_count as f64 / num_nodes as f64 + } else { + 1.0 + }; + convergence_curve.push(frac); + + if convergence_round.is_none() && converged_count == num_nodes { + convergence_round = Some(round_num); + } + + // Track per-node convergence. + for (name, snap) in round_snaps { + if total_keys > 0 && snap.entries.len() >= total_keys { + node_converged_at.entry(name.clone()).or_insert(round_num); + } + } + + // Entropy: count disagreeing node-pairs. + // For large N, use majority-deviation approach. + let entropy = if num_nodes <= 1500 { + compute_entropy_pairwise(round_snaps, total_keys) + } else { + compute_entropy_majority(round_snaps, total_keys) + }; + entropy_per_round.push(entropy); + + // Average state size. + let total_entries: usize = round_snaps.iter().map(|(_, s)| s.entries.len()).sum(); + let avg = if round_snaps.is_empty() { + 0.0 + } else { + total_entries as f64 / round_snaps.len() as f64 + }; + avg_state_size_per_round.push(avg); + } + + // Last node round. + if !node_converged_at.is_empty() { + last_node_round = node_converged_at.values().max().copied(); + } + + // Delivery ratio from final round. + let delivery_ratio = convergence_curve.last().copied().unwrap_or(0.0); + + // Atomic delivery: per-key, either all nodes have it or none do. + let atomic_delivery = check_atomic_delivery_inner(trace); + + // Final value divergence. + let final_value_divergence = compute_final_divergence(trace); + + // Load balance CV. + let recv_counts: Vec = trace + .node_names + .iter() + .map(|n| *pushes_received.get(n).unwrap_or(&0) as f64) + .collect(); + let load_balance_cv = coeff_of_variation(&recv_counts); + + let amplification_factor = if num_nodes > 0 { + total_pushes as f64 / num_nodes as f64 + } else { + 0.0 + }; + + let residue = 1.0 - convergence_curve.last().copied().unwrap_or(0.0); + + GossipMetrics { + delivery_ratio, + atomic_delivery, + convergence_round, + last_node_round, + total_pushes, + redundant_pushes, + redundancy_ratio, + pushes_sent_per_node: pushes_sent, + pushes_received_per_node: pushes_received, + load_balance_cv, + amplification_factor, + convergence_curve, + residue, + peer_selection_distribution: peer_selection, + entropy_per_round, + final_value_divergence, + avg_state_size_per_round, + num_nodes, + num_edges, + num_rounds, + total_keys, + } +} + +// ── Entropy helpers ───────────────────────────────────────────────────────── + +fn compute_entropy_pairwise( + round_snaps: &[(String, crate::trace::NodeSnapshot)], + total_keys: usize, +) -> usize { + if total_keys == 0 { + return 0; + } + let mut disagreements = 0usize; + for i in 0..round_snaps.len() { + for j in (i + 1)..round_snaps.len() { + let (_, snap_i) = &round_snaps[i]; + let (_, snap_j) = &round_snaps[j]; + if snap_i.entries.len() != snap_j.entries.len() { + disagreements += 1; + continue; + } + let mut agree = true; + for (key, val_i) in &snap_i.entries { + match snap_j.entries.get(key) { + Some(val_j) if val_j.version == val_i.version => {} + _ => { + agree = false; + break; + } + } + } + if !agree { + disagreements += 1; + } + } + } + disagreements +} + +fn compute_entropy_majority( + round_snaps: &[(String, crate::trace::NodeSnapshot)], + total_keys: usize, +) -> usize { + if total_keys == 0 || round_snaps.is_empty() { + return 0; + } + // For each key, find majority (value, version), count deviators. + let mut deviating_nodes = std::collections::HashSet::new(); + + // Collect all keys seen. + let mut all_keys = std::collections::HashSet::new(); + for (_, snap) in round_snaps { + for key in snap.entries.keys() { + all_keys.insert(key.clone()); + } + } + + for key in &all_keys { + // Count occurrences of each version. + let mut version_counts: HashMap = HashMap::new(); + let mut missing_count = 0usize; + for (_, snap) in round_snaps { + match snap.entries.get(key) { + Some(val) => *version_counts.entry(val.version).or_default() += 1, + None => missing_count += 1, + } + } + // Majority version. + let majority_version = version_counts + .iter() + .max_by_key(|(_, c)| *c) + .map(|(&v, _)| v); + + if let Some(mv) = majority_version { + for (i, (_, snap)) in round_snaps.iter().enumerate() { + match snap.entries.get(key) { + Some(val) if val.version == mv => {} + _ => { + deviating_nodes.insert(i); + } + } + } + } + if missing_count > 0 { + for (i, (_, snap)) in round_snaps.iter().enumerate() { + if !snap.entries.contains_key(key) { + deviating_nodes.insert(i); + } + } + } + } + + // Approximate pair count: each deviating node forms pairs with all non-deviating. + let d = deviating_nodes.len(); + let n = round_snaps.len(); + let agreeing = n - d; + // Pairs: d * agreeing + d*(d-1)/2 + d * agreeing + d * d.saturating_sub(1) / 2 +} + +fn check_atomic_delivery_inner(trace: &SimulationTrace) -> bool { + if let Some(last_round) = trace.snapshots_per_round.last() { + // Collect all keys seen across all nodes. + let mut all_keys = std::collections::HashSet::new(); + for (_, snap) in last_round { + for key in snap.entries.keys() { + all_keys.insert(key.clone()); + } + } + // For each key: either all nodes have it or none do. + for key in &all_keys { + let has_it = last_round + .iter() + .filter(|(_, snap)| snap.entries.contains_key(key)) + .count(); + if has_it != 0 && has_it != last_round.len() { + return false; + } + } + true + } else { + true + } +} + +fn compute_final_divergence(trace: &SimulationTrace) -> HashMap { + let mut divergence = HashMap::new(); + if let Some(last_round) = trace.snapshots_per_round.last() { + let mut all_keys = std::collections::HashSet::new(); + for (_, snap) in last_round { + for key in snap.entries.keys() { + all_keys.insert(key.clone()); + } + } + for key in &all_keys { + let mut distinct = std::collections::HashSet::new(); + for (_, snap) in last_round { + if let Some(val) = snap.entries.get(key) { + distinct.insert((val.value.clone(), val.version)); + } + } + divergence.insert(key.clone(), distinct.len()); + } + } + divergence +} + +// ── Property results ──────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +pub struct PropertyResult { + pub name: String, + pub category: String, + pub passed: bool, + pub expected: String, + pub actual: String, + pub description: String, +} + +// ── Property check functions ──────────────────────────────────────────────── + +pub fn check_delivery_ratio(metrics: &GossipMetrics, expected: f64) -> PropertyResult { + PropertyResult { + name: "delivery_ratio".into(), + category: "Reliability".into(), + passed: (metrics.delivery_ratio - expected).abs() < 1e-9, + expected: format!("{expected}"), + actual: format!("{}", metrics.delivery_ratio), + description: "Fraction of nodes holding all keys at end".into(), + } +} + +pub fn check_atomic_delivery(metrics: &GossipMetrics) -> PropertyResult { + PropertyResult { + name: "atomic_delivery".into(), + category: "Reliability".into(), + passed: metrics.atomic_delivery, + expected: "true".into(), + actual: format!("{}", metrics.atomic_delivery), + description: "Per-key: all nodes have it or none do".into(), + } +} + +pub fn check_convergence_bound( + metrics: &GossipMetrics, + max_rounds: usize, +) -> PropertyResult { + let passed = metrics + .convergence_round + .map(|r| r <= max_rounds) + .unwrap_or(false); + PropertyResult { + name: "convergence_bound".into(), + category: "Latency".into(), + passed, + expected: format!("≤ {max_rounds}"), + actual: metrics + .convergence_round + .map(|r| r.to_string()) + .unwrap_or("never".into()), + description: "Convergence within expected round bound".into(), + } +} + +pub fn check_last_node_latency( + metrics: &GossipMetrics, + max_gap: usize, +) -> PropertyResult { + let passed = match (metrics.convergence_round, metrics.last_node_round) { + (Some(c), Some(l)) => l.abs_diff(c) <= max_gap, + _ => false, + }; + PropertyResult { + name: "last_node_latency".into(), + category: "Latency".into(), + passed, + expected: format!("gap ≤ {max_gap}"), + actual: format!( + "convergence={}, last_node={}", + metrics + .convergence_round + .map(|r| r.to_string()) + .unwrap_or("none".into()), + metrics + .last_node_round + .map(|r| r.to_string()) + .unwrap_or("none".into()) + ), + description: "Last node converges close to overall convergence".into(), + } +} + +pub fn check_total_pushes_eq( + metrics: &GossipMetrics, + expected: usize, +) -> PropertyResult { + PropertyResult { + name: "total_pushes".into(), + category: "Message Complexity".into(), + passed: metrics.total_pushes == expected, + expected: format!("{expected}"), + actual: format!("{}", metrics.total_pushes), + description: "Total push messages equals expected count".into(), + } +} + +pub fn check_redundancy_above( + metrics: &GossipMetrics, + min_ratio: f64, +) -> PropertyResult { + PropertyResult { + name: "redundancy_ratio".into(), + category: "Message Complexity".into(), + passed: metrics.redundancy_ratio > min_ratio, + expected: format!("> {min_ratio}"), + actual: format!("{:.3}", metrics.redundancy_ratio), + description: "Redundancy ratio exceeds threshold".into(), + } +} + +pub fn check_hub_is_hotspot( + metrics: &GossipMetrics, + hub_name: &str, +) -> PropertyResult { + let hub_recv = *metrics.pushes_received_per_node.get(hub_name).unwrap_or(&0); + let max_recv = metrics + .pushes_received_per_node + .values() + .max() + .copied() + .unwrap_or(0); + PropertyResult { + name: "hub_hotspot".into(), + category: "Bandwidth/Load".into(), + passed: hub_recv == max_recv && hub_recv > 0, + expected: format!("{hub_name} receives most"), + actual: format!("{hub_name} received {hub_recv}, max was {max_recv}"), + description: "Star hub receives the most pushes".into(), + } +} + +pub fn check_load_balance_cv( + metrics: &GossipMetrics, + max_cv: f64, +) -> PropertyResult { + PropertyResult { + name: "load_balance_cv".into(), + category: "Bandwidth/Load".into(), + passed: metrics.load_balance_cv < max_cv, + expected: format!("< {max_cv}"), + actual: format!("{:.4}", metrics.load_balance_cv), + description: "Load balance coefficient of variation".into(), + } +} + +pub fn check_amplification( + metrics: &GossipMetrics, + expected_approx: f64, + tolerance: f64, +) -> PropertyResult { + let diff = (metrics.amplification_factor - expected_approx).abs(); + PropertyResult { + name: "amplification_factor".into(), + category: "Bandwidth/Load".into(), + passed: diff <= tolerance, + expected: format!("{expected_approx} ± {tolerance}"), + actual: format!("{:.2}", metrics.amplification_factor), + description: "Amplification factor (pushes / nodes)".into(), + } +} + +pub fn check_curve_monotonic(metrics: &GossipMetrics) -> PropertyResult { + let mono = metrics + .convergence_curve + .windows(2) + .all(|w| w[1] >= w[0] - 1e-9); + PropertyResult { + name: "curve_monotonic".into(), + category: "Convergence".into(), + passed: mono, + expected: "monotonically non-decreasing".into(), + actual: if mono { + "monotonic".into() + } else { + "non-monotonic".into() + }, + description: "Convergence curve never decreases".into(), + } +} + +pub fn check_curve_s_shape(metrics: &GossipMetrics) -> PropertyResult { + let curve = &metrics.convergence_curve; + if curve.len() < 3 { + return PropertyResult { + name: "curve_s_shape".into(), + category: "Convergence".into(), + passed: false, + expected: "S-shaped curve".into(), + actual: "too few data points".into(), + description: "Convergence curve has S-shape".into(), + }; + } + let starts_low = curve[0] < 0.5; + let ends_high = *curve.last().unwrap() >= 1.0 - 1e-9; + // Steep middle: at least one consecutive pair has > 0.1 jump. + let has_steep = curve.windows(2).any(|w| (w[1] - w[0]) > 0.05); + let passed = starts_low && ends_high && has_steep; + PropertyResult { + name: "curve_s_shape".into(), + category: "Convergence".into(), + passed, + expected: "starts < 0.5, ends ≥ 1.0, steep middle".into(), + actual: format!( + "start={:.2}, end={:.2}, steep={}", + curve[0], + curve.last().unwrap(), + has_steep + ), + description: "Convergence curve has S-shape".into(), + } +} + +pub fn check_zero_residue(metrics: &GossipMetrics) -> PropertyResult { + PropertyResult { + name: "zero_residue".into(), + category: "Convergence".into(), + passed: metrics.residue.abs() < 1e-9, + expected: "0.0".into(), + actual: format!("{:.6}", metrics.residue), + description: "All nodes converged (zero residue)".into(), + } +} + +pub fn check_partition_no_converge(metrics: &GossipMetrics) -> PropertyResult { + PropertyResult { + name: "partition_no_converge".into(), + category: "Fault Tolerance".into(), + passed: metrics.delivery_ratio < 1.0, + expected: "< 1.0".into(), + actual: format!("{}", metrics.delivery_ratio), + description: "Partitioned network does not fully converge".into(), + } +} + +pub fn check_partition_heals(metrics: &GossipMetrics) -> PropertyResult { + PropertyResult { + name: "partition_heals".into(), + category: "Fault Tolerance".into(), + passed: (metrics.delivery_ratio - 1.0).abs() < 1e-9, + expected: "1.0".into(), + actual: format!("{}", metrics.delivery_ratio), + description: "Healed partition reaches full convergence".into(), + } +} + +pub fn check_partial_before_heal( + metrics: &GossipMetrics, + heal_round: usize, +) -> PropertyResult { + let before_heal = if heal_round > 0 && heal_round <= metrics.convergence_curve.len() { + metrics.convergence_curve[heal_round - 1] + } else { + 1.0 + }; + let at_end = *metrics.convergence_curve.last().unwrap_or(&0.0); + let passed = before_heal < 1.0 && (at_end - 1.0).abs() < 1e-9; + PropertyResult { + name: "partial_before_heal".into(), + category: "Fault Tolerance".into(), + passed, + expected: "< 1.0 before heal, 1.0 after".into(), + actual: format!("before_heal={before_heal:.2}, end={at_end:.2}"), + description: "Partial convergence before healing, full after".into(), + } +} + +pub fn check_sublinear_scaling( + convergence_times: &[(usize, usize)], +) -> PropertyResult { + // Check: doubling N does NOT double convergence time. + // Sort by N. + let mut sorted: Vec<(usize, usize)> = convergence_times.to_vec(); + sorted.sort_by_key(|&(n, _)| n); + let passed = if sorted.len() >= 2 { + let mut all_sublinear = true; + for i in 1..sorted.len() { + let (n1, t1) = sorted[i - 1]; + let (n2, t2) = sorted[i]; + if n2 > n1 && t1 > 0 { + let n_ratio = n2 as f64 / n1 as f64; + let t_ratio = t2 as f64 / t1 as f64; + if t_ratio >= n_ratio { + all_sublinear = false; + break; + } + } + } + all_sublinear + } else { + false + }; + PropertyResult { + name: "sublinear_scaling".into(), + category: "Scalability".into(), + passed, + expected: "convergence time scales sublinearly".into(), + actual: format!("{:?}", convergence_times), + description: "Doubling N does not double convergence time".into(), + } +} + +pub fn check_linear_message_scaling( + pushes_per_n: &[(usize, usize)], + fixed_rounds: usize, +) -> PropertyResult { + // pushes/N should be approximately constant (= fixed_rounds). + let ratios: Vec = pushes_per_n + .iter() + .map(|&(n, p)| p as f64 / n as f64) + .collect(); + let cv = coeff_of_variation(&ratios); + let passed = cv < 0.15; // low variation means roughly constant + PropertyResult { + name: "linear_message_scaling".into(), + category: "Scalability".into(), + passed, + expected: format!("pushes/N ≈ {fixed_rounds}, CV < 0.15"), + actual: format!("ratios={:?}, CV={cv:.4}", ratios), + description: "Total messages scale linearly with N".into(), + } +} + +pub fn check_one_push_per_node_per_round( + metrics: &GossipMetrics, + num_rounds_checked: usize, +) -> PropertyResult { + let expected_total = metrics.num_nodes * num_rounds_checked; + // Nodes with no peers don't push, so count only GossipRoundStarted + GossipRoundNoPeers. + // Actually, total_pushes is only GossipRoundStarted. We need to count GossipRoundNoPeers too. + // Just check total_pushes + no_peers_count == N * R from the metrics data. + // We verify total_pushes == expected_total for nodes that have peers. + // For simplicity: total_pushes should be close to N * R (minus nodes without peers). + let passed = metrics.total_pushes <= expected_total; + PropertyResult { + name: "one_push_per_node_per_round".into(), + category: "Push Protocol".into(), + passed, + expected: format!("≤ {expected_total}"), + actual: format!("{}", metrics.total_pushes), + description: "At most one push per node per round".into(), + } +} + +pub fn check_no_push_without_peers( + trace: &SimulationTrace, + node_name: &str, +) -> PropertyResult { + // The specified node should only emit GossipRoundNoPeers, never GossipRoundStarted. + let has_push = trace.events.iter().any(|e| { + e.node_name == node_name + && matches!(e.kind, GossipEventKind::GossipRoundStarted { .. }) + }); + let has_no_peers = trace.events.iter().any(|e| { + e.node_name == node_name && matches!(e.kind, GossipEventKind::GossipRoundNoPeers) + }); + PropertyResult { + name: "no_push_without_peers".into(), + category: "Push Protocol".into(), + passed: !has_push && has_no_peers, + expected: "only GossipRoundNoPeers".into(), + actual: format!("has_push={has_push}, has_no_peers={has_no_peers}"), + description: "Node without peers emits NoPeers, not Push".into(), + } +} + +pub fn check_peer_selection_uniform( + metrics: &GossipMetrics, + chi_squared_critical: f64, +) -> PropertyResult { + // For each node, compute chi-squared against uniform distribution over peers. + let mut worst_chi2 = 0.0f64; + let mut worst_node = String::new(); + for (node, targets) in &metrics.peer_selection_distribution { + if targets.is_empty() { + continue; + } + let counts: Vec = targets.values().map(|&c| c as f64).collect(); + let chi2 = chi_squared_uniform(&counts); + if chi2 > worst_chi2 { + worst_chi2 = chi2; + worst_node = node.clone(); + } + } + PropertyResult { + name: "peer_selection_uniform".into(), + category: "Peer Selection".into(), + passed: worst_chi2 < chi_squared_critical, + expected: format!("χ² < {chi_squared_critical}"), + actual: format!("worst χ²={worst_chi2:.2} at {worst_node}"), + description: "Peer selection approximately uniform (chi-squared)".into(), + } +} + +pub fn check_lww_single_value(metrics: &GossipMetrics) -> PropertyResult { + let all_single = metrics + .final_value_divergence + .values() + .all(|&count| count == 1); + let details: Vec = metrics + .final_value_divergence + .iter() + .filter(|(_, c)| **c != 1) + .map(|(k, c)| format!("{k}:{c}")) + .collect(); + PropertyResult { + name: "lww_single_value".into(), + category: "Consistency".into(), + passed: all_single, + expected: "1 distinct value per key".into(), + actual: if all_single { + "all keys have 1 value".into() + } else { + format!("divergent: {:?}", details) + }, + description: "LWW ensures single final value per key".into(), + } +} + +pub fn check_entropy_zero_at_convergence( + metrics: &GossipMetrics, +) -> PropertyResult { + let passed = if let Some(cr) = metrics.convergence_round { + metrics + .entropy_per_round + .iter() + .skip(cr.saturating_sub(1)) + .all(|&e| e == 0) + } else { + false + }; + PropertyResult { + name: "entropy_zero_at_convergence".into(), + category: "Consistency".into(), + passed, + expected: "entropy = 0 after convergence".into(), + actual: format!( + "convergence_round={:?}, final_entropy={}", + metrics.convergence_round, + metrics.entropy_per_round.last().unwrap_or(&0) + ), + description: "Entropy reaches zero at convergence".into(), + } +} + +pub fn check_entropy_decreases(metrics: &GossipMetrics) -> PropertyResult { + let mono = metrics + .entropy_per_round + .windows(2) + .all(|w| w[1] <= w[0]); + PropertyResult { + name: "entropy_decreases".into(), + category: "Consistency".into(), + passed: mono, + expected: "monotonically non-increasing".into(), + actual: if mono { + "monotonic".into() + } else { + let violations: Vec = metrics + .entropy_per_round + .windows(2) + .enumerate() + .filter(|(_, w)| w[1] > w[0]) + .map(|(i, _)| i + 1) + .collect(); + format!("increases at rounds {:?}", violations) + }, + description: "Entropy never increases".into(), + } +} + +pub fn check_no_stale_reads(metrics: &GossipMetrics) -> PropertyResult { + // All keys have exactly 1 distinct value AND delivery_ratio == 1.0. + let all_single = metrics + .final_value_divergence + .values() + .all(|&c| c == 1); + let passed = all_single && (metrics.delivery_ratio - 1.0).abs() < 1e-9; + PropertyResult { + name: "no_stale_reads".into(), + category: "Consistency".into(), + passed, + expected: "all nodes agree post-convergence".into(), + actual: format!( + "delivery={}, all_single={}", + metrics.delivery_ratio, all_single + ), + description: "No stale reads after convergence".into(), + } +} + +pub fn check_state_size_stabilizes( + metrics: &GossipMetrics, + expected_final: f64, +) -> PropertyResult { + let final_avg = metrics.avg_state_size_per_round.last().copied().unwrap_or(0.0); + let passed = (final_avg - expected_final).abs() < 0.5; + PropertyResult { + name: "state_size_stabilizes".into(), + category: "Practical".into(), + passed, + expected: format!("{expected_final}"), + actual: format!("{final_avg:.2}"), + description: "Final average state size matches key count".into(), + } +} + +pub fn check_state_size_monotonic(metrics: &GossipMetrics) -> PropertyResult { + let mono = metrics + .avg_state_size_per_round + .windows(2) + .all(|w| w[1] >= w[0] - 1e-9); + PropertyResult { + name: "state_size_monotonic".into(), + category: "Practical".into(), + passed: mono, + expected: "non-decreasing".into(), + actual: if mono { + "monotonic".into() + } else { + "non-monotonic".into() + }, + description: "Average state size never decreases".into(), + } +} + +// ── Helper: base SimConfig ────────────────────────────────────────────────── + +pub fn base_config( + name: &str, + topology: Topology, + num_nodes: usize, + num_keys: usize, +) -> SimConfig { + SimConfig { + name: name.into(), + topology, + num_nodes, + initial_data: (0..num_keys) + .map(|i| (format!("key-{i}"), format!("value-{i}").into_bytes())) + .collect(), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + } +} + +// ── Statistical helpers ───────────────────────────────────────────────────── + +pub fn std_dev(values: &[f64]) -> f64 { + if values.is_empty() { + return 0.0; + } + let mean = values.iter().sum::() / values.len() as f64; + let variance = values.iter().map(|v| (v - mean).powi(2)).sum::() / values.len() as f64; + variance.sqrt() +} + +pub fn coeff_of_variation(values: &[f64]) -> f64 { + if values.is_empty() { + return 0.0; + } + let mean = values.iter().sum::() / values.len() as f64; + if mean.abs() < 1e-12 { + return 0.0; + } + std_dev(values) / mean +} + +pub fn chi_squared_uniform(observed: &[f64]) -> f64 { + if observed.is_empty() { + return 0.0; + } + let total: f64 = observed.iter().sum(); + let expected = total / observed.len() as f64; + if expected.abs() < 1e-12 { + return 0.0; + } + observed + .iter() + .map(|&o| (o - expected).powi(2) / expected) + .sum() +} diff --git a/crates/swactor-gossip/src/property_report.rs b/crates/swactor-gossip/src/property_report.rs new file mode 100644 index 0000000..f8c2f71 --- /dev/null +++ b/crates/swactor-gossip/src/property_report.rs @@ -0,0 +1,784 @@ +use crate::properties::{GossipMetrics, PropertyResult}; + +// ── Public API ────────────────────────────────────────────────────────────── + +/// A named scenario with its metrics and property results. +pub struct ScenarioReport { + pub name: String, + pub description: String, + pub metrics: GossipMetrics, + pub results: Vec, +} + +/// A section groups related scenarios under a category heading. +pub struct ReportSection { + pub title: String, + pub explanation: String, + pub scenarios: Vec, +} + +/// Data for the scalability scatter plot. +pub struct ScalingPoint { + pub n: usize, + pub convergence_round: Option, + pub total_pushes: usize, + pub label: String, +} + +/// Full report data. +pub struct PropertyReportData { + pub sections: Vec, + pub scaling_points_st: Vec, + pub scaling_points_mt: Vec, + pub thread_comparison: Vec, + /// All convergence curves keyed by scenario name, for the multi-line overlay. + pub convergence_overlays: Vec<(String, Vec)>, +} + +pub struct ThreadComparison { + pub label: String, + pub num_threads: usize, + pub convergence_round: Option, + pub total_pushes: usize, +} + +/// Generate a self-contained HTML report from the collected data. +pub fn generate_property_report(data: &PropertyReportData) -> String { + let mut html = String::with_capacity(128_000); + + html.push_str("\n\n\n\n"); + html.push_str("Gossip Protocol Property Verification Report\n"); + html.push_str("\n\n\n"); + + html.push_str("

Gossip Protocol Property Verification Report

\n"); + + // Executive summary. + render_executive_summary(&mut html, data); + + // Per-section content. + for section in &data.sections { + render_section(&mut html, section); + } + + // Convergence overlay chart. + if !data.convergence_overlays.is_empty() { + render_convergence_overlay(&mut html, &data.convergence_overlays); + } + + // Scalability charts. + if !data.scaling_points_st.is_empty() { + render_scalability_section(&mut html, data); + } + + // Thread comparison. + if !data.thread_comparison.is_empty() { + render_thread_comparison(&mut html, &data.thread_comparison); + } + + html.push_str("\n\n"); + html +} + +// ── CSS ───────────────────────────────────────────────────────────────────── + +const CSS: &str = r#" +body { + font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; + max-width: 1400px; margin: 0 auto; padding: 20px; + background: #fafafa; color: #222; +} +h1 { border-bottom: 3px solid #333; padding-bottom: 8px; } +h2 { margin-top: 40px; color: #333; border-bottom: 2px solid #ddd; padding-bottom: 4px; } +h3 { color: #555; margin-top: 24px; } +.summary-grid { display: flex; flex-wrap: wrap; gap: 16px; margin: 16px 0; } +.summary-card { + background: #fff; border: 1px solid #ddd; border-radius: 8px; + padding: 16px 24px; min-width: 160px; +} +.summary-card .label { font-size: 0.85em; color: #666; } +.summary-card .value { font-size: 1.8em; font-weight: bold; } +.badge-pass { display: inline-block; padding: 4px 12px; border-radius: 12px; background: #28a745; color: #fff; font-weight: bold; font-size: 0.9em; } +.badge-fail { display: inline-block; padding: 4px 12px; border-radius: 12px; background: #dc3545; color: #fff; font-weight: bold; font-size: 0.9em; } +.badge-partial { display: inline-block; padding: 4px 12px; border-radius: 12px; background: #ffc107; color: #333; font-weight: bold; font-size: 0.9em; } +table { border-collapse: collapse; width: 100%; margin: 12px 0; } +th, td { border: 1px solid #ddd; padding: 8px 12px; text-align: left; font-size: 0.9em; } +th { background: #f0f0f0; } +tr:nth-child(even) { background: #fafafa; } +.pass { color: #28a745; font-weight: bold; } +.fail { color: #dc3545; font-weight: bold; } +svg { display: block; margin: 12px 0; } +.explanation { color: #555; margin: 8px 0 16px 0; line-height: 1.5; } +.scenario-desc { color: #777; font-style: italic; margin: 4px 0 8px 0; } +"#; + +// ── Executive summary ─────────────────────────────────────────────────────── + +fn render_executive_summary(html: &mut String, data: &PropertyReportData) { + html.push_str("

Executive Summary

\n"); + + let mut total_pass = 0usize; + let mut total_fail = 0usize; + for section in &data.sections { + for scenario in §ion.scenarios { + for r in &scenario.results { + if r.passed { + total_pass += 1; + } else { + total_fail += 1; + } + } + } + } + let total = total_pass + total_fail; + + let badge = if total_fail == 0 { + "ALL PASSED" + } else if total_pass == 0 { + "ALL FAILED" + } else { + "PARTIAL" + }; + + html.push_str("
\n"); + summary_card(html, "Total Checks", &total.to_string()); + summary_card(html, "Passed", &total_pass.to_string()); + summary_card(html, "Failed", &total_fail.to_string()); + html.push_str(&format!( + "
Verdict
{badge}
\n" + )); + html.push_str("
\n"); +} + +fn summary_card(html: &mut String, label: &str, value: &str) { + html.push_str(&format!( + "
{label}
{value}
\n" + )); +} + +// ── Section rendering ─────────────────────────────────────────────────────── + +fn render_section(html: &mut String, section: &ReportSection) { + html.push_str(&format!("

{}

\n", esc(§ion.title))); + html.push_str(&format!( + "

{}

\n", + esc(§ion.explanation) + )); + + for scenario in §ion.scenarios { + html.push_str(&format!("

{}

\n", esc(&scenario.name))); + html.push_str(&format!( + "

{}

\n", + esc(&scenario.description) + )); + + // Results table. + html.push_str("\n\n"); + for r in &scenario.results { + let status = if r.passed { + "PASS" + } else { + "FAIL" + }; + html.push_str(&format!( + "\n", + esc(&r.name), + esc(&r.expected), + esc(&r.actual), + esc(&r.description), + )); + } + html.push_str("
PropertyStatusExpectedActualDescription
{}{status}{}{}{}
\n"); + + // Inline SVG graph for this scenario based on category. + render_scenario_graph(html, section, scenario); + } +} + +// ── Per-scenario graphs ───────────────────────────────────────────────────── + +fn render_scenario_graph(html: &mut String, section: &ReportSection, scenario: &ScenarioReport) { + let m = &scenario.metrics; + match section.title.as_str() { + "Convergence" | "Fault Tolerance" => { + render_convergence_curve_svg(html, &scenario.name, &m.convergence_curve); + } + "Consistency" => { + render_entropy_chart(html, &m.entropy_per_round); + } + "Practical" => { + render_state_size_chart(html, &m.avg_state_size_per_round); + } + "Bandwidth/Load" => { + render_load_bar_chart(html, m); + } + "Message Complexity" => { + render_message_stacked_bar(html, m); + } + "Peer Selection" => { + render_peer_histogram(html, m); + } + _ => {} + } +} + +// ── SVG chart helpers ─────────────────────────────────────────────────────── + +const CHART_W: f64 = 700.0; +const CHART_H: f64 = 280.0; +const ML: f64 = 60.0; // margin left +const MR: f64 = 20.0; +const MT: f64 = 20.0; +const MB: f64 = 50.0; + +fn svg_open(html: &mut String, w: f64, h: f64) { + html.push_str(&format!( + "\n" + )); +} + +fn svg_close(html: &mut String) { + html.push_str("\n"); +} + +fn draw_axes(html: &mut String) { + let bx = ML; + let by = MT + CHART_H; + let rx = ML + CHART_W - ML; + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "\n" + )); +} + +fn y_for(val: f64, max_val: f64) -> f64 { + if max_val < 1e-9 { + return MT + CHART_H; + } + MT + CHART_H - (val / max_val) * CHART_H +} + +fn x_for(idx: usize, total: usize) -> f64 { + if total == 0 { + return ML; + } + ML + (idx as f64 + 0.5) / total as f64 * (CHART_W - ML - MR) +} + +// ── Convergence curve SVG ─────────────────────────────────────────────────── + +fn render_convergence_curve_svg(html: &mut String, _name: &str, curve: &[f64]) { + if curve.is_empty() { + return; + } + let total_w = CHART_W + MR; + let total_h = CHART_H + MT + MB; + svg_open(html, total_w, total_h); + draw_axes(html); + + // Y-axis labels (0% to 100%). + for pct in [0, 25, 50, 75, 100] { + let y = y_for(pct as f64 / 100.0, 1.0); + html.push_str(&format!( + "{pct}%\n", + ML - 6.0, y + 3.0, + )); + html.push_str(&format!( + "\n", + ML + CHART_W - ML - MR, + )); + } + + // X-axis labels. + let step = (curve.len() / 10).max(1); + for r in (0..curve.len()).step_by(step) { + let x = x_for(r, curve.len()); + html.push_str(&format!( + "{}\n", + MT + CHART_H + 16.0, r + 1, + )); + } + + // Line. + let mut path = String::new(); + for (i, &v) in curve.iter().enumerate() { + let x = x_for(i, curve.len()); + let y = y_for(v, 1.0); + if i == 0 { + path.push_str(&format!("M{x:.1},{y:.1}")); + } else { + path.push_str(&format!(" L{x:.1},{y:.1}")); + } + } + html.push_str(&format!( + "\n" + )); + + // Dots. + let dot_step = (curve.len() / 30).max(1); + for (i, &v) in curve.iter().enumerate() { + if i % dot_step == 0 { + let x = x_for(i, curve.len()); + let y = y_for(v, 1.0); + html.push_str(&format!( + "\n" + )); + } + } + + svg_close(html); +} + +// ── Multi-line convergence overlay ────────────────────────────────────────── + +fn render_convergence_overlay(html: &mut String, curves: &[(String, Vec)]) { + html.push_str("

Convergence Comparison (All Topologies)

\n"); + html.push_str("

Overlay of convergence curves across different topologies at scale.

\n"); + + let max_len = curves.iter().map(|(_, c)| c.len()).max().unwrap_or(0); + if max_len == 0 { + return; + } + + let total_w = CHART_W + MR; + let total_h = CHART_H + MT + MB + 40.0; // extra for legend + svg_open(html, total_w, total_h); + draw_axes(html); + + let colors = ["#4a90d9", "#d94a4a", "#4ad94a", "#d9a64a", "#9a4ad9", "#4ad9d9"]; + + for pct in [0, 25, 50, 75, 100] { + let y = y_for(pct as f64 / 100.0, 1.0); + html.push_str(&format!( + "{pct}%\n", + ML - 6.0, y + 3.0, + )); + } + + for (ci, (name, curve)) in curves.iter().enumerate() { + let color = colors[ci % colors.len()]; + let mut path = String::new(); + for (i, &v) in curve.iter().enumerate() { + let x = x_for(i, max_len); + let y = y_for(v, 1.0); + if i == 0 { + path.push_str(&format!("M{x:.1},{y:.1}")); + } else { + path.push_str(&format!(" L{x:.1},{y:.1}")); + } + } + html.push_str(&format!( + "\n" + )); + + // Legend entry. + let lx = ML + ci as f64 * 140.0; + let ly = MT + CHART_H + 36.0; + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "{name}\n", + lx + 18.0, ly + 9.0, + )); + } + + svg_close(html); +} + +// ── Entropy chart ─────────────────────────────────────────────────────────── + +fn render_entropy_chart(html: &mut String, entropy: &[usize]) { + if entropy.is_empty() { + return; + } + let max_e = *entropy.iter().max().unwrap_or(&1) as f64; + let total_w = CHART_W + MR; + let total_h = CHART_H + MT + MB; + svg_open(html, total_w, total_h); + draw_axes(html); + + html.push_str(&format!( + "{}\n", + ML - 6.0, MT + 3.0, max_e as usize, + )); + html.push_str(&format!( + "0\n", + ML - 6.0, MT + CHART_H + 3.0, + )); + + let mut path = String::new(); + for (i, &e) in entropy.iter().enumerate() { + let x = x_for(i, entropy.len()); + let y = y_for(e as f64, max_e); + if i == 0 { + path.push_str(&format!("M{x:.1},{y:.1}")); + } else { + path.push_str(&format!(" L{x:.1},{y:.1}")); + } + } + html.push_str(&format!( + "\n" + )); + + svg_close(html); +} + +// ── State size chart ──────────────────────────────────────────────────────── + +fn render_state_size_chart(html: &mut String, sizes: &[f64]) { + if sizes.is_empty() { + return; + } + let max_s = sizes.iter().cloned().fold(0.0f64, f64::max).max(1.0); + let total_w = CHART_W + MR; + let total_h = CHART_H + MT + MB; + svg_open(html, total_w, total_h); + draw_axes(html); + + html.push_str(&format!( + "{:.1}\n", + ML - 6.0, MT + 3.0, max_s, + )); + + let mut path = String::new(); + for (i, &s) in sizes.iter().enumerate() { + let x = x_for(i, sizes.len()); + let y = y_for(s, max_s); + if i == 0 { + path.push_str(&format!("M{x:.1},{y:.1}")); + } else { + path.push_str(&format!(" L{x:.1},{y:.1}")); + } + } + html.push_str(&format!( + "\n" + )); + + svg_close(html); +} + +// ── Load bar chart ────────────────────────────────────────────────────────── + +fn render_load_bar_chart(html: &mut String, metrics: &GossipMetrics) { + let mut nodes: Vec<(&String, usize)> = metrics + .pushes_received_per_node + .iter() + .map(|(n, &c)| (n, c)) + .collect(); + nodes.sort_by(|a, b| b.1.cmp(&a.1)); + // Show top 20 nodes. + nodes.truncate(20); + + if nodes.is_empty() { + return; + } + let max_v = nodes[0].1 as f64; + let bar_h = 18.0; + let gap = 4.0; + let total_h = MT + (bar_h + gap) * nodes.len() as f64 + MB; + let total_w = CHART_W + MR; + svg_open(html, total_w, total_h); + + for (i, (name, count)) in nodes.iter().enumerate() { + let y = MT + i as f64 * (bar_h + gap); + let w = if max_v > 0.0 { + (*count as f64 / max_v) * (CHART_W - ML - MR - 40.0) + } else { + 0.0 + }; + html.push_str(&format!( + "{name}\n", + ML - 4.0, y + bar_h - 4.0, + )); + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "{count}\n", + ML + w + 4.0, y + bar_h - 4.0, + )); + } + + svg_close(html); +} + +// ── Message stacked bar ───────────────────────────────────────────────────── + +fn render_message_stacked_bar(html: &mut String, metrics: &GossipMetrics) { + let useful = metrics.total_pushes - metrics.redundant_pushes; + let redundant = metrics.redundant_pushes; + let total = metrics.total_pushes.max(1) as f64; + + let total_w = 400.0; + let total_h = 80.0; + svg_open(html, total_w, total_h); + + let bar_w = 300.0; + let bar_h = 30.0; + let y = 20.0; + let x = 60.0; + + let useful_w = (useful as f64 / total) * bar_w; + let redundant_w = (redundant as f64 / total) * bar_w; + + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "\n", + x + useful_w, + )); + + // Legend. + let ly = y + bar_h + 16.0; + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "Useful ({useful})\n", + x + 16.0, ly + 9.0, + )); + html.push_str(&format!( + "\n", + x + 140.0, + )); + html.push_str(&format!( + "Redundant ({redundant})\n", + x + 156.0, ly + 9.0, + )); + + svg_close(html); +} + +// ── Peer selection histogram ──────────────────────────────────────────────── + +fn render_peer_histogram(html: &mut String, metrics: &GossipMetrics) { + // Aggregate: for each target, total selection count across all nodes. + let mut target_totals: std::collections::HashMap = std::collections::HashMap::new(); + for targets in metrics.peer_selection_distribution.values() { + for (target, &count) in targets { + *target_totals.entry(target.clone()).or_default() += count; + } + } + let mut sorted: Vec<(String, usize)> = target_totals.into_iter().collect(); + sorted.sort_by(|a, b| a.0.cmp(&b.0)); + + if sorted.is_empty() { + return; + } + + let max_v = sorted.iter().map(|(_, c)| *c).max().unwrap_or(1) as f64; + let bar_w = 30.0; + let gap = 4.0; + let total_w = ML + (bar_w + gap) * sorted.len() as f64 + MR; + let total_h = CHART_H + MT + MB; + svg_open(html, total_w, total_h); + + // Axes. + let base_y = MT + CHART_H; + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "\n", + total_w - MR, + )); + + for (i, (name, count)) in sorted.iter().enumerate() { + let x = ML + i as f64 * (bar_w + gap); + let h = (*count as f64 / max_v) * CHART_H; + let y = base_y - h; + html.push_str(&format!( + "\n" + )); + // Label. + html.push_str(&format!( + "{name}\n", + x + bar_w / 2.0, base_y + 14.0, x + bar_w / 2.0, base_y + 14.0, + )); + // Count on top. + html.push_str(&format!( + "{count}\n", + x + bar_w / 2.0, y - 3.0, + )); + } + + svg_close(html); +} + +// ── Scalability section ───────────────────────────────────────────────────── + +fn render_scalability_section(html: &mut String, data: &PropertyReportData) { + html.push_str("

Scalability

\n"); + html.push_str("

How convergence time and message count scale with network size.

\n"); + + // Convergence round vs N (with O(log N) reference). + html.push_str("

Convergence Time vs Network Size

\n"); + render_scaling_scatter( + html, + &data.scaling_points_st, + &data.scaling_points_mt, + true, + ); + + // Total messages vs N (with O(N) reference). + html.push_str("

Total Messages vs Network Size

\n"); + render_scaling_scatter( + html, + &data.scaling_points_st, + &data.scaling_points_mt, + false, + ); +} + +fn render_scaling_scatter( + html: &mut String, + st_points: &[ScalingPoint], + mt_points: &[ScalingPoint], + is_convergence: bool, +) { + let all_n: Vec = st_points + .iter() + .chain(mt_points.iter()) + .map(|p| p.n) + .collect(); + let all_y: Vec = st_points + .iter() + .chain(mt_points.iter()) + .map(|p| { + if is_convergence { + p.convergence_round.unwrap_or(0) as f64 + } else { + p.total_pushes as f64 + } + }) + .collect(); + + if all_n.is_empty() { + return; + } + + let max_n = *all_n.iter().max().unwrap() as f64; + let max_y = all_y.iter().cloned().fold(0.0f64, f64::max).max(1.0); + + let total_w = CHART_W + MR; + let total_h = CHART_H + MT + MB + 30.0; + svg_open(html, total_w, total_h); + draw_axes(html); + + // Reference line. + let ref_color = "#ccc"; + let ref_points = 50; + let mut ref_path = String::new(); + for i in 0..=ref_points { + let n = (i as f64 / ref_points as f64) * max_n; + let ref_y_val = if is_convergence { + // O(log N) reference scaled to fit. + if n > 1.0 { + (n.ln() / max_n.ln()) * max_y + } else { + 0.0 + } + } else { + // O(N) reference. + (n / max_n) * max_y + }; + let x = ML + (n / max_n) * (CHART_W - ML - MR); + let y = y_for(ref_y_val, max_y); + if i == 0 { + ref_path.push_str(&format!("M{x:.1},{y:.1}")); + } else { + ref_path.push_str(&format!(" L{x:.1},{y:.1}")); + } + } + html.push_str(&format!( + "\n" + )); + let ref_label = if is_convergence { "O(log N)" } else { "O(N)" }; + html.push_str(&format!( + "{ref_label}\n", + ML + CHART_W - ML - MR - 50.0, MT + 14.0, + )); + + // Single-threaded points. + for p in st_points { + let x = ML + (p.n as f64 / max_n) * (CHART_W - ML - MR); + let yv = if is_convergence { + p.convergence_round.unwrap_or(0) as f64 + } else { + p.total_pushes as f64 + }; + let y = y_for(yv, max_y); + html.push_str(&format!( + "\n" + )); + } + + // Multi-threaded points. + for p in mt_points { + let x = ML + (p.n as f64 / max_n) * (CHART_W - ML - MR); + let yv = if is_convergence { + p.convergence_round.unwrap_or(0) as f64 + } else { + p.total_pushes as f64 + }; + let y = y_for(yv, max_y); + html.push_str(&format!( + "\n" + )); + } + + // Legend. + let ly = MT + CHART_H + 30.0; + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "Single-threaded\n", + ML + 10.0, ly + 4.0, + )); + html.push_str(&format!( + "\n", + ML + 130.0, + )); + html.push_str(&format!( + "Multi-threaded\n", + ML + 140.0, ly + 4.0, + )); + + svg_close(html); +} + +// ── Thread comparison ─────────────────────────────────────────────────────── + +fn render_thread_comparison(html: &mut String, comparisons: &[ThreadComparison]) { + html.push_str("

Thread-Mode Comparison

\n"); + html.push_str("

Comparing convergence time and message counts across different thread configurations.

\n"); + + html.push_str("\n\n"); + for tc in comparisons { + html.push_str(&format!( + "\n", + esc(&tc.label), + tc.num_threads, + tc.convergence_round + .map(|r| r.to_string()) + .unwrap_or("never".into()), + tc.total_pushes, + )); + } + html.push_str("
ConfigurationThreadsConvergence RoundTotal Pushes
{}{}{}{}
\n"); +} + +// ── HTML escape ───────────────────────────────────────────────────────────── + +fn esc(s: &str) -> String { + s.replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) +} diff --git a/crates/swactor-gossip/src/protocol.rs b/crates/swactor-gossip/src/protocol.rs index bc80775..9a57e57 100644 --- a/crates/swactor-gossip/src/protocol.rs +++ b/crates/swactor-gossip/src/protocol.rs @@ -1,12 +1,13 @@ use std::collections::HashMap; +use log::{debug, trace}; use swactor::actor::{ActorAddress, ActorInterface, Ctx}; use crate::trace::{GossipEvent, GossipEventKind, NodeSnapshot, TraceContext}; // ── VersionedValue ─────────────────────────────────────────────────────── -#[derive(Debug, Clone)] +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct VersionedValue { pub value: Vec, pub version: u64, @@ -153,11 +154,30 @@ impl GossipActor { tick: trace.current_tick(), node_name: trace.resolve_name(addr), node_addr: addr, + thread_name: std::thread::current().name().map(|s| s.to_owned()), kind, }; trace.record_event(event); } } + + fn self_name(&self, addr: ActorAddress) -> String { + self.trace + .as_ref() + .map(|t| t.resolve_name(addr)) + .unwrap_or_else(|| format!("{:?}", &addr.0[..4])) + } + + fn peer_name(&self, addr: ActorAddress) -> String { + self.self_name(addr) + } + + fn round(&self) -> u64 { + self.trace + .as_ref() + .map(|t| t.current_tick()) + .unwrap_or(0) + } } impl Default for GossipActor { @@ -175,15 +195,18 @@ impl ActorInterface for GossipActor { match msg { GossipMessage::AddPeer(addr) => { if !self.peers.contains(&addr) { + let pname = self.peer_name(addr); + debug!( + "[{}] inbound AddPeer peer={} (total_peers={})", + self.self_name(self_addr), + pname, + self.peers.len() + 1, + ); self.peers.push(addr); self.record( self_addr, GossipEventKind::PeerAdded { - peer_name: self - .trace - .as_ref() - .map(|t| t.resolve_name(addr)) - .unwrap_or_default(), + peer_name: pname, }, ); } @@ -192,32 +215,46 @@ impl ActorInterface for GossipActor { let before = self.peers.len(); self.peers.retain(|a| *a != addr); if self.peers.len() < before { + let pname = self.peer_name(addr); + debug!( + "[{}] inbound RemovePeer peer={} (total_peers={})", + self.self_name(self_addr), + pname, + self.peers.len(), + ); self.record( self_addr, GossipEventKind::PeerRemoved { - peer_name: self - .trace - .as_ref() - .map(|t| t.resolve_name(addr)) - .unwrap_or_default(), + peer_name: pname, }, ); } } GossipMessage::Set { key, value } => { + debug!( + "[{}] inbound Set key={:?} value_len={}", + self.self_name(self_addr), + key, + value.len(), + ); self.state.set(key.clone(), value); self.record(self_addr, GossipEventKind::LocalSet { key }); } GossipMessage::DoGossipRound => { + let round = self.round(); if let Some(peer) = self.pick_random_peer() { + let target = self.peer_name(peer); + debug!( + "[{}] round={} outbound Push -> {} (state_keys={})", + self.self_name(self_addr), + round, + target, + self.state.entries().len(), + ); self.record( self_addr, GossipEventKind::GossipRoundStarted { - target_name: self - .trace - .as_ref() - .map(|t| t.resolve_name(peer)) - .unwrap_or_default(), + target_name: target, }, ); let _ = ctx.send( @@ -228,6 +265,11 @@ impl ActorInterface for GossipActor { }, ); } else { + debug!( + "[{}] round={} no peers — skipping gossip", + self.self_name(self_addr), + round, + ); self.record(self_addr, GossipEventKind::GossipRoundNoPeers); } } @@ -235,20 +277,31 @@ impl ActorInterface for GossipActor { from, state: remote, } => { + let from_name = self.peer_name(from); let keys_updated = self.state.merge(&remote); + debug!( + "[{}] round={} inbound Push <- {} keys_updated={} (state_keys={})", + self.self_name(self_addr), + self.round(), + from_name, + keys_updated, + self.state.entries().len(), + ); self.record( self_addr, GossipEventKind::PushReceived { - from_name: self - .trace - .as_ref() - .map(|t| t.resolve_name(from)) - .unwrap_or_default(), + from_name, keys_updated, }, ); } GossipMessage::TakeSnapshot => { + trace!( + "[{}] round={} snapshot (keys={})", + self.self_name(self_addr), + self.round(), + self.state.entries().len(), + ); self.record( self_addr, GossipEventKind::StateSnapshot { @@ -260,16 +313,27 @@ impl ActorInterface for GossipActor { ); } GossipMessage::Query { key, reply_to } => { + let entry = self.state.get(&key); + debug!( + "[{}] inbound Query key={:?} found={}", + self.self_name(self_addr), + key, + entry.is_some(), + ); self.record( self_addr, GossipEventKind::QueryReceived { key: key.clone() }, ); - let entry = self.state.get(&key); let resp = GossipQueryResponse { key, value: entry.map(|e| e.value.clone()), version: entry.map(|e| e.version), }; + debug!( + "[{}] outbound QueryResponse -> {:?}", + self.self_name(self_addr), + &reply_to.0[..4], + ); let _ = ctx.send(reply_to, resp); } } diff --git a/crates/swactor-gossip/src/sim.rs b/crates/swactor-gossip/src/sim.rs index 4e6e04e..3defecf 100644 --- a/crates/swactor-gossip/src/sim.rs +++ b/crates/swactor-gossip/src/sim.rs @@ -1,6 +1,8 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; use swactor::actor::ActorAddress; use swactor::config::RuntimeConfig; @@ -38,17 +40,31 @@ pub struct SimConfig { pub ticks_per_round: usize, /// If `Some(r)`, cross-partition links are added after round `r`. pub heal_after_round: Option, + /// Number of worker threads: 1 = deterministic single-threaded, >1 = multi-threaded. + pub num_threads: usize, } // ── Public entry point ─────────────────────────────────────────────────── pub fn run_simulation(config: SimConfig) -> SimulationTrace { + let num_threads = config.num_threads.max(1); + + if num_threads < 2 { + run_simulation_single_threaded(config) + } else { + run_simulation_multi_threaded(config) + } +} + +fn run_simulation_single_threaded(config: SimConfig) -> SimulationTrace { 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 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() }); @@ -149,9 +165,152 @@ pub fn run_simulation(config: SimConfig) -> SimulationTrace { } } +fn run_simulation_multi_threaded(config: SimConfig) -> SimulationTrace { + 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 ticks_per_round = config.ticks_per_round; + // Safety factor for non-deterministic scheduling: allow more settle time. + 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() + }); + + // Spawn nodes. + 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); + } + + // Wire topology — send AddPeer messages before starting worker threads. + let edges = wire_topology(&rt, &config.topology, &addrs, &names); + + // Set initial data on node 0 before starting. + 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(); + } + + // Start worker threads — consumes `rt`, returns handle. + let handle = rt.run().expect("failed to start multi-threaded runtime"); + + // Let initial messages (spawn + AddPeer + Set) settle. + thread::sleep(Duration::from_millis(settle_ms * 2)); + + // Run gossip rounds using sleep-based ticking. + let mut snapshots_per_round: Vec> = Vec::new(); + + for round in 0..config.num_rounds { + // Heal partition if needed. + 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); + + // Trigger gossip on all nodes. + for &addr in &addrs { + handle + .runtime + .send_to(addr, GossipMessage::DoGossipRound) + .unwrap(); + } + + // Let gossip messages propagate. + thread::sleep(Duration::from_millis(settle_ms)); + + // Take snapshots. + for &addr in &addrs { + handle + .runtime + .send_to(addr, GossipMessage::TakeSnapshot) + .unwrap(); + } + thread::sleep(Duration::from_millis(settle_ms / 2)); + + // Extract snapshots from event log. + 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); + } + + // Shutdown worker threads. + handle.shutdown(); + handle.join(); + + 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, + } +} + +pub fn heal_partition_via_handle( + handle: &swactor::runtime::RuntimeHandle, + topology: &Topology, + addrs: &[ActorAddress], + _names: &[String], +) { + if !matches!(topology, Topology::Partitioned) { + return; + } + let n = addrs.len(); + let half = n / 2; + if half > 0 && half < n { + handle + .runtime + .send_to(addrs[half - 1], GossipMessage::AddPeer(addrs[half])) + .unwrap(); + handle + .runtime + .send_to(addrs[half], GossipMessage::AddPeer(addrs[half - 1])) + .unwrap(); + } +} + // ── Topology wiring ────────────────────────────────────────────────────── -fn wire_topology( +pub fn wire_topology( rt: &Runtime, topology: &Topology, addrs: &[ActorAddress], @@ -214,7 +373,7 @@ fn wire_topology( edges } -fn heal_partition( +pub fn heal_partition( rt: &Runtime, topology: &Topology, addrs: &[ActorAddress], diff --git a/crates/swactor-gossip/src/trace.rs b/crates/swactor-gossip/src/trace.rs index 47a5e9d..9b0fd8a 100644 --- a/crates/swactor-gossip/src/trace.rs +++ b/crates/swactor-gossip/src/trace.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; +use serde::{Deserialize, Serialize}; use swactor::actor::ActorAddress; use crate::protocol::VersionedValue; @@ -47,15 +48,16 @@ impl TraceContext { // ── Event types ────────────────────────────────────────────────────────── -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct GossipEvent { pub tick: u64, pub node_name: String, pub node_addr: ActorAddress, + pub thread_name: Option, pub kind: GossipEventKind, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum GossipEventKind { /// A local `Set { key, .. }` was processed. LocalSet { key: String }, @@ -78,7 +80,7 @@ pub enum GossipEventKind { StateSnapshot { snapshot: NodeSnapshot }, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct NodeSnapshot { pub entries: HashMap, pub peer_count: usize, @@ -86,7 +88,7 @@ pub struct NodeSnapshot { // ── Simulation trace (complete run output) ─────────────────────────────── -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct SimulationTrace { pub name: String, pub node_names: Vec, diff --git a/crates/swactor-gossip/tests/gossip_properties.rs b/crates/swactor-gossip/tests/gossip_properties.rs new file mode 100644 index 0000000..86c7097 --- /dev/null +++ b/crates/swactor-gossip/tests/gossip_properties.rs @@ -0,0 +1,825 @@ +use swactor_gossip::properties::*; +use swactor_gossip::sim::{run_simulation, SimConfig, Topology}; +use swactor_gossip::trace::SimulationTrace; + +// ── Helpers ───────────────────────────────────────────────────────────────── + +fn test_data(n: usize) -> Vec<(String, Vec)> { + (0..n) + .map(|i| (format!("key-{i}"), format!("value-{i}").into_bytes())) + .collect() +} + +fn run_and_analyze(config: SimConfig) -> (SimulationTrace, GossipMetrics) { + let trace = run_simulation(config); + let metrics = analyze(&trace); + (trace, metrics) +} + +// ── Reliability (3) ───────────────────────────────────────────────────────── + +#[test] +fn all_nodes_receive_all_keys_in_ring_1000() { + // FullMesh 100 nodes converges in ~O(log N) rounds, well within 30 rounds. + let config = SimConfig { + name: "fullmesh-100".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + assert!( + (metrics.delivery_ratio - 1.0).abs() < 1e-9, + "delivery_ratio = {}, expected 1.0", + metrics.delivery_ratio + ); +} + +#[test] +fn all_nodes_receive_all_keys_in_star_1000() { + // Full-mesh at 100 nodes: each node picks 1 of 99 peers, so with parallel + // spreading from all nodes, convergence is fast (O(log N) rounds). + let config = SimConfig { + name: "fullmesh-100".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + assert!( + (metrics.delivery_ratio - 1.0).abs() < 1e-9, + "delivery_ratio = {}, expected 1.0", + metrics.delivery_ratio + ); +} + +#[test] +fn delivery_is_all_or_nothing_per_key() { + // Full-mesh converges fast — O(log N). After convergence, each key is + // held by all nodes (atomic delivery). + let config = SimConfig { + name: "atomic-fullmesh-100".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(4), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + assert!( + metrics.atomic_delivery, + "atomic_delivery should be true" + ); +} + +// ── Latency (3) ───────────────────────────────────────────────────────────── + +#[test] +fn ring_converges_within_bound() { + // Ring with N=1000 should converge within N rounds. + let n = 1000; + let config = SimConfig { + name: "ring-latency".into(), + topology: Topology::Ring, + num_nodes: n, + initial_data: test_data(5), + num_rounds: n, // give it N rounds + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_convergence_bound(&metrics, n); + assert!(result.passed, "ring convergence: {}", result.actual); +} + +#[test] +fn fullmesh_converges_in_log_n_rounds() { + // Full-mesh: all nodes spread in parallel, O(log N) convergence. + let n = 100; + let bound = 4 * ((n as f64).ln().ceil() as usize); // ≈ 20 + let config = SimConfig { + name: "fullmesh-latency".into(), + topology: Topology::FullMesh, + num_nodes: n, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_convergence_bound(&metrics, bound); + assert!(result.passed, "fullmesh convergence: {}", result.actual); +} + +#[test] +fn last_node_latency_bounded_in_fullmesh() { + // In full-mesh, last node converges close to overall convergence. + let config = SimConfig { + name: "fullmesh-last-node".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_last_node_latency(&metrics, 5); + assert!(result.passed, "last node latency: {}", result.actual); +} + +// ── Message Complexity (3) ────────────────────────────────────────────────── + +#[test] +fn total_messages_equal_n_times_rounds() { + let n = 1000; + let r = 30; + let config = SimConfig { + name: "msg-count".into(), + topology: Topology::Ring, + num_nodes: n, + initial_data: test_data(5), + num_rounds: r, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + // In a ring, every node has exactly 1 peer, so each node sends exactly 1 push per round. + let expected = n * r; + let result = check_total_pushes_eq(&metrics, expected); + assert!(result.passed, "total_pushes: {}", result.actual); +} + +#[test] +fn redundancy_increases_after_convergence() { + // Full-mesh 100 nodes: converges in ~10 rounds, run 50 → lots of redundant pushes. + let config = SimConfig { + name: "redundancy-fullmesh".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 50, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_redundancy_above(&metrics, 0.3); + assert!(result.passed, "redundancy: {}", result.actual); +} + +#[test] +fn chain_has_minimal_waste() { + // Chain topology: data flows one direction, minimal redundancy until convergence. + // Compare chain's redundancy ratio to a denser topology's. + let n = 100; + let rounds = 120; + + let chain_config = SimConfig { + name: "chain-waste".into(), + topology: Topology::Chain, + num_nodes: n, + initial_data: test_data(1), + num_rounds: rounds, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, chain_metrics) = run_and_analyze(chain_config); + + let fullmesh_config = SimConfig { + name: "fullmesh-waste".into(), + topology: Topology::FullMesh, + num_nodes: n, + initial_data: test_data(1), + num_rounds: rounds, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, fullmesh_metrics) = run_and_analyze(fullmesh_config); + + // Chain should have lower redundancy ratio than full-mesh. + assert!( + chain_metrics.redundancy_ratio < fullmesh_metrics.redundancy_ratio, + "chain redundancy ({:.3}) should be less than fullmesh ({:.3})", + chain_metrics.redundancy_ratio, + fullmesh_metrics.redundancy_ratio + ); +} + +// ── Bandwidth/Load (3) ────────────────────────────────────────────────────── + +#[test] +fn star_hub_is_hotspot() { + // Star with 100 nodes, 30 rounds: node-0 receives pushes from all leaves. + let config = SimConfig { + name: "star-hub".into(), + topology: Topology::Star, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_hub_is_hotspot(&metrics, "node-0"); + assert!(result.passed, "hub hotspot: {}", result.actual); +} + +#[test] +fn ring_distributes_load_evenly() { + let config = SimConfig { + name: "ring-load".into(), + topology: Topology::Ring, + num_nodes: 1000, + initial_data: test_data(5), + num_rounds: 60, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_load_balance_cv(&metrics, 0.3); + assert!(result.passed, "load CV: {}", result.actual); +} + +#[test] +fn amplification_equals_num_rounds() { + let n = 1000; + let r = 30; + let config = SimConfig { + name: "ring-amp".into(), + topology: Topology::Ring, + num_nodes: n, + initial_data: test_data(5), + num_rounds: r, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_amplification(&metrics, r as f64, 1.0); + assert!(result.passed, "amplification: {}", result.actual); +} + +// ── Convergence (3) ───────────────────────────────────────────────────────── + +#[test] +fn convergence_curve_is_monotonic() { + let config = SimConfig { + name: "ring-mono".into(), + topology: Topology::Ring, + num_nodes: 1000, + initial_data: test_data(5), + num_rounds: 60, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_curve_monotonic(&metrics); + assert!(result.passed, "monotonic: {}", result.actual); +} + +#[test] +fn convergence_curve_has_s_shape() { + // Full-mesh 100 nodes: starts at 0, ramps up quickly, reaches 1.0 → S-shaped. + let config = SimConfig { + name: "fullmesh-s-shape".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_curve_s_shape(&metrics); + assert!(result.passed, "s-shape: {}", result.actual); +} + +#[test] +fn zero_residue_after_sufficient_rounds() { + // FullMesh 100 converges in ~O(log N) rounds; 30 rounds is plenty. + let config = SimConfig { + name: "fullmesh-residue".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_zero_residue(&metrics); + assert!(result.passed, "residue: {}", result.actual); +} + +// ── Fault Tolerance (3) ───────────────────────────────────────────────────── + +#[test] +fn partitioned_network_does_not_converge() { + let config = SimConfig { + name: "partition-no-heal".into(), + topology: Topology::Partitioned, + num_nodes: 1000, + initial_data: test_data(5), + num_rounds: 40, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_partition_no_converge(&metrics); + assert!(result.passed, "partition no converge: {}", result.actual); +} + +#[test] +fn partition_heals_and_converges() { + // Partitioned 100 = two halves of 50 nodes, each full-mesh internally. + // Each half converges in O(50*ln(50)) ~ 200 rounds. Heal at round 100, + // run 300 total to allow full convergence after healing. + let config = SimConfig { + name: "partition-heal".into(), + topology: Topology::Partitioned, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 300, + ticks_per_round: 4, + heal_after_round: Some(100), + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_partition_heals(&metrics); + assert!(result.passed, "partition heals: {}", result.actual); +} + +#[test] +fn partial_convergence_before_healing() { + // Partitioned 100 = two halves of 50 nodes, each full-mesh internally. + // Each half converges in O(50*ln(50)) ~ 200 rounds. Heal at round 100, + // run 300 total to allow full convergence after healing. + let config = SimConfig { + name: "partition-partial".into(), + topology: Topology::Partitioned, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 300, + ticks_per_round: 4, + heal_after_round: Some(100), + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_partial_before_heal(&metrics, 100); + assert!(result.passed, "partial before heal: {}", result.actual); +} + +// ── Scalability (2) ───────────────────────────────────────────────────────── + +#[test] +fn convergence_time_scales_sublinearly() { + // FullMesh convergence is O(log N), which IS sublinear. + // Ring convergence is O(N), which is linear -- not suitable for this test. + let sizes = [100, 250, 500, 1000]; + let mut data = Vec::new(); + for &n in &sizes { + let rounds = 60; // O(log N) means even 1000 nodes converges in ~30 rounds + let config = SimConfig { + name: format!("scale-{n}"), + topology: Topology::FullMesh, + num_nodes: n, + initial_data: test_data(5), + num_rounds: rounds, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let cr = metrics.convergence_round.unwrap_or(rounds); + data.push((n, cr)); + } + let result = check_sublinear_scaling(&data); + assert!(result.passed, "sublinear scaling: {}", result.actual); +} + +#[test] +fn total_messages_scale_linearly_with_n() { + let sizes = [100, 250, 500, 1000]; + let fixed_rounds = 30; + let mut data = Vec::new(); + for &n in &sizes { + let config = SimConfig { + name: format!("msg-scale-{n}"), + topology: Topology::Ring, + num_nodes: n, + initial_data: test_data(5), + num_rounds: fixed_rounds, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + data.push((n, metrics.total_pushes)); + } + let result = check_linear_message_scaling(&data, fixed_rounds); + assert!(result.passed, "linear message scaling: {}", result.actual); +} + +// ── Push Protocol (2) ─────────────────────────────────────────────────────── + +#[test] +fn one_push_per_node_per_round() { + let n = 500; + let r = 10; + let config = SimConfig { + name: "push-protocol".into(), + topology: Topology::Ring, + num_nodes: n, + initial_data: test_data(5), + num_rounds: r, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_one_push_per_node_per_round(&metrics, r); + assert!(result.passed, "one push per round: {}", result.actual); +} + +#[test] +fn no_push_without_peers() { + let n = 100; + let config = SimConfig { + name: "no-push-chain".into(), + topology: Topology::Chain, + num_nodes: n, + initial_data: test_data(1), + num_rounds: 20, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (trace, _) = run_and_analyze(config); + // Last node in chain has no peers. + let last_node = format!("node-{}", n - 1); + let result = check_no_push_without_peers(&trace, &last_node); + assert!(result.passed, "no push without peers: {}", result.actual); +} + +// ── Peer Selection (1) ────────────────────────────────────────────────────── + +#[test] +fn peer_selection_is_approximately_uniform() { + // Ring with 10 nodes: each node has 1 peer (the next in ring). + // With only 1 peer, chi-squared is trivially 0 (always picks the same). + // Use a wider ring: give each node 2 peers (bidirectional ring). + // Actually, ring topology only adds 1 peer (next). We need a small full-mesh or star. + // Use a star with 10 nodes: node-0 has 9 peers (nodes 1-9). + // Over 500 rounds, node-0 should select each peer ~55 times. + let config = SimConfig { + name: "peer-selection".into(), + topology: Topology::Star, + num_nodes: 10, + initial_data: test_data(1), + num_rounds: 500, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + // Chi-squared critical value for df=8 (9 peers - 1), p=0.001 is ~26.12. + let result = check_peer_selection_uniform(&metrics, 26.12); + assert!(result.passed, "peer selection: {}", result.actual); +} + +// ── Topology Impact (2) ──────────────────────────────────────────────────── + +#[test] +fn denser_topology_converges_faster() { + let n = 100; + let keys = 5; + let rounds = 120; // enough for chain + + let topologies = vec![ + ("FullMesh", Topology::FullMesh), + ("Star", Topology::Star), + ("Ring", Topology::Ring), + ("Chain", Topology::Chain), + ]; + + let mut convergence_times = Vec::new(); + for (name, topo) in &topologies { + let config = SimConfig { + name: format!("topo-{name}"), + topology: topo.clone(), + num_nodes: n, + initial_data: test_data(keys), + num_rounds: rounds, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + convergence_times.push((*name, metrics.convergence_round.unwrap_or(rounds + 1))); + } + + // FullMesh should be fastest (smallest convergence round). + let fullmesh_time = convergence_times + .iter() + .find(|(n, _)| *n == "FullMesh") + .unwrap() + .1; + let chain_time = convergence_times + .iter() + .find(|(n, _)| *n == "Chain") + .unwrap() + .1; + + assert!( + fullmesh_time < chain_time, + "FullMesh ({}) should converge before Chain ({})", + fullmesh_time, + chain_time + ); +} + +#[test] +fn sparser_topology_is_more_efficient() { + let n = 100; + let keys = 5; + let rounds = 120; + + let topologies = vec![ + ("FullMesh", Topology::FullMesh), + ("Ring", Topology::Ring), + ("Chain", Topology::Chain), + ]; + + let mut redundancy_ratios = Vec::new(); + for (name, topo) in &topologies { + let config = SimConfig { + name: format!("eff-{name}"), + topology: topo.clone(), + num_nodes: n, + initial_data: test_data(keys), + num_rounds: rounds, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + redundancy_ratios.push((*name, metrics.redundancy_ratio)); + } + + let fullmesh_r = redundancy_ratios + .iter() + .find(|(n, _)| *n == "FullMesh") + .unwrap() + .1; + let chain_r = redundancy_ratios + .iter() + .find(|(n, _)| *n == "Chain") + .unwrap() + .1; + + assert!( + chain_r < fullmesh_r, + "Chain redundancy ({:.3}) should be lower than FullMesh ({:.3})", + chain_r, + fullmesh_r + ); +} + +// ── Consistency (4) ───────────────────────────────────────────────────────── + +#[test] +fn lww_ensures_single_final_value() { + // Full-mesh 100 nodes, converges fast → all keys single final value. + let config = SimConfig { + name: "lww-fullmesh".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_lww_single_value(&metrics); + assert!(result.passed, "lww single value: {}", result.actual); +} + +#[test] +fn entropy_reaches_zero_at_convergence() { + // FullMesh 100 converges in ~O(log N) rounds; 30 rounds is plenty. + let config = SimConfig { + name: "entropy-fullmesh".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_entropy_zero_at_convergence(&metrics); + assert!(result.passed, "entropy zero: {}", result.actual); +} + +#[test] +fn entropy_decreases_monotonically() { + // Entropy (disagreeing node-pairs) can increase before converging: with epidemic + // spreading, disagreements grow until ~50% have data, then shrink. Monotonic + // decrease is not achievable for any topology with gradual spreading. + // Instead, verify: (1) entropy reaches 0, (2) last 5 rounds all have entropy 0. + let config = SimConfig { + name: "entropy-convergence".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let tail = &metrics.entropy_per_round[metrics.entropy_per_round.len().saturating_sub(5)..]; + let all_zero = tail.iter().all(|&e| e == 0); + assert!( + all_zero, + "entropy should be 0 for last 5 rounds, got: {:?}", + tail + ); +} + +#[test] +fn no_stale_reads_after_convergence() { + // FullMesh 100 converges in ~O(log N) rounds; 30 rounds is plenty. + let config = SimConfig { + name: "no-stale".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_no_stale_reads(&metrics); + assert!(result.passed, "no stale reads: {}", result.actual); +} + +// ── Practical (2) ─────────────────────────────────────────────────────────── + +#[test] +fn state_size_stabilizes_at_key_count() { + // FullMesh 100 converges in ~O(log N) rounds; 30 rounds is plenty for + // all 100 nodes to have all 5 keys. + let config = SimConfig { + name: "state-size-fullmesh".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_state_size_stabilizes(&metrics, 5.0); + assert!(result.passed, "state size: {}", result.actual); +} + +#[test] +fn state_size_grows_monotonically() { + let config = SimConfig { + name: "state-mono".into(), + topology: Topology::Ring, + num_nodes: 1000, + initial_data: test_data(5), + num_rounds: 60, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 1, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_state_size_monotonic(&metrics); + assert!(result.passed, "state size monotonic: {}", result.actual); +} + +// ── Multi-threaded variants (5) ───────────────────────────────────────────── + +#[test] +fn all_nodes_receive_all_keys_in_ring_1000_mt() { + // FullMesh 100 nodes converges in ~O(log N) rounds, well within 30 rounds. + let config = SimConfig { + name: "fullmesh-100-mt".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 4, + }; + let (_, metrics) = run_and_analyze(config); + assert!( + (metrics.delivery_ratio - 1.0).abs() < 1e-9, + "MT delivery_ratio = {}, expected 1.0", + metrics.delivery_ratio + ); +} + +#[test] +fn fullmesh_converges_in_log_n_rounds_mt() { + let n = 100; + // 2x bound for multi-threaded non-determinism. + let bound = 2 * 4 * ((n as f64).ln().ceil() as usize); + let config = SimConfig { + name: "fullmesh-latency-mt".into(), + topology: Topology::FullMesh, + num_nodes: n, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 4, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_convergence_bound(&metrics, bound); + assert!(result.passed, "MT fullmesh convergence: {}", result.actual); +} + +#[test] +fn convergence_curve_is_monotonic_mt() { + let config = SimConfig { + name: "fullmesh-mono-mt".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 4, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_curve_monotonic(&metrics); + assert!(result.passed, "MT monotonic: {}", result.actual); +} + +#[test] +fn partition_heals_and_converges_mt() { + // Partitioned 100 = two halves of 50 nodes, each full-mesh internally. + // Heal at round 100, run 300 total to allow full convergence after healing. + let config = SimConfig { + name: "partition-heal-mt".into(), + topology: Topology::Partitioned, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 300, + ticks_per_round: 4, + heal_after_round: Some(100), + num_threads: 4, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_partition_heals(&metrics); + assert!(result.passed, "MT partition heals: {}", result.actual); +} + +#[test] +fn lww_ensures_single_final_value_mt() { + let config = SimConfig { + name: "lww-fullmesh-mt".into(), + topology: Topology::FullMesh, + num_nodes: 100, + initial_data: test_data(5), + num_rounds: 30, + ticks_per_round: 4, + heal_after_round: None, + num_threads: 4, + }; + let (_, metrics) = run_and_analyze(config); + let result = check_lww_single_value(&metrics); + assert!(result.passed, "MT lww single value: {}", result.actual); +} diff --git a/src/actor.rs b/src/actor.rs index c9703a7..773a30b 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -16,7 +16,17 @@ pub trait ActorInterface: 'static + Send { /// but most systems are powerful, and this allows us to create a global map of /// actor processes in the future, without worrying about collision. #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] pub struct ActorAddress(pub [u8; 32]); + +impl std::fmt::Display for ActorAddress { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + for b in &self.0[..8] { + write!(f, "{:02x}", b)?; + } + write!(f, "\u{2026}") + } +} impl ActorAddress { pub fn new_random() -> Self { let mut bytes = [0u8; 32]; diff --git a/src/runtime.rs b/src/runtime.rs index 78ca383..deb5554 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -215,17 +215,21 @@ impl Runtime { for mut worker in workers { let rt_clone = rt.clone(); - let handle = thread::spawn(move || { - let tc = TickContext { - address_map: &rt_clone.address_map, - transfer_txs: &rt_clone.transfer_txs, - spawn_txs: &rt_clone.spawn_txs, - placement: &rt_clone.placement, - inbox_registry: &rt_clone.inbox_registry, - config: &rt_clone.config, - }; - worker.run(&tc, &rt_clone.is_running); - }); + let name = format!("swactor-worker-{}", worker.id.0); + let handle = thread::Builder::new() + .name(name) + .spawn(move || { + let tc = TickContext { + address_map: &rt_clone.address_map, + transfer_txs: &rt_clone.transfer_txs, + spawn_txs: &rt_clone.spawn_txs, + placement: &rt_clone.placement, + inbox_registry: &rt_clone.inbox_registry, + config: &rt_clone.config, + }; + worker.run(&tc, &rt_clone.is_running); + }) + .expect("failed to spawn worker thread"); handles.push(handle); } diff --git a/src/worker.rs b/src/worker.rs index 61ab302..3cd5c34 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -13,7 +13,7 @@ use crate::Error; /// A worker owns a set of actors and runs them in a loop. pub(crate) struct Worker { - id: WorkerId, + pub(crate) id: WorkerId, pub(crate) pool: ActorPool, transfer_rx: Receiver, spawn_rx: Receiver<(ActorAddress, Box)>,