From c2362bca14d64ba5164d6c9952422c69b41eb842 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 10 Jul 2026 23:57:10 +0800 Subject: [PATCH] perf(ecstore): slice in-memory shards instead of copying them twice (#4687) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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` 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)` and `Stream(Box)`, 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`, 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 * fix(ecstore): implement ShardSource for Cursor<&[u8]> used by the erasure bench `crates/ecstore/benches/erasure_benchmark.rs` builds `BitrotReader>`, 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 --------- Co-authored-by: heihutu --- crates/ecstore/src/erasure/coding/bitrot.rs | 178 ++++++++++++++++-- crates/ecstore/src/erasure/coding/decode.rs | 30 +-- crates/ecstore/src/io_support/bitrot.rs | 69 +++++-- .../src/set_disk/core/io_primitives.rs | 5 +- crates/ecstore/src/set_disk/mod.rs | 2 +- 5 files changed, 235 insertions(+), 49 deletions(-) diff --git a/crates/ecstore/src/erasure/coding/bitrot.rs b/crates/ecstore/src/erasure/coding/bitrot.rs index ee6915ff5..ec25b8cef 100644 --- a/crates/ecstore/src/erasure/coding/bitrot.rs +++ b/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 { + 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> {} + +impl ShardSource for std::io::Cursor<&[u8]> {} + +impl ShardSource for Box {} + +impl ShardSource for std::io::Cursor { + fn try_take_block(&mut self, n: usize) -> Option { + 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 { @@ -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 { + 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 BitrotReader +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 { - 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` 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` (and does not for `Cursor>`), 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 = (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 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 = 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 = 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 = 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"); + } } diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index 11a91ad85..e0120b9b9 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/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 { impl ParallelReader 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>>, e: Erasure, offset: usize, total_length: usize) -> Self { @@ -686,7 +685,7 @@ fn record_scheduled_read_cost( impl ParallelReader where - R: AsyncRead + Unpin + Send + Sync, + R: crate::erasure::coding::ShardSource, { #[cfg_attr(feature = "hotpath", hotpath::measure)] pub async fn read(&mut self) -> (Vec>>, Vec>) { @@ -1247,7 +1246,7 @@ where #[async_trait::async_trait] impl ShardStripeSource for ParallelReader 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( stage_metrics_enabled: bool, ) -> (Vec>>, Vec>) 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) 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) 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) 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) 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; + 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, } + /// 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> { 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> { match &mut *self { diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index bd1981750..0afbf63f8 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -33,7 +33,36 @@ use std::time::Instant; use tokio::io::{AsyncRead, ReadBuf}; use tracing::debug; -type BoxedObjectReader = Box; +/// A shard source for the bitrot reader. +/// +/// `InMemory` keeps the `Bytes` concrete instead of erasing it behind +/// `dyn AsyncRead`, so `BitrotReader` can slice the `[hash][data]` block straight +/// out of the page-cache copy rather than copying it into a scratch buffer first +/// (rustfs/backlog#1159). Everything else is a stream and keeps the old path. +pub enum ShardReader { + InMemory(Cursor), + Stream(Box), +} + +impl AsyncRead for ShardReader { + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut tokio::io::ReadBuf<'_>) -> Poll> { + match self.get_mut() { + Self::InMemory(cursor) => Pin::new(cursor).poll_read(cx, buf), + Self::Stream(reader) => Pin::new(reader).poll_read(cx, buf), + } + } +} + +impl crate::erasure::coding::ShardSource for ShardReader { + fn try_take_block(&mut self, n: usize) -> Option { + match self { + Self::InMemory(cursor) => cursor.try_take_block(n), + Self::Stream(_) => None, + } + } +} + +type BoxedObjectReader = ShardReader; type OpenObjectReaderFuture = Pin>> + Send>>; #[derive(Clone, Copy)] @@ -70,7 +99,7 @@ impl BitrotReaderSource { let mut rd = Cursor::new(data); let offset = u64::try_from(self.offset).map_err(|_| DiskError::FileCorrupt)?; rd.set_position(offset); - Ok(Some(Box::new(rd))) + Ok(Some(ShardReader::InMemory(rd))) } else if let Some(disk) = self.disk { open_disk_reader( &disk, @@ -272,7 +301,7 @@ async fn open_disk_reader( length: usize, use_mmap_read: bool, metrics_path: Option<&'static str>, -) -> disk::error::Result { +) -> disk::error::Result { let metrics_path = metrics_path.filter(|_| rustfs_io_metrics::get_stage_metrics_enabled()); let stage_metrics_enabled = metrics_path.is_some(); @@ -309,7 +338,7 @@ async fn open_disk_reader( "zero_copy_read_success" ); - return Ok(Box::new(Cursor::new(bytes))); + return Ok(ShardReader::InMemory(Cursor::new(bytes))); } Err(err) => { if let Some(metrics_path) = metrics_path { @@ -345,14 +374,18 @@ async fn open_disk_reader( Ok(wrap_first_read_metrics(reader, metrics_path)) } -fn wrap_first_read_metrics(reader: FileReader, metrics_path: Option<&'static str>) -> FileReader { +fn wrap_first_read_metrics(reader: FileReader, metrics_path: Option<&'static str>) -> ShardReader { if let Some(metrics_path) = metrics_path && rustfs_io_metrics::get_stage_metrics_enabled() { - return Box::new(FirstReadMetricsReader::new(reader, metrics_path, GET_STAGE_READER_STREAM_FIRST_READ)); + return ShardReader::Stream(Box::new(FirstReadMetricsReader::new( + reader, + metrics_path, + GET_STAGE_READER_STREAM_FIRST_READ, + ))); } - reader + ShardReader::Stream(reader) } fn bitrot_encoded_range(offset: usize, length: usize, shard_size: usize, checksum_algo: HashAlgorithm) -> (usize, usize) { @@ -387,7 +420,7 @@ pub async fn create_bitrot_reader( checksum_algo: HashAlgorithm, skip_verify: bool, use_mmap_read: bool, -) -> disk::error::Result>>> { +) -> disk::error::Result>> { create_bitrot_reader_with_stage_metrics( inline_data, disk, @@ -417,7 +450,7 @@ pub(crate) async fn create_bitrot_reader_with_stage_metrics( skip_verify: bool, use_mmap_read: bool, stage_metrics: Option, -) -> disk::error::Result>>> { +) -> disk::error::Result>> { create_bitrot_reader_from_bytes_with_stage_metrics( inline_data.map(Bytes::copy_from_slice), disk, @@ -450,7 +483,7 @@ pub async fn create_bitrot_reader_from_bytes( checksum_algo: HashAlgorithm, skip_verify: bool, use_mmap_read: bool, -) -> disk::error::Result>>> { +) -> disk::error::Result>> { create_bitrot_reader_from_bytes_with_stage_metrics( inline_data, disk, @@ -480,7 +513,7 @@ async fn create_bitrot_reader_from_bytes_with_stage_metrics( skip_verify: bool, use_mmap_read: bool, stage_metrics: Option, -) -> disk::error::Result>>> { +) -> disk::error::Result>> { let stage_metrics = stage_metrics.filter(|_| rustfs_io_metrics::get_stage_metrics_enabled()); let stage_metrics_enabled = stage_metrics.is_some(); @@ -527,7 +560,7 @@ pub fn create_deferred_bitrot_reader( checksum_algo: HashAlgorithm, skip_verify: bool, use_mmap_read: bool, -) -> BitrotReader> { +) -> BitrotReader { create_deferred_bitrot_reader_with_stripe_handle( inline_data, disk, @@ -558,7 +591,7 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle( checksum_algo: HashAlgorithm, skip_verify: bool, use_mmap_read: bool, -) -> (BitrotReader>, DeferredReaderStripeHandle) { +) -> (BitrotReader, DeferredReaderStripeHandle) { let stripe_stride = shard_size + checksum_algo.size(); let (offset, length) = bitrot_encoded_range(offset, length, shard_size, checksum_algo.clone()); let source = BitrotReaderSource { @@ -574,12 +607,10 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle( let deferred = DeferredObjectReader::new(source); let handle = deferred.stripe_handle(stripe_stride); - let reader = BitrotReader::new( - Box::new(deferred) as Box, - shard_size, - checksum_algo, - skip_verify, - ); + // The deferred parity reader opens its source lazily, so it cannot hand out an + // in-memory block up front; it stays on the streaming path. Parity shards are + // only read when a data shard fails, so the fast path is not needed here. + let reader = BitrotReader::new(ShardReader::Stream(Box::new(deferred)), shard_size, checksum_algo, skip_verify); (reader, handle) } diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index d3506770a..ff347e7cc 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -48,6 +48,7 @@ use crate::diagnostics::get::{ }; use crate::disk::OldCurrentSize; use crate::erasure::coding::BitrotReader; +use crate::io_support::bitrot::ShardReader; use crate::io_support::bitrot::{ BitrotReaderStageMetrics, DeferredReaderStripeHandle, create_bitrot_reader_with_stage_metrics, create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled, @@ -916,7 +917,7 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter( }); } -pub(in crate::set_disk) type ObjectBitrotReader = BitrotReader>; +pub(in crate::set_disk) type ObjectBitrotReader = BitrotReader; pub(in crate::set_disk) type BitrotReaderTask<'a> = Pin, DiskError>)> + Send + 'a>>; @@ -3954,7 +3955,7 @@ mod tests { fn test_object_bitrot_reader() -> ObjectBitrotReader { BitrotReader::new( - Box::new(Cursor::new(vec![1u8, 2, 3, 4])) as Box, + ShardReader::Stream(Box::new(Cursor::new(vec![1u8, 2, 3, 4]))), 4, HashAlgorithm::None, false, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 60af7e9a6..f67b4dd9a 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -192,7 +192,7 @@ type ListObjectsV2Info = StorageListObjectsV2Info; type ListObjectVersionsInfo = StorageListObjectVersionsInfo; type ObjectInfoOrErr = StorageObjectInfoOrErr; type WalkOptions = StorageWalkOptions bool>; -type InlineBitrotReader = coding::BitrotReader>; +type InlineBitrotReader = coding::BitrotReader; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_SET_DISK: &str = "set_disk";