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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
//! Passing a writer's changes on along its tree: every member gets them once, over a
//! link that works, and as soon as the member before it has them.
//!
//! The tree is the same on every node, computed from what they share: the graph the
//! member list gives (`topology::member_list_graph`: every member's neighbours, as if
//! all reached all) and what each member tells its neighbours it reaches
//! (`swarm::reach`). Every member's distance from the writer over that graph orders the
//! members, by distance and then pair key; a member's parent is the neighbour it reaches
//! that comes before it, the closest to the writer first, then by pair score
//! (`topology::pair_score`). Coming before rules out cycles, and taking only a neighbour
//! it reaches routes around links that don't work: one behind NAT that reaches a single
//! member takes that one, at whatever distance.
//!
//! A sync message that passes a writer's change on says whose (`SyncEnvelope::writer`).
//! Whoever applies one passes it on to its own children in that writer's tree
//! (`forward`), however it got it; so does the writer, to all its neighbours, with its
//! own writes. A member no member it reaches comes before (or whose parent didn't get
//! the change) gets it from the announcements (`swarm::relay`), as before.
//!
//! The graph is computed once per member list (tens of milliseconds for thousands of
//! members), the distances once per writer and graph (`MAX_WRITERS` kept).
use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
use std::sync::Arc;
use tracing::trace;
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::{reach, topology};
/// The writers whose distances are kept at most.
const MAX_WRITERS: usize = 256;
type Graph = BTreeMap<String, BTreeSet<String>>;
/// The distances from each writer, over the graph they were computed for.
#[derive(Default)]
pub(crate) struct Trees {
graph: Option<Arc<Graph>>,
/// The members by pair key, of that graph.
by_key: HashMap<u64, String>,
distances: HashMap<u64, Arc<HashMap<String, u32>>>,
}
/// The graph, and every member's distance from `writer` (by pair key) over it; None
/// while the member list isn't known, or `writer` isn't in it.
fn distances(state: &AppState, writer: u64) -> Option<(Arc<Graph>, Arc<HashMap<String, u32>>)> {
let graph = topology::member_list_graph(state)?;
let mut trees = state.trees.lock_or_recover();
if !trees
.graph
.as_ref()
.is_some_and(|known| Arc::ptr_eq(known, &graph))
{
trees.by_key = graph
.keys()
.map(|member| (topology::pair_key(member), member.clone()))
.collect();
trees.distances.clear();
trees.graph = Some(Arc::clone(&graph));
}
if let Some(known) = trees.distances.get(&writer) {
return Some((graph, Arc::clone(known)));
}
let from = trees.by_key.get(&writer)?.clone();
let mut distance = HashMap::from([(from.clone(), 0u32)]);
let mut queue = VecDeque::from([from]);
while let Some(member) = queue.pop_front() {
let next = distance[&member] + 1;
for neighbour in graph.get(&member).into_iter().flatten() {
if !distance.contains_key(neighbour) {
distance.insert(neighbour.clone(), next);
queue.push_back(neighbour.clone());
}
}
}
if trees.distances.len() >= MAX_WRITERS {
trees.distances.clear();
}
let distance = Arc::new(distance);
trees.distances.insert(writer, Arc::clone(&distance));
Some((graph, distance))
}
/// Our children in `writer`'s tree: our neighbours in the graph whose parent we are,
/// and whom we reach.
pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
let Some((graph, distance)) = distances(state, writer) else {
return Vec::new();
};
let rank = |member: &str| {
distance
.get(member)
.map(|d| (*d, topology::pair_key(member)))
};
let me = state.local_id();
let Some(my_rank) = rank(me) else {
return Vec::new();
};
let mut children = Vec::new();
for member in graph.get(me).into_iter().flatten() {
let Some(its_rank) = rank(member) else {
continue;
};
if its_rank <= my_rank {
continue;
}
// Whom it reaches, as it told us; not knowing, as the graph says.
let reaches = match state.get_peer_state(member).await {
Some(peer_state) => reach::of(state, peer_state.lock().await.reaches.as_ref()),
None => None,
};
let parent = graph
.get(member)
.into_iter()
.flatten()
.filter(|candidate| rank(candidate).is_some_and(|r| r < its_rank))
.filter(|candidate| {
reaches
.as_ref()
.is_none_or(|keys| keys.contains(&topology::pair_key(candidate)))
})
.min_by_key(|candidate| {
(
distance[*candidate],
topology::pair_score(candidate, member),
)
});
if parent.is_some_and(|parent| parent == me) && topology::reaches(state, member).await {
children.push(member.clone());
}
}
children
}
/// We applied a change of `writer`'s that `from` passed on: to our children in its
/// tree (but `from`), a sync message each, saying whose. Returns them.
pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Vec<String> {
let children: Vec<String> = children(state, writer)
.await
.into_iter()
.filter(|child| child != from)
.collect();
for child in &children {
if let Some(peer_state) = state.get_peer_state(child).await {
peer_state.lock().await.tree_writer = Some(writer);
}
state.stats.tree_forward();
state.request_sync(child);
}
if !children.is_empty() {
trace!(
writer,
children = children.len(),
"Passing a change on along the writer's tree"
);
}
children
}
|
