diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/links.rs | 212 |
1 files changed, 202 insertions, 10 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs index 04f2e2d..80e6833 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -8,6 +8,12 @@ //! may have none. Only members (in the member list) are answered, as the links name //! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing //! 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, or a +//! round trip that moved by a fifth), looked at every `WATCH_EVERY`, 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. use std::collections::HashMap; use std::sync::Mutex; @@ -32,6 +38,13 @@ pub(crate) const MAX_OTHERS: usize = 8; const ANSWER_EVERY: Duration = Duration::from_secs(5); /// Replies kept from members at most (one each), and requests out. const MAX_KEPT: usize = 4096; +/// Subscriptions a member keeps at most, and how long one lasts at most. +const MAX_SUBSCRIBERS: usize = 64; +const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300); +/// How often a member with subscribers looks whether its links changed. +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); /// The requests out and the replies in, and when each asker was last answered. #[derive(Default)] @@ -41,12 +54,17 @@ pub(crate) struct Links { #[derive(Default)] struct Inner { - /// Our LINKS out, by member: its nonce. - asked: HashMap<String, u64>, + /// Our LINKS out, by member: its nonce, and until when replies to it are taken (its + /// subscription's end, or `REPLY_WITHIN`). + asked: HashMap<String, (u64, Instant)>, /// The latest LINKS_REPLY of each member: when, and its links. replies: HashMap<String, (Instant, Vec<Link>)>, /// When we last answered each member. answered: HashMap<String, Instant>, + /// Who subscribed to our links: the nonce to answer with, and until when. + subscribers: HashMap<String, (u64, Instant)>, + /// The links subscribers were last sent (`watch`). + published: Option<Vec<Link>>, } impl Links { @@ -67,6 +85,19 @@ impl Links { }) .collect() } + + /// The members our subscriptions are with (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() + .asked + .iter() + .filter(|(_, (_, until))| now < *until) + .map(|(member, _)| member.clone()) + .collect() + } } /// This node's links: its neighbours, as it measures them. @@ -151,22 +182,61 @@ fn rtt_of(destination: Option<&Destination>) -> Option<Duration> { .map(|ms| Duration::from_secs_f64(ms / 1000.0)) } -/// Asks `member` for its links (a LINKS); its reply lands in `AppState::links`. +/// Asks `member` for its links (a LINKS), and with `subscribe` for its changes for +/// that long; its replies land in `AppState::links`. #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] -pub(crate) async fn ask(state: &AppState, member: &str) { - let nonce = crate::transport::paths::random_nonce(); - { +pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) { + let nonce = { + let now = Instant::now(); let mut inner = state.links.inner.lock_or_recover(); if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) { inner.asked.clear(); } - inner.asked.insert(member.to_owned(), nonce); + let until = now + subscribe.max(REPLY_WITHIN); + // The same nonce while its replies are taken: a renewal doesn't drop a reply on + // its way under the last one (in a chain of relays, every hop renews at once, + // `relay`), as `ask_via` keeps one per relay. + match inner.asked.get_mut(member) { + Some((nonce, taken)) if now < *taken => { + *taken = (*taken).max(until); + *nonce + } + _ => { + let nonce = crate::transport::paths::random_nonce(); + inner.asked.insert(member.to_owned(), (nonce, until)); + nonce + } + } + }; + 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), + }); + if let Err(error) = state.transport.send_message(member, &request).await { + debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); } +} + +/// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any +/// more. +#[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(); + match inner.asked.remove(member) { + Some((nonce, _)) => nonce, + None => return, + } + }; let (epoch, epoch_seq) = state.epoch_stamp(); let request = ParsedMessage::Links(LinksRequest { epoch, epoch_seq, nonce, + subscribe_secs: 0, }); if let Err(error) = state.transport.send_message(member, &request).await { debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); @@ -188,6 +258,17 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) { let now = Instant::now(); let mut inner = state.links.inner.lock_or_recover(); + // A subscription is taken, renewed or ended even when the reply is due later. + if request.subscribe_secs == 0 { + inner.subscribers.remove(peer); + } else if inner.subscribers.len() < MAX_SUBSCRIBERS || inner.subscribers.contains_key(peer) + { + let lasts = + Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION); + inner + .subscribers + .insert(peer.to_owned(), (request.nonce, now + lasts)); + } if inner .answered .get(peer) @@ -214,13 +295,17 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) } } -/// A LINKS_REPLY from `peer`: kept if it answers our LINKS to it. +/// 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(); - if inner.asked.get(peer) != Some(&reply.nonce) { + let taken = inner + .asked + .get(peer) + .is_some_and(|(nonce, until)| *nonce == reply.nonce && Instant::now() < *until); + if !taken { return; } - inner.asked.remove(peer); reply.links.truncate(MAX_LINKS); for link in &mut reply.links { link.others.truncate(MAX_OTHERS); @@ -232,3 +317,110 @@ pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) { .replies .insert(peer.to_owned(), (Instant::now(), reply.links)); } + +/// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's +/// links; sent to each of them when they changed since they were last sent. Runs for +/// the node's life. +pub(crate) async fn watch(state: AppState) { + let mut every = tokio::time::interval(WATCH_EVERY); + every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + every.tick().await; + let subscribers: Vec<(String, u64)> = { + let now = Instant::now(); + let mut inner = state.links.inner.lock_or_recover(); + inner.subscribers.retain(|_, (_, until)| now < *until); + if inner.subscribers.is_empty() { + inner.published = None; + continue; + } + inner + .subscribers + .iter() + .map(|(peer, (nonce, _))| (peer.clone(), *nonce)) + .collect() + }; + let links = own_links(&state).await; + { + let mut inner = state.links.inner.lock_or_recover(); + if inner + .published + .as_ref() + .is_some_and(|published| !changed(published, &links)) + { + continue; + } + inner.published = Some(links.clone()); + } + let (epoch, epoch_seq) = state.epoch_stamp(); + for (peer, nonce) in subscribers { + let reply = ParsedMessage::LinksReply(LinksReply { + epoch, + epoch_seq, + nonce, + links: links.clone(), + }); + if let Err(error) = state.transport.send_message(&peer, &reply).await { + debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); + } + } + } +} + +/// Whether `new` says something `old` didn't: another peer, transport, relay or +/// dormancy, or a round trip measured or moved by more than a fifth (and 1 ms). Round +/// trips wobble; small moves aren't news. +fn changed(old: &[Link], new: &[Link]) -> bool { + let moved = |a: Option<u32>, b: Option<u32>| match (a, b) { + (Some(a), Some(b)) => { + let by = a.abs_diff(b); + by > 1000 && by > a.max(b) / 5 + } + (a, b) => a.is_some() != b.is_some(), + }; + old.len() != new.len() + || old.iter().zip(new).any(|(a, b)| { + a.peer != b.peer + || a.transport != b.transport + || a.relayed != b.relayed + || a.dormant != b.dormant + || moved(a.rtt_us, b.rtt_us) + || a.others.len() != b.others.len() + || a.others.iter().zip(&b.others).any(|(x, y)| { + x.transport != y.transport + || x.relayed != y.relayed + || moved(x.rtt_us, y.rtt_us) + }) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn link(peer: &str, rtt_us: Option<u32>) -> Link { + Link { + peer: peer.into(), + transport: Some("libp2p".into()), + relayed: false, + rtt_us, + dormant: false, + others: Vec::new(), + } + } + + #[test] + fn only_news_is_a_change() { + let old = [link("a", Some(10_000))]; + assert!(!changed(&old, &[link("a", Some(11_000))])); + assert!(changed(&old, &[link("a", Some(13_000))])); + assert!(changed(&old, &[link("a", None)])); + assert!(changed(&old, &[link("b", Some(10_000))])); + assert!(changed(&old, &[])); + let mut dormant = link("a", Some(10_000)); + dormant.dormant = true; + assert!(changed(&old, &[dormant])); + // Small round trips: a millisecond's wobble isn't news. + assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))])); + } +} |