From 8366f7c5de474753dbfbf88431917e79c15e5389 Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 05:21:25 +0100 Subject: replace eager links by trees --- src/swarm/relay.rs | 97 +++++++++--------------------------------------------- 1 file changed, 16 insertions(+), 81 deletions(-) (limited to 'src/swarm/relay.rs') diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 7dbc5b0..f4eaf5a 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -19,15 +19,9 @@ //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). //! -//! Links start *lazy*: changes are announced over them and pulled. A pull that brought -//! changes we lacked makes the link *eager* in that direction (Plumtree's GRAFT): our -//! STATUS asks the neighbour for its changes as soon as it applies them (`eager`), and -//! it sends them right away as sync messages, without announcing or waiting -//! (`forward_eagerly`). Changes we had already over an eager link make it lazy again -//! (Plumtree's PRUNE): another path brought them first. So eager links form where -//! changes actually flow, over links that deliver, and the announcements on the others -//! are what catches whatever they miss. A link starts lazy again whenever its sync does -//! (`PeerState::reset_sync`: a new neighbour, dormancy, a new run of the peer). +//! Changes that say whose they are (the writer's own, and what passes on along its tree) +//! go at once to the receiver's children in the writer's tree (`swarm::tree`), which +//! the announcement round then leaves out. //! //! In a full mesh too: the member list makes every member the writer's neighbour, but //! that says nothing about whether the writer reaches them. One behind NAT may reach a @@ -74,7 +68,6 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option) if !state.initial_sync_done.load(Ordering::Acquire) { return; } - forward_eagerly(state, from).await; let sent = match writer { Some(writer) => crate::swarm::tree::forward(state, writer, from).await, None => Vec::new(), @@ -106,20 +99,6 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option) }); } -/// What `from` sent us, at once to the neighbours that asked for our changes as we apply -/// them (eager links): a sync message each, with what it lacks (the sync worker's). -async fn forward_eagerly(state: &AppState, from: &str) { - for (peer, peer_state) in state.peers_snapshot().await { - if peer == from || !topology::is_neighbour(state, &peer) { - continue; - } - if peer_state.lock().await.eager_to { - state.stats.eager_forward(); - state.request_sync(&peer); - } - } -} - /// One announcement round. The peer whose changes we announce has them already, so a /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. @@ -140,8 +119,7 @@ async fn announce_now(state: &AppState, pending: &Pending) { /// announce has them already, so with a single sender it is skipped; with several, each /// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an /// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have; and -/// one we forward to eagerly, or `sent` the changes along a writer's tree, which has -/// them already. +/// one `sent` the changes along a writer's tree, which has them already. async fn unreached( state: &AppState, senders: &HashSet, @@ -163,15 +141,13 @@ async fn unreached( { continue; } - let peer_state = peer_state.lock().await; - // Sent what we have as we applied it (`forward_eagerly`). - let eager = peer_state.eager_to; let has_ours = peer_state + .lock() + .await .last_heads .as_ref() .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1); - drop(peer_state); - if !eager && !has_ours { + if !has_ours { unreached.push(peer); } } @@ -255,62 +231,21 @@ pub(crate) async fn pull_if_behind( true } -/// A sync message from `peer` was applied (`changed`: it brought changes we lacked): a -/// pull from it is answered, and the next waiting neighbour's turn comes. A pull that -/// brought changes makes the link eager (`graft`). -pub(crate) async fn pulled(state: &AppState, peer: &str, changed: bool) { - let next = { +/// 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.current.as_ref().is_none_or(|(p, _)| p != peer) { return; } pulling.current = None; - !pulling.waiting.is_empty() - }; - if changed { - graft(state, peer).await; - } - if next { - let state = state.clone(); - tokio::spawn(async move { pull_next(&state).await }); - } -} - -/// `peer` brought changes we lacked when we pulled: from now on we want them as it -/// applies them, and our STATUS says so (Plumtree's GRAFT). -async fn graft(state: &AppState, peer: &str) { - if !set_eager_from(state, peer, true).await { - return; - } - state.stats.link_grafted(); - debug!(peer = %peer, "Link eager: we want its changes as it applies them"); - if let Err(error) = announce_heads(state, peer, false).await { - debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is eager failed"); - } -} - -/// Changes we had already came from `peer` over an eager link: another path brings them -/// first, so announcements will do; our STATUS says so (Plumtree's PRUNE). -pub(crate) async fn prune(state: &AppState, peer: &str) { - if !set_eager_from(state, peer, false).await { - return; - } - state.stats.link_pruned(); - debug!(peer = %peer, "Link lazy: its changes reach us first another way"); - if let Err(error) = announce_heads(state, peer, false).await { - debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed"); + if pulling.waiting.is_empty() { + return; + } } -} - -/// Sets whether we want `peer`'s changes eagerly; whether that changed it. -async fn set_eager_from(state: &AppState, peer: &str, eager: bool) -> bool { - let Some(peer_state) = state.get_peer_state(peer).await else { - return false; - }; - let mut peer_state = peer_state.lock().await; - let changed = peer_state.eager_from != eager; - peer_state.eager_from = eager; - changed + let state = state.clone(); + tokio::spawn(async move { pull_next(&state).await }); } /// Pulls from the first waiting neighbour that still has heads we lack. -- cgit v1.3.1