diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/links.rs | 446 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 12 |
2 files changed, 434 insertions, 24 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs index 1d5436e..80def7a 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -15,8 +15,20 @@ //! 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. +//! +//! A member the asker can't reach (on none of its transports: one on Tor alone, asked +//! from one on libp2p alone) is asked through a peer that can (`relay_for`): a LINKS +//! naming the members (`of`) asks the receiver to ask them on the sender's behalf, and +//! to pass their links on (`relayed`). The receiver asks each itself, or through a peer +//! of its own while hops are left (`hops`), for as long as the sender asked, and holds +//! what arrives a while to send it together: the window (`batch_ms`) is the asker's for +//! the whole way (at most `MAX_BATCH`), shared out among the hops that may be left, so +//! the links reach the asker within it however long the chain (every hop holds +//! `MIN_BATCH` at least). Such a +//! subscription isn't ended by a message: it runs out, at every hop, unless the asker +//! renews it, so one that is gone (or no longer looks) stops it within a lease. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Mutex; use std::time::Duration; @@ -28,7 +40,7 @@ use crate::state::AppState; use crate::swarm::{members, topology}; use crate::transport::paths::{self, ClockOffset}; use crate::transport::{ - Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, + Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, RelayedLinks, }; /// Links one reply lists at most (a member has about `--neighbours`, more if others @@ -47,6 +59,15 @@ const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300); const WATCH_EVERY: Duration = Duration::from_secs(2); /// A reply to a LINKS that doesn't subscribe is taken for this long. const REPLY_WITHIN: Duration = Duration::from_secs(30); +/// Members a relayed LINKS names at most, and that one LINKS_REPLY passes on at most. +const MAX_RELAYED: usize = 64; +/// How long relayed links are held to be sent together, along the whole way: the +/// asker's window (at most `MAX_BATCH`), each hop holding at least `MIN_BATCH`. +pub(crate) const DEFAULT_BATCH: Duration = Duration::from_millis(250); +const MIN_BATCH: Duration = Duration::from_millis(10); +const MAX_BATCH: Duration = Duration::from_secs(5); +/// Nodes a relayed LINKS goes through at most, the first included. +const MAX_HOPS: u8 = 3; /// The requests out and the replies in, and when each asker was last answered. #[derive(Default)] @@ -69,6 +90,26 @@ struct Inner { subscribers: HashMap<String, (u64, Instant)>, /// The links subscribers were last sent (`watch`). published: Option<Vec<Link>>, + /// Our relayed LINKS out, by the peer asked to relay: its nonce, until when replies + /// to it are taken, and the members asked for through it. + via: HashMap<String, (u64, Instant, HashSet<String>)>, + /// What we ask on others' behalf, by member asked for: who asked (`Relaying`). + relaying: HashMap<String, HashMap<String, Relaying>>, + /// Relayed links due to an asker, by asker: the members whose links changed. + due: HashMap<String, HashSet<String>>, +} + +/// One asker's relayed subscription to a member. +#[derive(Clone, Copy)] +struct Relaying { + /// The nonce to answer with. + nonce: u64, + /// Until when, if still wanted. + until: Instant, + /// How long its links are held to be sent with others. + window: Duration, + /// Asked for once (no subscription): ends with the first links passed on. + once: bool, } impl Links { @@ -90,16 +131,24 @@ impl Links { .collect() } - /// The members our subscriptions are with (not run out). + /// The members our subscriptions are with, and the peers relaying for us (not run + /// out). #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) fn subscribed(&self) -> Vec<String> { let now = Instant::now(); - self.inner - .lock_or_recover() + let inner = self.inner.lock_or_recover(); + inner .asked .iter() .filter(|(_, (_, until))| now < *until) .map(|(member, _)| member.clone()) + .chain( + inner + .via + .iter() + .filter(|(_, (_, until, _))| now < *until) + .map(|(relay, _)| relay.clone()), + ) .collect() } } @@ -222,18 +271,156 @@ pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) { epoch_seq, nonce, subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX), + of: Vec::new(), + batch_ms: 0, + hops: 0, }); if let Err(error) = state.transport.send_message(member, &request).await { debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); } } +/// Asks `members` for their links (`ask`): those we can reach ourselves, the others +/// through peers that can (`relay_for`, `ask_via`), with our window for what they pass +/// on (`DEFAULT_BATCH`). +#[cfg_attr(not(feature = "dashboard"), allow(dead_code))] +pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: Duration) { + let (direct, through) = split(state, members, None).await; + for member in direct { + let state = state.clone(); + tokio::spawn(async move { ask(&state, &member, subscribe).await }); + } + for (relay, members) in through { + ask_via( + state, + &relay, + members, + subscribe, + DEFAULT_BATCH, + MAX_HOPS - 1, + ) + .await; + } +} + +/// `members` we can reach ourselves, and the others by the peer to ask them through +/// (none for those no peer leads to); not `exclude` (who asked us). +async fn split( + state: &AppState, + members: Vec<String>, + exclude: Option<&str>, +) -> (Vec<String>, HashMap<String, Vec<String>>) { + let mut direct = Vec::new(); + let mut through: HashMap<String, Vec<String>> = HashMap::new(); + for member in members { + if member == state.local_id() || Some(member.as_str()) == exclude { + continue; + } + if state.transport.can_reach(&member) { + direct.push(member); + } else if let Some(relay) = relay_for(state, &member, exclude).await { + through.entry(relay).or_default().push(member); + } + } + (direct, through) +} + +/// The peer to ask `member` through, which we can't reach: an active member we reach +/// that, by the member list, is on a transport `member` is on; else one on a transport +/// we aren't on (it may know one that is). Of several, the one with the lowest pair score +/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`. +async fn relay_for(state: &AppState, member: &str, exclude: Option<&str>) -> Option<String> { + let theirs = topology::transports_of(state, member); + let ours = topology::transports_of(state, state.local_id()); + let mut best: Option<(bool, u64, String)> = None; + for (peer, peer_state) in state.peers_snapshot().await { + if peer == member || Some(peer.as_str()) == exclude || !state.transport.can_reach(&peer) { + continue; + } + { + let peer_state = peer_state.lock().await; + if peer_state.liveness.is_dormant() + || !matches!(peer_state.standing, crate::state::Standing::Member { .. }) + { + continue; + } + } + let on = topology::transports_of(state, &peer); + let shares = !on.is_disjoint(&theirs); + if !shares && on.is_subset(&ours) { + continue; + } + let score = topology::pair_score(&peer, member); + // Sharing a transport with the member first, then the lowest score. + let better = best.as_ref().is_none_or(|(best_shares, best_score, _)| { + (shares, std::cmp::Reverse(score)) > (*best_shares, std::cmp::Reverse(*best_score)) + }); + if better { + best = Some((shares, score, peer)); + } + } + best.map(|(_, _, peer)| peer) +} + +/// Asks `relay` to ask `members` on our behalf (a LINKS naming them), with `subscribe`, +/// passing their links on within `window`, through `hops` more nodes at most. One nonce +/// per relay while its replies are taken, so its replies for every member we asked it +/// for are taken. +async fn ask_via( + state: &AppState, + relay: &str, + members: Vec<String>, + subscribe: Duration, + window: Duration, + hops: u8, +) { + let members: Vec<String> = members.into_iter().take(MAX_RELAYED).collect(); + let nonce = { + let now = Instant::now(); + let mut inner = state.links.inner.lock_or_recover(); + if inner.via.len() >= MAX_KEPT && !inner.via.contains_key(relay) { + inner.via.retain(|_, (_, until, _)| now < *until); + } + let until = now + subscribe.max(REPLY_WITHIN); + let entry = inner.via.entry(relay.to_owned()).or_insert_with(|| { + ( + crate::transport::paths::random_nonce(), + until, + HashSet::new(), + ) + }); + if now >= entry.1 { + entry.0 = crate::transport::paths::random_nonce(); + entry.2.clear(); + } + entry.1 = entry.1.max(until); + entry.2.extend(members.iter().cloned()); + entry.0 + }; + let (epoch, epoch_seq) = state.epoch_stamp(); + let request = ParsedMessage::Links(LinksRequest { + epoch, + epoch_seq, + nonce, + subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX), + of: members, + batch_ms: u32::try_from(window.as_millis()).unwrap_or(u32::MAX), + hops, + }); + if let Err(error) = state.transport.send_message(relay, &request).await { + debug!(relay = %relay, error = %format!("{error:#}"), "Relayed LINKS failed"); + } +} + /// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any -/// more. +/// more. A peer relaying for us isn't told: what it asks for us runs out by itself. #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) async fn unsubscribe(state: &AppState, member: &str) { let nonce = { let mut inner = state.links.inner.lock_or_recover(); + if inner.via.remove(member).is_some() && !inner.asked.contains_key(member) { + return; + } match inner.asked.remove(member) { Some((nonce, _)) => nonce, None => return, @@ -245,6 +432,9 @@ pub(crate) async fn unsubscribe(state: &AppState, member: &str) { epoch_seq, nonce, subscribe_secs: 0, + of: Vec::new(), + batch_ms: 0, + hops: 0, }); if let Err(error) = state.transport.send_message(member, &request).await { debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); @@ -263,6 +453,10 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member"); return; } + if !request.of.is_empty() { + relay(state, peer, request).await; + return; + } { let now = Instant::now(); let mut inner = state.links.inner.lock_or_recover(); @@ -298,33 +492,236 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) epoch_seq, nonce: request.nonce, links: own_links(state).await, + relayed: Vec::new(), }); if let Err(error) = state.transport.send_message(peer, &reply).await { debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); } } -/// A LINKS_REPLY from `peer`: kept if it answers our latest LINKS to it, while its -/// replies are taken (a subscription's go on). -pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) { - let mut inner = state.links.inner.lock_or_recover(); - let taken = inner - .asked - .get(peer) - .is_some_and(|(nonce, until)| *nonce == reply.nonce && Instant::now() < *until); - if !taken { - return; +/// A relayed LINKS from `asker` (`answer`): the members it names are asked on its +/// behalf, ourselves or through a peer of ours while hops are left, and what we know of +/// them already is passed on at once (within its window); the rest as it arrives. +async fn relay(state: &AppState, asker: &str, request: LinksRequest) { + let now = Instant::now(); + let lease = Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION); + let once = lease.is_zero(); + // What is left of the asker's window, shared out among this hop and those that may + // follow: ours is a share, the rest goes on. + let left = Duration::from_millis(u64::from(request.batch_ms)).min(MAX_BATCH); + let window = (left / (u32::from(request.hops) + 1)).max(MIN_BATCH); + let rest = left.saturating_sub(window); + let members: Vec<String> = request + .of + .into_iter() + .filter(|member| member != asker && member != state.local_id()) + .take(MAX_RELAYED) + .collect(); + let known: Vec<String> = { + let mut inner = state.links.inner.lock_or_recover(); + for member in &members { + if inner.relaying.len() >= MAX_KEPT && !inner.relaying.contains_key(member) { + inner + .relaying + .retain(|_, askers| askers.values().any(|relaying| now < relaying.until)); + } + let askers = inner.relaying.entry(member.clone()).or_default(); + askers.retain(|_, relaying| now < relaying.until); + if askers.len() >= MAX_SUBSCRIBERS && !askers.contains_key(asker) { + continue; + } + askers.insert( + asker.to_owned(), + Relaying { + nonce: request.nonce, + // Asked once: for as long as its reply is awaited. + until: now + if once { REPLY_WITHIN } else { lease }, + window, + once, + }, + ); + } + members + .iter() + .filter(|member| { + inner + .replies + .get(*member) + .is_some_and(|(at, _)| now.saturating_duration_since(*at) < REPLY_WITHIN) + }) + .cloned() + .collect() + }; + for member in known { + forward(state, &member); + } + let (direct, through) = split(state, members, Some(asker)).await; + for member in direct { + // Asked once while we subscribe anyway (our own dashboard, or another asker's): + // what comes is passed on, and a LINKS with 0 would end that subscription. + let subscribed = once + && state + .links + .inner + .lock_or_recover() + .asked + .get(&member) + .is_some_and(|(_, until)| now + REPLY_WITHIN < *until); + if !subscribed { + ask(state, &member, lease).await; + } } - reply.links.truncate(MAX_LINKS); - for link in &mut reply.links { - link.others.truncate(MAX_OTHERS); + if request.hops > 0 { + for (next, members) in through { + ask_via(state, &next, members, lease, rest, request.hops - 1).await; + } + } +} + +/// `member`'s links changed (or arrived): due to those who asked us for them +/// (`relaying`), each sent with what else is due to it at the end of its window. +fn forward(state: &AppState, member: &str) { + let now = Instant::now(); + let mut starting = Vec::new(); + { + let mut inner = state.links.inner.lock_or_recover(); + let askers: Vec<(String, Duration)> = match inner.relaying.get_mut(member) { + Some(askers) => { + askers.retain(|_, relaying| now < relaying.until); + askers + .iter() + .map(|(asker, relaying)| (asker.clone(), relaying.window)) + .collect() + } + None => return, + }; + for (asker, window) in askers { + let due = inner.due.entry(asker.clone()).or_default(); + if due.is_empty() { + starting.push((asker, window)); + } + due.insert(member.to_owned()); + } + } + for (asker, window) in starting { + let state = state.clone(); + tokio::spawn(async move { + tokio::time::sleep(window).await; + pass_on(&state, &asker).await; + }); + } +} + +/// What is due to `asker`: the links of the members it asked for, as we have them, in +/// as few LINKS_REPLYs as `MAX_RELAYED` allows (one per nonce it asked with). +async fn pass_on(state: &AppState, asker: &str) { + let now = Instant::now(); + let mut by_nonce: HashMap<u64, Vec<RelayedLinks>> = HashMap::new(); + { + let mut inner = state.links.inner.lock_or_recover(); + let due = inner.due.remove(asker).unwrap_or_default(); + for member in due { + let Some(relaying) = inner + .relaying + .get(&member) + .and_then(|askers| askers.get(asker)) + .copied() + .filter(|relaying| now < relaying.until) + else { + continue; + }; + let Some((at, links)) = inner.replies.get(&member) else { + continue; + }; + let age_ms = + u32::try_from(now.saturating_duration_since(*at).as_millis()).unwrap_or(u32::MAX); + by_nonce + .entry(relaying.nonce) + .or_default() + .push(RelayedLinks { + member: member.clone(), + age_ms, + links: links.clone(), + }); + if relaying.once + && let Some(askers) = inner.relaying.get_mut(&member) + { + askers.remove(asker); + } + } + } + let (epoch, epoch_seq) = state.epoch_stamp(); + for (nonce, relayed) in by_nonce { + for chunk in relayed.chunks(MAX_RELAYED) { + let reply = ParsedMessage::LinksReply(LinksReply { + epoch, + epoch_seq, + nonce, + links: Vec::new(), + relayed: chunk.to_vec(), + }); + if let Err(error) = state.transport.send_message(asker, &reply).await { + debug!(asker = %asker, error = %format!("{error:#}"), "Relayed LINKS_REPLY failed"); + } + } + } +} + +/// A LINKS_REPLY from `peer`: its own links kept if it answers our latest LINKS to it, +/// and those it relays if it answers our latest relayed one, while their replies are +/// taken (a subscription's go on); then passed on to whoever asked us for them. +pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) { + let now = Instant::now(); + let mut arrived = Vec::new(); + { + let mut inner = state.links.inner.lock_or_recover(); + let own = inner + .asked + .get(peer) + .is_some_and(|(nonce, until)| *nonce == reply.nonce && now < *until); + let via: Option<HashSet<String>> = inner + .via + .get(peer) + .filter(|(nonce, until, _)| *nonce == reply.nonce && now < *until) + .map(|(_, _, members)| members.clone()); + let mut keep = |member: String, age: Duration, mut links: Vec<Link>| { + links.truncate(MAX_LINKS); + for link in &mut links { + link.others.truncate(MAX_OTHERS); + } + if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(&member) { + inner.replies.clear(); + } + let at = now.checked_sub(age).unwrap_or(now); + // Not older news over newer. + if inner + .replies + .get(&member) + .is_some_and(|(known, _)| *known > at) + { + return; + } + inner.replies.insert(member.clone(), (at, links)); + arrived.push(member); + }; + if own { + keep(peer.to_owned(), Duration::ZERO, reply.links); + } + if let Some(members) = via { + for relayed in reply.relayed.into_iter().take(MAX_RELAYED) { + if members.contains(&relayed.member) { + keep( + relayed.member, + Duration::from_millis(u64::from(relayed.age_ms)), + relayed.links, + ); + } + } + } } - if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) { - inner.replies.clear(); + for member in arrived { + forward(state, &member); } - inner - .replies - .insert(peer.to_owned(), (Instant::now(), reply.links)); } /// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's @@ -374,6 +771,7 @@ pub(crate) async fn watch(state: AppState) { epoch_seq, nonce, links: links.clone(), + relayed: Vec::new(), }); if let Err(error) = state.transport.send_message(&peer, &reply).await { debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index 946fc33..df75f46 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -152,6 +152,18 @@ pub(crate) fn neighbours_of(state: &AppState, member: &str) -> Option<BTreeSet<S .then(|| view.neighbours_of(state.local_id(), member)) } +/// The transports `member` is on, by the member list (the schemes of its addresses); +/// ours as we are on them. None known: empty. +pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> { + let view = state.neighbours.read_or_recover(); + match view.as_ref() { + Some(view) if member == state.local_id() => view.local_transports.clone(), + Some(view) => view.transports.get(member).cloned().unwrap_or_default(), + None if member == state.local_id() => state.transport.transports().into_iter().collect(), + None => BTreeSet::new(), + } +} + impl NeighbourView { fn neighbours_of(&self, local: &str, member: &str) -> BTreeSet<String> { if member == local { |