diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/swarm/sync.rs | 24 |
1 files changed, 23 insertions, 1 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 5a4cba4..868fead 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -11,7 +11,7 @@ use tokio::time::Instant; use tracing::{debug, info, trace, warn}; use crate::locks::MutexExt; -use crate::state::{AppState, Liveness, Standing}; +use crate::state::{AppState, Liveness, PeerState, Standing}; use crate::swarm::peer::{peer_swarm, send_hello_once}; use crate::swarm::source; use crate::swarm::throughput::SyncBytes; @@ -190,6 +190,28 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { (session, encoded) }; + // Off the sync worker: a slow send (a large message, a busy Mycelium) mustn't hold + // up every other peer's. One message per peer is in flight (`sending`), so a peer's + // messages still go out in order. + let state = state.clone(); + let peer = peer.to_owned(); + tokio::spawn(async move { + if let Err(error) = send_generated(&state, &peer, &peer_state_mutex, session, encoded).await + { + warn!(peer = %peer, error = %format!("{error:#}"), "Synchronization processing failed"); + } + }); + Ok(()) +} + +/// Hands a message generated from session `session` to the transport and settles it. +async fn send_generated( + state: &AppState, + peer: &str, + peer_state_mutex: &tokio::sync::Mutex<PeerState>, + session: u64, + encoded: transport::EncodedMessage, +) -> Result<()> { // No state locks are held across the send. match state.transport.send_encoded(&encoded).await { Ok(()) => { |