aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/dissemination.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/dissemination.rs')
-rw-r--r--src/swarm/dissemination.rs100
1 files changed, 100 insertions, 0 deletions
diff --git a/src/swarm/dissemination.rs b/src/swarm/dissemination.rs
index 1d92b06..93ad4e9 100644
--- a/src/swarm/dissemination.rs
+++ b/src/swarm/dissemination.rs
@@ -10,6 +10,7 @@
//! first, and so how soon.
use async_trait::async_trait;
+use tracing::trace;
use crate::document::ChangeHash;
use crate::state::AppState;
@@ -48,4 +49,103 @@ pub(crate) trait Dissemination: Send + Sync + 'static {
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,
+}
+
+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()),
+ }
+ }
}