From d9d398522a142283b537670068cc6418402d9546 Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 06:02:19 +0100 Subject: soak with churn test --- src/simtest/bench.rs | 4 +- src/simtest/soak.rs | 244 ++++++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 243 insertions(+), 5 deletions(-) (limited to 'src/simtest') diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs index 2a96007..e8e452c 100644 --- a/src/simtest/bench.rs +++ b/src/simtest/bench.rs @@ -122,7 +122,7 @@ fn lossy(sim: &Sim, _: &[String]) { /// from 20 to 250 ms, each message within a fifth of its link's. So what the uniform /// 50–500 ms of `lossy` gives that real links don't: messages on one link overtake each /// other only within the jitter, and the fast links to a node are fast every time. -fn mycelium_like(sim: &Sim, ids: &[String]) { +pub(super) fn mycelium_like(sim: &Sim, ids: &[String]) { let mut rng = crate::testrand::Rng::new(7); for (i, from) in ids.iter().enumerate() { for to in &ids[i + 1..] { @@ -694,7 +694,7 @@ async fn trees(out: &mut String, nodes: &[AppState], net: &crate::transport::sim /// says it has them; a session whose peer never sends one (along a tree, with one /// writer) keeps all. And of those, how many because another parent brings them /// (`SyncSession::known_to_peer`): a session with a neighbour we count as covered. -async fn sent_kept(out: &mut String, nodes: &[AppState]) { +pub(super) async fn sent_kept(out: &mut String, nodes: &[AppState]) { let (mut most, mut sum) = (0, 0); let (mut most_elsewhere, mut sum_elsewhere) = (0, 0); let mut sessions = 0; diff --git a/src/simtest/soak.rs b/src/simtest/soak.rs index db4ba60..eb024d3 100644 --- a/src/simtest/soak.rs +++ b/src/simtest/soak.rs @@ -1,12 +1,16 @@ //! Many nodes under realistic conditions for a while. Ignored by default (a few -//! minutes in a debug build): `cargo test soak -- --ignored --nocapture`. +//! minutes in a debug build): `cargo test soak -- --ignored --nocapture`; the soak with +//! churn takes over ten minutes (`soak_with_churn`). use std::time::Duration; use serde_json::json; +use tokio::time::Instant; -use super::{Sim, write}; -use crate::state::AppState; +use super::{SYNC_REPLY_TIMEOUT, Sim, write}; +use crate::locks::MutexExt; +use crate::node::NodeConfig; +use crate::state::{AppState, Timing}; use crate::swarm::bootstrap::Start; use crate::swarm::members; use crate::swarm::stats::Propagation; @@ -99,3 +103,237 @@ pub(super) fn percentile_bound(propagation: &Propagation, quantile: f64) -> Opti } None } + +/// The soak with churn: its members, their keepalive (a day of keepalive-scaled timers in +/// minutes), and each workload's length (one writer, then writers in turn), two writes +/// a second. +const CHURN_NODES: usize = 30; +const CHURN_KEEPALIVE: Duration = Duration::from_millis(500); +const CHURN_WORKLOAD: Duration = Duration::from_secs(300); +const CHURN_WRITE_EVERY: Duration = Duration::from_millis(500); +/// A member away (all to and from it lost) for `AWAY_FOR` every `AWAY_EVERY`, in turns +/// (`n10` to `n19`); one restarted (its process gone, a new one on the same ID, joining +/// again) every `RESTART_EVERY`, in turns (`n20` to `n24`). +const AWAY_EVERY: Duration = Duration::from_secs(20); +const AWAY_FOR: Duration = Duration::from_secs(15); +const RESTART_EVERY: Duration = Duration::from_secs(120); +/// How long the quiet phase at the end is watched. +const CHURN_QUIET: Duration = Duration::from_secs(30); + +/// What grows over a long run, and what churn sets off: 30 members on Mycelium's round +/// trips (`bench::mycelium_like`: each link its own, 2% lost, 1% twice), keepalive +/// 500 ms; one writer for 5 minutes, then writers in turn for 5; a member away for 15 s +/// every 20 s, a restart every 2 minutes, a member joining halfway, a remake near the +/// end. Asserted: every node on the remake's epoch, converged; peers and the tables +/// kept per peer no larger than the swarm; tasks alive within twice their number after +/// the warm-up. Reported: propagation, what counts as sent per session, heals, the +/// links recorded at the end (`swarm::working_links`), and what the quiet phase sends. +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +#[ignore] +async fn soak_with_churn() { + let sim = Sim::new(31); + let ids: Vec = (0..CHURN_NODES).map(|i| format!("n{i:02}")).collect(); + super::bench::mycelium_like(&sim, &ids); + let timing = |config: &mut NodeConfig| { + config.timing = Timing::new( + CHURN_KEEPALIVE, + SYNC_REPLY_TIMEOUT, + crate::transport::sim::DELIVERY_LIMIT, + ); + }; + let genesis = ids[0].as_str(); + let mut nodes = vec![sim.start_with(genesis, &[], Start::Genesis, timing)]; + for id in &ids[1..CHURN_NODES - 1] { + nodes.push(sim.start_with(id, &[genesis], Start::Join, timing)); + } + { + let all: Vec<&AppState> = nodes.iter().collect(); + let all = all.as_slice(); + sim.wait_ready(all, Duration::from_secs(120)).await; + let count = all.len(); + sim.wait_until( + "every node knows every member", + Duration::from_secs(120), + 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, Duration::from_secs(120)).await; + } + let warm_tasks = tokio::runtime::Handle::current() + .metrics() + .num_alive_tasks(); + + let start = Instant::now(); + let total = CHURN_WORKLOAD * 2; + let (mut next_away, mut next_restart) = (start + AWAY_EVERY, start + RESTART_EVERY); + let (mut away_turn, mut restart_turn) = (0, 0); + let mut away: Option<(String, Instant)> = None; + let (mut joined, mut remade) = (false, false); + let mut writes = 0usize; + while start.elapsed() < total { + let now = Instant::now(); + if let Some((id, until)) = &away + && now >= *until + { + sim.net.bring_up(id); + away = None; + } + if away.is_none() && now >= next_away { + let id = ids[10 + away_turn % 10].clone(); + sim.net.take_down(&id); + away = Some((id, now + AWAY_FOR)); + away_turn += 1; + next_away += AWAY_EVERY; + } + if now >= next_restart { + let id = ids[20 + restart_turn % 5].clone(); + let restarted = sim.restart_with(&id, &[genesis], Start::Join, timing); + for node in &mut nodes { + if node.alias.as_deref() == Some(id.as_str()) { + *node = restarted.clone(); + } + } + restart_turn += 1; + next_restart += RESTART_EVERY; + } + if !joined && start.elapsed() >= CHURN_WORKLOAD { + nodes.push(sim.start_with(&ids[CHURN_NODES - 1], &[genesis], Start::Join, timing)); + joined = true; + } + if !remade && start.elapsed() >= total - CHURN_WORKLOAD / 5 { + let (status, report) = + crate::remake::testing::start(nodes[0].clone(), json!({ "kind": "compact" }), None) + .await; + assert!(status.is_success(), "remake refused: {status} {report}"); + remade = true; + } + let writer = if start.elapsed() < CHURN_WORKLOAD { + &nodes[1] + } else { + &nodes[writes % nodes.len()] + }; + if writer.is_ready().await { + write(writer, &format!("churn/{writes}"), json!(writes)).await; + } + writes += 1; + tokio::time::sleep(CHURN_WRITE_EVERY).await; + } + if let Some((id, _)) = away { + sim.net.bring_up(&id); + } + + let all: Vec<&AppState> = nodes.iter().collect(); + let all = all.as_slice(); + sim.wait_until( + "every node on the remake's epoch, ready", + Duration::from_secs(180), + move || async move { + for node in all { + if node.local_epoch().generation != 1 || !node.is_ready().await { + return false; + } + } + true + }, + ) + .await; + sim.wait_converged(all, Duration::from_secs(120)).await; + + let mut out = String::new(); + let mut propagation = Propagation::default(); + let (mut peers, mut rtts, mut throughputs, mut traffic, mut heals) = (0, 0, 0, 0, 0); + for node in all { + let theirs = node.stats.snapshot().propagation; + for (sum, count) in propagation.buckets.iter_mut().zip(theirs.buckets) { + *sum += count; + } + propagation.count += theirs.count; + propagation.sum_ms += theirs.sum_ms; + let known = node.peers_snapshot().await; + peers = peers.max(known.len()); + for (_, peer_state) in &known { + heals += peer_state.lock().await.incidents.heals; + } + rtts = rtts.max(node.lock_rtt().len()); + throughputs = throughputs.max(node.throughput.lock_or_recover().len()); + traffic = traffic.max(node.traffic.len()); + } + let recorded = { + let replica = nodes[0].doc.read().await; + members::members(&replica.doc) + .iter() + .filter(|member| { + members::links(&replica.doc, member) != members::RecordedLinks::default() + }) + .count() + }; + let end_tasks = tokio::runtime::Handle::current() + .metrics() + .num_alive_tasks(); + println!( + "soak with churn: {writes} writes, {away_turn} times away, {restart_turn} restarts; \ + propagation: {} changes, mean {} ms, p99 within {}", + propagation.count, + propagation.sum_ms / propagation.count.max(1), + percentile_bound(&propagation, 0.99) + .map_or("more than an hour".to_owned(), |ms| format!("{ms} ms")), + ); + super::bench::sent_kept(&mut out, &nodes).await; + print!("{out}"); + println!( + " per node at most: {peers} peers, round trips of {rtts}, throughput of \ + {throughputs}, traffic of {traffic}; {heals} heals; {recorded} members with links \ + recorded; tasks alive {warm_tasks} after the warm-up, {end_tasks} at the end" + ); + + // The quiet phase: what the swarm sends with nothing to pass on. + let sent = |nodes: &[AppState]| -> [u64; 4] { + let mut sum = [0; 4]; + for node in nodes { + let total = node.traffic.total().sent; + for (sum, counter) in + sum.iter_mut() + .zip([total.sync, total.status, total.status_reply, total.paths]) + { + *sum += counter.messages; + } + } + sum + }; + let before = sent(&nodes); + tokio::time::sleep(CHURN_QUIET).await; + let after = sent(&nodes); + let per_node_minute = |i: usize| { + (after[i] - before[i]) as f64 * 60.0 / CHURN_QUIET.as_secs_f64() / nodes.len() as f64 + }; + println!( + " quiet, per node and minute: {:.1} SYNC, {:.1} STATUS, {:.1} STATUS_REPLY, \ + {:.1} PING/PONG", + per_node_minute(0), + per_node_minute(1), + per_node_minute(2), + per_node_minute(3), + ); + + for (what, most) in [ + ("peers", peers), + ("round trips", rtts), + ("throughputs", throughputs), + ] { + assert!( + most <= CHURN_NODES, + "{most} {what} kept by a node of {CHURN_NODES}" + ); + } + assert!( + end_tasks <= warm_tasks * 2, + "tasks alive grew from {warm_tasks} to {end_tasks}" + ); +} -- cgit v1.3.1