diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/dashboard/mod.rs | 38 | ||||
| -rw-r--r-- | src/node.rs | 3 | ||||
| -rw-r--r-- | src/swarm/links.rs | 212 | ||||
| -rw-r--r-- | src/transport/mod.rs | 4 | ||||
| -rw-r--r-- | src/transport/wire.rs | 15 |
5 files changed, 245 insertions, 27 deletions
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index dd454ec..f0cbbfd 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -10,8 +10,9 @@ //! themselves: they alone changing is no change, and the page counts them on. //! //! The map: the graph the member list makes (`topology::intended_graph`, as this node -//! sees it), and the links every member measures, asked for with LINKS every -//! `LINKS_EVERY` while anyone looks (`swarm::links`), the same answers for all clients. +//! sees it), and the links every member measures: subscribed to with LINKS while anyone +//! looks (renewed every `LINKS_EVERY`, ended when nobody does; `swarm::links`), so a +//! member tells of a change as it happens; the same answers for all clients. use std::collections::{BTreeMap, BTreeSet}; use std::convert::Infallible; @@ -34,10 +35,14 @@ use crate::transport::Link; /// How often the snapshot is built and compared, while anyone looks. const TICK: Duration = Duration::from_secs(2); -/// How often members are asked for their links, and the graph recomputed. -const LINKS_EVERY: Duration = Duration::from_secs(10); +/// How often the subscriptions to the members' links are renewed (each renewal is +/// answered: a reply at least this often), and the graph recomputed; changes come in +/// between, as they happen (`swarm::links`). +const LINKS_EVERY: Duration = Duration::from_secs(30); +/// How long a subscription lasts unless renewed: a node that is gone stops it. +const LINKS_LEASE: Duration = Duration::from_secs(90); /// A member's links older than this are left out (it didn't answer since). -const LINKS_KEPT_FOR: Duration = Duration::from_secs(30); +const LINKS_KEPT_FOR: Duration = Duration::from_secs(90); /// The producer runs on this long after the last request for the state. const LINGER: Duration = Duration::from_secs(60); /// Members asked for their links at most, each round. @@ -164,14 +169,27 @@ impl Hub { if let Some(changed) = document_changes(&mut changes) { let _ = self.documents.send(Arc::new(changed.to_string())); } - { + let unwanted = || { + self.patches.receiver_count() == 0 + && Instant::now() >= self.control.lock_or_recover().wanted_until + }; + if *self.closing.borrow() { + self.control.lock_or_recover().running = false; + return; + } + if unwanted() { + // Nobody looks: the members needn't tell us any more. Unless someone + // came meanwhile, then it goes on (and subscribes again). + for member in self.state.links.subscribed() { + links::unsubscribe(&self.state, &member).await; + } let mut control = self.control.lock_or_recover(); - let unwanted = - self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until; - if unwanted || *self.closing.borrow() { + if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until { control.running = false; return; } + drop(control); + graph_due = Instant::now(); } } } @@ -232,7 +250,7 @@ impl Graph { .take(MAX_ASKED) { let state = state.clone(); - tokio::spawn(async move { links::ask(&state, &member).await }); + tokio::spawn(async move { links::ask(&state, &member, LINKS_LEASE).await }); } } } diff --git a/src/node.rs b/src/node.rs index be722e6..0b8f663 100644 --- a/src/node.rs +++ b/src/node.rs @@ -171,6 +171,9 @@ impl Node { }); trace!("Sync worker task spawned"); + // Subscribers to our links (`swarm::links`; any build has them). + tokio::spawn(crate::swarm::links::watch(state.clone())); + let peer_worker_state = state.clone(); tokio::spawn(async move { peer::run_peer_worker(peer_worker_state).await; 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))])); + } +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 7fbbd73..903dd07 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -403,6 +403,10 @@ pub(crate) struct LinksRequest { pub(crate) epoch_seq: u32, /// Repeated in the reply: which request it answers. pub(crate) nonce: u64, + /// For this long (seconds), a LINKS_REPLY again whenever the links change (a + /// subscription; the next LINKS renews it). 0: this reply only, and a subscription + /// ends. + pub(crate) subscribe_secs: u32, } /// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them. diff --git a/src/transport/wire.rs b/src/transport/wire.rs index dc94ab3..40e32a2 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 10; +pub(crate) const VERSION: u8 = 11; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -409,6 +409,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { epoch, epoch_seq: 1, nonce: 3, + subscribe_secs: 60, }), ParsedMessage::LinksReply(crate::transport::LinksReply { epoch, @@ -599,7 +600,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("0a0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("0b0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -619,7 +620,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "0a0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "0b0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -635,7 +636,7 @@ mod tests { }); assert_eq!( hex(&encode(&status).unwrap()), - format!("0a03969290c400{epoch_hex}01c3c2c2") + format!("0b03969290c400{epoch_hex}01c3c2c2") ); let who = ParsedMessage::Who(IdentityProof { @@ -644,7 +645,7 @@ mod tests { }); assert_eq!( hex(&encode(&who).unwrap()), - format!("0a0692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + format!("0b0692c420{}c440{}", "08".repeat(32), "09".repeat(64)) ); let ParsedMessage::Me(proof) = decode( &encode(&ParsedMessage::Me(IdentityProof { @@ -658,10 +659,10 @@ mod tests { }; assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); - assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0a0807"); + assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0b0807"); assert_eq!( hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()), - "0a09cfffffffffffffffff" + "0b09cfffffffffffffffff" ); let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else { panic!("not a PONG"); |