aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/relay.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/relay.rs')
-rw-r--r--src/swarm/relay.rs97
1 files changed, 16 insertions, 81 deletions
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<u64>)
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<u64>)
});
}
-/// 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<String>,
@@ -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.