add propagation benchmark

1 files changed, 394 insertions, 0 deletions

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),
+    );
+}