1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
//! Whom each neighbour reaches: a node's 2-hop neighbourhood, as measured, not as the
//! member list says.
//!
//! A node knows its own links, and from them whom it reaches: the members it has an
//! active session with (handshake done, neither dormant nor unreachable). It tells its
//! neighbours so in its STATUS (`Status::reaches`, each member by its pair key,
//! `topology::pair_key`, 8 bytes instead of a node ID): in every keepalive STATUS and
//! STATUS_REPLY, and soon (the peer worker looks every 5 s) to every neighbour when it
//! changed,
//! but not in announcements of heads, which go often. So every node knows whom its
//! neighbours reach, as of their latest STATUS: for as long as that is fresh
//! (`held_for`), then not, until the next. Nothing goes into the document, and nobody
//! knows more than two hops out.
//!
//! What it is for: passing changes on around links that don't work, and to exactly the
//! members the sender doesn't reach; a member a neighbour reaches is one worth trying,
//! and a way to it. Not for choosing multipoint relays (as OLSR does): over the member
//! list's graph, which is close to random, neighbourhoods hardly overlap, and nearly
//! every neighbour would be one (13.7 of 13.7 at 250 members and more), saving next to
//! nothing but in a full mesh.
use std::collections::HashMap;
use std::time::Duration;
use tokio::time::Instant;
use tracing::debug;
use crate::locks::MutexExt;
use crate::state::{AppState, HelloState, Standing};
use crate::swarm::peer::announce_heads;
use crate::swarm::{members, topology};
/// Whom we reach, as last told our neighbours.
#[derive(Default)]
pub(crate) struct Reach {
published: Option<Vec<u64>>,
}
/// The pair keys of the members we reach now, sorted: those with an active session.
async fn measure(state: &AppState) -> Vec<u64> {
let mut reached = Vec::new();
for (peer, peer_state) in state.peers_snapshot().await {
let peer_state = peer_state.lock().await;
if peer_state.hello_state == HelloState::Established
&& !peer_state.liveness.is_dormant()
&& !peer_state.is_unreachable()
&& matches!(peer_state.standing, Standing::Member { .. })
{
reached.push(topology::pair_key(&peer));
}
}
reached.sort_unstable();
reached
}
/// Whom we reach, as we tell it (`Status::reaches`); empty until first measured.
pub(crate) fn published(state: &AppState) -> Vec<u64> {
state
.reach
.lock_or_recover()
.published
.clone()
.unwrap_or_default()
}
/// With the peer worker's periodic pass (every 5 s): whom we reach now; if it changed,
/// our neighbours are told at once (a STATUS each, unanswered if the heads are the
/// same). A change that is undone within a pass isn't told.
pub(crate) async fn check(state: &AppState) {
let now = measure(state).await;
{
let mut reach = state.reach.lock_or_recover();
if reach.published.as_ref() == Some(&now) {
return;
}
reach.published = Some(now.clone());
}
debug!(
reached = now.len(),
"Whom we reach changed; telling our neighbours"
);
for (peer, peer_state) in state.peers_snapshot().await {
let established = peer_state.lock().await.hello_state == HelloState::Established;
if !established || !topology::is_neighbour(state, &peer) {
continue;
}
if let Err(error) = tell(state, &peer).await {
debug!(peer = %peer, error = %format!("{error:#}"), "Telling whom we reach failed");
}
}
}
/// A STATUS to `peer` with whom we reach: as an announcement (unanswered if the heads
/// are the same), but with `reaches`.
async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> {
announce_heads(state, peer, true).await
}
/// How long what a neighbour said it reaches holds: past two of its regular STATUSes,
/// each of which says it again.
pub(crate) fn held_for(state: &AppState) -> Duration {
state.timing.status_refresh * 2 + state.timing.keepalive
}
/// The members `peer` reaches, as it last said, while that holds (`held_for`); by
/// pair key.
pub(crate) fn of(state: &AppState, said: Option<&(Instant, Vec<u64>)>) -> Option<Vec<u64>> {
said.filter(|(at, _)| at.elapsed() < held_for(state))
.map(|(_, reaches)| reaches.clone())
}
/// The members of our member list by pair key, to name what neighbours say they reach.
pub(crate) async fn names(state: &AppState) -> HashMap<u64, String> {
let replica = state.doc.read().await;
members::members(&replica.doc)
.into_iter()
.map(|member| (topology::pair_key(&member), member))
.collect()
}
|
