diff options
| -rw-r--r-- | docs/measurements.md | 3 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 2 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 18 |
3 files changed, 20 insertions, 3 deletions
diff --git a/docs/measurements.md b/docs/measurements.md index a8c2730..894ccd3 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -122,7 +122,8 @@ hour; otherwise it starts over with no history, and `observedSecs` says so. ## Propagation How long changes written elsewhere took to reach this node, from the writer's commit -stamp to the apply here, for every change in every applied sync message once caught up. +stamp to the apply here, for every change in every applied sync message and every delta +fetched as a blob ([`peers.md`](peers.md)), once caught up. The apply is timed by the peers' clocks: this node's own offset from them (`clock.offsetMs`, Clocks) is taken off. A histogram with buckets at 100 ms, 250 ms, 500 ms, 1 s, 2.5 s, 5 s, 10 s, 30 s, 1 min, diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index de63962..c322305 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -144,7 +144,7 @@ pub(crate) struct Relay { pub(crate) nothing_new_change_bytes: u64, pub(crate) nothing_new_bloom_bytes: u64, /// Changes from elsewhere that came not along a writer's tree (pulled, with an - /// answer, sent again), and their delays summed (as `Propagation`). + /// answer, sent again, fetched as a delta), and their delays summed (as `Propagation`). pub(crate) off_tree: u64, pub(crate) off_tree_ms: u64, } diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 24b867b..dcb29a7 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -633,6 +633,10 @@ async fn download_and_apply_delta( let doc = Arc::clone(&state.doc); let peer_state = state.get_peer_state(peer).await; let stats = Arc::clone(&state.stats); + // As for sync messages (`receive`): not while catching up. + let measure_propagation = state.initial_sync_done.load(Ordering::Acquire); + let clock = Arc::clone(&state.clock); + let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed); tokio::task::spawn_blocking(move || -> Result<bool> { let mut replica = doc.blocking_write(); if replica.epoch != epoch { @@ -646,11 +650,23 @@ async fn download_and_apply_delta( bail!("{missing} of the offered heads are still missing after loading"); } stats.heads_seen(replica.heads().len()); + let changed = replica.heads() != heads_before.as_slice(); + if changed && measure_propagation { + // As in `receive`; a delta goes along no writer's tree. + let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms; + let delays: Vec<i64> = + crate::document::changes::commit_times(&replica.doc, &heads_before) + .into_iter() + .map(|time| now_ms - time) + .collect(); + stats.off_tree(&delays); + stats.propagated(delays); + } if let Some(peer_state) = peer_state { // Sync from it made progress: no heal for a while (`keepalive`). peer_state.blocking_lock().last_sync_rx = Some(Instant::now()); } - Ok(replica.heads() != heads_before.as_slice()) + Ok(changed) }) .await .context("Delta apply task panicked")? |