diff options
| -rw-r--r-- | src/swarm/blob.rs | 56 |
1 files changed, 55 insertions, 1 deletions
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 454687f..6febc92 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -23,7 +23,9 @@ //! checked against the manifest; a peer that sends a bad one is dropped, a chunk //! that doesn't arrive in time (three times as long as its chunks took lately, where //! another peer could send it) is asked of another peer, as are all of a peer's once -//! we no longer reach it (`swarm::reach`). The caller checks the whole +//! we no longer reach it (`swarm::reach`). The last chunks, once all are asked for, +//! are asked of a second source too where one has room, if they are a second late +//! ([`DOUBLE_AFTER`]). The caller checks the whole //! (for a change: its hash). //! //! The chunk size is the serving node's (`--blob-chunk`); the manifest carries it. @@ -76,6 +78,9 @@ const CHUNK_TIMEOUT: Duration = Duration::from_secs(60); const CHUNK_TIMEOUT_MIN: Duration = Duration::from_secs(5); /// A download gives up after this long without any chunk arriving. const STALL_TIMEOUT: Duration = Duration::from_secs(180); +/// At the end of a download, a chunk in flight this long is asked of a second source +/// with room too: what comes quickly isn't fetched twice. +const DOUBLE_AFTER: Duration = Duration::from_secs(1); /// Bad chunks after which a peer is no longer asked. const MAX_BAD: u32 = 2; @@ -556,6 +561,8 @@ pub(crate) async fn fetch( let mut done = 0usize; let mut sources: Vec<Source> = Vec::new(); let mut last_progress = Instant::now(); + // Chunks asked of a second source at the end. + let mut doubled: HashSet<u32> = HashSet::new(); while done < total { // Fill up the sources, then their windows. @@ -595,6 +602,46 @@ pub(crate) async fn fetch( } request(state, &source.peer, id, batch, delta).await; } + // The last chunks: all are asked for, and a source has room. A chunk another one + // has had in flight for `DOUBLE_AFTER` is asked of it too, once (`doubled`): the + // download ends with the first of the two, not the slower. + if wanted.next().is_none() && sources.len() > 1 { + for at in 0..sources.len() { + let free = sources[at] + .window + .saturating_sub(sources[at].in_flight.len()); + if free == 0 { + continue; + } + let mut others: Vec<(Instant, u32)> = sources + .iter() + .enumerate() + .filter(|(other, _)| *other != at) + .flat_map(|(_, other)| { + other.in_flight.iter().map(|(index, sent)| (*sent, *index)) + }) + .filter(|(sent, index)| { + now.saturating_duration_since(*sent) >= DOUBLE_AFTER + && !doubled.contains(index) + && !sources[at].in_flight.contains_key(index) + }) + .collect(); + others.sort_unstable(); + let batch: Vec<u32> = others + .into_iter() + .take(free) + .map(|(_, index)| index) + .collect(); + if batch.is_empty() { + continue; + } + for index in &batch { + doubled.insert(*index); + sources[at].in_flight.insert(*index, now); + } + request(state, &sources[at].peer, id, batch, delta).await; + } + } let answer = tokio::time::timeout(Duration::from_millis(500), answers.recv()).await; // A deflated chunk inflated, to no more than a chunk; one that doesn't is bad. @@ -634,6 +681,13 @@ pub(crate) async fn fetch( last_progress = Instant::now(); // A peer that delivers may have one more in flight. source.window = (source.window + 1).min(MAX_WINDOW); + // Asked of two at the end: the other's copy is no longer awaited (nor + // counted against it as late). + if doubled.contains(&index) { + for other in &mut sources { + other.in_flight.remove(&index); + } + } } Ok(Some((from, BlobBody::Missing))) => { sources.retain(|source| source.peer != from); |