perf(ecstore): slice in-memory shards instead of copying them twice (#4687)

* perf(ecstore): slice in-memory shards instead of copying them twice (backlog#1159)

The GET path reads a shard out of the page cache into a `Bytes`, then
`open_disk_reader` erased it behind `Box<dyn AsyncRead>` by wrapping it in
a `Cursor`. Downstream, `BitrotReader` could only get it back by copying:
once out of the `Cursor` into its scratch buffer, and once from there into
the caller's buffer. CPU profiling of a cached 1 MiB GET (device reads = 0)
attributed 8.23% of the whole server to `Cursor::poll_read` alone — a copy
of data that was already sitting in memory.

Keep the source concrete instead of erasing it. `ShardReader` is an enum of
`InMemory(Cursor<Bytes>)` and `Stream(Box<dyn AsyncRead ...>)`, and the new
`ShardSource::try_take_block(n)` lets an in-memory source hand over the
`[hash][data]` block as a slice. `read_appending` uses it to verify the hash
on the slice and `extend_from_slice` the shard straight into the caller's
buffer: one copy instead of two.

`try_take_block` defaults to `None`, so a streaming source keeps the old
path byte for byte, along with its short-read and EOF semantics. A source
that cannot serve `n` bytes declines rather than truncating, so a partial
block still becomes UnexpectedEof rather than a short shard. The hash is
still checked before anything is appended, so a corrupt shard never reaches
the caller's buffer on either path. The deferred parity reader opens its
source lazily and stays on the streaming path; parity is only read when a
data shard fails.

Tests gate equivalence and non-vacuity:
  * `try_take_block` fires for `Cursor<Bytes>`, advances the position exactly
    as a read of the same length would, declines when fewer than `n` bytes
    remain, and returns `None` for a non-`Bytes` source — without this the
    equivalence test below would silently compare one path against itself;
  * both paths return identical bytes for the same shard;
  * a corrupt shard fails on the fast path too, appending nothing.

Verified: clippy --tests -D warnings clean; `erasure::` 215 passed, 0 failed;
`set_disk::core::io_primitives` 49 and `io_support::` 22 pass.

Stacked on #4681 (`read_appending`), which this builds on.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): implement ShardSource for Cursor<&[u8]> used by the erasure bench

`crates/ecstore/benches/erasure_benchmark.rs` builds
`BitrotReader<Cursor<&[u8]>>`, which the new `ShardSource` bound on
`ParallelReader`/`decode` does not accept. `cargo clippy --tests` does not
compile bench targets, so this only surfaced in CI's `--all-targets` run.

A borrowed slice carries no `Bytes` to hand out, so it takes the default
`try_take_block` and keeps the old streaming copy path — no behavior change.

Verified with the same target set CI uses:
`cargo clippy -p rustfs-ecstore --all-targets -- -D warnings` clean.

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/erasure/coding/bitrot.rs

@@ -20,6 +20,46 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tracing::error;
 use uuid::Uuid;
 
+/// A shard source that may already hold its bytes in memory.
+///
+/// The GET path reads a shard out of the page cache into a `Bytes` and then, in
+/// the old code, copied it twice more: once out of the `Cursor` wrapping it into
+/// the reader's scratch buffer, and once from there into the caller's buffer.
+/// `try_take_block` lets an in-memory source hand over the `[hash][data]` block
+/// as a slice instead, collapsing that to a single copy (rustfs/backlog#1159:
+/// `Cursor::poll_read` was 8.23% of GET CPU).
+///
+/// The default says "not in memory", so a streaming source keeps the old path and
+/// its short-read/EOF semantics unchanged.
+pub trait ShardSource: AsyncRead + Send + Sync + Unpin {
+    /// The next `n` bytes, consumed from the source, or `None` when the source is
+    /// not in memory or holds fewer than `n` bytes left. Advancing must match what
+    /// an `AsyncRead` of `n` bytes would have done, so the two can be mixed.
+    fn try_take_block(&mut self, _n: usize) -> Option<bytes::Bytes> {
+        None
+    }
+}
+
+/// Borrowed and owned byte slices are ordinary streaming sources: they carry no
+/// `Bytes` to hand out, so they take the default and keep the old copy path.
+impl ShardSource for std::io::Cursor<Vec<u8>> {}
+
+impl ShardSource for std::io::Cursor<&[u8]> {}
+
+impl ShardSource for Box<dyn AsyncRead + Send + Sync + Unpin> {}
+
+impl ShardSource for std::io::Cursor<bytes::Bytes> {
+    fn try_take_block(&mut self, n: usize) -> Option<bytes::Bytes> {
+        let pos = usize::try_from(self.position()).ok()?;
+        let end = pos.checked_add(n)?;
+        if end > self.get_ref().len() {
+            return None;
+        }
+        self.set_position(end as u64);
+        Some(self.get_ref().slice(pos..end))
+    }
+}
+
 pin_project! {
     /// BitrotReader reads (hash+data) blocks from an async reader and verifies hash integrity.
     pub struct BitrotReader<R> {
@@ -132,6 +172,24 @@ where
         Ok(out.len())
     }
 
+    /// Map a completed no-hash read to the shared short-shard contract: a full
+    /// buffer returns its length, a short read is UnexpectedEof (backlog#799 B2).
+    fn finish_len(&self, data_len: usize, want: usize) -> std::io::Result<usize> {
+        if data_len < want {
+            error!("bitrot reader short shard read: id={} got {} of {} bytes", self.id, data_len, want);
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                format!("short shard read: got {data_len} of {want} bytes"),
+            ));
+        }
+        Ok(data_len)
+    }
+}
+
+impl<R> BitrotReader<R>
+where
+    R: ShardSource,
+{
     /// Same contract as [`Self::read`], but **appends** `want` bytes into `out`'s
     /// spare capacity instead of demanding an initialized `&mut [u8]`
     /// (rustfs/backlog#1159).
@@ -178,10 +236,31 @@ where
             return self.finish_len(out.len() - start, want);
         }
 
-        // Hashed path: identical to `read` — one pass pulls `[hash][data]` into
+        let need = hash_size + want;
+
+        // In-memory fast path: the block is already resident, so slice it instead of
+        // copying it into the scratch buffer first (rustfs/backlog#1159). One copy
+        // (`extend_from_slice`) instead of two. A source that cannot serve `need`
+        // bytes returns `None` and falls through, keeping the short-read contract.
+        if let Some(block) = self.inner.try_take_block(need) {
+            let (hash, data) = block.split_at(hash_size);
+            if !self.skip_verify {
+                let verify_start = std::time::Instant::now();
+                let actual_hash = self.hash_algo.hash_encode(data);
+                self.last_verify_duration = verify_start.elapsed();
+                if actual_hash.as_ref() != hash {
+                    error!("bitrot reader hash mismatch, id={} data_len={}, out_len={}", self.id, data.len(), want);
+                    return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch"));
+                }
+            }
+            // Only after verification: a corrupt shard must not reach the caller.
+            out.extend_from_slice(data);
+            return Ok(want);
+        }
+
+        // Streaming path: identical to `read` — one pass pulls `[hash][data]` into
         // the scratch buffer — except the shard lands in `out` by `extend_from_slice`
         // rather than `copy_from_slice` into a pre-zeroed buffer. Same single copy.
-        let need = hash_size + want;
         if self.buf.len() < need {
             self.buf.resize(need, 0);
         }
@@ -209,19 +288,6 @@ where
         out.extend_from_slice(data);
         Ok(want)
     }
-
-    /// Map a completed no-hash read to the shared short-shard contract: a full
-    /// buffer returns its length, a short read is UnexpectedEof (backlog#799 B2).
-    fn finish_len(&self, data_len: usize, want: usize) -> std::io::Result<usize> {
-        if data_len < want {
-            error!("bitrot reader short shard read: id={} got {} of {} bytes", self.id, data_len, want);
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::UnexpectedEof,
-                format!("short shard read: got {data_len} of {want} bytes"),
-            ));
-        }
-        Ok(data_len)
-    }
 }
 
 pin_project! {
@@ -611,6 +677,7 @@ impl BitrotWriterWrapper {
 
 #[cfg(test)]
 mod tests {
+    use super::ShardSource;
     use super::{
         BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
     };
@@ -1322,4 +1389,85 @@ mod tests {
             .expect_err("want > shard_size must be rejected");
         assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
     }
+
+    /// The in-memory fast path (rustfs/backlog#1159) must be *equivalent* to the
+    /// streaming path, not merely present. `Cursor<Bytes>` slices the
+    /// `[hash][data]` block instead of copying it into the scratch buffer; if that
+    /// ever diverged, GET would return different bytes.
+    ///
+    /// The first assertion is the non-vacuity gate: it proves `try_take_block`
+    /// actually fires for `Cursor<Bytes>` (and does not for `Cursor<Vec<u8>>`), so
+    /// the equivalence below is really comparing two different code paths.
+    #[tokio::test]
+    async fn in_memory_fast_path_fires_and_matches_the_streaming_path() {
+        use bytes::Bytes;
+        use std::io::Cursor;
+
+        const SHARD: usize = 4096;
+        let algo = HashAlgorithm::HighwayHash256;
+        let data: Vec<u8> = (0..SHARD).map(|i| (i * 17 + 3) as u8).collect();
+        let mut encoded = Vec::new();
+        BitrotWriter::new(&mut encoded, SHARD, algo.clone())
+            .write(&data)
+            .await
+            .expect("write shard");
+
+        // Non-vacuity: the fast path exists for Bytes and not for Vec.
+        let mut mem = Cursor::new(Bytes::from(encoded.clone()));
+        assert!(
+            ShardSource::try_take_block(&mut mem, 8).is_some(),
+            "Cursor<Bytes> must be able to hand out a block, otherwise the fast path is dead code"
+        );
+        assert_eq!(mem.position(), 8, "taking a block must advance like a read of the same length");
+        let mut streamed = Cursor::new(encoded.clone());
+        assert!(
+            ShardSource::try_take_block(&mut streamed, 8).is_none(),
+            "a non-Bytes source must stay on the streaming path"
+        );
+        // A block larger than what is left must decline rather than truncate.
+        let mut short = Cursor::new(Bytes::from_static(b"1234"));
+        assert!(ShardSource::try_take_block(&mut short, 5).is_none());
+
+        // Equivalence: same bytes out of both paths.
+        let mut via_mem: Vec<u8> = Vec::with_capacity(SHARD);
+        BitrotReader::new(Cursor::new(Bytes::from(encoded.clone())), SHARD, algo.clone(), false)
+            .read_appending(&mut via_mem, SHARD)
+            .await
+            .expect("in-memory read");
+
+        let mut via_stream: Vec<u8> = Vec::with_capacity(SHARD);
+        BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
+            .read_appending(&mut via_stream, SHARD)
+            .await
+            .expect("streaming read");
+
+        assert_eq!(via_mem, via_stream, "the two paths must return identical bytes");
+        assert_eq!(via_mem, data);
+    }
+
+    /// A corrupt shard must fail on the fast path too — the slice is verified
+    /// before anything is appended, exactly as on the streaming path.
+    #[tokio::test]
+    async fn in_memory_fast_path_rejects_a_corrupt_shard_without_appending() {
+        use bytes::Bytes;
+        use std::io::Cursor;
+
+        const SHARD: usize = 4096;
+        let algo = HashAlgorithm::HighwayHash256;
+        let mut encoded = Vec::new();
+        BitrotWriter::new(&mut encoded, SHARD, algo.clone())
+            .write(&vec![5u8; SHARD])
+            .await
+            .expect("write shard");
+        let last = encoded.len() - 1;
+        encoded[last] ^= 0xff;
+
+        let mut out: Vec<u8> = Vec::with_capacity(SHARD);
+        let err = BitrotReader::new(Cursor::new(Bytes::from(encoded)), SHARD, algo, false)
+            .read_appending(&mut out, SHARD)
+            .await
+            .expect_err("a corrupt shard must not verify on the fast path either");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert!(out.is_empty(), "corrupt bytes must never reach the caller's buffer");
+    }
 }

crates/ecstore/src/erasure/coding/decode.rs

@@ -34,7 +34,6 @@ use std::io;
 use std::io::ErrorKind;
 use std::pin::Pin;
 use std::time::{Duration, Instant};
-use tokio::io::AsyncRead;
 use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
 use tracing::{error, warn};
@@ -270,7 +269,7 @@ fn read_shard<'a, R>(
     metrics_path: Option<&'static str>,
 ) -> ShardReadFuture<'a>
 where
-    R: AsyncRead + Unpin + Send + Sync + 'a,
+    R: crate::erasure::coding::ShardSource + 'a,
 {
     let role = shard_role(index, data_shards);
     if let Some(reader) = reader {
@@ -395,7 +394,7 @@ pub(crate) struct ParallelReader<R> {
 
 impl<R> ParallelReader<R>
 where
-    R: AsyncRead + Unpin + Send + Sync,
+    R: crate::erasure::coding::ShardSource,
 {
     // Readers should handle disk errors before being passed in, ensuring each reader reaches the available number of BitrotReaders
     pub fn new(readers: Vec<Option<BitrotReader<R>>>, e: Erasure, offset: usize, total_length: usize) -> Self {
@@ -686,7 +685,7 @@ fn record_scheduled_read_cost(
 
 impl<R> ParallelReader<R>
 where
-    R: AsyncRead + Unpin + Send + Sync,
+    R: crate::erasure::coding::ShardSource,
 {
     #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub async fn read(&mut self) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>) {
@@ -1247,7 +1246,7 @@ where
 #[async_trait::async_trait]
 impl<R> ShardStripeSource for ParallelReader<R>
 where
-    R: AsyncRead + Unpin + Send + Sync,
+    R: crate::erasure::coding::ShardSource,
 {
     async fn read_next_stripe(&mut self) -> StripeReadState {
         let read_quorum = self.data_shards;
@@ -1275,7 +1274,7 @@ async fn read_stripe_timed<R>(
     stage_metrics_enabled: bool,
 ) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>)
 where
-    R: AsyncRead + Unpin + Send + Sync,
+    R: crate::erasure::coding::ShardSource,
 {
     let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
     let out = reader.read().await;
@@ -1422,7 +1421,7 @@ impl Erasure {
     ) -> (usize, Option<std::io::Error>)
     where
         W: AsyncWrite + Send + Sync + Unpin,
-        R: AsyncRead + Unpin + Send + Sync,
+        R: crate::erasure::coding::ShardSource,
     {
         self.decode_inner(writer, readers, offset, length, total_length, None, Vec::new())
             .await
@@ -1439,7 +1438,7 @@ impl Erasure {
     ) -> (usize, Option<std::io::Error>)
     where
         W: AsyncWrite + Send + Sync + Unpin,
-        R: AsyncRead + Unpin + Send + Sync,
+        R: crate::erasure::coding::ShardSource,
     {
         self.decode_inner(writer, readers, offset, length, total_length, Some(read_costs), Vec::new())
             .await
@@ -1461,7 +1460,7 @@ impl Erasure {
     ) -> (usize, Option<std::io::Error>)
     where
         W: AsyncWrite + Send + Sync + Unpin,
-        R: AsyncRead + Unpin + Send + Sync,
+        R: crate::erasure::coding::ShardSource,
     {
         self.decode_inner(writer, readers, offset, length, total_length, read_costs, deferred_handles)
             .await
@@ -1582,7 +1581,7 @@ impl Erasure {
     ) -> (usize, Option<std::io::Error>)
     where
         W: AsyncWrite + Send + Sync + Unpin,
-        R: AsyncRead + Unpin + Send + Sync,
+        R: crate::erasure::coding::ShardSource,
     {
         if readers.len() != self.data_shards + self.parity_shards {
             record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid);
@@ -1806,10 +1805,11 @@ mod tests {
         atomic::{AtomicUsize, Ordering},
     };
     use std::task::{Context, Poll};
+    use tokio::io::AsyncRead;
     use tokio::io::ReadBuf;
     use tokio::time::{Instant as TokioInstant, Sleep};
 
-    type BoxedShardReader = Box<dyn AsyncRead + Send + Sync + Unpin>;
+    type BoxedShardReader = crate::io_support::bitrot::ShardReader;
 
     /// Counts the raw bytes pulled from a shard stream, to prove which shards
     /// a decode path actually touches (backlog#923 call-count evidence).
@@ -1818,6 +1818,10 @@ mod tests {
         bytes_read: Arc<AtomicUsize>,
     }
 
+    /// Streaming test source: no in-memory block, so it exercises the same path a
+    /// real disk stream takes.
+    impl crate::erasure::coding::ShardSource for CountingShardReader {}
+
     impl AsyncRead for CountingShardReader {
         fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
             let before = buf.filled().len();
@@ -1853,7 +1857,7 @@ mod tests {
                     .map(|(_, len)| buf[..*len].to_vec())
                     .unwrap_or_else(|| buf.clone());
                 readers.push(Some(BitrotReader::new(
-                    Box::new(Cursor::new(bytes)) as BoxedShardReader,
+                    crate::io_support::bitrot::ShardReader::Stream(Box::new(Cursor::new(bytes))),
                     shard_size,
                     hash_algo.clone(),
                     false,
@@ -1893,6 +1897,8 @@ mod tests {
         TimedOut,
     }
 
+    impl crate::erasure::coding::ShardSource for TestShardReader {}
+
     impl AsyncRead for TestShardReader {
         fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
             match &mut *self {