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
120
121
122
123
124
125
126
127
128
129
130
131
|
//! 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) that answer
//! (`PeerState::answer_overdue`: none left a keepalive or sync message unanswered for
//! `Timing::unreached_after`, asked again halfway, with nothing else heard from it; a
//! member gone shows within seconds, not after `dormant_after`). 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 to every neighbour when it changed (the peer worker looks
//! every 5 s, and at once when a member we reach stops answering), 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 the session lasts (a
//! neighbour that goes dormant, or starts anew, says it again). Not for a set time:
//! while a pair exchanges changes it sends no keepalives, and so no STATUS, for as
//! long as the writes go on. 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 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
/// that answer.
async fn measure(state: &AppState) -> Vec<u64> {
let now = Instant::now();
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()
&& !peer_state.answer_overdue(now)
&& 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), and at once when a member we said
/// we reach stops answering (`reached_stops_answering`): whom we reach now; if it changed,
/// our neighbours are told at once (a STATUS each, unanswered unless they lack some of
/// our heads). 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");
}
}
}
/// Whether `peer`, whose answer is overdue (`PeerState::answer_overdue`), is one we
/// said we reach: then the peer worker checks at once (`check`).
pub(crate) fn reached_stops_answering(state: &AppState, peer: &str) -> bool {
state
.reach
.lock_or_recover()
.published
.as_ref()
.is_some_and(|keys| keys.contains(&topology::pair_key(peer)))
}
/// A STATUS to `peer` with whom we reach: as an announcement (unanswered unless it lacks
/// some of our heads), but with `reaches`.
async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> {
announce_heads(state, peer, true).await
}
/// The members a peer reaches, as it last said in this session (`said`, kept with its
/// state, cleared when the session ends: `PeerState::reset_sync`); by pair key.
pub(crate) fn of(said: Option<&(Instant, Vec<u64>)>) -> Option<Vec<u64>> {
said.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()
}
|
