diff --git a/Cargo.lock b/Cargo.lock index 2fb380f..01fe5b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1361,6 +1361,7 @@ name = "ripunzip" version = "0.2.2" dependencies = [ "anyhow", + "arbitrary", "clap 4.0.29", "criterion", "env_logger", diff --git a/Cargo.toml b/Cargo.toml index 142da95..5d3fff8 100644 --- a/Cargo.toml +++ b/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" diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index 244ca09..dbae21e 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -1198,6 +1198,7 @@ name = "ripunzip" version = "0.2.2" dependencies = [ "anyhow", + "arbitrary", "clap", "env_logger", "indicatif", diff --git a/fuzz/fuzz_targets/fuzz_ripunzip.rs b/fuzz/fuzz_targets/fuzz_ripunzip.rs index da6d136..5218539 100644 --- a/fuzz/fuzz_targets/fuzz_ripunzip.rs +++ b/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, server_type: ripunzip_test_utils::ServerType, + uri_options: UnzipUriOptions, }, } @@ -98,8 +99,8 @@ 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); @@ -107,7 +108,7 @@ fuzz_target!(|input: Inputs| { let ripunzip = ripunzip::UnzipEngine::for_uri( uri, options, - readahead_limit, + uri_options, progress_reporter, || {}, )?; @@ -141,7 +142,11 @@ fn recursive_lsdir(dir: &Path) -> HashSet { .collect() } -fn create_zip(output: &mut Vec, zip_members: &HashMap>, compression_method: CompressionMethod) { +fn create_zip( + output: &mut Vec, + zip_members: &HashMap>, + compression_method: CompressionMethod, +) { let compression_method = match compression_method { CompressionMethod::Stored => zip::CompressionMethod::Stored, CompressionMethod::Deflated => zip::CompressionMethod::Deflated, @@ -149,8 +154,7 @@ fn create_zip(output: &mut Vec, zip_members: &HashMap 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(); diff --git a/src/lib.rs b/src/lib.rs index 056ab4b..99a7a7b 100644 --- a/src/lib.rs +++ b/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; diff --git a/src/main.rs b/src/main.rs index e095239..df2063c 100644 --- a/src/main.rs +++ b/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, + + /// 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() + } } } diff --git a/src/unzip/http_range_reader.rs b/src/unzip/http_range_reader.rs index 82f05cc..a77bd56 100644 --- a/src/unzip/http_range_reader.rs +++ b/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 { - log::info!("Fetch range 0x{:x}", offset); + pub(crate) fn fetch_range(&self, offset: Range) -> Result { + 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"); diff --git a/src/unzip/mod.rs b/src/unzip/mod.rs index 8c5e181..343991e 100644 --- a/src/unzip/mod.rs +++ b/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, + /// 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, + /// The maximum amount of data to read per HTTP stream. + pub range_per_stream: Option, + /// The maximum number of parallel HTTP streams to have. + pub maximum_streams: Option, +} + /// Options for unzipping. pub struct UnzipOptions { /// The destination directory. @@ -218,11 +258,19 @@ impl UnzipEngine

{ }) } - /// 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 @@ -230,14 +278,23 @@ impl UnzipEngine

{ pub fn for_uri( uri: &str, options: UnzipOptions, - readahead_limit: Option, + uri_options: UnzipUriOptions, progress_reporter: P, callback_on_rewind: F, ) -> Result { + 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) = 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, || {}, ) diff --git a/src/unzip/seekable_http_reader.rs b/src/unzip/seekable_http_reader.rs index 0202cd6..bcf2c09 100644 --- a/src/unzip/seekable_http_reader.rs +++ b/src/unzip/seekable_http_reader.rs @@ -7,9 +7,10 @@ // except according to those terms. use std::{ - cmp::min, + cmp::{max, min}, collections::BTreeMap, io::{BufReader, ErrorKind, Read, Seek, SeekFrom}, + num::{NonZeroU64, NonZeroUsize}, sync::{Arc, Condvar, Mutex}, }; @@ -21,19 +22,6 @@ use super::{ http_range_reader::{self, RangeFetcher}, }; -/// 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 MAX_BLOCK: usize = 1024 * 1024; - /// A hint to the [`SeekableHttpReaderEngine`] about the expected access pattern. #[derive(Debug, Clone, Copy, Eq, PartialEq)] pub(crate) enum AccessPattern { @@ -88,12 +76,14 @@ impl CacheCell { /// Internal state of the [`SeekableHttpReaderEngine`], in a separate struct /// because access is protected by a mutex. -#[derive(Default)] struct State { /// The expected pattern of seeks and reads; a hint from the user. access_pattern: AccessPattern, /// Maximum size of the "cache" readahead_limit: Option, + /// The block size we read each time we do a read from the underlying + /// stream. + max_block: NonZeroUsize, /// Current size of the cache current_size: usize, /// The readahead "cache", which is not really a cache in the strict sense, @@ -105,26 +95,36 @@ struct State { /// we skipped over, in order to service any subsequent requests for those /// positions. cache: BTreeMap, - /// Whether a read from the underlying HTTP stream is afoot. Only one thread - /// can be doing a read at a time. - read_in_progress: bool, + /// Whether a read from the underlying HTTP stream is afoot on each of the + /// read streams. Only one thread can be doing a read at a time on each + /// stream. + read_in_progress: Vec, /// Some statistics about how we're doing. stats: SeekableHttpReaderStatistics, } impl State { - fn new(readahead_limit: Option, access_pattern: AccessPattern) -> Self { + fn new( + readahead_limit: Option, + access_pattern: AccessPattern, + parallel_readers: usize, + max_block: NonZeroUsize, + ) -> Self { // Grow the readahead limit if it's less than block size, because we // must always store one block in order to service the most recent read. let readahead_limit = match readahead_limit { - Some(readahead_limit) if readahead_limit > MAX_BLOCK => Some(readahead_limit), - Some(_) => Some(MAX_BLOCK), + Some(readahead_limit) if readahead_limit > max_block.get() => Some(readahead_limit), + Some(_) => Some(max_block.get()), _ => None, }; Self { readahead_limit, access_pattern, - ..Default::default() + max_block, + read_in_progress: vec![false; parallel_readers], + cache: BTreeMap::new(), + stats: SeekableHttpReaderStatistics::default(), + current_size: 0, } } @@ -161,8 +161,9 @@ impl State { let discard_read_data = matches!(self.access_pattern, AccessPattern::SequentialIsh); let mut block_to_discard = None; let mut return_value = None; - for (possible_block_start, block) in - self.cache.range_mut(pos - min(pos, MAX_BLOCK as u64)..=pos) + for (possible_block_start, block) in self + .cache + .range_mut(pos - min(pos, self.max_block.get() as u64)..=pos) { let block_offset = pos as usize - *possible_block_start as usize; let block_len = block.len(); @@ -207,8 +208,8 @@ impl std::fmt::Debug for State { /// Items related to reading from the underlying HTTP streams. This is /// in a separate struct because it's protected by a mutex. struct ReadingMaterials { - range_fetcher: RangeFetcher, - reader: Option<(BufReader, u64)>, // second item in tuple is current reader pos + reader: BufReader, + reader_pos: u64, } /// A type which can produce objects that can be [`Read`] and [`Seek`] even @@ -220,8 +221,13 @@ struct ReadingMaterials { pub(crate) struct SeekableHttpReaderEngine { /// Total stream length len: u64, - /// Facilities to read from the underlying HTTP stream(s) - reader: Mutex, + /// The object which can create streams to fetch HTTP ranges. + range_fetcher: Mutex, + /// Facilities to read from the underlying HTTP stream(s). + /// Items may only be added to or removed from this vec while the State + /// mutex is held. However the items within the vec can be used even + /// without the state mutex held. + readers: Vec>>, /// Overall state of this object, mostly related to the readahead cache /// of blocks we already read, but also with the all-important boolean /// stating whether any thread is already reading on the underlying stream. @@ -230,6 +236,8 @@ pub(crate) struct SeekableHttpReaderEngine { /// readahead cache and all other threads should consider if their read /// request can be serviced. read_completed: Condvar, + /// Maximum read block size. + max_block: NonZeroUsize, } /// Some results about the success (or otherwise) of this reader. @@ -254,21 +262,40 @@ impl SeekableHttpReaderEngine { pub(crate) fn new( uri: String, readahead_limit: Option, + parallelism_limit: Option, access_pattern: AccessPattern, + max_block: NonZeroUsize, + range_per_reader: NonZeroU64, ) -> Result, Error> { let range_fetcher = RangeFetcher::new(uri).map_err(Error::RangeFetcherError)?; if !range_fetcher.accepts_ranges() { return Err(Error::AcceptRangesNotSupported); } let len = range_fetcher.len(); + // Calculate how many HTTP streams we want going at once, if and when + // we get to AccessPattern::SequentialIsh. + let mut parallel_readers = (len / range_per_reader) as usize; + if let Some(parallelism_limit) = parallelism_limit { + parallel_readers = min(parallel_readers, parallelism_limit.get()); + } + parallel_readers = max(1, parallel_readers); + println!("Parallel readers: {}", parallel_readers); + let mut readers = Vec::with_capacity(parallel_readers); + for _ in 0..parallel_readers { + readers.push(Mutex::new(None)); + } Ok(Arc::new(Self { len, - reader: Mutex::new(ReadingMaterials { - range_fetcher, - reader: None, - }), - state: Mutex::new(State::new(readahead_limit, access_pattern)), + range_fetcher: Mutex::new(range_fetcher), + readers, + state: Mutex::new(State::new( + readahead_limit, + access_pattern, + parallel_readers, + max_block, + )), read_completed: Condvar::new(), + max_block, })) } @@ -334,8 +361,12 @@ impl SeekableHttpReaderEngine { if let Some(bytes_read_from_cache) = state.read_from_cache(pos, buf) { return Ok(bytes_read_from_cache); } + // - If no, we will consider doing an actual read. + // Work out which reader is relevant to our pos + let currently_relevant_reader = self.calculate_relevant_reader(&state, pos); + let end_of_reader_range = self.end_of_reader_range(&state, currently_relevant_reader); // - If no, check if read in progress - let mut read_in_progress = state.read_in_progress; + let mut read_in_progress = state.read_in_progress[currently_relevant_reader]; // Is there read in progress? while read_in_progress { // - If yes, release CACHE mutex, WAIT on condvar atomically @@ -344,59 +375,67 @@ impl SeekableHttpReaderEngine { if let Some(bytes_read_from_cache) = state.read_from_cache(pos, buf) { return Ok(bytes_read_from_cache); } - read_in_progress = state.read_in_progress; + read_in_progress = state.read_in_progress[currently_relevant_reader]; } state.stats.cache_misses += 1; // - If no: + // we're actually likely to do a read. // set read in progress - state.read_in_progress = true; + state.read_in_progress[currently_relevant_reader] = true; // claim READER mutex - let mut reading_stuff = self.reader.lock().unwrap(); + let mut reading_stuff = self.readers[currently_relevant_reader].lock().unwrap(); // release STATE mutex drop(state); // perform read // First check if we need to rewind. - if let Some((_, readerpos)) = reading_stuff.reader.as_ref() { - if pos < *readerpos { + if let Some(reading_materials_inner) = reading_stuff.as_mut() { + if pos < reading_materials_inner.reader_pos { log::info!( "New reader will be required at 0x{:x} - old reader pos was 0x{:x}", pos, - *readerpos + reading_materials_inner.reader_pos ); - reading_stuff.reader = None; + *reading_stuff = None; } } let mut reader_created = false; - if reading_stuff.reader.is_none() { + if reading_stuff.is_none() { log::info!("create_reader"); - reading_stuff.reader = Some(( - BufReader::new( - reading_stuff - .range_fetcher - .fetch_range(pos) + *reading_stuff = Some(ReadingMaterials { + reader: BufReader::new( + self.range_fetcher + .lock() + .unwrap() + .fetch_range(pos..end_of_reader_range) .map_err(|e| std::io::Error::new(ErrorKind::Unsupported, e.to_string()))?, ), - pos, - )); + reader_pos: pos, + }); reader_created = true; }; - let (reader, reader_pos) = reading_stuff.reader.as_mut().unwrap(); - if pos > *reader_pos { - log::info!("Read: fast-forward from 0x{:x} to 0x{:x}", *reader_pos, pos); + let reading_stuff = reading_stuff.as_mut().unwrap(); + if pos > reading_stuff.reader_pos { + log::info!( + "Read: fast-forward from 0x{:x} to 0x{:x}", + reading_stuff.reader_pos, + pos + ); } - while pos >= *reader_pos { + while pos >= reading_stuff.reader_pos { // Fast forward beyond the desired position, recording any reads in the cache // for later. - let to_read = min(MAX_BLOCK, self.len as usize - *reader_pos as usize); + let to_read = min( + self.max_block.get(), + self.len as usize - reading_stuff.reader_pos as usize, + ); let mut new_block = vec![0u8; to_read]; - reader.read_exact(&mut new_block)?; + reading_stuff.reader.read_exact(&mut new_block)?; // claim STATE mutex let mut state = self.state.lock().unwrap(); - state.insert(*reader_pos, new_block); - // Tell any waiting threads they should re-check the cache + state.insert(reading_stuff.reader_pos, new_block); self.read_completed.notify_all(); - *reader_pos += to_read as u64; + reading_stuff.reader_pos += to_read as u64; } // Because the above condition is >=, and because we know the request was not // to read at the very end of the file, we know we now have some data in the @@ -410,7 +449,7 @@ impl SeekableHttpReaderEngine { state.stats.num_http_streams += 1; } // set read not in progress - state.read_in_progress = false; + state.read_in_progress[currently_relevant_reader] = false; // release STATE mutex // release READER mutex Ok(bytes_read) @@ -435,17 +474,25 @@ impl SeekableHttpReaderEngine { state.stats ); if matches!(access_pattern, AccessPattern::SequentialIsh) { - if state.read_in_progress { + if state.read_in_progress.iter().any(|b| *b) { panic!("Must not call set_expected_access_pattern while a read is in progress"); } - // If we're switching to a sequential pattern, recreate - // the reader at position zero. - log::info!("create_reader_at_zero"); + // If we're switching to a sequential pattern, recreate all our + // readers at the start of their respective ranges. + log::info!("creating sequential readers at the starts of ranges"); { - let mut reading_materials = self.reader.lock().unwrap(); - let new_reader = reading_materials.range_fetcher.fetch_range(0); - if let Ok(new_reader) = new_reader { - reading_materials.reader = Some((BufReader::new(new_reader), 0)); + for reader_num in 0..self.readers.len() { + let mut reading_materials = self.readers[reader_num].lock().unwrap(); + let start_of_reader_range = self.start_of_reader_range(&state, reader_num); + let reader_range = + start_of_reader_range..self.end_of_reader_range(&state, reader_num); + let new_reader = self.range_fetcher.lock().unwrap().fetch_range(reader_range); + if let Ok(new_reader) = new_reader { + *reading_materials = Some(ReadingMaterials { + reader: BufReader::new(new_reader), + reader_pos: start_of_reader_range, + }); + } } } state.stats.num_http_streams += 1; @@ -457,6 +504,45 @@ impl SeekableHttpReaderEngine { pub(crate) fn get_stats(&self) -> SeekableHttpReaderStatistics { self.state.lock().unwrap().stats.clone() } + + /// Determine which of the many underlying readers is the right one + /// for an upcoming read. + fn calculate_relevant_reader(&self, state: &State, pos: u64) -> usize { + match state.access_pattern { + AccessPattern::RandomAccess => 0usize, // always use a single reader + AccessPattern::SequentialIsh => { + let num_readers = self.readers.len(); + let range_covered_by_each_reader = self.len / num_readers as u64; + min((pos / range_covered_by_each_reader) as usize, num_readers) + } + } + } + + fn start_of_reader_range(&self, state: &State, reader: usize) -> u64 { + match state.access_pattern { + AccessPattern::RandomAccess => 0u64, + AccessPattern::SequentialIsh => { + let num_readers = self.readers.len(); + let range_covered_by_each_reader = self.len / num_readers as u64; + reader as u64 * range_covered_by_each_reader + } + } + } + + fn end_of_reader_range(&self, state: &State, reader: usize) -> u64 { + match state.access_pattern { + AccessPattern::RandomAccess => self.len(), + AccessPattern::SequentialIsh => { + let num_readers = self.readers.len(); + let range_covered_by_each_reader = self.len / num_readers as u64; + if reader == num_readers - 1 { + self.len() + } else { + (reader + 1) as u64 * range_covered_by_each_reader + } + } + } + } } impl Drop for SeekableHttpReaderEngine { @@ -515,7 +601,10 @@ impl HasLength for SeekableHttpReader { #[cfg(test)] mod tests { - use std::io::{Read, Seek, SeekFrom}; + use std::{ + io::{Read, Seek, SeekFrom}, + num::{NonZeroU64, NonZeroUsize}, + }; use test_log::test; use httptest::{matchers::*, responders::*, Expectation, Server}; @@ -557,7 +646,10 @@ mod tests { let mut seekable_http_reader = SeekableHttpReaderEngine::new( server.url("/foo").to_string(), readahead_limit, + None, access_pattern, + NonZeroUsize::new(4096).unwrap(), + NonZeroU64::new(10 * 1024 * 1024).unwrap(), ) .unwrap() .create_reader();