aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/hyparview.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/hyparview.rs')
-rw-r--r--src/swarm/hyparview.rs457
1 files changed, 457 insertions, 0 deletions
diff --git a/src/swarm/hyparview.rs b/src/swarm/hyparview.rs
new file mode 100644
index 0000000..6e363e3
--- /dev/null
+++ b/src/swarm/hyparview.rs
@@ -0,0 +1,457 @@
+//! HyParView (Leitão, Pereira and Rodrigues, "HyParView: a membership protocol for
+//! reliable gossip-based broadcast", 2007), slimmed down, as an experiment
+//! (`--topology hyparview`, `swarm::topology::Topology`).
+//!
+//! Each node keeps an active view: its neighbours, about `k` of them (the swarm's
+//! neighbours, as with the member list's graph), each of which keeps it in its own: a
+//! link both ends agreed to (NEIGHBOUR, answered by NEIGHBOUR_REPLY), not one every
+//! node computes alike. A node with fewer asks others from its passive view, one at a
+//! time per missing neighbour; one that has none asks with high priority, and is taken
+//! even where that means dropping another (told so: DISCONNECT). A neighbour that fails
+//! (judged unreachable: unanswered HELLOs, dormant, not alive lately) leaves the active
+//! view, and another is asked. Every neighbour is asked again now and then whether it
+//! still is (`REAFFIRM_EVERY`): the paper's links are TCP connections, whose ends know
+//! when the other closed them; ours, without, could be left in one end's view.
+//!
+//! The paper's passive view comes from its own messages (JOIN, FORWARD_JOIN, SHUFFLE),
+//! as its nodes know no member list. Ours has one, in the document, so the passive view
+//! is drawn from it (`passive_view`): every member we may reach that isn't in the
+//! active view. Those messages could replace it later without changing the rest.
+//!
+//! There is no graph every node shares (`Topology::graph`): what needs one does nothing
+//! (writers' trees, recorded links, indirect probes), so changes pass on by a
+//! dissemination that needs none (`--dissemination plumtree`), or by announcements.
+//!
+//! A member being asked counts as a neighbour already: it gets HELLOs, and NEIGHBOUR
+//! goes once the handshake is done (a swarm's members only take messages from peers
+//! that did one). Refused, or unanswered for `ASK_FOR`, it is not asked again for
+//! `REFUSED_FOR`, and stops counting.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+
+use async_trait::async_trait;
+use tokio::time::Instant;
+use tracing::debug;
+
+use crate::locks::MutexExt;
+use crate::state::{AppState, HelloState};
+use crate::swarm::topology::{self, Inputs, Topology};
+use crate::transport::{NeighbourAnswer, NeighbourRequest, ParsedMessage};
+
+/// How long a member asked to be a neighbour has to agree, the handshake before
+/// included: over Tor, a few round trips of up to 3 s.
+const ASK_FOR: Duration = Duration::from_secs(20);
+
+/// A member that refused, didn't answer, or left isn't asked again for this long.
+const REFUSED_FOR: Duration = Duration::from_secs(60);
+
+/// A member one neighbour short asks this seldom: in a swarm whose members mostly have
+/// `k`, those that refuse are full, and asking each in turn kept NEIGHBOURs going in a
+/// quiet swarm (up to 300 in 20 s at 50 members). Two or more short, it asks at once.
+const ONE_SHORT_EVERY: Duration = Duration::from_secs(300);
+
+/// Each neighbour is asked again this often whether it still is (NEIGHBOUR, as a new
+/// ask would be): a NEIGHBOUR_REPLY or DISCONNECT lost on a lossy link leaves a link in
+/// one end's view only, and nothing else would tell that end: the other still answers
+/// it. A neighbour that has us agrees at once; one that doesn't takes us if it has room,
+/// else refuses, and we drop it.
+const REAFFIRM_EVERY: Duration = Duration::from_secs(60);
+
+/// A member asked to be a neighbour.
+#[derive(Debug)]
+struct Asking {
+ since: Instant,
+ /// With high priority: we had no neighbour when we asked.
+ high: bool,
+ /// NEIGHBOUR sent (after the handshake).
+ sent: bool,
+}
+
+/// The views.
+#[derive(Debug, Default)]
+struct Views {
+ /// Our neighbours: each agreed, or asked us and we agreed.
+ active: BTreeSet<String>,
+ /// Asked, not answered yet.
+ asking: BTreeMap<String, Asking>,
+ /// Refused, unanswered or left lately: since when.
+ refused: BTreeMap<String, Instant>,
+ /// When we last asked while one neighbour short (`ONE_SHORT_EVERY`).
+ asked_one_short: Option<Instant>,
+ /// When the neighbours were last asked whether they still are (`REAFFIRM_EVERY`).
+ reaffirmed: Option<Instant>,
+}
+
+#[derive(Default)]
+pub(crate) struct HyParView {
+ views: Mutex<Views>,
+ /// Bumped whenever the views change what `neighbours` gives (`Topology::version`).
+ version: AtomicU64,
+}
+
+impl HyParView {
+ /// The members we could ask to be neighbours: from the member list, every member on
+ /// one of our transports that we don't judge unreachable, not us. The paper keeps
+ /// these from its own messages instead (`passive_view`, and the module's docs).
+ fn reachable_members(local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
+ inputs
+ .transports
+ .iter()
+ .filter(|(member, on)| {
+ member.as_str() != local
+ && !on.is_disjoint(inputs.local_transports)
+ && !(inputs.unreachable)(member)
+ })
+ .map(|(member, _)| member.clone())
+ .collect()
+ }
+
+ /// The passive view: whom we may ask when we lack neighbours. Drawn from the member
+ /// list (`reachable_members`), not from JOIN and SHUFFLE as in the paper.
+ fn passive_view(views: &Views, reachable: &BTreeSet<String>) -> Vec<String> {
+ reachable
+ .iter()
+ .filter(|member| {
+ !views.active.contains(*member)
+ && !views.asking.contains_key(*member)
+ && !views.refused.contains_key(*member)
+ })
+ .cloned()
+ .collect()
+ }
+
+ fn changed(&self, state: &AppState) {
+ self.version.fetch_add(1, Ordering::Relaxed);
+ state.topology_changed.notify_one();
+ }
+
+ async fn send(state: &AppState, peer: &str, message: ParsedMessage) {
+ if let Err(error) = state.transport.send_message(peer, &message).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "A HyParView message failed");
+ }
+ }
+
+ fn answer(state: &AppState, agreed: bool) -> NeighbourAnswer {
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ NeighbourAnswer {
+ epoch,
+ epoch_seq,
+ agreed,
+ }
+ }
+}
+
+/// `count` of `members`, picked at random.
+fn some_of(mut members: Vec<String>, count: usize) -> Vec<String> {
+ let count = count.min(members.len());
+ for picked in 0..count {
+ let rest = (members.len() - picked) as u64;
+ let other = picked + (crate::transport::paths::random_nonce() % rest) as usize;
+ members.swap(picked, other);
+ }
+ members.truncate(count);
+ members
+}
+
+#[async_trait]
+impl Topology for HyParView {
+ fn name(&self) -> &'static str {
+ "hyparview"
+ }
+
+ fn graph(
+ &self,
+ members: &[String],
+ k: usize,
+ unreachable: &dyn Fn(&str) -> bool,
+ stable: &dyn Fn(&str) -> bool,
+ ) -> Option<BTreeMap<String, BTreeSet<String>>> {
+ let _ = (members, k, unreachable, stable);
+ None
+ }
+
+ /// The active view, and those being asked.
+ fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
+ let _ = (local, inputs);
+ let views = self.views.lock_or_recover();
+ views
+ .active
+ .iter()
+ .chain(views.asking.keys())
+ .cloned()
+ .collect()
+ }
+
+ /// Failed neighbours dropped, unanswered asks given up, NEIGHBOUR sent to those
+ /// asked once the handshake is done, and new members asked while we lack
+ /// neighbours.
+ async fn maintain(&self, state: &AppState, inputs: &Inputs<'_>) {
+ let local = state.local_id().to_owned();
+ let reachable = Self::reachable_members(&local, inputs);
+ let now = Instant::now();
+ let mut to_leave = Vec::new();
+ let (changed, to_ask) = {
+ let mut views = self.views.lock_or_recover();
+ let mut changed = false;
+ // Told all the same: one we judge unreachable may only be silent towards
+ // us (a link that works one way), and kept us in its view, it would be our
+ // neighbour at its end alone.
+ let failed: Vec<String> = views
+ .active
+ .iter()
+ .filter(|member| !reachable.contains(*member))
+ .cloned()
+ .collect();
+ if !failed.is_empty() {
+ debug!(
+ dropped = failed.len(),
+ "Neighbours judged unreachable left the active view"
+ );
+ for member in failed {
+ views.active.remove(&member);
+ to_leave.push(member);
+ }
+ changed = true;
+ }
+ let given_up: Vec<String> = views
+ .asking
+ .iter()
+ .filter(|(member, asking)| {
+ !reachable.contains(*member) || now.duration_since(asking.since) > ASK_FOR
+ })
+ .map(|(member, _)| member.clone())
+ .collect();
+ for member in given_up {
+ // Asked: told we gave up, should it agree after all (its answer lost,
+ // or still on its way).
+ if views
+ .asking
+ .remove(&member)
+ .is_some_and(|asking| asking.sent)
+ {
+ to_leave.push(member.clone());
+ }
+ views.refused.insert(member, now);
+ changed = true;
+ }
+ views
+ .refused
+ .retain(|_, since| now.duration_since(*since) < REFUSED_FOR);
+ // More than one beyond `k` (asks sent before others asked us came back
+ // agreed): the excess dropped, at random, and told.
+ if views.active.len() > inputs.k + 1 {
+ let excess = views.active.len() - inputs.k;
+ let all: Vec<String> = views.active.iter().cloned().collect();
+ for member in some_of(all, excess) {
+ views.active.remove(&member);
+ views.refused.insert(member.clone(), now);
+ to_leave.push(member);
+ }
+ changed = true;
+ }
+ let missing = inputs
+ .k
+ .saturating_sub(views.active.len() + views.asking.len());
+ // One short, with neighbours: not again before `ONE_SHORT_EVERY`.
+ let one_short = missing == 1 && !views.active.is_empty();
+ let waiting = one_short
+ && views
+ .asked_one_short
+ .is_some_and(|at| now.duration_since(at) < ONE_SHORT_EVERY);
+ if missing > 0 && !waiting {
+ if one_short {
+ views.asked_one_short = Some(now);
+ }
+ // With high priority only the first, while we have no neighbour: the
+ // others would make room for us where none is needed.
+ let mut high = views.active.is_empty() && views.asking.is_empty();
+ for member in some_of(Self::passive_view(&views, &reachable), missing) {
+ views.asking.insert(
+ member,
+ Asking {
+ since: now,
+ high,
+ sent: false,
+ },
+ );
+ high = false;
+ changed = true;
+ }
+ }
+ let mut to_ask: Vec<(String, bool)> = views
+ .asking
+ .iter()
+ .filter(|(_, asking)| !asking.sent)
+ .map(|(member, asking)| (member.clone(), asking.high))
+ .collect();
+ if views
+ .reaffirmed
+ .is_none_or(|at| now.duration_since(at) >= REAFFIRM_EVERY)
+ {
+ views.reaffirmed = Some(now);
+ to_ask.extend(views.active.iter().map(|member| (member.clone(), false)));
+ }
+ (changed, to_ask)
+ };
+ if changed {
+ self.version.fetch_add(1, Ordering::Relaxed);
+ }
+ for member in to_leave {
+ Self::send(
+ state,
+ &member,
+ ParsedMessage::Disconnect(Self::answer(state, false)),
+ )
+ .await;
+ }
+ for (member, high) in to_ask {
+ let ready = match state.get_peer_state(&member).await {
+ Some(peer_state) => peer_state.lock().await.hello_state == HelloState::Established,
+ None => false,
+ };
+ if !ready {
+ continue;
+ }
+ // A neighbour asked again isn't asking: nothing to mark.
+ if let Some(asking) = self.views.lock_or_recover().asking.get_mut(&member) {
+ asking.sent = true;
+ }
+ let (epoch, epoch_seq) = state.epoch_stamp();
+ Self::send(
+ state,
+ &member,
+ ParsedMessage::Neighbour(NeighbourRequest {
+ epoch,
+ epoch_seq,
+ high,
+ }),
+ )
+ .await;
+ }
+ }
+
+ async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) {
+ match message {
+ ParsedMessage::Neighbour(request) => {
+ let k = topology::k(state);
+ let (agreed, dropped) = {
+ let mut views = self.views.lock_or_recover();
+ if views.active.contains(peer) {
+ (true, None)
+ } else if request.high || views.active.len() < k {
+ // Full: room made by dropping another, at random.
+ let dropped = if views.active.len() >= k {
+ let others: Vec<String> = views.active.iter().cloned().collect();
+ let dropped = some_of(others, 1).pop();
+ if let Some(dropped) = &dropped {
+ views.active.remove(dropped);
+ views.refused.insert(dropped.clone(), Instant::now());
+ }
+ dropped
+ } else {
+ None
+ };
+ views.active.insert(peer.to_owned());
+ views.asking.remove(peer);
+ views.refused.remove(peer);
+ // Asks not sent yet no longer needed: given up.
+ let room = k.saturating_sub(views.active.len());
+ let unsent: Vec<String> = views
+ .asking
+ .iter()
+ .filter(|(_, asking)| !asking.sent)
+ .map(|(member, _)| member.clone())
+ .collect();
+ let sent = views.asking.len() - unsent.len();
+ for member in unsent.into_iter().skip(room.saturating_sub(sent)) {
+ views.asking.remove(&member);
+ }
+ (true, dropped)
+ } else {
+ (false, None)
+ }
+ };
+ if let Some(dropped) = dropped {
+ Self::send(
+ state,
+ &dropped,
+ ParsedMessage::Disconnect(Self::answer(state, false)),
+ )
+ .await;
+ }
+ Self::send(
+ state,
+ peer,
+ ParsedMessage::NeighbourReply(Self::answer(state, agreed)),
+ )
+ .await;
+ if agreed {
+ self.changed(state);
+ }
+ }
+ ParsedMessage::NeighbourReply(answer) => {
+ {
+ let mut views = self.views.lock_or_recover();
+ // Not asked: a neighbour asked again whether it still is
+ // (`REAFFIRM_EVERY`), dropped if it refused; or one we gave up on,
+ // and told so (`maintain`), which undoes its agreeing.
+ if views.asking.remove(peer).is_none() {
+ if answer.agreed || !views.active.remove(peer) {
+ return;
+ }
+ views.refused.insert(peer.to_owned(), Instant::now());
+ drop(views);
+ self.changed(state);
+ return;
+ }
+ if answer.agreed {
+ views.active.insert(peer.to_owned());
+ } else {
+ views.refused.insert(peer.to_owned(), Instant::now());
+ }
+ }
+ self.changed(state);
+ }
+ ParsedMessage::Disconnect(_) => {
+ {
+ let mut views = self.views.lock_or_recover();
+ views.active.remove(peer);
+ views.asking.remove(peer);
+ views.refused.insert(peer.to_owned(), Instant::now());
+ }
+ self.changed(state);
+ }
+ _ => {}
+ }
+ }
+
+ fn version(&self) -> u64 {
+ self.version.load(Ordering::Relaxed)
+ }
+
+ /// Those we had as neighbours asked first: the handshakes are done, so NEIGHBOUR
+ /// goes with the next `maintain`. Up to `k` of them, at random, as `k` isn't known
+ /// here: the most `maintain` asks itself is what we lack.
+ fn seed(&self, neighbours: &BTreeSet<String>) {
+ let now = Instant::now();
+ let mut views = self.views.lock_or_recover();
+ for member in neighbours {
+ views.asking.insert(
+ member.clone(),
+ Asking {
+ since: now,
+ high: false,
+ sent: false,
+ },
+ );
+ }
+ }
+
+ #[cfg(test)]
+ fn active(&self) -> Option<BTreeSet<String>> {
+ Some(self.views.lock_or_recover().active.clone())
+ }
+}