diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/node.rs | 6 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 9 | ||||
| -rw-r--r-- | src/swarm/peer/lanes.rs | 60 | ||||
| -rw-r--r-- | src/swarm/peer/mod.rs | 3 |
4 files changed, 75 insertions, 3 deletions
diff --git a/src/node.rs b/src/node.rs index 9200d16..5b55c8d 100644 --- a/src/node.rs +++ b/src/node.rs @@ -136,13 +136,17 @@ impl Node { let Self { state, sync_rx } = self; let consumer_state = state.clone(); + let lanes = Arc::new(peer::SyncLanes::start(&state)); tokio::spawn(async move { let inner_state = consumer_state.clone(); let _ = consumer_state .transport .start_consumer(Box::new(move |sender, msg, _payload_len| { let state_clone = inner_state.clone(); - Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await }) + let lanes = Arc::clone(&lanes); + Box::pin(async move { + peer::handle_message(&state_clone, &lanes, sender, msg).await + }) })) .await; }); diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 684a358..73cc629 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -12,12 +12,17 @@ use super::handshake::{ discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply, }; use super::keepalive::{heal_if_diverged, local_keepalive, note_keepalive}; +use super::lanes::SyncLanes; use crate::state::AppState; -use crate::swarm::sync::{SyncEvent, handle_sync}; +use crate::swarm::sync::SyncEvent; use crate::transport::{self, ParsedMessage}; +/// Handles `message` from `peer` in arrival order, except that a sync message is only +/// queued here, on `lanes`: everything before it (admission, liveness, the peer's epoch) +/// is done in order, applying it isn't. pub(crate) async fn handle_message( state: &AppState, + lanes: &SyncLanes, peer: String, message: transport::ParsedMessage, ) -> Result<()> { @@ -130,7 +135,7 @@ pub(crate) async fn handle_message( return Ok(()); } crate::swarm::source::synced_from(state, &peer); - handle_sync(state, &peer, envelope).await?; + lanes.enqueue(&peer, envelope); } transport::ParsedMessage::Blob(blob) => { crate::swarm::blob::handle(state, &peer, blob).await; diff --git a/src/swarm/peer/lanes.rs b/src/swarm/peer/lanes.rs new file mode 100644 index 0000000..88639ea --- /dev/null +++ b/src/swarm/peer/lanes.rs @@ -0,0 +1,60 @@ +//! Sync lanes: inbound sync messages are applied off the consumer, so one slow apply (a +//! large delta takes seconds) doesn't hold up keepalives, HELLOs and blob chunks behind +//! it. Each sender's messages go to the same lane, in arrival order; a few lanes are +//! enough, since applying takes the document's write lock anyway. + +use std::collections::hash_map::DefaultHasher; +use std::hash::{Hash, Hasher}; + +use tokio::sync::mpsc::{self, error::TrySendError}; +use tracing::{debug, warn}; + +use crate::state::AppState; +use crate::swarm::sync::handle_sync; +use crate::transport::SyncEnvelope; + +const LANES: usize = 4; +/// Sync messages waiting per lane. A full lane drops what arrives: its sender gets no +/// answer and resends after its reply timeout, by which time the backlog is gone. +const QUEUED_PER_LANE: usize = 256; + +pub(crate) struct SyncLanes { + lanes: Vec<mpsc::Sender<(String, SyncEnvelope)>>, +} + +impl SyncLanes { + /// Starts a task per lane. + pub(crate) fn start(state: &AppState) -> Self { + let lanes = (0..LANES) + .map(|_| { + let (tx, mut rx) = mpsc::channel::<(String, SyncEnvelope)>(QUEUED_PER_LANE); + let state = state.clone(); + tokio::spawn(async move { + while let Some((peer, envelope)) = rx.recv().await { + if let Err(error) = handle_sync(&state, &peer, envelope).await { + warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process sync message"); + } + } + }); + tx + }) + .collect(); + Self { lanes } + } + + /// Queues `envelope` from `peer` on its lane. + pub(crate) fn enqueue(&self, peer: &str, envelope: SyncEnvelope) { + let mut hasher = DefaultHasher::new(); + peer.hash(&mut hasher); + let lane = &self.lanes[hasher.finish() as usize % LANES]; + match lane.try_send((peer.to_owned(), envelope)) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + debug!(peer = %peer, "Sync lane full; dropping the message (its sender resends)") + } + Err(TrySendError::Closed(_)) => { + warn!(peer = %peer, "Sync lane stopped; dropping the message") + } + } + } +} diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs index f0222c4..2e5c1dc 100644 --- a/src/swarm/peer/mod.rs +++ b/src/swarm/peer/mod.rs @@ -8,6 +8,7 @@ //! * `forget`: swarm-wide forget records. //! * `worker`: the periodic peer worker. //! * `inbound`: dispatch of inbound messages. +//! * `lanes`: sync messages applied off the consumer. mod admit; mod epoch; @@ -15,6 +16,7 @@ mod forget; mod handshake; mod inbound; mod keepalive; +mod lanes; mod worker; pub(crate) use forget::{forget_unless_alive, forgotten_path}; @@ -23,4 +25,5 @@ pub(crate) use handshake::{ }; pub(crate) use inbound::handle_message; pub(crate) use keepalive::send_ping; +pub(crate) use lanes::SyncLanes; pub(crate) use worker::run_peer_worker; |