1 files changed, 120 insertions, 0 deletions
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 9b18b2c..426507b 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -657,3 +657,123 @@ fn report(
         );
     }
 }
+
+/// Values written in `bench_bridge`, unless `BENCH_BRIDGE_WRITES` says otherwise.
+const BRIDGE_WRITES: usize = 200;
+/// Their size: random, so nothing compresses them.
+const BRIDGE_VALUE: usize = 256 * 1024;
+/// A hundred a second.
+const BRIDGE_EVERY: Duration = Duration::from_millis(10);
+
+/// A writer on a fast transport (5 ms round trips, 1 GiB/s), a bridge on it and a slow
+/// one (500 ms round trips, 50 MiB/s), and a node on the slow one alone: a writes 256
+/// KiB values at 100 a second (25 MiB/s, a quarter of the fast link, half of the slow
+/// one), b passes them on to z. What z gets, as sync messages and as blobs (deltas),
+/// against what was written, and how late.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+#[ignore = "a benchmark: cargo benchmarks"]
+async fn bench_bridge() {
+    println!();
+    let writes = std::env::var("BENCH_BRIDGE_WRITES")
+        .ok()
+        .and_then(|writes| writes.parse().ok())
+        .unwrap_or(BRIDGE_WRITES);
+    let sim = Sim::with_identity(71);
+    sim.net.set_default_link(LinkParams {
+        delay: (Duration::from_micros(2_000), Duration::from_micros(3_000)),
+        rate: Some(1 << 30),
+        ..LinkParams::default()
+    });
+    sim.net_for("sim2").set_default_link(LinkParams {
+        delay: (Duration::from_millis(240), Duration::from_millis(260)),
+        rate: Some(50 << 20),
+        ..LinkParams::default()
+    });
+    let timing = |config: &mut crate::node::NodeConfig| {
+        config.timing = Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
+        config.neighbours_k = 12;
+    };
+    let a = sim.start_on_with("a", &["sim"], &[], Start::Genesis, timing);
+    let b = sim.start_on_with("b", &["sim", "sim2"], &["a"], Start::Join, timing);
+    let z = sim.start_on_with("z", &["sim2"], &["b"], Start::Join, timing);
+    let all = [&a, &b, &z];
+    sim.wait_ready(&all, SETTLE).await;
+    sim.wait_until("every node knows every member", SETTLE, || async {
+        for node in all {
+            if members::members(&node.doc.read().await.doc).len() < 3 {
+                return false;
+            }
+        }
+        true
+    })
+    .await;
+    sim.wait_converged(&all, SETTLE).await;
+
+    let received = |node: &AppState| node.traffic.total().received;
+    let (z0, b0) = (received(&z), b.traffic.total().sent);
+    let p0 = z.stats.snapshot().propagation;
+    let mut rng = crate::testrand::Rng::new(72);
+    let started = Instant::now();
+    for i in 0..writes {
+        let value: String = (0..BRIDGE_VALUE / 16)
+            .map(|_| format!("{:016x}", rng.next_u64()))
+            .collect();
+        write(&a, &format!("bridge/{i}"), json!(value)).await;
+        // At the pace, however long a write took.
+        let due = started + BRIDGE_EVERY * (i as u32 + 1);
+        tokio::time::sleep_until(due).await;
+    }
+    let written_in = started.elapsed();
+    let written = Instant::now();
+    sim.wait_converged(&all, SETTLE).await;
+    let converged = written.elapsed();
+
+    let (z1, b1) = (received(&z), b.traffic.total().sent);
+    let p1 = z.stats.snapshot().propagation;
+    let diff = |c1: Counter, c0: Counter| (c1.messages - c0.messages, c1.bytes - c0.bytes);
+    let mib = |bytes: u64| bytes as f64 / f64::from(1 << 20);
+    let total = (writes * BRIDGE_VALUE) as u64;
+    let (z_sync, z_blob) = (diff(z1.sync, z0.sync), diff(z1.blob, z0.blob));
+    let (b_sync, b_blob) = (diff(b1.sync, b0.sync), diff(b1.blob, b0.blob));
+    let mut propagation = Propagation::default();
+    for ((sum, before), after) in propagation
+        .buckets
+        .iter_mut()
+        .zip(p0.buckets)
+        .zip(p1.buckets)
+    {
+        *sum = after - before;
+    }
+    propagation.count = p1.count - p0.count;
+    propagation.sum_ms = p1.sum_ms - p0.sum_ms;
+    let within = |quantile| {
+        propagation
+            .within(quantile)
+            .map_or("beyond the last bucket".to_owned(), |ms| format!("{ms} ms"))
+    };
+    println!(
+        "\nfast writer, bridge, slow node ({writes} values of {} KiB, {} a second)\n  \
+         written: {:.1} MiB in {:.1} s; z converged {} ms after the last write\n  \
+         b sent (to a and z): SYNC {} messages, {:.1} MiB; BLOB {} messages, {:.1} MiB\n  \
+         z received: SYNC {} messages, {:.1} MiB; BLOB {} messages, {:.1} MiB: {:.2} times what was written\n  \
+         z's delays: {} changes, mean {} ms, half within {}, 99% within {}",
+        BRIDGE_VALUE / 1024,
+        1000 / BRIDGE_EVERY.as_millis(),
+        mib(total),
+        written_in.as_secs_f64(),
+        converged.as_millis(),
+        b_sync.0,
+        mib(b_sync.1),
+        b_blob.0,
+        mib(b_blob.1),
+        z_sync.0,
+        mib(z_sync.1),
+        z_blob.0,
+        mib(z_blob.1),
+        (z_sync.1 + z_blob.1) as f64 / total as f64,
+        propagation.count,
+        propagation.sum_ms / propagation.count.max(1),
+        within(0.5),
+        within(0.99),
+    );
+}