aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/state/mod.rs2
-rw-r--r--src/state/peer.rs13
-rw-r--r--src/swarm/peer/admit.rs3
-rw-r--r--src/swarm/peer/handshake.rs15
-rw-r--r--src/swarm/peer/keepalive.rs24
-rw-r--r--src/swarm/peer/worker.rs8
-rw-r--r--src/swarm/topology/mod.rs22
7 files changed, 56 insertions, 31 deletions
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 8f3a841..a2d5d74 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -143,7 +143,7 @@ pub(crate) struct AppState {
/// Our neighbours, once known (`swarm::topology`).
pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,
/// Something the neighbours are computed from may have changed: the member list or a
- /// peer's reachability (`swarm::topology::run_worker`).
+ /// peer's state (`swarm::topology::run_worker`, `swarm::topology::change_peer`).
pub(crate) topology_changed: Arc<tokio::sync::Notify>,
/// While rejoining a newer epoch, the previous document, served read-only (and
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 95a9dc3..99f912b 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -288,6 +288,19 @@ impl PeerState {
self.unreachable_since.is_some()
}
+ /// What choosing neighbours reads of this peer (`swarm::topology::refresh`): whether
+ /// it is unreachable, whether it was ever heard from, whether it is a member, and
+ /// whether it syncs with us. A change to any may change the neighbours
+ /// (`swarm::topology::change_peer`).
+ pub(crate) fn neighbour_inputs(&self) -> impl PartialEq + use<> {
+ (
+ self.is_unreachable(),
+ self.last_seen.is_some(),
+ matches!(self.standing, Standing::Member { .. }),
+ self.syncs_with_us,
+ )
+ }
+
/// The wait before the next HELLO: a probe's if unreachable.
pub(crate) fn next_hello_wait(&mut self) -> Duration {
if self.is_unreachable() {
diff --git a/src/swarm/peer/admit.rs b/src/swarm/peer/admit.rs
index 1800775..6e02150 100644
--- a/src/swarm/peer/admit.rs
+++ b/src/swarm/peer/admit.rs
@@ -4,6 +4,7 @@ use tracing::{debug, info, trace, warn};
use super::handshake::remember_peer;
use crate::state::{AppState, HelloState, Standing};
+use crate::swarm::topology;
use crate::transport::ParsedMessage;
/// With a swarm key, only peers that proved it get past this point. A HELLO or
@@ -50,7 +51,7 @@ pub(super) async fn admit(state: &AppState, peer: &str, message: &ParsedMessage)
if peer_state.standing.is_admitted() {
warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership");
}
- peer_state.standing = Standing::Stranger;
+ topology::change_peer(state, &mut peer_state, |p| p.standing = Standing::Stranger);
peer_state.hello_state = HelloState::NotSent;
peer_state.reset_sync();
}
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 225a7c9..13a1264 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -147,11 +147,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
let mut peer_state = peer_state_mutex.lock().await;
if peer_state.hello_state == HelloState::Sending {
- let was_unreachable = peer_state.unreachable_since.is_some();
- peer_state.hello_sent(Instant::now());
- if !was_unreachable && peer_state.unreachable_since.is_some() {
- state.topology_changed.notify_one();
- }
+ topology::change_peer(state, &mut peer_state, |p| p.hello_sent(Instant::now()));
let wait = peer_state.next_hello_wait();
peer_state.hello_state = HelloState::AwaitingReply {
next_attempt: Instant::now() + wait,
@@ -402,11 +398,12 @@ pub(super) async fn establish_peer(
}
if !matches!(peer_state.standing, Standing::Member { .. }) {
- peer_state.standing = Standing::Member {
- swarm_id: swarm_id.clone(),
- };
// A member our list may lack yet: a neighbour then (`topology::refresh`).
- state.topology_changed.notify_one();
+ topology::change_peer(state, &mut peer_state, |p| {
+ p.standing = Standing::Member {
+ swarm_id: swarm_id.clone(),
+ }
+ });
}
// Stop our retries and sync from a clean slate. A new handshake from the peer always
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 40b8ed3..5e03d0e 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -91,11 +91,9 @@ pub(super) async fn heard_from(
let mut peer_state = peer_state.lock().await;
also(&mut peer_state);
let previously_seen = peer_state.last_seen;
- if peer_state.unreachable_since.is_some() {
- // Reachable again (`mark_seen` clears it).
- state.topology_changed.notify_one();
- }
- (peer_state.mark_seen(now), previously_seen)
+ // Reachable (again).
+ let reactivated = topology::change_peer(state, &mut peer_state, |p| p.mark_seen(now));
+ (reactivated, previously_seen)
};
crate::swarm::source::heard_from(state, peer);
if reactivated {
@@ -187,22 +185,20 @@ pub(super) async fn status_received(state: &AppState, peer: &str) {
pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
- // Its farewell: dormant now, as if silent for long (a message from it, as its
- // HELLO when it's back, makes it active again).
- let left = remote.leaving && peer_state.say_farewell(Instant::now());
+ let left = topology::change_peer(state, &mut peer_state, |p| {
+ // A neighbour that only connects out stays one as long as it says so.
+ p.syncs_with_us = Some(remote.syncs_with_you);
+ // Its farewell: dormant now, as if silent for long (a message from it, as
+ // its HELLO when it's back, makes it active again).
+ remote.leaving && p.say_farewell(Instant::now())
+ });
let became_ready = remote.ready && peer_state.reports_ready != Some(true);
peer_state.reports_ready = Some(remote.ready);
- let changed =
- peer_state.syncs_with_us.replace(remote.syncs_with_you) != Some(remote.syncs_with_you);
peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
drop(peer_state);
if left {
info!(peer = %peer, "Peer said farewell; dormant until it is heard from again");
}
- if changed || left {
- // A neighbour that only connects out stays one as long as it says so.
- state.topology_changed.notify_one();
- }
// Off the consumer: telling which heads we lack takes the document's lock.
{
let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone());
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 87d2b3c..797f091 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -51,8 +51,11 @@ pub(crate) async fn run_peer_worker(state: AppState) {
let (became_dormant, keepalive, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
- let became_dormant =
- ready && neighbour && peer_state.check_dormancy(&state.timing, now);
+ let became_dormant = ready
+ && neighbour
+ && topology::change_peer(&state, &mut peer_state, |p| {
+ p.check_dormancy(&state.timing, now)
+ });
let keepalive = if neighbour {
peer_state.take_keepalive_due(&state.timing, now)
} else {
@@ -72,7 +75,6 @@ pub(crate) async fn run_peer_worker(state: AppState) {
}
if became_dormant {
- state.topology_changed.notify_one();
warn!(
peer = %peer,
silent_secs = state.timing.dormant_after.as_secs(),
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index ea71712..e494f34 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -30,7 +30,7 @@ use std::collections::{BTreeMap, BTreeSet};
use tracing::info;
use crate::locks::RwLockExt;
-use crate::state::AppState;
+use crate::state::{AppState, PeerState};
use crate::swarm::members;
mod graph;
@@ -164,10 +164,26 @@ pub(crate) struct NeighbourChange {
pub(crate) removed: Vec<String>,
}
+/// Runs `change` on a peer's state, and has the neighbours recomputed (`run_worker`) if
+/// it changed what they are chosen from (`PeerState::neighbour_inputs`). Changes to a
+/// peer that may matter to the neighbours go through here.
+pub(crate) fn change_peer<T>(
+ state: &AppState,
+ peer_state: &mut PeerState,
+ change: impl FnOnce(&mut PeerState) -> T,
+) -> T {
+ let before = peer_state.neighbour_inputs();
+ let result = change(peer_state);
+ if peer_state.neighbour_inputs() != before {
+ state.topology_changed.notify_one();
+ }
+ result
+}
+
/// Recomputes the neighbours whenever something they depend on may have changed
/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`),
-/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a
-/// fallback. A burst of notifications coalesces into one recomputation.
+/// a peer's state (`change_peer`), and every 5 s as a fallback (`swarm::peer`). A burst
+/// of notifications coalesces into one recomputation.
pub(crate) async fn run_worker(state: AppState) {
loop {
state.topology_changed.notified().await;