aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/topology/graph.rs
blob: fb33170b9c843402cd1a60aefb13e5e482ba2739 (plain)
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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
//! The graph itself, from the members alone: a ring by hash, plus each member's
//! lowest-scoring pairs, stable members preferred; per transport, and across them. Pure:
//! the same members give every node the same graph.

use sha2::{Digest, Sha256};

use super::*;

pub(super) const RING_LABEL: &[u8] = b"continuum/topology-ring/1";

pub(super) const PAIR_LABEL: &[u8] = b"continuum/topology-pair/2";

/// How many members may prefer the same stable member, as a multiple of `k`. Without a
/// bound, 20 stable members among 2000 get about 500 neighbours each: every
/// announcement in the swarm lands on them, and the backbone saturates first. Members
/// that find a stable member full prefer the next one, then go by pair score, so the
/// graph stays connected and symmetric.
pub(super) const STABLE_FANIN_FACTOR: usize = 4;

/// The transports `addresses` are on: their schemes (`mycelium` of `mycelium:<key>`);
/// the empty one for an address without (a transport whose identities are addresses).
pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> {
    addresses
        .iter()
        .map(|address| {
            address
                .split_once(':')
                .map_or("", |(scheme, _)| scheme)
                .to_owned()
        })
        .collect()
}

/// The neighbours of `local` among `members` (by the transports each is on), on each of
/// `local_transports`: on each, its neighbours among the members on it
/// (`neighbours`), all together.
pub(crate) fn neighbours_across(
    members: &BTreeMap<String, BTreeSet<String>>,
    local: &str,
    local_transports: &BTreeSet<String>,
    k: usize,
    unreachable: &dyn Fn(&str) -> bool,
    stable: &dyn Fn(&str) -> bool,
) -> BTreeSet<String> {
    let mut all = BTreeSet::new();
    for transport in local_transports {
        let on_it: Vec<String> = members
            .iter()
            .filter(|(member, transports)| *member != local && transports.contains(transport))
            .map(|(member, _)| member.clone())
            .collect();
        all.extend(neighbours(&on_it, local, k, unreachable, stable));
    }
    all
}

/// The neighbours of `local` among `members` (which may or may not contain `local`),
/// skipping members for which `unreachable` holds.
pub(crate) fn neighbours(
    members: &[String],
    local: &str,
    k: usize,
    unreachable: &dyn Fn(&str) -> bool,
    stable: &dyn Fn(&str) -> bool,
) -> BTreeSet<String> {
    let mut all: Vec<String> = members.to_vec();
    all.push(local.to_owned());
    graph(&all, k, &|m: &str| m != local && unreachable(m), stable)
        .remove(local)
        .unwrap_or_default()
}

/// Every reachable member's neighbours. Hashes each member once and scores pairs by
/// mixing the two hashes (`pair_score`), so the N² scores are a few nanoseconds each:
/// tens of milliseconds for thousands of members. `stable` members (`swarm::stability`)
/// are preferred for half of each member's scored slots, each by at most
/// `STABLE_FANIN_FACTOR * k` members.
///
/// Each reachable member walks the ring (by key hash) both ways to the first reachable
/// member, which gives a ring over the reachable members, and adds the `k - 2`
/// reachable members with the lowest pair score. A link exists if either end chose it.
/// With at most `k + 1` reachable members, the reachable members form a full mesh.
///
/// Each node judges reachability by itself, so nodes may briefly disagree; that only
/// adds links (whoever chose a link initiates, the other answers), it never cuts one.
/// The preference budget is spent in key order, which every node shares, so nodes that
/// agree on reachability compute the same links.
pub(crate) fn graph(
    members: &[String],
    k: usize,
    unreachable: &dyn Fn(&str) -> bool,
    stable: &dyn Fn(&str) -> bool,
) -> BTreeMap<String, BTreeSet<String>> {
    let k = k.max(2);
    let all: BTreeSet<&str> = members.iter().map(String::as_str).collect();
    let live: BTreeSet<&str> = all.iter().copied().filter(|m| !unreachable(m)).collect();
    let mut links: BTreeMap<String, BTreeSet<String>> = live
        .iter()
        .map(|m| ((*m).to_owned(), BTreeSet::new()))
        .collect();

    let mut link = |a: &str, b: &str| {
        if a != b {
            links.get_mut(a).expect("live member").insert(b.to_owned());
            links.get_mut(b).expect("live member").insert(a.to_owned());
        }
    };

    if live.len() <= k + 1 {
        for a in &live {
            for b in &live {
                link(a, b);
            }
        }
        return links;
    }

    // The ring over all members, so its order doesn't change when one goes offline.
    let mut ring: Vec<([u8; 32], &str)> =
        all.iter().map(|m| (hash(RING_LABEL, &[*m]), *m)).collect();
    ring.sort_unstable();
    let n = ring.len();
    let position: BTreeMap<&str, usize> =
        ring.iter().enumerate().map(|(i, (_, m))| (*m, i)).collect();
    let first_live = |from: usize, step: usize| {
        (1..n)
            .map(|d| ring[(from + d * step) % n].1)
            .find(|m| live.contains(m))
            .expect("more than k + 1 live members")
    };

    // Live members in key order, with their pair keys; and those of them that are stable.
    let keyed: Vec<(&str, u64)> = live.iter().map(|m| (*m, pair_key(m))).collect();
    let stable_keyed: Vec<(&str, u64)> = keyed.iter().copied().filter(|(m, _)| stable(m)).collect();

    // How many members have preferred each stable member so far (`keyed` is in key
    // order, the same on every node).
    let budget = STABLE_FANIN_FACTOR * k;
    let mut preferred_by: BTreeMap<&str, usize> = BTreeMap::new();
    // Half the slots (rounded up) prefer stable members, so long-running nodes form a
    // backbone; the rest go by pair score alone, keeping paths short. A stable member
    // enough others prefer already is skipped, so no hub grows with the swarm.
    let stable_slots = (k - 2).div_ceil(2);

    for &(member, key) in &keyed {
        let at = position[member];
        let next = first_live(at, 1);
        let prev = first_live(at, n - 1);
        link(member, next);
        link(member, prev);
        let skip = [member, next, prev];
        // The stable candidates in score order (there are few).
        let mut stable_candidates = scored(&stable_keyed, key, skip);
        stable_candidates.sort_unstable();
        let preferred: Vec<&str> = stable_candidates
            .iter()
            .map(|(_, other)| *other)
            .filter(|other| preferred_by.get(other).copied().unwrap_or(0) < budget)
            .take(stable_slots)
            .collect();
        for other in preferred.iter().copied() {
            *preferred_by.entry(other).or_default() += 1;
        }
        // The rest: the lowest scores among the others that aren't preferred, which are
        // among the `need + preferred` lowest overall. Selecting those is linear; only
        // they are sorted.
        let need = k - 2 - preferred.len();
        let mut candidates = scored(&keyed, key, skip);
        let lowest = (need + preferred.len()).min(candidates.len());
        if lowest < candidates.len() {
            candidates.select_nth_unstable(lowest);
            candidates.truncate(lowest);
        }
        candidates.sort_unstable();
        let rest = candidates
            .iter()
            .map(|(_, other)| *other)
            .filter(|other| !preferred.contains(other))
            .take(need);
        for partner in preferred.iter().copied().chain(rest) {
            link(member, partner);
        }
    }
    links
}

/// `others` but those in `skip`, scored as pairs with the member whose key is `key`.
pub(super) fn scored<'a>(
    others: &[(&'a str, u64)],
    key: u64,
    skip: [&str; 3],
) -> Vec<(u64, &'a str)> {
    others
        .iter()
        .filter(|(other, _)| !skip.contains(other))
        .map(|&(other, other_key)| (mix_pair(key, other_key), other))
        .collect()
}

/// Symmetric: the same for (a, b) and (b, a).
pub(crate) fn pair_score(a: &str, b: &str) -> u64 {
    mix_pair(pair_key(a), pair_key(b))
}

/// What a member contributes to its pair scores: computed once per member.
pub(super) fn pair_key(member: &str) -> u64 {
    let digest = hash(PAIR_LABEL, &[member]);
    u64::from_be_bytes(digest[..8].try_into().expect("8 bytes"))
}

/// The score of a pair from its members' keys, symmetric: SplitMix64's finalizer over
/// the lower key and the mixed higher one. Keys are uniform (SHA-256), so for each member
/// the others' scores are in an order unrelated to anyone else's.
pub(super) fn mix_pair(a: u64, b: u64) -> u64 {
    let (low, high) = if a <= b { (a, b) } else { (b, a) };
    mix(low ^ mix(high))
}

pub(super) fn mix(mut z: u64) -> u64 {
    z = z.wrapping_add(0x9e37_79b9_7f4a_7c15);
    z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
    z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
    z ^ (z >> 31)
}

pub(super) fn hash(label: &[u8], fields: &[&str]) -> [u8; 32] {
    let mut hasher = Sha256::new();
    for field in [label]
        .into_iter()
        .chain(fields.iter().map(|f| f.as_bytes()))
    {
        hasher.update((field.len() as u32).to_be_bytes());
        hasher.update(field);
    }
    hasher.finalize().into()
}