diff options
| -rw-r--r-- | docs/dashboard.md | 7 | ||||
| -rw-r--r-- | src/swarm/links.rs | 30 |
2 files changed, 33 insertions, 4 deletions
diff --git a/docs/dashboard.md b/docs/dashboard.md index d641603..821c9e8 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -356,9 +356,10 @@ which transport, whether only through a relay, the round trip (as the paths laye measures it), whether it holds the peer dormant or the path is down, the peer's other transports with a path up (each with its fastest path's round trip, and whether it goes only through a relay), and the peer's clock against the member's ([`measurements.md`](measurements.md), -Per peer, `clock`). A member answers a LINKS from members only, at most every 5 s (one -asking it to relay, each time), and every build answers, with or without the feature -(`swarm::links`). +Per peer, `clock`). A member answers a LINKS from members only, at most every 5 s, and +asks for each peer that has it relay at most 512 members at once and 32 a second on +average (the rest of a request is dropped); every build answers, with or without the +feature (`swarm::links`). **Clocks.** The clocks in the links chain every member's clock to this node's, along the narrowest bounds, and the map takes each against their median: a member more than diff --git a/src/swarm/links.rs b/src/swarm/links.rs index cdbae67..4b89e94 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -66,6 +66,11 @@ const MAX_RELAYED: usize = 64; 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); +/// Members a peer may have us ask for it, at most: as many at once (`RELAY_BURST`), and +/// on average this many per second (`RELAY_RATE`): a dashboard asking every member +/// (`MAX_ASKED` there, 512) every 30 s needs less. The rest of a request is dropped. +const RELAY_BURST: f64 = 512.0; +const RELAY_RATE: f64 = 32.0; /// Nodes a relayed LINKS goes through at most, the first included. const MAX_HOPS: u8 = 3; @@ -97,6 +102,9 @@ struct Inner { relaying: HashMap<String, HashMap<String, Relaying>>, /// Relayed links due to an asker, by asker: the members whose links changed. due: HashMap<String, HashSet<String>>, + /// How many members each peer may still have us ask for it (`RELAY_BURST`, + /// `RELAY_RATE`), as of when. + relay_budget: HashMap<String, (f64, Instant)>, } /// One asker's relayed subscription to a member. @@ -544,7 +552,7 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) { 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 + let mut members: Vec<String> = request .of .into_iter() .filter(|member| member != asker && member != state.local_id()) @@ -552,6 +560,26 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) { .collect(); let known: Vec<String> = { let mut inner = state.links.inner.lock_or_recover(); + // Within the asker's budget: each member costs one. + if inner.relay_budget.len() >= MAX_KEPT && !inner.relay_budget.contains_key(asker) { + inner.relay_budget.retain(|_, (budget, at)| { + *budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE + < RELAY_BURST + }); + } + let (budget, at) = inner + .relay_budget + .entry(asker.to_owned()) + .or_insert((RELAY_BURST, now)); + *budget = (*budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE) + .min(RELAY_BURST); + *at = now; + let allowed = (*budget as usize).min(members.len()); + *budget -= allowed as f64; + if allowed < members.len() { + debug!(asker = %asker, asked = members.len(), allowed, "Relayed LINKS over its budget; asking for part of it"); + members.truncate(allowed); + } for member in &members { if inner.relaying.len() >= MAX_KEPT && !inner.relaying.contains_key(member) { inner |