parallel propagation benchmark

1 files changed, 54 insertions, 14 deletions

diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index babd8e9..d4a254f 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -12,6 +12,7 @@
 //! 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::fmt::Write;
 use std::time::Duration;
 
 use serde_json::json;
@@ -35,6 +36,10 @@ const KEEPALIVE: Duration = Duration::from_secs(5);
 const QUIET: Duration = Duration::from_secs(20);
 /// Only the scenarios whose names contain this, if set: `BENCH_SCENARIO=NAT`.
 const ONLY: &str = "BENCH_SCENARIO";
+/// If set, all scenarios at once (`BENCH_PARALLEL=1`): several times faster on many
+/// cores, but they share them, so delays (and what depends on timing) may come out
+/// higher than alone; message counts barely.
+const PARALLEL: &str = "BENCH_PARALLEL";
 const SETTLE: Duration = Duration::from_secs(180);
 
 /// How the swarm is made.
@@ -134,15 +139,28 @@ const SCENARIOS: &[Scenario] = &[
     },
 ];
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 16)]
 #[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!();
     let only = std::env::var(ONLY).unwrap_or_default();
-    for (seed, scenario) in (70..).zip(SCENARIOS) {
-        if scenario.name.contains(&only) {
-            run(seed, scenario).await;
+    let chosen: Vec<(u64, &'static Scenario)> = (70..)
+        .zip(SCENARIOS)
+        .filter(|(_, scenario)| scenario.name.contains(&only))
+        .collect();
+    if std::env::var_os(PARALLEL).is_some() {
+        // All at once, each reported when all are done, in order.
+        let running: Vec<_> = chosen
+            .into_iter()
+            .map(|(seed, scenario)| tokio::spawn(run(seed, scenario)))
+            .collect();
+        for report in running {
+            print!("{}", report.await.expect("scenario panicked"));
+        }
+    } else {
+        for (seed, scenario) in chosen {
+            print!("{}", run(seed, scenario).await);
         }
     }
 }
@@ -236,7 +254,9 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 9] {
     ]
 }
 
-async fn run(seed: u64, scenario: &Scenario) {
+/// Runs `scenario`; returns its report.
+async fn run(seed: u64, scenario: &'static Scenario) -> String {
+    let mut out = String::new();
     let sim = Sim::with_identity(seed);
     let ids: Vec<String> = (0..scenario.nodes).map(|i| format!("n{i:03}")).collect();
     (scenario.net)(&sim, &ids);
@@ -290,7 +310,8 @@ async fn run(seed: u64, scenario: &Scenario) {
     let n = halves.len();
     let quiet = (halves[n.saturating_sub(3)], halves[n - 1]);
 
-    println!(
+    let _ = writeln!(
+        out,
         "\n{} ({} writes at {} a second each)",
         scenario.name,
         WRITES,
@@ -314,21 +335,36 @@ async fn run(seed: u64, scenario: &Scenario) {
         let written = Instant::now();
         sim.wait_converged(all, SETTLE).await;
         let converged = written.elapsed();
-        report(workload, quiet, (start, Totals::of(&nodes)), converged);
+        report(
+            &mut out,
+            workload,
+            quiet,
+            (start, Totals::of(&nodes)),
+            converged,
+        );
     }
+    out
 }
 
 /// 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) {
+fn report(
+    out: &mut String,
+    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!(
+    let _ = writeln!(
+        out,
         "  {workload}: {:.1} s, converged {} ms after the last write",
         took.as_secs_f64(),
         converged.as_millis()
     );
-    println!(
+    let _ = writeln!(
+        out,
         "    {:<13} {:>9} {:>9} {:>11}   (quiet phase, as long: msgs)",
         "kind", "msgs", "/write", "bytes/write"
     );
@@ -351,20 +387,23 @@ fn report(workload: &str, quiet: (Totals, Totals), during: (Totals, Totals), con
         let bytes = (d1.bytes - d0.bytes) as f64;
         all_messages += messages;
         all_bytes += bytes;
-        println!(
+        let _ = writeln!(
+            out,
             "    {kind:<13} {messages:>9.0} {:>9.1} {:>11.0}   ({quiet_messages:.0})",
             messages / writes,
             bytes / writes
         );
     }
-    println!(
+    let _ = writeln!(
+        out,
         "    {:<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!(
+    let _ = writeln!(
+        out,
         "    relay per write: {:.1} announcement rounds, {:.1} pulls ({:.1} queued), {:.1} along writers' trees",
         (r1.announce_rounds - r0.announce_rounds) as f64 / writes,
         (r1.pulls - r0.pulls) as f64 / writes,
@@ -387,7 +426,8 @@ fn report(workload: &str, quiet: (Totals, Totals), during: (Totals, Totals), con
         super::soak::percentile_bound(&propagation, quantile)
             .map_or("more than an hour".to_owned(), |ms| format!("{ms} ms"))
     };
-    println!(
+    let _ = writeln!(
+        out,
         "    propagation: {} arrivals, mean {} ms, half within {}, 99% within {}",
         propagation.count,
         propagation.sum_ms / propagation.count.max(1),