diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/links.rs | 24 | ||||
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 11 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 97 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 21 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 18 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 1 |
6 files changed, 35 insertions, 137 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs index e0d483d..c430d58 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -1,9 +1,8 @@ //! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a //! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its //! neighbours and the peers it holds dormant (`own_links`), each with the transport its -//! messages go over, whether only through a relay, the round trip, whether it is -//! dormant, and which way the link is eager (`swarm::relay`): as the member measures -//! them, which no other node can. +//! messages go over, whether only through a relay, the round trip and whether it is +//! dormant: as the member measures them, which no other node can. //! //! Every node answers, whatever its build: one with a dashboard asks any member, which //! may have none. Only members (in the member list) are answered, as the links name @@ -11,8 +10,8 @@ //! is written to the document: asked for while someone looks, forgotten after. //! //! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again -//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, eager, -//! or a round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone +//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a +//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone //! subscribes (round trips change without telling), until the subscription runs out //! (the asker renews it with its next LINKS, so one that is gone stops it by not //! renewing) or a LINKS with 0 ends it. @@ -232,14 +231,7 @@ impl Links { pub(crate) async fn own_links(state: &AppState) -> Vec<Link> { let mut links = Vec::new(); for (peer, peer_state) in state.peers_snapshot().await { - let (dormant, eager_to, eager_from) = { - let peer_state = peer_state.lock().await; - ( - peer_state.liveness.is_dormant(), - peer_state.eager_to, - peer_state.eager_from, - ) - }; + let dormant = peer_state.lock().await.liveness.is_dormant(); if !dormant && !topology::is_neighbour(state, &peer) { continue; } @@ -264,8 +256,6 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> { .rtt(&peer) .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)), dormant, - eager_to, - eager_from, clock: state.transport.clock_offset(&peer), peer, }); @@ -1020,8 +1010,6 @@ fn changed(old: &[Link], new: &[Link]) -> bool { || a.relayed != b.relayed || a.down != b.down || a.dormant != b.dormant - || a.eager_to != b.eager_to - || a.eager_from != b.eager_from || moved(a.rtt_us, b.rtt_us) || clock_moved(a.clock, b.clock) || a.others.len() != b.others.len() @@ -1074,8 +1062,6 @@ mod tests { down: false, rtt_us, dormant: false, - eager_to: false, - eager_from: false, others: Vec::new(), clock: None, } diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index c04b1ad..8362d73 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -179,12 +179,9 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta stamp.epoch_seq, ) }; - let (syncs_with_you, eager) = match state.get_peer_state(peer).await { - Some(peer_state) => { - let peer_state = peer_state.lock().await; - (peer_state.syncs_on(epoch), peer_state.eager_from) - } - None => (false, false), + let syncs_with_you = match state.get_peer_state(peer).await { + Some(peer_state) => peer_state.lock().await.syncs_on(epoch), + None => false, }; transport::Status { heads, @@ -194,7 +191,6 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta syncs_with_you, leaving: state.is_closing(), announcing: false, - eager, reaches: Some(crate::swarm::reach::published(state)), } } @@ -232,7 +228,6 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport peer_state.reports_ready = Some(remote.ready); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); peer_state.last_heads = Some((remote.epoch, remote.heads.clone())); - peer_state.eager_to = remote.eager; if let Some(reaches) = &remote.reaches { peer_state.reaches = Some((Instant::now(), reaches.clone())); } 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. diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index dd56bb0..8a00629 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -102,12 +102,9 @@ pub(crate) struct Relay { pub(crate) pulls: u64, /// Announcements of heads we lacked that had to wait for the pull in progress. pub(crate) pulls_queued: u64, - /// Links made eager (a pull brought changes over them) and lazy again (changes came - /// over them that another path brought first). - pub(crate) grafts: u64, - pub(crate) prunes: u64, - /// Changes forwarded over eager links: a sync message asked for per neighbour. - pub(crate) eager_forwards: u64, + /// Changes passed on along writers' trees (`swarm::tree`): a sync message asked for + /// per child. + pub(crate) tree_forwards: u64, } /// When the initial sync finished, and what it took. @@ -190,16 +187,8 @@ impl NodeStats { self.lock().relay.pulls_queued += 1; } - pub(crate) fn link_grafted(&self) { - self.lock().relay.grafts += 1; - } - - pub(crate) fn link_pruned(&self) { - self.lock().relay.prunes += 1; - } - - pub(crate) fn eager_forward(&self) { - self.lock().relay.eager_forwards += 1; + pub(crate) fn tree_forward(&self) { + self.lock().relay.tree_forwards += 1; } /// A sync message of `bytes` arrived. diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 403e0c1..187dacc 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -376,12 +376,11 @@ pub(crate) async fn handle_sync( let clock = Arc::clone(&state.clock); let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed); - // Some((caught_up, heads changed, a duplicate)) when applied, None when the - // message belongs to another epoch. - tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool, bool)>> { + // Some((caught_up, heads changed)) when applied, None when the message belongs + // to another epoch. + tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> { // Before the lock: decoding a large message takes a while. let message = crate::document::sync::decode(&message)?; - let carries_changes = message.carries_changes(); let mut replica = doc.blocking_write(); // Authoritative check, under the same lock as the apply: the document may @@ -411,9 +410,6 @@ pub(crate) async fn handle_sync( let heads_after = replica.heads().to_vec(); stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; - // Changes we had already, over a link we asked to be eager: the link isn't - // needed (`relay::prune`). - let duplicate = carries_changes && !changed && peer_state.eager_from; if changed && measure_propagation { // Now by the peers' clocks (their median): this node's own clock skew // doesn't count; the writer's still does. @@ -434,14 +430,13 @@ pub(crate) async fn handle_sync( // before it registered itself). check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc), changed, - duplicate, ))) }) .await .context("Sync task panicked")?? }; - let Some((caught_up, changed, duplicate)) = applied else { + let Some((caught_up, changed)) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; @@ -460,10 +455,7 @@ pub(crate) async fn handle_sync( state.notify_store_changed(); - crate::swarm::relay::pulled(state, peer, changed).await; - if duplicate { - crate::swarm::relay::prune(state, peer).await; - } + crate::swarm::relay::pulled(state, peer); // Caught up: our other neighbours may lack what the source had. if changed || just_caught_up { crate::swarm::relay::announce(state, peer, writer.filter(|_| changed)).await; diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs index 489bc22..1dff92e 100644 --- a/src/swarm/tree.rs +++ b/src/swarm/tree.rs @@ -146,6 +146,7 @@ pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Vec<St if let Some(peer_state) = state.get_peer_state(child).await { peer_state.lock().await.tree_writer = Some(writer); } + state.stats.tree_forward(); state.request_sync(child); } if !children.is_empty() { |