From 9a165fd06a1b52bb943a84f26adf8dec72056143 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 19:21:13 +0100 Subject: add pull queue --- src/swarm/peer/keepalive.rs | 2 +- src/swarm/relay.rs | 110 +++++++++++++++++++++++++++++++------------- src/swarm/sync.rs | 6 ++- 3 files changed, 84 insertions(+), 34 deletions(-) (limited to 'src/swarm') diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 4ce326e..a593ac9 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -66,7 +66,7 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp { let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone()); tokio::spawn(async move { - crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await + crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await; }); } if became_ready { diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 34451c9..7c56c90 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -18,7 +18,7 @@ //! //! In a full mesh the writer reaches everyone itself, so nothing is announced. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::sync::atomic::Ordering; use std::time::Duration; @@ -42,7 +42,11 @@ const ANNOUNCE_DELAY: Duration = Duration::from_millis(200); /// After changes from `from` were applied and our heads changed: tell our other /// neighbours, in the next announcement round. pub(crate) async fn announce(state: &AppState, from: &str) { - if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) { + // While the view is unknown everyone counts as a neighbour, but that isn't a full + // mesh: the others may not hear from the writer. + if !state.initial_sync_done.load(Ordering::Acquire) + || (topology::view_known(state) && topology::is_full_mesh(state)) + { return; } let schedule = { @@ -98,15 +102,31 @@ async fn announce_now(state: &AppState, senders: &HashSet) { } } -/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time). -pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transport::Keepalive) { +/// Pulls in progress: one neighbour at a time, so each change arrives about once. The +/// neighbours that announced heads we lack meanwhile wait their turn, with what they +/// announced, instead of until their next keepalive. +#[derive(Default)] +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, +} + +/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time; a +/// busy slot queues the peer). Returns whether a pull started. +pub(crate) async fn pull_if_behind( + state: &AppState, + peer: &str, + remote: &transport::Keepalive, +) -> bool { if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) { - return; + return false; } // The common case, the same heads, needs only the published heads. let ours = HeadsSummary::of(state.stamp.borrow().heads.to_vec()); if ours == remote.heads { - return; + return false; } let missing = match &remote.heads { HeadsSummary::Heads(heads) => { @@ -116,7 +136,7 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp .iter() .any(|head| replica.doc.get_change_by_hash(head).is_none()); if !missing { - return; + return false; } Some(heads.clone()) } @@ -124,47 +144,75 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp HeadsSummary::Digest(_) => None, }; + let Some(peer_state) = state.get_peer_state(peer).await else { + return false; + }; + if !matches!(peer_state.lock().await.standing, Standing::Member { .. }) { + return false; + } + let now = Instant::now(); { let mut pulling = state.pulling.lock_or_recover(); let busy = matches!( - &*pulling, + &pulling.current, Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT ); if busy { - return; + pulling.waiting.insert(peer.to_owned(), remote.clone()); + return false; } - *pulling = Some((peer.to_owned(), now)); + pulling.current = Some((peer.to_owned(), now)); + pulling.waiting.remove(peer); } - let Some(peer_state) = state.get_peer_state(peer).await else { - return; - }; - { - let mut peer_state = peer_state.lock().await; - if !matches!(peer_state.standing, Standing::Member { .. }) { - return; - } - match missing { - Some(heads) => { - // Automerge asks for whatever these heads need that we don't have. - peer_state.sync.pull_from(heads); - } - // A fresh session's opening carries our heads and bloom filter; the answer - // brings what we lack. - None => peer_state.reset_sync(), - } + match missing { + // Automerge asks for whatever these heads need that we don't have. + Some(heads) => peer_state.lock().await.sync.pull_from(heads), + // A fresh session's opening carries our heads and bloom filter; the answer + // brings what we lack. + None => peer_state.lock().await.reset_sync(), } debug!(peer = %peer, "Neighbour has changes we lack; pulling"); let _ = state .sync_events .send(SyncEvent::PeerNeedsSync(peer.to_owned())); + true } -/// A sync message from `peer` was applied: a pull from it is answered. +/// A sync message from `peer` was applied: a pull from it is answered, and the next +/// waiting neighbour's turn comes. pub(crate) fn pulled(state: &AppState, peer: &str) { - let mut pulling = state.pulling.lock_or_recover(); - if pulling.as_ref().is_some_and(|(p, _)| p == peer) { - *pulling = None; + { + let mut pulling = state.pulling.lock_or_recover(); + if !pulling.current.as_ref().is_some_and(|(p, _)| p == peer) { + return; + } + pulling.current = None; + if pulling.waiting.is_empty() { + return; + } + } + let state = state.clone(); + tokio::spawn(async move { pull_next(&state).await }); +} + +/// Pulls from the first waiting neighbour that still has heads we lack. +async fn pull_next(state: &AppState) { + loop { + let (peer, remote) = { + let mut pulling = state.pulling.lock_or_recover(); + if pulling.current.is_some() { + return; + } + let Some(peer) = pulling.waiting.keys().next().cloned() else { + return; + }; + let remote = pulling.waiting.remove(&peer).expect("just listed"); + (peer, remote) + }; + if pull_if_behind(state, &peer, &remote).await { + return; + } } } diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 868fead..0fbff05 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -400,7 +400,8 @@ pub(crate) async fn handle_sync( source::reported(state, peer, from_ready, None); - if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) { + let just_caught_up = caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel); + if just_caught_up { info!(peer = %peer, "Initial sync complete"); state .stats @@ -411,7 +412,8 @@ pub(crate) async fn handle_sync( state.notify_store_changed(); crate::swarm::relay::pulled(state, peer); - if changed { + // Caught up: our other neighbours may lack what the source had. + if changed || just_caught_up { crate::swarm::relay::announce(state, peer).await; } -- cgit v1.3.1