Re-implement the prefetcher using backpressure mechanism

mountpoint-s3/src/checksums.rs

@@ -63,7 +63,7 @@ impl ChecksummedBytes {
     /// This operation just increases the reference count and sets a few indices,
     /// so there will be no validation and the checksum will not be recomputed.
     pub fn split_off(&mut self, at: usize) -> ChecksummedBytes {
-        assert!(at < self.len());
+        assert!(at <= self.len());
 
         let start = self.range.start;
         let prefix_range = start..(start + at);
@@ -77,6 +77,27 @@ impl ChecksummedBytes {
         }
     }
 
+    /// Splits the checksummed bytes into two at the given index.
+    ///
+    /// Afterwards self contains elements [at, len), and the returned Bytes contains elements [0, at).
+    ///
+    /// This operation just increases the reference count and sets a few indices,
+    /// so there will be no validation and the checksum will not be recomputed.
+    pub fn split_to(&mut self, at: usize) -> ChecksummedBytes {
+        assert!(at <= self.len());
+
+        let start = self.range.start;
+        let prefix_range = start..(start + at);
+        let suffix_range = (start + at)..self.range.end;
+
+        self.range = suffix_range;
+        Self {
+            buffer: self.buffer.clone(),
+            range: prefix_range,
+            checksum: self.checksum,
+        }
+    }
+
     /// Returns a slice of self for the provided range.
     ///
     /// This operation just increases the reference count and sets a few indices,
@@ -321,6 +342,26 @@ mod tests {
         assert_eq!(expected_checksum, new_checksummed_bytes.checksum);
     }
 
+    #[test]
+    fn test_split_to() {
+        let split_to_at = 4;
+        let bytes = Bytes::from_static(b"some bytes");
+        let expected = bytes.clone();
+        let expected_checksum = crc32c::checksum(&expected);
+        let mut checksummed_bytes = ChecksummedBytes::new(bytes);
+
+        let mut expected_part2 = expected.clone();
+        let expected_part1 = expected_part2.split_to(split_to_at);
+        let new_checksummed_bytes = checksummed_bytes.split_to(split_to_at);
+
+        assert_eq!(expected, checksummed_bytes.buffer);
+        assert_eq!(expected, new_checksummed_bytes.buffer);
+        assert_eq!(expected_part1, new_checksummed_bytes.buffer_slice());
+        assert_eq!(expected_part2, checksummed_bytes.buffer_slice());
+        assert_eq!(expected_checksum, checksummed_bytes.checksum);
+        assert_eq!(expected_checksum, new_checksummed_bytes.checksum);
+    }
+
     #[test]
     fn test_slice() {
         let range = 3..7;

mountpoint-s3/src/cli.rs

@@ -649,11 +649,25 @@ pub fn create_s3_client(args: &CliArgs) -> anyhow::Result<(S3CrtClient, EventLoo
         user_agent.key_value("mp-nw-interfaces", &interfaces.len().to_string());
     }
 
+    // This is a weird looking number! We really want our first request size to be 1MiB,
+    // which is a common IO size. But Linux's readahead will try to read an extra 128k on on
+    // top of a 1MiB read, which we'd have to wait for a second request to service. Because
+    // FUSE doesn't know the difference between regular reads and readahead reads, it will
+    // send us a READ request for that 128k, so we'll have to block waiting for it even if
+    // the application doesn't want it. This is all in the noise for sequential IO, but
+    // waiting for the readahead hurts random IO. So we add 128k to the first request size
+    // to avoid the latency hit of the second request.
+    //
+    // Note the CRT does not respect this value right now, they always return chunks of part size
+    // but this is the first window size we prefer.
+    let initial_read_window_size = 1024 * 1024 + 128 * 1024;
     let mut client_config = S3ClientConfig::new()
         .auth_config(auth_config)
         .throughput_target_gbps(throughput_target_gbps)
         .read_part_size(args.read_part_size.unwrap_or(args.part_size) as usize)
         .write_part_size(args.write_part_size.unwrap_or(args.part_size) as usize)
+        .read_backpressure(true)
+        .initial_read_window(initial_read_window_size)
         .user_agent(user_agent);
     #[cfg(feature = "multiple-nw-iface")]
     if let Some(interfaces) = &args.bind {

mountpoint-s3/src/fs.rs

@@ -1349,6 +1349,8 @@ mod tests {
         let bucket = "bucket";
         let client = Arc::new(MockClient::new(MockClientConfig {
             bucket: bucket.to_owned(),
+            enable_backpressure: true,
+            initial_read_window_size: 1024 * 1024,
             ..Default::default()
         }));
         // Create "dir1" in the client to avoid creating it locally

mountpoint-s3/src/fs/error.rs

@@ -129,9 +129,12 @@ impl<E: std::error::Error + Send + Sync + 'static> From<PrefetchReadError<E>> fo
                 GetObjectError::PreconditionFailed,
             )) => err!(libc::ESTALE, "object was mutated remotely"),
             PrefetchReadError::Integrity(e) => err!(libc::EIO, source:e, "integrity error"),
+            PrefetchReadError::PartReadFailed(e) => err!(libc::EIO, source:e, "part read failed"),
             PrefetchReadError::GetRequestFailed(_)
             | PrefetchReadError::GetRequestTerminatedUnexpectedly
-            | PrefetchReadError::GetRequestReturnedWrongOffset { .. } => {
+            | PrefetchReadError::GetRequestReturnedWrongOffset { .. }
+            | PrefetchReadError::BackpressurePreconditionFailed
+            | PrefetchReadError::ReadWindowIncrement => {
                 err!(libc::EIO, source:err, "get request failed")
             }
         }