diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/node.rs | 4 | ||||
| -rw-r--r-- | src/state.rs | 3 | ||||
| -rw-r--r-- | src/swarm/members.rs | 2 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 16 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 7 | ||||
| -rw-r--r-- | src/swarm/topology.rs | 20 |
6 files changed, 47 insertions, 5 deletions
diff --git a/src/node.rs b/src/node.rs index b3e7463..82b99fc 100644 --- a/src/node.rs +++ b/src/node.rs @@ -103,6 +103,7 @@ impl Node { files: config.files, config: Arc::new(config.config_report), neighbours: Default::default(), + topology_changed: Default::default(), pulling: Default::default(), announce_pending: Default::default(), rtt: Default::default(), @@ -155,6 +156,9 @@ impl Node { tokio::spawn(remake::run_worker(state.clone())); trace!("Remake worker task spawned"); + tokio::spawn(swarm::topology::run_worker(state.clone())); + trace!("Topology worker task spawned"); + state } } diff --git a/src/state.rs b/src/state.rs index af01cf0..ee7fa7a 100644 --- a/src/state.rs +++ b/src/state.rs @@ -632,6 +632,9 @@ pub(crate) struct AppState { pub(crate) announce_pending: Arc<std::sync::Mutex<Option<std::collections::HashSet<String>>>>, /// 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`). + pub(crate) topology_changed: Arc<tokio::sync::Notify>, /// While rejoining a newer epoch, the previous document, served read-only (and /// marked stale) instead of answering 503. diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 221437e..895e0cc 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -25,7 +25,7 @@ use crate::state::{AppState, unix_millis}; const META_ROOT: &str = "_continuum"; -fn members_dir() -> Vec<String> { +pub(crate) fn members_dir() -> Vec<String> { vec![META_ROOT.to_owned(), "members".to_owned()] } diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 7385ae2..e8248bf 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -299,7 +299,11 @@ 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(); + } let wait = match peer_state.liveness { Liveness::Dormant { .. } => state.timing.dormant_probe, Liveness::Active => peer_state.hello_backoff.next(), @@ -413,7 +417,7 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S /// neighbours get no more HELLOs, keepalives or sync openings from us; their session is /// reset, so we don't keep pushing changes to them either. They are still answered if /// they contact us. -async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { +pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { for peer in &change.added { remember_peer(state, peer).await; if let Err(error) = send_hello_once(state, peer).await { @@ -492,6 +496,7 @@ 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(), @@ -529,9 +534,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { .unwrap_or_else(|poisoned| poisoned.into_inner()) .retain(|peer| known.contains_key(peer)); } - if let Some(change) = topology::refresh(&state).await { - apply_neighbour_change(&state, change).await; - } + // Fallback for what doesn't notify (`topology::run_worker`). + state.topology_changed.notify_one(); last_forget_check = Some(Instant::now()); } } @@ -802,6 +806,10 @@ pub(crate) async fn handle_message( let peer_state_mutex = state.get_or_create_peer_state(&peer).await; let mut peer_state = peer_state_mutex.lock().await; 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(Instant::now()), previously_seen) }; crate::swarm::source::heard_from(state, &peer); diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 93af6f7..d4503bb 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -608,6 +608,13 @@ pub(crate) async fn run_change_publisher(state: AppState) { warn!(path = ?path, "Concurrent writes produced a conflict"); } + if paths + .iter() + .any(|path| crate::swarm::topology::affects_neighbours(path)) + { + state.topology_changed.notify_one(); + } + // Fails only when nobody is subscribed. let _ = state.change_events.send(Arc::new(ChangeSet { paths })); } diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index 57856b6..b40b120 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -107,6 +107,26 @@ pub(crate) struct NeighbourChange { pub(crate) removed: Vec<String>, } +/// Recomputes the neighbours whenever something they depend on may have changed +/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`), +/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a +/// fallback, which also covers unreachability expiring after `dormant_probe`. A burst of +/// notifications coalesces into one recomputation. +pub(crate) async fn run_worker(state: AppState) { + loop { + state.topology_changed.notified().await; + if let Some(change) = refresh(&state).await { + crate::swarm::peer::apply_neighbour_change(&state, change).await; + } + } +} + +/// Whether a change at `path` may change the neighbours: one in the member list, or the +/// root (the whole document replaced, as after an epoch switch). +pub(crate) fn affects_neighbours(path: &[String]) -> bool { + path.is_empty() || path.starts_with(&members::members_dir()) +} + /// Recomputes the neighbours once ready, if the member list, the reachable members or /// `k` changed. pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { |