Create multiple HTTP streams intentionally.

Cargo.toml

@@ -31,6 +31,7 @@ reqwest = { version = "0.11.13", features = ["blocking"] }
 tempfile = "3.3.0"
 thiserror = "1.0.37"
 zip = "0.6.3"
+arbitrary = { version = "1.2.0", features = [ "derive" ] }
 
 [dev-dependencies]
 hexdump = "0.1.1"

fuzz/fuzz_targets/fuzz_ripunzip.rs

@@ -16,11 +16,11 @@ use httptest::Server;
 use libfuzzer_sys::arbitrary;
 use libfuzzer_sys::fuzz_target;
 use std::collections::{HashMap, HashSet};
+use ripunzip::UnzipUriOptions;
 use std::fs::File;
 use std::io::Cursor;
 use std::io::Write;
 use std::path::Path;
-use itertools::Itertools;
 
 /// Segments of the filename of each zip member. We don't allow
 /// totally arbitrary filenames, because that tends to throw up permissions
@@ -63,6 +63,7 @@ enum Method {
     Uri {
         readahead_limit: Option<usize>,
         server_type: ripunzip_test_utils::ServerType,
+        uri_options: UnzipUriOptions,
     },
 }
 
@@ -98,16 +99,16 @@ fuzz_target!(|input: Inputs| {
             ripunzip.unzip()
         }
         Method::Uri {
-            readahead_limit,
             server_type,
+            uri_options,
         } => {
             let server = Server::run();
             ripunzip_test_utils::set_up_server(&server, zip_data, server_type);
             let uri = &server.url("/foo").to_string();
             let ripunzip = ripunzip::UnzipEngine::for_uri(
                 uri,
                 options,
-                readahead_limit,
+                uri_options,
                 progress_reporter,
                 || {},
             )?;
@@ -141,16 +142,19 @@ fn recursive_lsdir(dir: &Path) -> HashSet<std::path::PathBuf> {
         .collect()
 }
 
-fn create_zip(output: &mut Vec<u8>, zip_members: &HashMap<ZipMemberFilename, Vec<u8>>, compression_method: CompressionMethod) {
+fn create_zip(
+    output: &mut Vec<u8>,
+    zip_members: &HashMap<ZipMemberFilename, Vec<u8>>,
+    compression_method: CompressionMethod,
+) {
     let compression_method = match compression_method {
         CompressionMethod::Stored => zip::CompressionMethod::Stored,
         CompressionMethod::Deflated => zip::CompressionMethod::Deflated,
         CompressionMethod::Bzip2 => zip::CompressionMethod::Bzip2,
         CompressionMethod::Zstd => zip::CompressionMethod::Zstd,
     };
     let mut zip = zip::ZipWriter::new(Cursor::new(output));
-    let options =
-        zip::write::FileOptions::default().compression_method(compression_method);
+    let options = zip::write::FileOptions::default().compression_method(compression_method);
     for (name, data) in zip_members.into_iter() {
         zip.start_file(name, options).unwrap();
         zip.write(&data).unwrap();

src/lib.rs

@@ -12,3 +12,4 @@ pub use unzip::NullProgressReporter;
 pub use unzip::UnzipEngine;
 pub use unzip::UnzipOptions;
 pub use unzip::UnzipProgressReporter;
+pub use unzip::UnzipUriOptions;

src/main.rs

@@ -6,12 +6,12 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use std::{fmt::Write, fs::File, path::PathBuf};
+use std::{fmt::Write, fs::File, num::NonZeroUsize, path::PathBuf};
 
 use anyhow::Result;
 use clap::{Parser, Subcommand};
 use indicatif::{ProgressBar, ProgressState, ProgressStyle};
-use ripunzip::{UnzipEngine, UnzipOptions, UnzipProgressReporter};
+use ripunzip::{UnzipEngine, UnzipOptions, UnzipProgressReporter, UnzipUriOptions};
 
 /// Unzip all files within a zip file as quickly as possible.
 #[derive(Parser, Debug)]
@@ -51,6 +51,11 @@ enum Commands {
         /// problem, but may make transfers much less efficient by requiring multiple HTTP streams.
         #[arg(long, value_name = "BYTES")]
         readahead_limit: Option<usize>,
+
+        /// Whether to create just a single HTTP stream for the main part of the download.
+        /// If false, multiple HTTP streams may be created to download in parallel.
+        #[arg(long)]
+        single_stream: bool,
     },
 }
 
@@ -69,14 +74,26 @@ fn main() -> Result<()> {
         Commands::Uri {
             uri,
             readahead_limit,
-        } => UnzipEngine::for_uri(
-            uri,
-            options,
-            *readahead_limit,
-            ProgressDisplayer::new(),
-            report_on_insufficient_readahead_size,
-        )?
-        .unzip(),
+            single_stream,
+        } => {
+            let unzip_uri_options = UnzipUriOptions {
+                readahead_limit: *readahead_limit,
+                maximum_streams: if *single_stream {
+                    Some(NonZeroUsize::new(1).unwrap())
+                } else {
+                    None
+                },
+                ..Default::default()
+            };
+            UnzipEngine::for_uri(
+                uri,
+                options,
+                unzip_uri_options,
+                ProgressDisplayer::new(),
+                report_on_insufficient_readahead_size,
+            )?
+            .unzip()
+        }
     }
 }
 

src/unzip/http_range_reader.rs

@@ -6,7 +6,7 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use std::{cmp::min, io::Read};
+use std::{cmp::min, io::Read, ops::Range};
 
 use reqwest::blocking::{Client, Response};
 use thiserror::Error;
@@ -74,17 +74,17 @@ impl RangeFetcher {
     /// of the resource but discard bytes before that point. (Clearly that
     /// can be expensive if you only care about a few bytes later in a
     /// resource.)
-    pub(crate) fn fetch_range(&self, offset: u64) -> Result<Response, Error> {
-        log::info!("Fetch range 0x{:x}", offset);
+    pub(crate) fn fetch_range(&self, offset: Range<u64>) -> Result<Response, Error> {
+        log::info!("Fetch range {:?}", offset);
         let mut builder = self.client.get(&self.uri);
         if self.accept_ranges {
-            let range_header = format!("bytes={}-{}", offset, self.len());
+            let range_header = format!("bytes={}-{}", offset.start, offset.end);
             builder = builder.header(reqwest::header::RANGE, range_header);
         }
         let mut response = builder.send().map_err(Error::HttpGet)?;
-        if !self.accept_ranges && offset > 0 {
+        if !self.accept_ranges && offset.start > 0 {
             // Read and discard data prior to 'offset'
-            let mut to_read = offset as usize;
+            let mut to_read = offset.start as usize;
             let mut throwaway = [0u8; 4096];
             while to_read > 0 {
                 let bytes_read = response
@@ -150,13 +150,13 @@ mod tests {
                 .respond_with(status_code(200).body(body))
         });
         assert_eq!(accept_ranges, range_fetcher.accepts_ranges());
-        let mut resp = range_fetcher.fetch_range(0u64).unwrap();
+        let mut resp = range_fetcher.fetch_range(0u64..10u64).unwrap();
         let mut throwaway = [0u8; 10];
         resp.read_exact(&mut throwaway).unwrap();
         assert_eq!(std::str::from_utf8(&throwaway).unwrap(), "0123456789");
 
         // Test read only a range
-        let mut resp = range_fetcher.fetch_range(4u64).unwrap();
+        let mut resp = range_fetcher.fetch_range(4u64..10u64).unwrap();
         let mut throwaway = [0u8; 6];
         resp.read_exact(&mut throwaway).unwrap();
         assert_eq!(std::str::from_utf8(&throwaway).unwrap(), "456789");

src/unzip/mod.rs

@@ -14,6 +14,7 @@ use std::{
     borrow::Cow,
     fs::{File, Permissions},
     io::{ErrorKind, Read, Seek},
+    num::{NonZeroU64, NonZeroUsize},
     path::{Path, PathBuf},
     sync::{Arc, Mutex},
 };
@@ -29,6 +30,45 @@ use self::{
     seekable_http_reader::{AccessPattern, SeekableHttpReader, SeekableHttpReaderEngine},
 };
 
+/// This is how much we read from the underlying HTTP stream in a given thread,
+/// before signalling other threads that they may wish to continue with their
+/// CPU-bound unzipping. Empirically determined.
+/// 128KB = 172ms
+/// 512KB = 187ms
+/// 1024KB = 152ms
+/// 2048KB = 170ms
+/// If we set this too high, we starve multiple threads - they can't start
+/// acting on the data to unzip their files until the read is complete. If we
+/// set this too low, the cache structure (a `BTreeMap`) becomes dominant in
+/// CPU usage.
+const DEFAULT_MAX_BLOCK: usize = 1024 * 1024;
+
+/// We create multiple parallel HTTP streams. Each HTTP stream reads this much
+/// of the end file. If we set this too low, we waste parallelism. If we set
+/// it too high, each HTTP stream takes too long to ramp up based on the TCP
+/// slow start algorithm (or equivalents)
+const DEFAULT_RANGE_PER_READER: u64 = 1024 * 1024 * 1024;
+
+/// Options for downloading from a URI. If in doubt, use the [`Default`]
+/// implementation. These parameters are typically only useful for testing
+/// or fuzzing or performance optimization, though you may wish to set
+/// [`readahead_limit`] to limit RAM consumption.
+#[derive(Default, Debug, Clone)]
+#[cfg_attr(feature = "fuzzing", derive(arbitrary::Arbitrary))]
+pub struct UnzipUriOptions {
+    /// Maximum RAM consumption of stored blocks of zip data that are yet
+    /// to be read by the unzip algorithm.
+    pub readahead_limit: Option<usize>,
+    /// How many bytes to read in one go, if we're reading ahead of the
+    /// current zip file request. This is almost never worth tinkering
+    /// with unless you're testing pathological cases.
+    pub max_block: Option<NonZeroUsize>,
+    /// The maximum amount of data to read per HTTP stream.
+    pub range_per_stream: Option<NonZeroU64>,
+    /// The maximum number of parallel HTTP streams to have.
+    pub maximum_streams: Option<NonZeroUsize>,
+}
+
 /// Options for unzipping.
 pub struct UnzipOptions {
     /// The destination directory.
@@ -218,26 +258,43 @@ impl<P: UnzipProgressReporter> UnzipEngine<P> {
         })
     }
 
-    /// Create an unzip engine which knows how to unzip a URI.
+    /// Create an unzip engine which knows how to unzip a URI. This will try
+    /// to download and unzip in parallel, if the HTTP server supports
+    /// the `Range` header.
     /// Parameters:
     /// - the URI
     /// - unzip options
     /// - how big a readahead buffer to create in memory.
+    /// - optionally, a maximum block size to read. This is only useful in
+    ///   testing and fuzzing cases where we want to simulate maximal parallelism
+    ///   even with small zip files. Leave at `None` if in doubt.
+    /// - how large a chunk of the file should be downloaded on each separate
+    ///   HTTP stream. Leave at `None` if in doubt.
+    /// - a maximum number of streams to create
     /// - a progress reporter (set of callbacks)
     /// - an additional callback to warn if performance was impaired by
     ///   rewinding the HTTP stream. (This implies the readahead buffer was
     ///   too small.)
     pub fn for_uri<F: Fn() + 'static>(
         uri: &str,
         options: UnzipOptions,
-        readahead_limit: Option<usize>,
+        uri_options: UnzipUriOptions,
         progress_reporter: P,
         callback_on_rewind: F,
     ) -> Result<Self> {
+        let max_block = uri_options
+            .max_block
+            .unwrap_or_else(|| NonZeroUsize::new(DEFAULT_MAX_BLOCK).unwrap());
+        let range_per_reader = uri_options
+            .range_per_stream
+            .unwrap_or_else(|| NonZeroU64::new(DEFAULT_RANGE_PER_READER).unwrap());
         let seekable_http_reader = SeekableHttpReaderEngine::new(
             uri.to_string(),
-            readahead_limit,
+            uri_options.readahead_limit,
+            uri_options.maximum_streams,
             AccessPattern::RandomAccess,
+            max_block,
+            range_per_reader,
         );
         let (compressed_length, zipfile): (u64, Box<dyn UnzipEngineImpl>) =
             match seekable_http_reader {
@@ -390,7 +447,7 @@ mod tests {
     use test_log::test;
     use zip::{write::FileOptions, ZipWriter};
 
-    use crate::{NullProgressReporter, UnzipEngine, UnzipOptions};
+    use crate::{NullProgressReporter, UnzipEngine, UnzipOptions, UnzipUriOptions};
     use ripunzip_test_utils::*;
 
     fn create_zip_file(path: &Path) {
@@ -484,7 +541,7 @@ mod tests {
         UnzipEngine::for_uri(
             &server.url("/foo").to_string(),
             options,
-            None,
+            UnzipUriOptions::default(),
             NullProgressReporter,
             || {},
         )
@@ -509,7 +566,7 @@ mod tests {
         UnzipEngine::for_uri(
             &server.url("/foo").to_string(),
             options,
-            None,
+            UnzipUriOptions::default(),
             NullProgressReporter,
             || {},
         )