From db3c4e997f03a660ac37a6bdf86c6215726a71bd Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 01:47:47 +0100 Subject: rename PING/PONG to STATUS/STATUS_REPLY --- src/swarm/pair.rs | 2 +- src/swarm/peer/inbound.rs | 34 ++++++++++++++++++---------------- src/swarm/peer/keepalive.rs | 36 ++++++++++++++++++------------------ src/swarm/peer/mod.rs | 4 ++-- src/swarm/peer/worker.rs | 6 +++--- src/swarm/relay.rs | 12 ++++++------ src/swarm/rtt.rs | 42 +++++++++++++++++++++--------------------- src/swarm/source.rs | 4 ++-- src/swarm/stats.rs | 2 +- 9 files changed, 72 insertions(+), 70 deletions(-) (limited to 'src/swarm') diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs index c1a196b..41fc219 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). +//! keepalive STATUSes). //! //! 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..4c7e7a8 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_status}; use super::lanes::SyncLanes; use crate::state::AppState; use crate::swarm::sync::SyncEvent; @@ -143,20 +143,22 @@ 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) => { + 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) => { + // It answers our STATUS: a round-trip sample. + state + .lock_rtt() + .status_reply_received(&peer, Instant::now()); + on_status(state, &peer, &remote, reactivated).await; } transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) => {} } @@ -164,19 +166,19 @@ pub(crate) async fn handle_message( 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..810c377 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -1,4 +1,4 @@ -//! Keepalives (PING / PONG) and the self-healing re-handshake when they show our +//! Keepalives (STATUS / STATUS_REPLY) and the self-healing re-handshake when they show our //! documents differ while sync is idle. use std::sync::atomic::Ordering; @@ -14,37 +14,37 @@ 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. -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()); +/// Sends `peer` a STATUS with our status, noting the time for its round trip. +pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> { + let status = local_status(state, peer).await; + state.lock_rtt().status_sent(peer, Instant::now()); state .transport - .send_message(peer, &ParsedMessage::Ping(keepalive)) + .send_message(peer, &ParsedMessage::Status(status)) .await } -/// How long shutting down waits for the last PINGs (`farewell`). +/// 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 +/// 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); - let mut pings = tokio::task::JoinSet::new(); + 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 { send_status(&state, &peer).await }); } } - let all_sent = async { while pings.join_next().await.is_some() {} }; + let all_sent = async { while statuses.join_next().await.is_some() {} }; let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).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 +64,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport:: } None => false, }; - transport::Keepalive { + transport::Status { heads, epoch, epoch_seq, @@ -73,8 +73,8 @@ 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) { +/// 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 +103,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..415cddf 100644 --- a/src/swarm/peer/mod.rs +++ b/src/swarm/peer/mod.rs @@ -2,7 +2,7 @@ //! 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`: 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 +24,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::{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..c68e2ea 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -10,7 +10,7 @@ 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_status; use crate::locks::MutexExt; use crate::state::AppState; use crate::swarm::sync::SyncEvent; @@ -83,9 +83,9 @@ 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 + && let Err(error) = send_status(&state, peer).await { - debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed"); + debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive STATUS 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) { .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, + waiting: HashMap, } /// 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..01d10ee 100644 --- a/src/swarm/rtt.rs +++ b/src/swarm/rtt.rs @@ -1,4 +1,4 @@ -//! Round-trip times to peers, measured from traffic we send anyway: PING → PONG and +//! Round-trip times to peers, measured from traffic we send anyway: STATUS → STATUS_REPLY and //! HELLO → HELLO_REPLY. No probes of our own. //! //! A sample only counts when exactly one request was outstanding: with two, we can't tell @@ -18,7 +18,7 @@ use tokio::time::Instant; #[derive(Debug, Default)] pub(crate) struct RttTable { - pings: HashMap, + statuses: HashMap, hellos: HashMap, smoothed: HashMap, } @@ -36,12 +36,12 @@ 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 status_sent(&mut self, peer: &str, now: Instant) { + Self::sent(&mut self.statuses, peer, now); } - pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) { - if let Some(sample) = Self::answered(&mut self.pings, peer, now) { + pub(crate) fn status_reply_received(&mut self, peer: &str, now: Instant) { + if let Some(sample) = Self::answered(&mut self.statuses, peer, now) { self.record(peer, sample, now); } } @@ -68,7 +68,7 @@ impl RttTable { .smoothed .get(peer) .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?; - let waiting = [self.pings.get(peer), self.hellos.get(peer)] + let waiting = [self.statuses.get(peer), self.hellos.get(peer)] .into_iter() .flatten() .map(|outstanding| now.saturating_duration_since(outstanding.sent)) @@ -128,31 +128,31 @@ mod tests { 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.status_sent("a", start); + table.status_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.status_sent("a", at(100)); + table.status_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 STATUSes out: the STATUS_REPLY can't be attributed. + table.status_sent("a", at(300)); + table.status_sent("a", at(400)); + table.status_reply_received("a", at(450)); assert_eq!( table.effective("a", at(450), FRESH), Some(Duration::from_millis(125)) ); - // An unsolicited PONG changes nothing. - table.pong_received("b", start); + // An unsolicited STATUS_REPLY changes nothing. + table.status_reply_received("b", start); assert_eq!(table.effective("b", start, FRESH), None); } @@ -160,12 +160,12 @@ mod tests { fn unanswered_requests_raise_it_and_old_samples_expire() { let mut table = RttTable::default(); let start = Instant::now(); - table.ping_sent("a", start); - table.pong_received("a", start + Duration::from_millis(50)); + table.status_sent("a", start); + table.status_reply_received("a", start + Duration::from_millis(50)); - // A PING unanswered for 40 s: at least 40 s. + // A STATUS unanswered for 40 s: at least 40 s. let sent = start + Duration::from_secs(60); - table.ping_sent("a", sent); + table.status_sent("a", sent); let later = sent + Duration::from_secs(40); assert_eq!( table.effective("a", later, FRESH), 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, -- cgit v1.3.1