From e293047c25d7167f25af0a611b33545f12225b05 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 20:36:17 +0100 Subject: peer worker sleeps until deadline, not fixed time --- src/node.rs | 1 + src/state/mod.rs | 5 +++ src/state/peer.rs | 73 +++++++++++++++++++++++++++++++++----------- src/swarm/peer/handshake.rs | 8 +++-- src/swarm/peer/keepalive.rs | 10 ++++++ src/swarm/peer/mod.rs | 2 +- src/swarm/peer/worker.rs | 74 +++++++++++++++++++++++++++++++++++++++------ src/swarm/source.rs | 16 ++++++++++ src/swarm/sync.rs | 11 +++++++ src/swarm/topology/mod.rs | 2 ++ 10 files changed, 170 insertions(+), 32 deletions(-) (limited to 'src') diff --git a/src/node.rs b/src/node.rs index 6f55ed5..6de8f0c 100644 --- a/src/node.rs +++ b/src/node.rs @@ -129,6 +129,7 @@ impl Node { neighbours: Default::default(), topology_changed: Default::default(), join_progress: Default::default(), + peers_due: Default::default(), pulling: Default::default(), announce_pending: Default::default(), rtt: Default::default(), diff --git a/src/state/mod.rs b/src/state/mod.rs index 2902234..ba90269 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -154,6 +154,8 @@ pub(crate) struct AppState { /// Something a joining node waits for may have happened: the swarm ID learned, the /// initial sync done, a rejoin ended, a seed found (`swarm::bootstrap`). pub(crate) join_progress: Arc, + /// When the peer worker is next due (`swarm::peer::Due`). + pub(crate) peers_due: Arc, /// While rejoining a newer epoch, the previous document, served read-only (and /// marked stale) instead of answering 503. @@ -319,6 +321,8 @@ impl AppState { let old_lifecycle = lifecycle.clone(); *lifecycle = NodeLifecycle::Ready { swarm_id }; debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}"); + // Dormancy is judged once ready, and the periodic checks run at once. + self.peers_due.now(); } NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => { debug!("Node already ready with the same identity"); @@ -366,6 +370,7 @@ impl AppState { /// State for a peer we haven't known so far (not inserted). pub(crate) fn new_peer_state(&self, pk: &str) -> Arc> { let leads = crate::swarm::pair::leads(self.local_id(), pk); + self.peers_due.now(); Arc::new(Mutex::new(PeerState::new(leads, &self.timing))) } diff --git a/src/state/peer.rs b/src/state/peer.rs index 3e1332d..cb42dc1 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -392,6 +392,26 @@ impl PeerState { timing: &Timing, now: Instant, ) -> Option { + if self.keepalive_due_at(timing).is_none_or(|at| now < at) { + return None; + } + self.last_keepalive_at = Some(now); + let status = self + .last_status_at + .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh); + Some(if status { + Keepalive::Status + } else { + Keepalive::Ping + }) + } + + /// When the next keepalive is due (`take_keepalive_due`), if any is: none to a + /// dormant peer, nor before the handshake. + pub(crate) fn keepalive_due_at(&self, timing: &Timing) -> Option { + if self.liveness.is_dormant() || self.hello_state != HelloState::Established { + return None; + } let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) { (None, None) => self.leads, (None, Some(_)) => true, @@ -407,27 +427,44 @@ impl PeerState { } else { timing.keepalive + timing.keepalive / 2 }; - let due = !self.liveness.is_dormant() - && self.hello_state == HelloState::Established - && now.saturating_duration_since(self.added_at) >= timing.keepalive - && self.quiet_for(now) >= quiet_enough - && self - .last_keepalive_at - .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive); - if !due { - return None; - } - self.last_keepalive_at = Some(now); - let status = self - .last_status_at - .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh); - Some(if status { - Keepalive::Status - } else { - Keepalive::Ping + let quiet_since = self.last_seen.unwrap_or(self.added_at); + let at = (self.added_at + timing.keepalive).max(quiet_since + quiet_enough); + Some(match self.last_keepalive_at { + Some(sent) => at.max(sent + timing.keepalive), + None => at, }) } + /// When the peer worker next has something to do for this peer + /// (`swarm::peer::run_peer_worker`): a keepalive (to a `neighbour`), dormancy (of a + /// neighbour, once `ready`), a HELLO retry, an overdue sync answer. None if nothing + /// is due until something changes. + pub(crate) fn next_due( + &self, + timing: &Timing, + neighbour: bool, + ready: bool, + ) -> Option { + let dormancy = (ready && neighbour && !self.liveness.is_dormant()) + .then(|| self.last_seen.unwrap_or(self.added_at) + timing.dormant_after); + let keepalive = if neighbour { + self.keepalive_due_at(timing) + } else { + None + }; + let hello = match self.hello_state { + HelloState::AwaitingReply { next_attempt } => Some(next_attempt), + _ => None, + }; + let reply = (!self.liveness.is_dormant() && self.hello_state == HelloState::Established) + .then(|| self.sync.reply.as_ref().map(|wait| wait.deadline)) + .flatten(); + [dormancy, keepalive, hello, reply] + .into_iter() + .flatten() + .min() + } + /// Self-healing check on a STATUS or STATUS_REPLY. Returns true if a re-handshake /// should start. pub(crate) fn needs_heal( diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs index 13a1264..896ffdf 100644 --- a/src/swarm/peer/handshake.rs +++ b/src/swarm/peer/handshake.rs @@ -149,9 +149,9 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { if peer_state.hello_state == HelloState::Sending { topology::change_peer(state, &mut peer_state, |p| p.hello_sent(Instant::now())); let wait = peer_state.next_hello_wait(); - peer_state.hello_state = HelloState::AwaitingReply { - next_attempt: Instant::now() + wait, - }; + let next_attempt = Instant::now() + wait; + peer_state.hello_state = HelloState::AwaitingReply { next_attempt }; + state.peers_due.by(next_attempt); } } @@ -413,6 +413,8 @@ pub(super) async fn establish_peer( if from_hello || peer_state.hello_state != HelloState::Established { peer_state.hello_state = HelloState::Established; peer_state.reset_sync(); + // Keepalives start. + state.peers_due.now(); info!(peer = %peer, from_hello, "Handshake complete; sync state reset"); } diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 11817cc..8597dcd 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -71,6 +71,9 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes if ping { // Its keepalive (or a path probe): the next is ours. peer_state.keepalive_received_at = Some(now); + if let Some(at) = peer_state.keepalive_due_at(&state.timing) { + state.peers_due.by(at); + } } }) .await; @@ -181,6 +184,9 @@ pub(super) async fn status_received(state: &AppState, peer: &str) { let now = Instant::now(); peer_state.keepalive_received_at = Some(now); peer_state.last_status_at = Some(now); + if let Some(at) = peer_state.keepalive_due_at(&state.timing) { + state.peers_due.by(at); + } } } @@ -195,6 +201,10 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport // its HELLO when it's back, makes it active again). remote.leaving && p.say_farewell(Instant::now()) }); + if left { + // Probed from now on. + state.peers_due.now(); + } let became_ready = remote.ready && peer_state.reports_ready != Some(true); peer_state.reports_ready = Some(remote.ready); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs index 326f3a9..589f48c 100644 --- a/src/swarm/peer/mod.rs +++ b/src/swarm/peer/mod.rs @@ -27,4 +27,4 @@ pub(crate) use handshake::{ pub(crate) use inbound::handle_message; pub(crate) use keepalive::{announce, farewell, send_status}; pub(crate) use lanes::SyncLanes; -pub(crate) use worker::run_peer_worker; +pub(crate) use worker::{Due, run_peer_worker}; diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index 9b62d24..546f867 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -1,11 +1,11 @@ -//! The peer worker: every `PEER_TICK`, keepalives due, dormancy, HELLO retries and -//! overdue sync replies; every `FORGET_CHECK_INTERVAL` (and at once when the node -//! becomes ready), forget records, the member registry, stability, measurements and -//! the state file. +//! The peer worker: keepalives due, dormancy, HELLO retries and overdue sync replies, +//! when the earliest of them is due (or something may have made one due sooner, +//! `Due`); every `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready), +//! forget records, the member registry, stability, measurements and the state file. use std::time::Duration; -use tokio::time::{Instant, MissedTickBehavior}; +use tokio::time::Instant; use tracing::{debug, info, warn}; use super::forget::{AppliedForgets, apply_forget_records, forget_strangers}; @@ -15,14 +15,49 @@ use crate::locks::MutexExt; use crate::state::{AppState, Keepalive}; use crate::swarm::topology; -const PEER_TICK: Duration = Duration::from_millis(500); +/// The least time between two passes, however often something is due. +const MIN_PASS_GAP: Duration = Duration::from_millis(20); const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5); -pub(crate) async fn run_peer_worker(state: AppState) { - let mut tick = tokio::time::interval(PEER_TICK); - tick.set_missed_tick_behavior(MissedTickBehavior::Delay); +/// When the peer worker is next due. What makes something of its due sooner than it +/// knows (a HELLO sent, a sync message waiting for its answer, a keepalive received, a +/// new peer, new neighbours) says so (`by`); what only puts things off needn't. +#[derive(Default)] +pub(crate) struct Due { + /// Its next pass while it waits; None while it runs. + next: std::sync::Mutex>, + wake: tokio::sync::Notify, +} + +impl Due { + /// Something is due at `at`: a pass then at the latest. + pub(crate) fn by(&self, at: Instant) { + if self.next.lock_or_recover().is_none_or(|next| at < next) { + self.wake.notify_one(); + } + } + + /// A pass now. + pub(crate) fn now(&self) { + self.wake.notify_one(); + } + + fn passing(&self) { + *self.next.lock_or_recover() = None; + } + /// Until `at`, or until something is due sooner. + async fn until(&self, at: Instant) { + *self.next.lock_or_recover() = Some(at); + tokio::select! { + () = self.wake.notified() => {} + () = tokio::time::sleep_until(at) => {} + } + } +} + +pub(crate) async fn run_peer_worker(state: AppState) { info!("Starting peer maintenance worker"); let mut last_forget_check: Option = None; @@ -30,7 +65,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { let mut was_ready = false; loop { - tick.tick().await; + state.peers_due.passing(); + let mut next = Instant::now() + FORGET_CHECK_INTERVAL; // While joining, keep handshakes fast; dormancy would slow probes to minutes. let ready = state.is_ready().await; @@ -103,6 +139,17 @@ pub(crate) async fn run_peer_worker(state: AppState) { if let Err(error) = retry_hello_if_due(&state, peer).await { warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed"); } + + let due = peer_state_mutex + .lock() + .await + .next_due(&state.timing, neighbour, ready); + let probe = crate::swarm::source::probe_at(&state, peer); + next = [Some(next), due, probe] + .into_iter() + .flatten() + .min() + .unwrap_or(next); } if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) { @@ -126,5 +173,12 @@ pub(crate) async fn run_peer_worker(state: AppState) { state.topology_changed.notify_one(); last_forget_check = Some(Instant::now()); } + if let Some(checked) = last_forget_check { + next = next.min(checked + FORGET_CHECK_INTERVAL); + } + + let passed = Instant::now(); + state.peers_due.until(next).await; + tokio::time::sleep_until(passed + MIN_PASS_GAP).await; } } diff --git a/src/swarm/source.rs b/src/swarm/source.rs index e098dc1..1288d86 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -146,6 +146,7 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt last_probe: None, refusals: 0, }); + state.peers_due.by(now + state.timing.catch_up_probe); None } @@ -285,6 +286,21 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool { due } +/// When a STATUS to `peer` is next due (`probe_due`), if it is the source. +pub(crate) fn probe_at(state: &AppState, peer: &str) -> Option { + if !catching_up(state) { + return None; + } + let probe = state.timing.catch_up_probe; + let source = lock(state); + let current = source.as_ref().filter(|current| current.peer == peer)?; + let heard = current.last_heard + probe; + Some(match current.last_probe { + Some(last) => heard.max(last + probe), + None => heard, + }) +} + /// No catch-up going on: no source, no head start, nobody avoided. When one ends, and /// before one starts (a rejoin: `Host::reset_initial_sync`). pub(crate) fn reset(state: &AppState) { diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index c643004..c8564c2 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -198,6 +198,17 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let Some(encoded) = generated else { return Ok(()); }; + // Its answer is due. + let reply_due = peer_state_mutex + .lock() + .await + .sync + .reply + .as_ref() + .map(|wait| wait.deadline); + if let Some(at) = reply_due { + state.peers_due.by(at); + } (session, encoded) }; diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index e494f34..30fa56e 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -368,6 +368,8 @@ pub(crate) async fn refresh(state: &AppState) -> Option { outbound, neighbours: computed, }); + // New neighbours are kept alive and judged. + state.peers_due.now(); Some(change) } -- cgit v1.3.1