diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/node.rs | 1 | ||||
| -rw-r--r-- | src/state.rs | 3 | ||||
| -rw-r--r-- | src/swarm/source.rs | 26 |
3 files changed, 28 insertions, 2 deletions
diff --git a/src/node.rs b/src/node.rs index 8a49c56..57d961c 100644 --- a/src/node.rs +++ b/src/node.rs @@ -113,6 +113,7 @@ impl Node { stale: Default::default(), catch_up_source: Default::default(), catch_up_preference: Default::default(), + catch_up_avoided: Default::default(), }; for actor in config.earlier_actors { state.remake.add_own_actor(actor); diff --git a/src/state.rs b/src/state.rs index 5ec1c5a..78b058c 100644 --- a/src/state.rs +++ b/src/state.rs @@ -673,6 +673,9 @@ pub(crate) struct AppState { pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>, /// Since when a closer peer has had a head start to become the source. pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>, + /// A catch-up source given up for not answering our sync messages, and until when it + /// isn't chosen again (`swarm::source`). + pub(crate) catch_up_avoided: Arc<std::sync::Mutex<Option<(String, Instant)>>>, } /// The peer a catching-up node syncs with first (`swarm::source`). diff --git a/src/swarm/source.rs b/src/swarm/source.rs index d5b572a..9eac445 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -21,7 +21,9 @@ //! So it is also given up after three of our sync messages in a row went unanswered, //! once nothing from it can still be on the way: no sync message from it for //! `Timing::delivery_limit`, after which Mycelium has delivered or dropped anything it -//! sent before. Meanwhile, every minute without a sync message from it is logged as a +//! sent before. It isn't chosen again for another `delivery_limit`, so another peer +//! takes over rather than the same one stalling again; if it is the only candidate, +//! catching up waits that long before trying it again. Meanwhile, every minute without a sync message from it is logged as a //! warning. use std::sync::atomic::Ordering; @@ -56,6 +58,13 @@ fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant .unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn lock_avoided(state: &AppState) -> std::sync::MutexGuard<'_, Option<(String, Instant)>> { + state + .catch_up_avoided + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + fn catching_up(state: &AppState) -> bool { !state.initial_sync_done.load(Ordering::Acquire) } @@ -74,12 +83,20 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt return None; } let stall = state.timing.catch_up_probe * UNANSWERED_PROBES; + let now = Instant::now(); + let avoided_until = lock_avoided(state) + .as_ref() + .filter(|(avoided, until)| avoided == peer && now < *until) + .map(|(_, until)| *until); + if let Some(until) = avoided_until { + // Given up for not answering; the timer brings it back once it may be chosen. + return Some(until - now); + } if peer_state.reports_ready == Some(false) || peer_state.syncs_with_us == Some(false) { // Not a candidate; its next message re-evaluates this, the timer is a fallback. return Some(stall); } - let now = Instant::now(); let mut source = lock(state); let quiet = match &*source { Some(current) if current.peer == peer => return None, @@ -171,6 +188,10 @@ pub(crate) fn reply_timed_out(state: &AppState, peer: &str) { let timeouts = current.reply_timeouts; if give_up { *source = None; + *lock_avoided(state) = Some(( + peer.to_owned(), + Instant::now() + state.timing.delivery_limit, + )); } give_up.then_some((timeouts, silent)) }; @@ -273,6 +294,7 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool { pub(crate) fn finished(state: &AppState, from: &str) { *lock(state) = None; *lock_preference(state) = None; + *lock_avoided(state) = None; let state = state.clone(); let from = from.to_owned(); tokio::spawn(async move { |