blob window grows

1 files changed, 10 insertions, 5 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 694fff0..11bf3dc 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -16,7 +16,8 @@
 //!   sends at a time.
 //! * Fetching: the manifest from the first peer that has it (closest first, and only
 //!   from a peer asked; of content, only with the size its reference names), then the
-//!   chunks, a few in flight per peer from up to [`MAX_SOURCES`] peers. Every chunk is
+//!   chunks, a few in flight per peer at first, more as they come (up to
+//!   [`MAX_WINDOW`]), from up to [`MAX_SOURCES`] peers. Every chunk is
 //!   checked against the manifest; a peer that sends a bad one is dropped, a chunk
 //!   that doesn't arrive in time is asked of another peer. The caller checks the whole
 //!   (for a change: its hash).
@@ -41,8 +42,12 @@ use crate::transport::{BlobBody, BlobId, BlobMessage, DeltaSpec, ParsedMessage};
 const MAX_SERVING: usize = 8;
 /// Peers a download fetches chunks from at once.
 const MAX_SOURCES: usize = 4;
-/// Chunks in flight per peer.
+/// Chunks in flight per peer at first, and after a chunk timed out (from 1).
 const WINDOW: usize = 4;
+/// At most this many in flight per peer: each chunk delivered allows one more, so a
+/// link with a long round trip isn't left idle (4 chunks of 64 KiB per 400 ms round
+/// trip are about 640 KiB/s; 32 about 5 MiB/s).
+const MAX_WINDOW: usize = 32;
 /// Largest blob a download of unknown size accepts (a change: an epoch's snapshot is as
 /// large as the document). Content has a known size, and only that is accepted.
 const MAX_SIZE: u64 = 4 << 30;
@@ -321,7 +326,7 @@ async fn serve(
         };
         return reply(state, peer, id, body).await;
     }
-    for index in chunks.into_iter().take(WINDOW * 2) {
+    for index in chunks.into_iter().take(MAX_WINDOW) {
         let start = index as usize * chunk_size as usize;
         if start >= prepared.bytes.len() {
             continue;
@@ -546,8 +551,8 @@ pub(crate) async fn fetch(
                 chunks[slot] = Some(data);
                 done += 1;
                 last_progress = Instant::now();
-                // A peer that delivers gets its full window back.
-                source.window = WINDOW;
+                // A peer that delivers may have one more in flight.
+                source.window = (source.window + 1).min(MAX_WINDOW);
             }
             Ok(Some((from, BlobBody::Missing))) => {
                 sources.retain(|source| source.peer != from);