1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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")
}
}
}
}
|
