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/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 ++ 7 files changed, 109 insertions(+), 14 deletions(-) (limited to 'src/swarm') 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