summaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/pair.rs2
-rw-r--r--src/swarm/peer/inbound.rs34
-rw-r--r--src/swarm/peer/keepalive.rs36
-rw-r--r--src/swarm/peer/mod.rs4
-rw-r--r--src/swarm/peer/worker.rs6
-rw-r--r--src/swarm/relay.rs12
-rw-r--r--src/swarm/rtt.rs42
-rw-r--r--src/swarm/source.rs4
-rw-r--r--src/swarm/stats.rs2
9 files changed, 72 insertions, 70 deletions
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<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..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<String, Outstanding>,
+ statuses: HashMap<String, Outstanding>,
hellos: HashMap<String, Outstanding>,
smoothed: HashMap<String, Sample>,
}
@@ -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,
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 20:16:44 +0000