diff options
| -rw-r--r-- | docs/development.md | 8 | ||||
| -rw-r--r-- | src/simtest/bench.rs | 394 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 1 | ||||
| -rw-r--r-- | src/simtest/soak.rs | 2 |
4 files changed, 404 insertions, 1 deletions
diff --git a/docs/development.md b/docs/development.md index d7f6068..cb42492 100644 --- a/docs/development.md +++ b/docs/development.md @@ -17,6 +17,14 @@ needs at least one: a transport left out takes its tests and dependencies with i (`--features libp2p` builds neither `reqwest` and the daemon supervision of Mycelium, nor arti; among Mycelium's tests is a swarm simulated on its envelope). +The benchmarks are ignored tests named `bench_*`: the document's (writes and sync +messages, `document/bench.rs`), the remakes' replay (`remake/bench.rs`), and what +passing changes on costs over the simulated network (`simtest/bench.rs`: messages and +bytes per write by kind, propagation delays, for a full mesh, a writer behind NAT, two +transports and partial meshes of 50 and 100 nodes; about a quarter of an hour). One of +them alone: `cargo test --release --features mycelium,libp2p,tor,dashboard +bench_propagation -- --ignored --nocapture`. + The simulations send the wire encoding alone, as libp2p and Tor do; a simulated net can put messages into a transport's envelope instead (`SimNet::with_codec`), and every message kind is round-tripped through each codec (`wire::samples`). diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs new file mode 100644 index 0000000..8c24d5c --- /dev/null +++ b/src/simtest/bench.rs @@ -0,0 +1,394 @@ +//! What passing changes on costs (`swarm::relay`): whole nodes on the simulated network, +//! a series of writes, and the messages, bytes and delays it took, per scenario. A +//! benchmark, not a test: `cargo benchmarks` (release, with the other `bench_*`) prints +//! a table per scenario to compare before and after a change; nothing is asserted but +//! that the nodes converge. +//! +//! Each scenario starts its swarm and waits until it has settled (every member known, +//! the same heads, then no sync message for a while), then measures a quiet phase (no +//! writes), then each workload: 100 writes at 5 a second, and until every node has +//! them. What a workload sent is shown as it is, beside what the quiet phase sent in as +//! long: not taken off, since writes make keepalives unneeded (a pair that exchanges +//! anything needs none), so the quiet phase's count isn't part of a workload's. + +use std::time::Duration; + +use serde_json::json; +use tokio::time::Instant; + +use super::{SYNC_REPLY_TIMEOUT, Sim, write}; +use crate::state::{AppState, Timing}; +use crate::swarm::bootstrap::Start; +use crate::swarm::members; +use crate::swarm::stats::{Propagation, Relay}; +use crate::transport::sim::{DELIVERY_LIMIT, LinkParams}; +use crate::transport::traffic::{ByKind, Counter}; + +/// Writes per workload, and how many a second. +const WRITES: usize = 100; +const WRITE_EVERY: Duration = Duration::from_millis(200); +/// Longer than the tests' (`simtest::KEEPALIVE`), so that keepalives don't drown what +/// the writes cost; as long as the relay tests'. +const KEEPALIVE: Duration = Duration::from_secs(5); +/// The quiet phase: long enough to see the keepalives' rate. +const QUIET: Duration = Duration::from_secs(20); +const SETTLE: Duration = Duration::from_secs(180); + +/// How the swarm is made. +struct Scenario { + name: &'static str, + nodes: usize, + /// The transports of node `i` (`sim`, `sim2`). + on: fn(usize) -> &'static [&'static str], + /// Before the nodes start: links cut, delays. + net: fn(&Sim, &[String]), + /// The node that writes in the one-writer workload. + writer: usize, +} + +fn sim_only(_: usize) -> &'static [&'static str] { + &["sim"] +} + +fn clean(_: &Sim, _: &[String]) {} + +/// Node 1 reaches node 2 alone, as one behind NAT with a single link. +fn writer_behind_nat(sim: &Sim, ids: &[String]) { + let cut = LinkParams { + loss: 1.0, + ..LinkParams::default() + }; + for (i, id) in ids.iter().enumerate() { + if i != 1 && i != 2 { + sim.net.set_link(&ids[1], id, cut.clone()); + sim.net.set_link(id, &ids[1], cut.clone()); + } + } +} + +/// Nodes 0 to 2 on both nets (bridges), the first half of the rest on `sim`, the +/// other half on `sim2`. +fn two_transports(i: usize) -> &'static [&'static str] { + match i { + 0..3 => &["sim", "sim2"], + 3..16 => &["sim"], + _ => &["sim2"], + } +} + +/// As the soak test's: 2% lost, 1% twice, 50 to 500 ms. +fn lossy(sim: &Sim, _: &[String]) { + sim.net.set_default_link(LinkParams { + loss: 0.02, + duplicate: 0.01, + delay: (Duration::from_millis(50), Duration::from_millis(500)), + ..LinkParams::default() + }); +} + +const SCENARIOS: &[Scenario] = &[ + Scenario { + name: "full mesh, 13 nodes", + nodes: 13, + on: sim_only, + net: clean, + writer: 1, + }, + Scenario { + name: "writer behind NAT (one link), 13 nodes", + nodes: 13, + on: sim_only, + net: writer_behind_nat, + writer: 1, + }, + Scenario { + name: "two transports, 3 bridges, 30 nodes", + nodes: 30, + on: two_transports, + net: clean, + writer: 5, + }, + Scenario { + name: "partial mesh, 50 nodes", + nodes: 50, + on: sim_only, + net: clean, + writer: 1, + }, + Scenario { + name: "partial mesh, lossy and slow links, 50 nodes", + nodes: 50, + on: sim_only, + net: lossy, + writer: 1, + }, + Scenario { + name: "partial mesh, 100 nodes", + nodes: 100, + on: sim_only, + net: clean, + writer: 1, + }, +]; + +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +#[ignore = "a benchmark: cargo benchmarks"] +async fn bench_propagation() { + // Past the test harness's `test … ...`, so each report is on a line of its own. + println!(); + for (seed, scenario) in (70..).zip(SCENARIOS) { + run(seed, scenario).await; + } +} + +/// What all nodes have sent, counted, and their relay counters and propagation delays +/// summed: at one moment. +#[derive(Clone, Copy)] +struct Totals { + at: Instant, + sent: ByKind, + relay: Relay, + propagation: Propagation, +} + +impl Totals { + fn of(nodes: &[AppState]) -> Self { + let mut totals = Self { + at: Instant::now(), + sent: ByKind::default(), + relay: Relay::default(), + propagation: Propagation::default(), + }; + for node in nodes { + let sent = node.traffic.total().sent; + for (total, theirs) in counters_mut(&mut totals.sent) + .into_iter() + .zip(counters(&sent)) + { + total.messages += theirs.messages; + total.bytes += theirs.bytes; + } + let stats = node.stats.snapshot(); + let relay = &mut totals.relay; + relay.announce_rounds += stats.relay.announce_rounds; + relay.pulls += stats.relay.pulls; + relay.pulls_queued += stats.relay.pulls_queued; + relay.grafts += stats.relay.grafts; + relay.prunes += stats.relay.prunes; + relay.eager_forwards += stats.relay.eager_forwards; + let propagation = &mut totals.propagation; + for (sum, count) in propagation + .buckets + .iter_mut() + .zip(stats.propagation.buckets) + { + *sum += count; + } + propagation.count += stats.propagation.count; + propagation.sum_ms += stats.propagation.sum_ms; + } + totals + } +} + +/// The kinds, in the order of the report. +const KINDS: [&str; 9] = [ + "SYNC", + "STATUS", + "STATUS_REPLY", + "BLOB", + "HELLO", + "HELLO_REPLY", + "PING/PONG", + "WHO/ME", + "LINKS", +]; + +fn counters(by: &ByKind) -> [Counter; 9] { + [ + by.sync, + by.status, + by.status_reply, + by.blob, + by.hello, + by.hello_reply, + by.paths, + by.identity, + by.links, + ] +} + +fn counters_mut(by: &mut ByKind) -> [&mut Counter; 9] { + [ + &mut by.sync, + &mut by.status, + &mut by.status_reply, + &mut by.blob, + &mut by.hello, + &mut by.hello_reply, + &mut by.paths, + &mut by.identity, + &mut by.links, + ] +} + +async fn run(seed: u64, scenario: &Scenario) { + let sim = Sim::with_identity(seed); + let ids: Vec<String> = (0..scenario.nodes).map(|i| format!("n{i:03}")).collect(); + (scenario.net)(&sim, &ids); + let timing = |config: &mut crate::node::NodeConfig| { + config.timing = Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT); + // As in production (`--neighbours`). + config.neighbours_k = 12; + }; + let mut nodes = vec![sim.start_on_with(&ids[0], (scenario.on)(0), &[], Start::Genesis, timing)]; + for (i, id) in ids.iter().enumerate().skip(1) { + // The writer behind NAT reaches node 2 alone. + let seed = if i == 1 { + ids[2].as_str() + } else { + ids[0].as_str() + }; + nodes.push(sim.start_on_with(id, (scenario.on)(i), &[seed], Start::Join, timing)); + } + let all: Vec<&AppState> = nodes.iter().collect(); + let all = all.as_slice(); + sim.wait_ready(all, SETTLE).await; + let count = scenario.nodes; + sim.wait_until( + "every node knows every member", + SETTLE, + move || async move { + for node in all { + if members::members(&node.doc.read().await.doc).len() < count { + return false; + } + } + true + }, + ) + .await; + sim.wait_converged(all, SETTLE).await; + // Neighbours chosen, handshakes done, what joining set off over: no sync message for + // two keepalives (registrations and alive stamps still pass for a while). + let settling = Instant::now(); + loop { + let before = Totals::of(&nodes); + tokio::time::sleep(KEEPALIVE * 2).await; + let after = Totals::of(&nodes); + if after.sent.sync.messages == before.sent.sync.messages || settling.elapsed() > SETTLE { + break; + } + } + + let before = Totals::of(&nodes); + tokio::time::sleep(QUIET).await; + let quiet = (before, Totals::of(&nodes)); + + println!( + "\n{} ({} writes at {} a second each)", + scenario.name, + WRITES, + 1000 / WRITE_EVERY.as_millis() + ); + for (w, (workload, writer)) in [ + ("one writer", Some(scenario.writer)), + ("writers in turn", None), + ] + .into_iter() + .enumerate() + { + let start = Totals::of(&nodes); + for i in 0..WRITES { + let node = &nodes[writer.unwrap_or(i % nodes.len())]; + // A path of each workload's own: writing what is there already changes + // nothing, and nothing would pass. + write(node, &format!("bench/{w}/{i}"), json!(i)).await; + tokio::time::sleep(WRITE_EVERY).await; + } + let written = Instant::now(); + sim.wait_converged(all, SETTLE).await; + let converged = written.elapsed(); + report(workload, quiet, (start, Totals::of(&nodes)), converged); + } +} + +/// The workload's counts (`during`), and the quiet phase's for as long. +fn report(workload: &str, quiet: (Totals, Totals), during: (Totals, Totals), converged: Duration) { + let took = during.1.at - during.0.at; + let scale = took.as_secs_f64() / (quiet.1.at - quiet.0.at).as_secs_f64(); + let writes = WRITES as f64; + println!( + " {workload}: {:.1} s, converged {} ms after the last write", + took.as_secs_f64(), + converged.as_millis() + ); + println!( + " {:<13} {:>9} {:>9} {:>11} (quiet phase, as long: msgs)", + "kind", "msgs", "/write", "bytes/write" + ); + let rows = KINDS + .iter() + .zip( + counters(&quiet.0.sent) + .into_iter() + .zip(counters(&quiet.1.sent)), + ) + .zip( + counters(&during.0.sent) + .into_iter() + .zip(counters(&during.1.sent)), + ); + let (mut all_messages, mut all_bytes) = (0.0, 0.0); + for ((kind, (q0, q1)), (d0, d1)) in rows { + let quiet_messages = (q1.messages - q0.messages) as f64 * scale; + let messages = (d1.messages - d0.messages) as f64; + let bytes = (d1.bytes - d0.bytes) as f64; + all_messages += messages; + all_bytes += bytes; + println!( + " {kind:<13} {messages:>9.0} {:>9.1} {:>11.0} ({quiet_messages:.0})", + messages / writes, + bytes / writes + ); + } + println!( + " {:<13} {all_messages:>9.0} {:>9.1} {:>11.0}", + "all", + all_messages / writes, + all_bytes / writes + ); + let (r0, r1) = (during.0.relay, during.1.relay); + println!( + " relay per write: {:.1} announcement rounds, {:.1} pulls ({:.1} queued), {:.2} grafts, {:.2} prunes, {:.1} eager forwards", + (r1.announce_rounds - r0.announce_rounds) as f64 / writes, + (r1.pulls - r0.pulls) as f64 / writes, + (r1.pulls_queued - r0.pulls_queued) as f64 / writes, + (r1.grafts - r0.grafts) as f64 / writes, + (r1.prunes - r0.prunes) as f64 / writes, + (r1.eager_forwards - r0.eager_forwards) as f64 / writes, + ); + let mut propagation = Propagation::default(); + let (p0, p1) = (during.0.propagation, during.1.propagation); + for ((sum, a), b) in propagation + .buckets + .iter_mut() + .zip(p0.buckets) + .zip(p1.buckets) + { + *sum = b - a; + } + propagation.count = p1.count - p0.count; + propagation.sum_ms = p1.sum_ms - p0.sum_ms; + let within = |quantile| { + super::soak::percentile_bound(&propagation, quantile) + .map_or("more than an hour".to_owned(), |ms| format!("{ms} ms")) + }; + println!( + " propagation: {} arrivals, mean {} ms, half within {}, 99% within {}", + propagation.count, + propagation.sum_ms / propagation.count.max(1), + within(0.5), + within(0.99), + ); +} diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 267768f..dc23440 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -9,6 +9,7 @@ //! `RUST_LOG=continuum=debug cargo test simtest -- --nocapture` shows the nodes' logs //! (all nodes interleaved; messages name peers by ID). +mod bench; mod deltas; mod identity; mod links; diff --git a/src/simtest/soak.rs b/src/simtest/soak.rs index 6906238..db4ba60 100644 --- a/src/simtest/soak.rs +++ b/src/simtest/soak.rs @@ -88,7 +88,7 @@ async fn soak_25_nodes() { /// The upper bound of the bucket in which the `quantile` of changes is reached; None if /// only in the last (unbounded) one. -fn percentile_bound(propagation: &Propagation, quantile: f64) -> Option<u64> { +pub(super) fn percentile_bound(propagation: &Propagation, quantile: f64) -> Option<u64> { let wanted = (propagation.count as f64 * quantile).ceil() as u64; let mut seen = 0; for (bound, count) in propagation.buckets() { |