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
|
//! How a change reaches every member: whom a node sends a writer's changes to at once,
//! and whom it leaves to the announcement rounds (`swarm::relay`, which repair what it
//! misses: announcements, pulls, the slow round). Writers' trees (`swarm::tree`) are
//! one way; others can be tried against them in the benchmark (`docs/future-work.md`,
//! Swappable topology and dissemination). One per swarm: a node's part in passing a
//! change on is worked out by itself and by its neighbours, and must agree.
//!
//! What every way keeps: the base underneath, pairs of neighbours syncing with
//! Automerge, which delivers everything in the end; a way only decides whom to push to
//! first, and so how soon.
use async_trait::async_trait;
use tracing::trace;
use crate::document::ChangeHash;
use crate::state::AppState;
use crate::swarm::tree::Plan;
#[async_trait]
pub(crate) trait Dissemination: Send + Sync + 'static {
/// Its name (in logs, and once chosen by the swarm, in the document).
fn name(&self) -> &'static str;
/// Our part in passing on `writer`'s changes (by its pair key, `topology::pair_key`):
/// whom we send them to at once, and how the announcement rounds treat our other
/// neighbours (`Plan`). None: we have no part, and announcements do it all.
async fn plan(&self, state: &AppState, writer: u64) -> Option<Plan>;
/// `writer`'s `changes` came from `from` (none: we had them already, and they came
/// as the writer's now): passed on at once to those we send them to, the others
/// counted as having them where they get them elsewhere. Returns our part, as
/// `plan`.
async fn forward(
&self,
state: &AppState,
writer: u64,
from: &str,
changes: &[ChangeHash],
) -> Option<Plan>;
/// What a sync message from `peer` says about how well this works (`/admin/stats`):
/// whether it came as `writer`'s and `carried` changes, and the new changes it
/// brought, by author. Not while catching up.
async fn note_arrival(
&self,
state: &AppState,
peer: &str,
writer: Option<u64>,
carried: bool,
authors: &[(String, Vec<ChangeHash>)],
);
/// `peer` asks us to send it `writer`'s changes at once (`eager`, a GRAFT) or to
/// leave them to announcements (a PRUNE), for a way that keeps such links
/// (PlumTree). By default nothing.
async fn linked(&self, state: &AppState, peer: &str, writer: u64, eager: bool) {
let _ = (state, peer, writer, eager);
}
/// `peer` told us of `heads` we lack (a STATUS: an announcement, or its keepalive's),
/// for a way that minds which neighbour has changes first (PlumTree). By default
/// nothing.
fn announced(&self, peer: &str, heads: &[ChangeHash]) {
let _ = (peer, heads);
}
/// The neighbours whose links are lazy, for a way that keeps such links (PlumTree),
/// for the benchmark to report. By default none.
#[cfg(test)]
fn lazy(&self) -> Option<std::collections::HashSet<String>> {
None
}
}
/// What every way does with its `plan` for `writer`'s `changes` from `from` (none: we
/// had them already): those it leaves to others (`Plan::covered`) count as having them
/// (`SyncSession::known_to_peer`): else, with writers in turn, each change coming along
/// another way, the next message we pass on to one would carry all we got since our
/// last to it, mostly what its other parents brought it; should one lack them after
/// all, it asks, and gets them (`resend_if_needed`). Those it sends to
/// (`Plan::children`, but `from`) get them at once, with those of them that `changes`
/// build on, if they were counted so lately (`SyncSession::with_predecessors`).
pub(crate) async fn pass_on(
state: &AppState,
writer: u64,
from: &str,
changes: &[ChangeHash],
mut plan: Plan,
) -> Plan {
if !changes.is_empty() {
for neighbour in &plan.covered {
if let Some(peer_state) = state.get_peer_state(neighbour).await {
peer_state
.lock()
.await
.sync
.known_to_peer(changes, tokio::time::Instant::now());
}
}
}
plan.children.retain(|child| child != from);
let children = &plan.children;
let replica = if changes.is_empty() {
None
} else {
Some(state.doc.read().await)
};
for child in children {
if let Some(peer_state) = state.get_peer_state(child).await {
let mut peer_state = peer_state.lock().await;
peer_state.tree_writer = Some(writer);
if let Some(replica) = &replica {
peer_state.sync.with_predecessors(
&replica.doc,
changes,
tokio::time::Instant::now(),
);
}
}
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"
);
}
plan
}
/// The ways to choose from (`--dissemination`); one per swarm.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub(crate) enum Kind {
/// Writers' trees (`swarm::tree`).
#[default]
WritersTrees,
/// PlumTree (`swarm::plumtree`).
Plumtree,
/// Flooding (`swarm::flood`).
Flood,
}
impl Kind {
pub(crate) fn make(self) -> std::sync::Arc<dyn Dissemination> {
match self {
Kind::WritersTrees => std::sync::Arc::new(crate::swarm::tree::WritersTrees),
Kind::Plumtree => std::sync::Arc::new(crate::swarm::plumtree::PlumTree::default()),
Kind::Flood => std::sync::Arc::new(crate::swarm::flood::Flood),
}
}
}
|
