summaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/bootstrap.rs1
-rw-r--r--src/swarm/pair.rs4
-rw-r--r--src/swarm/peer/inbound.rs41
-rw-r--r--src/swarm/peer/keepalive.rs140
-rw-r--r--src/swarm/peer/mod.rs5
-rw-r--r--src/swarm/peer/worker.rs35
-rw-r--r--src/swarm/relay.rs12
-rw-r--r--src/swarm/rtt.rs177
-rw-r--r--src/swarm/source.rs4
-rw-r--r--src/swarm/stats.rs2
-rw-r--r--src/swarm/sync.rs16
11 files changed, 272 insertions, 165 deletions
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 7c4c444..a9ab1b2 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -119,6 +119,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() {
state.mark_ready(swarm_id.clone()).await?;
info!(swarm_id = %swarm_id, "Node is ready");
+ crate::swarm::peer::announce(state);
return Ok(());
}
}
diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs
index c1a196b..0d45b32 100644
--- a/src/swarm/pair.rs
+++ b/src/swarm/pair.rs
@@ -1,5 +1,5 @@
-//! Which side of a pair of nodes leads, for duties only one side needs to do (sending
-//! keepalive PINGs).
+//! Which side of a pair of nodes leads, for what one side needs to start (the first
+//! keepalive; then the two take turns, `PeerState::take_keepalive_due`).
//!
//! Both sides decide on their own, from the two keys alone, and agree. The leader is
//! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index b04aad9..2e89739 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -11,7 +11,7 @@ use super::epoch::observe_peer_epoch;
use super::handshake::{
discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
};
-use super::keepalive::{heal_if_diverged, local_keepalive, note_keepalive};
+use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status, status_received};
use super::lanes::SyncLanes;
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
@@ -30,6 +30,11 @@ pub(crate) async fn handle_message(
debug!("Ignoring message from self");
return Ok(());
}
+ // PING and PONG only say the peer is alive (`keepalive::note_alive`).
+ if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) {
+ note_alive(state, &peer, &message).await;
+ return Ok(());
+ }
// WHO and ME end at the identity layer (`transport::identity`).
let Some((epoch, epoch_seq)) = message.epoch_stamp() else {
return Ok(());
@@ -143,40 +148,42 @@ pub(crate) async fn handle_message(
transport::ParsedMessage::Blob(blob) => {
crate::swarm::blob::handle(state, &peer, blob).await;
}
- transport::ParsedMessage::Ping(remote) => {
- let local = on_keepalive(state, &peer, &remote, reactivated).await;
+ transport::ParsedMessage::Status(remote) => {
+ status_received(state, &peer).await;
+ let local = on_status(state, &peer, &remote, reactivated).await;
if let Err(error) = state
.transport
- .send_message(&peer, &ParsedMessage::Pong(local))
+ .send_message(&peer, &ParsedMessage::StatusReply(local))
.await
{
- debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
+ debug!(peer = %peer, error = %format!("{error:#}"), "STATUS_REPLY failed");
}
}
- transport::ParsedMessage::Pong(remote) => {
- // It answers our PING: a round-trip sample.
- state.lock_rtt().pong_received(&peer, Instant::now());
- on_keepalive(state, &peer, &remote, reactivated).await;
+ transport::ParsedMessage::StatusReply(remote) => {
+ on_status(state, &peer, &remote, reactivated).await;
}
- transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) => {}
+ transport::ParsedMessage::Who(_)
+ | transport::ParsedMessage::Me(_)
+ | transport::ParsedMessage::Ping(_)
+ | transport::ParsedMessage::Pong(_) => {}
}
Ok(())
}
-/// What a PING and a PONG have in common: a handshake if needed, the peer's keepalive
-/// noted, and the heal if our documents differ. Returns our keepalive for the peer.
-async fn on_keepalive(
+/// What a STATUS and a STATUS_REPLY have in common: a handshake if needed, the peer's status
+/// noted, and the heal if our documents differ. Returns our status for the peer.
+async fn on_status(
state: &AppState,
peer: &str,
- remote: &transport::Keepalive,
+ remote: &transport::Status,
reactivated: bool,
-) -> transport::Keepalive {
+) -> transport::Status {
if reactivated || peer_swarm(state, peer).await.is_none() {
ensure_handshake(state, peer).await;
}
- note_keepalive(state, peer, remote).await;
- let local = local_keepalive(state, peer).await;
+ note_status(state, peer, remote).await;
+ let local = local_status(state, peer).await;
heal_if_diverged(state, peer, &local, remote).await;
local
}
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index ce2951b..c8a93eb 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -1,12 +1,19 @@
-//! Keepalives (PING / PONG) and the self-healing re-handshake when they show our
-//! documents differ while sync is idle.
+//! Keepalives and the self-healing re-handshake.
+//!
+//! A quiet pair's keepalive (`PeerState::take_keepalive_due`) is a PING, answered with a
+//! PONG by the peer's paths layer (`transport::paths`), over the same path: it tells both
+//! sides the other is alive, and measures the path (`swarm::rtt`). Every `Timing::status_refresh` it is a
+//! STATUS instead, answered with a STATUS_REPLY: each side's state (heads, readiness,
+//! epoch), which also go out as they change (relay announcements, the epoch after a
+//! switch, readiness: `announce`). When a STATUS shows our documents differ while sync
+//! is idle, the pair re-handshakes (`heal_if_diverged`).
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::Result;
use tokio::time::Instant;
-use tracing::warn;
+use tracing::{debug, info, warn};
use super::handshake::ensure_handshake;
use crate::state::{AppState, Liveness, Standing};
@@ -14,37 +21,111 @@ use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, ParsedMessage};
-/// Sends `peer` a PING with our keepalive, noting the time for its round trip.
+/// Sends `peer` a STATUS with our status.
+pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
+ let status = local_status(state, peer).await;
+ if let Some(peer_state) = state.get_peer_state(peer).await {
+ peer_state.lock().await.last_status_at = Some(Instant::now());
+ }
+ state
+ .transport
+ .send_message(peer, &ParsedMessage::Status(status))
+ .await
+}
+
+/// Sends `peer` a PING (the paths layer measures the path with it).
pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
- let keepalive = local_keepalive(state, peer).await;
- state.lock_rtt().ping_sent(peer, Instant::now());
+ let nonce = crate::transport::paths::random_nonce();
state
.transport
- .send_message(peer, &ParsedMessage::Ping(keepalive))
+ .send_message(peer, &ParsedMessage::Ping(nonce))
.await
}
-/// How long shutting down waits for the last PINGs (`farewell`).
+/// A PING or PONG from `peer`: it is alive, and after a PONG the paths layer has a
+/// new round-trip time to it (`swarm::rtt`). A transport without the paths layer (`Transport::answers_pings`: in tests)
+/// leaves answering a PING to us. Only peers we know already: a PING makes none.
+pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMessage) {
+ let now = Instant::now();
+ match message {
+ ParsedMessage::Ping(nonce) if !state.transport.answers_pings() => {
+ let pong = ParsedMessage::Pong(*nonce);
+ if let Err(error) = state.transport.send_message(peer, &pong).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
+ }
+ }
+ ParsedMessage::Pong(_) => {
+ if let Some(rtt) = state.transport.rtt(peer) {
+ state.lock_rtt().measured(peer, rtt, now);
+ }
+ }
+ _ => {}
+ }
+ let Some(peer_state) = state.get_peer_state(peer).await else {
+ return;
+ };
+ let reactivated = {
+ let mut peer_state = peer_state.lock().await;
+ if matches!(message, ParsedMessage::Ping(_)) {
+ // Its keepalive (or a path probe): the next is ours.
+ peer_state.keepalive_received_at = Some(now);
+ }
+ if peer_state.unreachable_since.is_some() {
+ // Reachable again (`mark_seen` clears it).
+ state.topology_changed.notify_one();
+ }
+ peer_state.mark_seen(now)
+ };
+ crate::swarm::source::heard_from(state, peer);
+ if reactivated {
+ info!(peer = %peer, "Dormant peer is active again; re-handshaking");
+ ensure_handshake(state, peer).await;
+ }
+}
+
+/// How long shutting down waits for the last STATUSes (`farewell`).
const FAREWELL_TIMEOUT: Duration = Duration::from_secs(2);
-/// At shutdown: one last PING to each neighbour, saying we aren't ready. A neighbour
-/// catching up from us picks another source now, instead of once we go dormant. Bounded
-/// in time: our transport may be gone already.
-pub(crate) async fn farewell(state: &AppState) {
- state.stopping.store(true, Ordering::Release);
- let mut pings = tokio::task::JoinSet::new();
+/// A STATUS to each neighbour, at most `within` (all sent: none).
+async fn status_to_neighbours(state: &AppState, within: Option<Duration>) {
+ let mut statuses = tokio::task::JoinSet::new();
for (peer, _) in state.peers_snapshot().await {
if topology::is_neighbour(state, &peer) {
let state = state.clone();
- pings.spawn(async move { send_ping(&state, &peer).await });
+ statuses.spawn(async move {
+ if let Err(error) = send_status(&state, &peer).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "STATUS failed");
+ }
+ });
}
}
- let all_sent = async { while pings.join_next().await.is_some() {} };
- let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await;
+ let all_sent = async { while statuses.join_next().await.is_some() {} };
+ match within {
+ Some(within) => {
+ let _ = tokio::time::timeout(within, all_sent).await;
+ }
+ None => all_sent.await,
+ }
+}
+
+/// Our readiness changed (we became ready, or ready again after a rejoin): a STATUS
+/// to each neighbour now, rather than with the next refresh. A peer opens sync towards
+/// us only once it knows we are ready.
+pub(crate) fn announce(state: &AppState) {
+ let state = state.clone();
+ tokio::spawn(async move { status_to_neighbours(&state, None).await });
+}
+
+/// At shutdown: one last STATUS to each neighbour, saying we aren't ready. A neighbour
+/// catching up from us picks another source now, instead of once we go dormant. Bounded
+/// in time: our transport may be gone already.
+pub(crate) async fn farewell(state: &AppState) {
+ state.stopping.store(true, Ordering::Release);
+ status_to_neighbours(state, Some(FAREWELL_TIMEOUT)).await;
}
-/// Our keepalive for `peer`.
-pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
+/// Our status for `peer`.
+pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Status {
// Heads and epoch are published together (`Replica::stamps`), so they always match,
// and reading them doesn't wait for a sync message being applied.
let (heads, epoch, epoch_seq) = {
@@ -64,7 +145,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
}
None => false,
};
- transport::Keepalive {
+ transport::Status {
heads,
epoch,
epoch_seq,
@@ -73,8 +154,19 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
}
}
-/// What a peer's keepalive says about it.
-pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) {
+/// A STATUS (not a reply) from `peer`: its keepalive, and a STATUS between us
+/// (`Timing::status_refresh`).
+pub(super) async fn status_received(state: &AppState, peer: &str) {
+ if let Some(peer_state) = state.get_peer_state(peer).await {
+ let mut peer_state = peer_state.lock().await;
+ let now = Instant::now();
+ peer_state.keepalive_received_at = Some(now);
+ peer_state.last_status_at = Some(now);
+ }
+}
+
+/// What a peer's status says about it.
+pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
let became_ready = remote.ready && peer_state.reports_ready != Some(true);
@@ -103,8 +195,8 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp
pub(super) async fn heal_if_diverged(
state: &AppState,
peer: &str,
- local: &transport::Keepalive,
- remote: &transport::Keepalive,
+ local: &transport::Status,
+ remote: &transport::Status,
) {
// Heads of documents from different epochs are never equal and never converge.
// And a peer that isn't our neighbour isn't ours to keep in step.
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 0249f3b..326f3a9 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -2,7 +2,8 @@
//! swarm-wide forget records, and dispatch of inbound messages.
//!
//! * `handshake`: discovering peers, HELLO / HELLO_REPLY, membership.
-//! * `keepalive`: PING / PONG and the self-healing re-handshake.
+//! * `keepalive`: PING / PONG and STATUS / STATUS_REPLY, and the self-healing
+//! re-handshake.
//! * `epoch`: the epochs peers advertise.
//! * `admit`: admission with a swarm key.
//! * `forget`: swarm-wide forget records.
@@ -24,6 +25,6 @@ pub(crate) use handshake::{
apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once,
};
pub(crate) use inbound::handle_message;
-pub(crate) use keepalive::{farewell, send_ping};
+pub(crate) use keepalive::{announce, farewell, send_status};
pub(crate) use lanes::SyncLanes;
pub(crate) use worker::run_peer_worker;
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 057e4bd..f3d0e84 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -10,9 +10,9 @@ use tracing::{debug, info, warn};
use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
use super::handshake::retry_hello_if_due;
-use super::keepalive::send_ping;
+use super::keepalive::{send_ping, send_status};
use crate::locks::MutexExt;
-use crate::state::AppState;
+use crate::state::{AppState, Keepalive};
use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
@@ -50,16 +50,20 @@ pub(crate) async fn run_peer_worker(state: AppState) {
// Only neighbours are kept alive and judged; others we only answer.
let neighbour = topology::is_neighbour(&state, peer);
- let (became_dormant, ping_due, unanswered) = {
+ let (became_dormant, keepalive, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
let became_dormant =
ready && neighbour && peer_state.check_dormancy(&state.timing, now);
- let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now);
+ let keepalive = if neighbour {
+ peer_state.take_keepalive_due(&state.timing, now)
+ } else {
+ None
+ };
if !neighbour {
peer_state.stop_hello_retries();
}
let unanswered = peer_state.take_overdue_sync_reply(now);
- (became_dormant, ping_due, unanswered)
+ (became_dormant, keepalive, unanswered)
};
if let Some(changes) = unanswered {
@@ -80,12 +84,21 @@ pub(crate) async fn run_peer_worker(state: AppState) {
);
}
- let probe_due =
- !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
- if (ping_due || probe_due)
- && let Err(error) = send_ping(&state, peer).await
- {
- debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed");
+ // The catch-up source gets STATUSes: we need its readiness, and whether it
+ // syncs with us.
+ let keepalive = match keepalive {
+ None if crate::swarm::source::probe_due(&state, peer, Instant::now()) => {
+ Some(Keepalive::Status)
+ }
+ keepalive => keepalive,
+ };
+ let sent = match keepalive {
+ Some(Keepalive::Ping) => send_ping(&state, peer).await,
+ Some(Keepalive::Status) => send_status(&state, peer).await,
+ None => Ok(()),
+ };
+ if let Err(error) = sent {
+ debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed");
}
if let Err(error) = retry_hello_if_due(&state, peer).await {
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 202ae40..6172f67 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -2,7 +2,7 @@
//!
//! A node pushes its own changes to its neighbours, but nothing it receives goes any
//! further by itself. So after applying changes from one neighbour, a node *announces*
-//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100
+//! its new heads to its other neighbours with a STATUS (a keepalive: heads, about 100
//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the
//! announced heads as the peer's heads in its sync state, Automerge asks for what's
//! missing (`need`), and the announcer answers with exactly those changes. Each node
@@ -11,7 +11,7 @@
//!
//! Announcements are coalesced: the first applied change schedules one announcement
//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
-//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the
+//! round costs one STATUS and one STATUS_REPLY per neighbour (plus the pulls), whatever the
//! write rate; the delay is the price per hop. Without coalescing, every applied sync
//! message would announce to every neighbour, and announcements would outnumber the
//! data they announce (about 15 to 1 at 250 nodes).
@@ -27,7 +27,7 @@ use tracing::debug;
use crate::locks::MutexExt;
use crate::state::{AppState, Standing};
-use crate::swarm::peer::send_ping;
+use crate::swarm::peer::send_status;
use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, HeadsSummary};
@@ -97,7 +97,7 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
.cloned()
.collect();
for peer in neighbours {
- if let Err(error) = send_ping(state, &peer).await {
+ if let Err(error) = send_status(state, &peer).await {
debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
}
}
@@ -111,7 +111,7 @@ pub(crate) struct Pulling {
/// The neighbour we are pulling from, and since when.
current: Option<(String, Instant)>,
/// The latest keepalive of each neighbour waiting to be pulled from.
- waiting: HashMap<String, transport::Keepalive>,
+ waiting: HashMap<String, transport::Status>,
}
/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time; a
@@ -119,7 +119,7 @@ pub(crate) struct Pulling {
pub(crate) async fn pull_if_behind(
state: &AppState,
peer: &str,
- remote: &transport::Keepalive,
+ remote: &transport::Status,
) -> bool {
if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
return false;
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
index ed6cc73..cf91ac2 100644
--- a/src/swarm/rtt.rs
+++ b/src/swarm/rtt.rs
@@ -1,15 +1,16 @@
-//! Round-trip times to peers, measured from traffic we send anyway: PING → PONG and
-//! HELLO → HELLO_REPLY. No probes of our own.
+//! Round-trip times to peers. Mostly the paths layer's (`transport::paths`): the
+//! smoothed round-trip time of the path a peer's messages go over, measured by PINGs
+//! (keepalives, and the paths layer's own), taken whenever a PONG from the peer arrives
+//! (`measured`). Before a peer's path is measured (while joining: no keepalives yet),
+//! HELLO → HELLO_REPLY gives one, smoothed (exponential moving average, weight 1/4);
+//! a HELLO_REPLY only counts when exactly one HELLO was outstanding: with two, we can't
+//! tell which one the answer belongs to (an answer to the older one would look too
+//! fast).
//!
-//! A sample only counts when exactly one request was outstanding: with two, we can't tell
-//! which one the answer belongs to (an answer to the older one would look too fast).
-//! Samples are smoothed (exponential moving average, weight 1/4) to ride out jitter.
-//!
-//! What we use is the *effective* round-trip time: the smoothed value, raised to the age
-//! of the oldest request still unanswered (the round trip takes at least that long), so
-//! a peer that stops answering drifts away by itself. Without a recent sample (nodes
-//! move; an old value says little) it is unknown. Silence alone doesn't count: we don't
-//! ask non-neighbours anything, so their quiet is ours.
+//! Without a recent value (nodes move; an old value says little) it is unknown. A lost
+//! PING or HELLO is no sample: a peer that stops answering goes dormant, and is given up
+//! as a catch-up source, by itself. Silence alone doesn't count: we don't ask
+//! non-neighbours anything, so their quiet is ours.
use std::collections::HashMap;
use std::time::Duration;
@@ -18,9 +19,11 @@ use tokio::time::Instant;
#[derive(Debug, Default)]
pub(crate) struct RttTable {
- pings: HashMap<String, Outstanding>,
hellos: HashMap<String, Outstanding>,
- smoothed: HashMap<String, Sample>,
+ /// From the paths layer, as last measured.
+ from_paths: HashMap<String, Sample>,
+ /// From HELLOs, smoothed.
+ from_hellos: HashMap<String, Sample>,
}
#[derive(Debug, Clone, Copy)]
@@ -36,85 +39,73 @@ struct Outstanding {
}
impl RttTable {
- pub(crate) fn ping_sent(&mut self, peer: &str, now: Instant) {
- Self::sent(&mut self.pings, peer, now);
- }
-
- pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) {
- if let Some(sample) = Self::answered(&mut self.pings, peer, now) {
- self.record(peer, sample, now);
- }
+ /// The paths layer's round-trip time to `peer` (`Transport::rtt`), as a PONG from
+ /// it arrived.
+ pub(crate) fn measured(&mut self, peer: &str, rtt: Duration, now: Instant) {
+ self.from_paths
+ .insert(peer.to_owned(), Sample { rtt, at: now });
}
pub(crate) fn hello_sent(&mut self, peer: &str, now: Instant) {
- Self::sent(&mut self.hellos, peer, now);
+ self.hellos
+ .entry(peer.to_owned())
+ .and_modify(|outstanding| outstanding.count += 1)
+ .or_insert(Outstanding {
+ sent: now,
+ count: 1,
+ });
}
pub(crate) fn hello_reply_received(&mut self, peer: &str, now: Instant) {
- if let Some(sample) = Self::answered(&mut self.hellos, peer, now) {
- self.record(peer, sample, now);
+ let Some(outstanding) = self.hellos.remove(peer) else {
+ return;
+ };
+ if outstanding.count != 1 {
+ return;
}
+ let sample = now.saturating_duration_since(outstanding.sent);
+ let rtt = match self.from_hellos.get(peer) {
+ Some(old) => (old.rtt * 3 + sample) / 4,
+ None => sample,
+ };
+ self.from_hellos
+ .insert(peer.to_owned(), Sample { rtt, at: now });
}
- /// The effective round-trip time: the smoothed value if sampled within `fresh_for`,
- /// raised to the age of the oldest unanswered request; None without a recent sample.
+ /// The round-trip time: the paths layer's if measured within `fresh_for`, else
+ /// HELLOs' if sampled within it; None without either.
pub(crate) fn effective(
&self,
peer: &str,
now: Instant,
fresh_for: Duration,
) -> Option<Duration> {
- let sample = self
- .smoothed
+ let fresh = |sample: &&Sample| now.saturating_duration_since(sample.at) < fresh_for;
+ self.from_paths
.get(peer)
- .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?;
- let waiting = [self.pings.get(peer), self.hellos.get(peer)]
- .into_iter()
- .flatten()
- .map(|outstanding| now.saturating_duration_since(outstanding.sent))
- .max()
- .unwrap_or_default();
- Some(sample.rtt.max(waiting))
+ .filter(fresh)
+ .or_else(|| self.from_hellos.get(peer).filter(fresh))
+ .map(|sample| sample.rtt)
}
- /// The effective round-trip time of every peer that has one.
+ /// The round-trip time of every peer that has one.
pub(crate) fn all_effective(
&self,
now: Instant,
fresh_for: Duration,
) -> impl Iterator<Item = (&str, Duration)> {
- self.smoothed.keys().filter_map(move |peer| {
+ let mut peers: Vec<&String> = self
+ .from_paths
+ .keys()
+ .chain(self.from_hellos.keys())
+ .collect();
+ peers.sort_unstable();
+ peers.dedup();
+ peers.into_iter().filter_map(move |peer| {
self.effective(peer, now, fresh_for)
.map(|rtt| (peer.as_str(), rtt))
})
}
-
- fn sent(map: &mut HashMap<String, Outstanding>, peer: &str, now: Instant) {
- map.entry(peer.to_owned())
- .and_modify(|outstanding| outstanding.count += 1)
- .or_insert(Outstanding {
- sent: now,
- count: 1,
- });
- }
-
- fn answered(
- map: &mut HashMap<String, Outstanding>,
- peer: &str,
- now: Instant,
- ) -> Option<Duration> {
- let outstanding = map.remove(peer)?;
- (outstanding.count == 1).then(|| now.saturating_duration_since(outstanding.sent))
- }
-
- fn record(&mut self, peer: &str, sample: Duration, now: Instant) {
- let rtt = match self.smoothed.get(peer) {
- Some(old) => (old.rtt * 3 + sample) / 4,
- None => sample,
- };
- self.smoothed
- .insert(peer.to_owned(), Sample { rtt, at: now });
- }
}
#[cfg(test)]
@@ -124,55 +115,57 @@ mod tests {
const FRESH: Duration = Duration::from_secs(300);
#[test]
- fn samples_only_unambiguous_answers_and_smooths() {
+ fn hellos_sample_only_unambiguous_answers_and_smooth() {
let mut table = RttTable::default();
let start = Instant::now();
let at = |ms| start + Duration::from_millis(ms);
- table.ping_sent("a", start);
- table.pong_received("a", at(100));
+ table.hello_sent("a", start);
+ table.hello_reply_received("a", at(100));
assert_eq!(
table.effective("a", at(100), FRESH),
Some(Duration::from_millis(100))
);
-
- table.ping_sent("a", at(100));
- table.pong_received("a", at(300));
+ table.hello_sent("a", at(100));
+ table.hello_reply_received("a", at(300));
assert_eq!(
table.effective("a", at(300), FRESH),
Some(Duration::from_millis(125))
);
-
- // Two PINGs out: the PONG can't be attributed.
- table.ping_sent("a", at(300));
- table.ping_sent("a", at(400));
- table.pong_received("a", at(450));
+ // Two HELLOs out: the HELLO_REPLY can't be attributed.
+ table.hello_sent("a", at(300));
+ table.hello_sent("a", at(400));
+ table.hello_reply_received("a", at(410));
assert_eq!(
- table.effective("a", at(450), FRESH),
+ table.effective("a", at(410), FRESH),
Some(Duration::from_millis(125))
);
-
- // An unsolicited PONG changes nothing.
- table.pong_received("b", start);
+ // An unsolicited HELLO_REPLY changes nothing.
+ table.hello_reply_received("b", start);
assert_eq!(table.effective("b", start, FRESH), None);
}
#[test]
- fn unanswered_requests_raise_it_and_old_samples_expire() {
+ fn the_path_wins_nothing_waits_and_old_values_expire() {
let mut table = RttTable::default();
let start = Instant::now();
- table.ping_sent("a", start);
- table.pong_received("a", start + Duration::from_millis(50));
-
- // A PING unanswered for 40 s: at least 40 s.
- let sent = start + Duration::from_secs(60);
- table.ping_sent("a", sent);
- let later = sent + Duration::from_secs(40);
+ table.hello_sent("a", start);
+ table.hello_reply_received("a", start + Duration::from_millis(80));
+ table.measured("a", Duration::from_millis(20), start);
assert_eq!(
- table.effective("a", later, FRESH),
- Some(Duration::from_secs(40))
+ table.effective("a", start, FRESH),
+ Some(Duration::from_millis(20))
);
-
- // No sample within the freshness window: unknown.
+ // A HELLO unanswered for long raises nothing.
+ table.hello_sent("a", start);
+ assert_eq!(
+ table.effective("a", start + Duration::from_secs(40), FRESH),
+ Some(Duration::from_millis(20))
+ );
+ assert_eq!(
+ table.all_effective(start, FRESH).collect::<Vec<_>>(),
+ [("a", Duration::from_millis(20))]
+ );
+ // No value within the freshness window: unknown.
let stale = start + FRESH + Duration::from_secs(1);
assert_eq!(table.effective("a", stale, FRESH), None);
}
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index 699b03b..d80e4d7 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -12,7 +12,7 @@
//! A source must be ready and willing to sync with us, as far as we know (its sync
//! messages and keepalives say so). It counts as alive while any message from it
//! arrives, so a slow transfer of a large document never looks like a stall. When it
-//! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or
+//! is quiet for `Timing::catch_up_probe`, we send it a STATUS; after three unanswered probes, or
//! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the
//! first time may be a handshake still under way), the next peer takes over.
//!
@@ -268,7 +268,7 @@ pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us:
}
}
-/// Whether to PING `peer` now: it is the source and has been quiet for a probe interval.
+/// Whether to send `peer` a STATUS now: it is the source and has been quiet for a probe interval.
pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
if !catching_up(state) {
return false;
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index 308c2a5..b5548d5 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -96,7 +96,7 @@ pub(crate) struct SavedSize {
/// Passing changes on (`swarm::relay`), since start.
#[derive(Debug, Clone, Copy, Default)]
pub(crate) struct Relay {
- /// Announcement rounds (each a PING to the neighbours).
+ /// Announcement rounds (each a STATUS to the neighbours).
pub(crate) announce_rounds: u64,
/// Pulls started.
pub(crate) pulls: u64,
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 011f049..df4bb2d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -39,7 +39,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
return Ok(());
}
let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
- trace!("Peer was forgotten; skipping synchronization");
+ trace!("Peer was forgotten; skipping sync");
return Ok(());
};
@@ -50,7 +50,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
if matches!(peer_state.liveness, Liveness::Dormant { .. }) {
// Reactivation re-handshakes, and PeerValidated then restarts sync.
- trace!("Skipping synchronization with dormant peer");
+ trace!("Skipping sync with dormant peer");
return Ok(());
}
@@ -59,7 +59,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
.is_some_and(|epoch| epoch != state.local_epoch())
{
// Unrelated document histories must never be synced.
- trace!("Skipping synchronization with peer on another epoch");
+ trace!("Skipping sync with peer on another epoch");
return Ok(());
}
@@ -222,7 +222,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
tokio::spawn(async move {
if let Err(error) = send_generated(&state, &peer, &peer_state_mutex, session, encoded).await
{
- warn!(peer = %peer, error = %format!("{error:#}"), "Synchronization processing failed");
+ warn!(peer = %peer, error = %format!("{error:#}"), "Sync processing failed");
}
});
Ok(())
@@ -393,7 +393,7 @@ pub(crate) async fn handle_sync(
peer_state
.sync
.receive(&mut replica.doc, message)
- .context("Automerge synchronization failed")?;
+ .context("Automerge sync failed")?;
peer_state.last_sync_rx = Some(Instant::now());
if let Some(wait) = &peer_state.sync.reply {
throughput
@@ -551,7 +551,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
let mut dirty_peers = HashSet::<String>::new();
let mut document_dirty = false;
- info!("Starting synchronization worker");
+ info!("Starting the sync worker");
loop {
if dirty_peers.is_empty() && !document_dirty {
@@ -567,7 +567,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
}
},
None => {
- warn!("Synchronization event channel closed");
+ warn!("Sync event channel closed");
return;
}
}
@@ -617,7 +617,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
warn!(
peer = %peer,
error = %format!("{error:#}"),
- "Synchronization processing failed"
+ "Sync processing failed"
);
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 07:37:36 +0000