aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/links.rs70
1 files changed, 60 insertions, 10 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index a12a1de..d7df0d5 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -19,11 +19,15 @@ use tracing::debug;
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::{members, topology};
-use crate::transport::{Connection, Link, LinksReply, LinksRequest, ParsedMessage};
+use crate::transport::{
+ Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage,
+};
/// Links one reply lists at most (a member has about `--neighbours`, more if others
/// prefer it).
pub(crate) const MAX_LINKS: usize = 256;
+/// Other transports in a link at most.
+pub(crate) const MAX_OTHERS: usize = 8;
/// A member answers each other one at most this often.
const ANSWER_EVERY: Duration = Duration::from_secs(5);
/// Replies kept from members at most (one each), and requests out.
@@ -74,19 +78,17 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
}
let dormant = peer_state.lock().await.liveness.is_dormant();
let destination = state.transport.destination(&peer);
+ let transport = destination
+ .as_ref()
+ .and_then(|destination| scheme_of(&destination.address));
links.push(Link {
- transport: destination.as_ref().and_then(|destination| {
- destination
- .address
- .split_once(':')
- .map(|(scheme, _)| scheme.to_owned())
- }),
+ others: others(destination.as_ref(), transport.as_deref()),
+ transport,
relayed: destination
.as_ref()
.is_some_and(|destination| destination.connection == Some(Connection::Relayed)),
- rtt_us: state
- .transport
- .rtt(&peer)
+ rtt_us: rtt_of(destination.as_ref())
+ .or_else(|| state.transport.rtt(&peer))
.map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
dormant,
peer,
@@ -97,6 +99,51 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
links
}
+/// The scheme of `address` (`mycelium:…`): its transport.
+fn scheme_of(address: &str) -> Option<String> {
+ address.split_once(':').map(|(scheme, _)| scheme.to_owned())
+}
+
+/// The transports besides `used` the peer proved itself at, with a path up: each with
+/// the round trip of its fastest path, if any is measured.
+fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> {
+ let mut others: Vec<OtherPath> = Vec::new();
+ for path in destination.map_or(&[][..], |destination| &destination.paths[..]) {
+ let Some(transport) = scheme_of(&path.address) else {
+ continue;
+ };
+ if path.down || used == Some(transport.as_str()) {
+ continue;
+ }
+ let rtt_us = path
+ .rtt_ms
+ .filter(|ms| ms.is_finite() && *ms >= 0.0)
+ .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32);
+ match others.iter_mut().find(|other| other.transport == transport) {
+ Some(other) => {
+ other.rtt_us = match (other.rtt_us, rtt_us) {
+ (Some(a), Some(b)) => Some(a.min(b)),
+ (a, b) => a.or(b),
+ };
+ }
+ None => others.push(OtherPath { transport, rtt_us }),
+ }
+ }
+ others.sort_by(|a, b| a.transport.cmp(&b.transport));
+ others.truncate(MAX_OTHERS);
+ others
+}
+
+/// The round-trip time of the path messages to the peer go over (`destination`'s
+/// first path, best first), if measured: `Transport::rtt` takes the first address the
+/// peer proved itself at, which needn't be the one in use.
+fn rtt_of(destination: Option<&Destination>) -> Option<Duration> {
+ let path = destination?.paths.first()?;
+ path.rtt_ms
+ .filter(|ms| ms.is_finite() && *ms >= 0.0)
+ .map(|ms| Duration::from_secs_f64(ms / 1000.0))
+}
+
/// Asks `member` for its links (a LINKS); its reply lands in `AppState::links`.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) async fn ask(state: &AppState, member: &str) {
@@ -168,6 +215,9 @@ pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) {
}
inner.asked.remove(peer);
reply.links.truncate(MAX_LINKS);
+ for link in &mut reply.links {
+ link.others.truncate(MAX_OTHERS);
+ }
if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) {
inner.replies.clear();
}