aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docs/dashboard.md6
-rw-r--r--src/swarm/links.rs143
2 files changed, 135 insertions, 14 deletions
diff --git a/docs/dashboard.md b/docs/dashboard.md
index 4092886..1fe51dc 100644
--- a/docs/dashboard.md
+++ b/docs/dashboard.md
@@ -355,6 +355,12 @@ renewal keeps its nonce, so what is on its way under the last one still counts.
renewal lost on the way leaves the hops behind it to run out at the old time: with 90 s
renewed every 30 s, one lost costs nothing.
+A way of asking a member that brings none of its links within 20 s (the member doesn't
+answer this node after all, or the peer relaying doesn't pass anything on) is left out
+for that member for 5 minutes: the next renewal goes the next way (directly, then the
+peer picked as above, then the next one), and the old way runs out by itself. Once
+every way has failed, all are tried again. Every hop does the same for what it asks on.
+
The links a member reports: its neighbours and the peers it holds dormant (a neighbour
that goes dormant is one no more, but stays listed until it is heard from again), by
which transport, whether only through a relay, the round trip (as the paths layer
diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index 14c5cf6..c430d58 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -29,6 +29,11 @@
//! `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.
+//!
+//! A way of asking a member that brings none of its links (`FAILED_AFTER`: we don't
+//! reach it after all, or the relay doesn't) is left out a while for that member, and
+//! the next ask (the renewal) goes the next way: so the old way runs out by itself, at
+//! no cost but the wait. Every hop does so for what it asks on.
use std::collections::{HashMap, HashSet};
use std::sync::Mutex;
@@ -80,6 +85,12 @@ 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;
+/// A way of asking a member (directly, or through a peer) that brought none of its
+/// links for this long after we asked has failed it (`split`): the next ask goes
+/// another way. More than a window (`MAX_BATCH`) along a few hops of slow transports.
+const FAILED_AFTER: Duration = Duration::from_secs(20);
+/// How long a way that failed a member is left out for that member.
+const FAILED_KEPT: Duration = Duration::from_secs(300);
/// The requests out and the replies in, and when each asker was last answered.
#[derive(Default)]
@@ -112,6 +123,18 @@ struct Inner {
/// 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)>,
+ /// How we ask each member (`split`).
+ routes: HashMap<String, Route>,
+}
+
+/// How we ask a member: the way we last did, since when, whether its links came since,
+/// and the ways that failed it lately (`FAILED_AFTER`), each since when.
+struct Route {
+ /// Through this peer; None: directly.
+ via: Option<String>,
+ asked: Instant,
+ arrived: bool,
+ failed: HashMap<Option<String>, Instant>,
}
/// One asker's relayed subscription to a member.
@@ -333,7 +356,7 @@ pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) {
/// 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;
+ let (direct, through) = split(state, members, None, true).await;
for member in direct {
let state = state.clone();
tokio::spawn(async move { ask(&state, &member, subscribe).await });
@@ -352,11 +375,15 @@ pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: D
}
/// `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).
+/// (none for those no peer leads to, nor without `through_peers`); not `exclude` (who
+/// asked us). A way that brought none of a member's links since we last asked it that
+/// way (`FAILED_AFTER`) is left out for a while (`FAILED_KEPT`): directly, the next
+/// relay, and so on; once every way failed, they are all tried again.
async fn split(
state: &AppState,
members: Vec<String>,
exclude: Option<&str>,
+ through_peers: bool,
) -> (Vec<String>, HashMap<String, Vec<String>>) {
let mut direct = Vec::new();
let mut through: HashMap<String, Vec<String>> = HashMap::new();
@@ -364,25 +391,111 @@ async fn split(
if member == state.local_id() || Some(member.as_str()) == exclude {
continue;
}
- if topology::reaches(state, &member).await {
- direct.push(member);
- } else if let Some(relay) = relay_for(state, &member, exclude).await {
- through.entry(relay).or_default().push(member);
+ let failed = failed_ways(state, &member);
+ let mut via = way(state, &member, exclude, through_peers, &failed).await;
+ if via.is_none() && !failed.is_empty() {
+ forget_failed(state, &member);
+ via = way(state, &member, exclude, through_peers, &HashMap::new()).await;
+ }
+ let Some(via) = via else {
+ continue;
+ };
+ asked_by(state, &member, via.clone());
+ match via {
+ None => direct.push(member),
+ Some(relay) => through.entry(relay).or_default().push(member),
}
}
(direct, through)
}
+/// How to ask `member`, but the `failed` ways: Some(None) directly, Some(Some(peer))
+/// through `peer` (if `through_peers`); None if no way is left.
+async fn way(
+ state: &AppState,
+ member: &str,
+ exclude: Option<&str>,
+ through_peers: bool,
+ failed: &HashMap<Option<String>, Instant>,
+) -> Option<Option<String>> {
+ if !failed.contains_key(&None) && topology::reaches(state, member).await {
+ return Some(None);
+ }
+ if !through_peers {
+ return None;
+ }
+ relay_for(state, member, exclude, failed).await.map(Some)
+}
+
+/// The ways that failed `member` lately; the one we last asked it by among them if it
+/// brought nothing within `FAILED_AFTER`.
+fn failed_ways(state: &AppState, member: &str) -> HashMap<Option<String>, Instant> {
+ let now = Instant::now();
+ let mut inner = state.links.inner.lock_or_recover();
+ let Some(route) = inner.routes.get_mut(member) else {
+ return HashMap::new();
+ };
+ route
+ .failed
+ .retain(|_, since| now.saturating_duration_since(*since) < FAILED_KEPT);
+ if !route.arrived && now.saturating_duration_since(route.asked) >= FAILED_AFTER {
+ debug!(member = %member, via = ?route.via, "LINKS brought nothing that way; asking another");
+ route.failed.insert(route.via.clone(), now);
+ }
+ route.failed.clone()
+}
+
+/// Every way may be tried again for `member`.
+fn forget_failed(state: &AppState, member: &str) {
+ if let Some(route) = state.links.inner.lock_or_recover().routes.get_mut(member) {
+ route.failed.clear();
+ }
+}
+
+/// We ask `member` `via` now: the wait for its links starts over, unless we asked it so
+/// already and are still waiting.
+fn asked_by(state: &AppState, member: &str, via: Option<String>) {
+ let now = Instant::now();
+ let mut inner = state.links.inner.lock_or_recover();
+ if inner.routes.len() >= MAX_KEPT && !inner.routes.contains_key(member) {
+ inner.routes.clear();
+ }
+ let route = inner
+ .routes
+ .entry(member.to_owned())
+ .or_insert_with(|| Route {
+ via: via.clone(),
+ asked: now,
+ arrived: false,
+ failed: HashMap::new(),
+ });
+ if route.via != via || route.arrived {
+ route.via = via;
+ route.asked = now;
+ route.arrived = false;
+ }
+}
+
/// The peer to ask `member` through, which we can't reach: an active member we reach
/// (that answers us) 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> {
+/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`, nor one
+/// that failed `member` (`failed`).
+async fn relay_for(
+ state: &AppState,
+ member: &str,
+ exclude: Option<&str>,
+ failed: &HashMap<Option<String>, Instant>,
+) -> 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) {
+ if peer == member
+ || Some(peer.as_str()) == exclude
+ || failed.contains_key(&Some(peer.clone()))
+ || !state.transport.can_reach(&peer)
+ {
continue;
}
{
@@ -625,7 +738,7 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
for member in known {
forward(state, &member);
}
- let (direct, through) = split(state, members, Some(asker)).await;
+ let (direct, through) = split(state, members, Some(asker), request.hops > 0).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.
@@ -641,10 +754,8 @@ async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
ask(state, &member, lease).await;
}
}
- if request.hops > 0 {
- for (next, members) in through {
- ask_via(state, &next, members, lease, rest, request.hops - 1).await;
- }
+ for (next, members) in through {
+ ask_via(state, &next, members, lease, rest, request.hops - 1).await;
}
}
@@ -783,6 +894,10 @@ pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) {
inner.replies.clear();
}
let at = now.checked_sub(age).unwrap_or(now);
+ // Arrived, if older news: the way works.
+ if let Some(route) = inner.routes.get_mut(&member) {
+ route.arrived = true;
+ }
// Not older news over newer.
if inner
.replies