peer worker sleeps until deadline, not fixed time

1 files changed, 64 insertions, 10 deletions

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<Option<Instant>>,
+    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<Instant> = 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;
     }
 }