aboutsummaryrefslogtreecommitdiffstats
path: root/src/simtest/soak.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/simtest/soak.rs')
-rw-r--r--src/simtest/soak.rs244
1 files changed, 241 insertions, 3 deletions
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<String> = (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}"
+ );
+}