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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
//! 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, then whether it reaches a member a hop closer, then pair key;
//! a member's parent is the neighbour it reaches that comes before it, the closest to
//! the writer first, then one that reaches closer, 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: where a writer behind NAT reaches a
//! single member, that one comes first and the others take it.
//!
//! 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.
///
/// Members are ordered by distance from the writer, then whether they reach a member a
/// hop closer (as they say: else, while it hasn't the change from its own side, it is
/// as good as farther), then pair key; a member's parent is the neighbour it reaches
/// that comes before it, by the same, then pair score. So where the writer reaches few
/// (behind NAT), those it reaches come first among the members as far from it, and the
/// others take them as parents. What a member reaches we know of ourselves and our
/// neighbours, as they said; of others, the graph is taken as right (if we're wrong,
/// a change goes twice or by announcement).
pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
let Some((graph, distance)) = distances(state, writer) else {
return Vec::new();
};
let me = state.local_id();
if !distance.contains_key(me) {
return Vec::new();
}
// Whom we and our neighbours reach, by pair key.
let mut reaches: HashMap<&str, Vec<u64>> = HashMap::new();
reaches.insert(me, reach::published(state));
for member in graph.get(me).into_iter().flatten() {
if let Some(peer_state) = state.get_peer_state(member).await
&& let Some(keys) = reach::of(state, peer_state.lock().await.reaches.as_ref())
{
reaches.insert(member.as_str(), keys);
}
}
let reached = |by: &str, member: &str| {
reaches
.get(by)
.is_none_or(|keys| keys.contains(&topology::pair_key(member)))
};
// Whether a member reaches one a hop closer to the writer (the writer: yes).
let closer = |member: &str| {
let d = distance[member];
d == 0
|| graph.get(member).into_iter().flatten().any(|other| {
distance.get(other.as_str()) == Some(&(d - 1)) && reached(member, other)
})
};
let rank = |member: &str| {
distance
.get(member)
.map(|d| (*d, !closer(member), topology::pair_key(member)))
};
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;
}
let parent = graph
.get(member)
.into_iter()
.flatten()
.filter(|candidate| rank(candidate).is_some_and(|r| r < its_rank))
.filter(|candidate| reached(member, candidate))
.min_by_key(|candidate| {
let (d, far, _) = rank(candidate).expect("ranked");
(d, far, 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
}
|
