diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index 8b7ff1c10..c9e7d73cb 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/crates/ecstore/src/erasure/coding/erasure.rs @@ -71,10 +71,16 @@ impl EncodedBlock { const MODERN_MAX_TOTAL_SHARDS: usize = ::ORDER; const MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64; +const LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 16; +// Vec growth may retain twice the requested logical length. Keeping the logical +// workspace at half the budget bounds each cached workspace's shard allocation to 1 MiB. +const LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE: usize = 512 * 1024; type ModernReedSolomonCache = RwLock>>; +type LegacyReedSolomonCache = RwLock>>; static MODERN_REED_SOLOMON_CACHE: OnceLock = OnceLock::new(); +static LEGACY_REED_SOLOMON_CACHE: OnceLock = OnceLock::new(); /// Errors returned when constructing an [`Erasure`] codec. #[derive(Debug, thiserror::Error)] @@ -141,43 +147,61 @@ pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize { struct LegacyReedSolomonEncoder { data_shards: usize, parity_shards: usize, - encoder_cache: std::sync::RwLock>, - decoder_cache: std::sync::RwLock>, -} - -impl Clone for LegacyReedSolomonEncoder { - fn clone(&self) -> Self { - Self { - data_shards: self.data_shards, - parity_shards: self.parity_shards, - encoder_cache: std::sync::RwLock::new(None), - decoder_cache: std::sync::RwLock::new(None), - } - } + cache_workspaces: bool, + encoder_cache: RwLock>, + decoder_cache: RwLock>, } impl LegacyReedSolomonEncoder { - fn new(_data_shards: usize, _parity_shards: usize) -> io::Result { + fn new(data_shards: usize, parity_shards: usize) -> io::Result { + Self::with_workspace_cache(data_shards, parity_shards, false) + } + + fn with_workspace_cache(data_shards: usize, parity_shards: usize, cache_workspaces: bool) -> io::Result { Ok(Self { - data_shards: _data_shards, - parity_shards: _parity_shards, - encoder_cache: std::sync::RwLock::new(None), - decoder_cache: std::sync::RwLock::new(None), + data_shards, + parity_shards, + cache_workspaces, + encoder_cache: RwLock::new(None), + decoder_cache: RwLock::new(None), }) } + fn logical_shard_bytes_upper_bound(&self, shard_len: usize) -> Option { + let aligned_shard_len = shard_len.checked_add(63)?.checked_div(64)?.checked_mul(64)?; + let high_rate_decoder_work_count = self + .parity_shards + .checked_next_power_of_two()? + .checked_add(self.data_shards)? + .checked_next_power_of_two()?; + let low_rate_decoder_work_count = self + .data_shards + .checked_next_power_of_two()? + .checked_add(self.parity_shards)? + .checked_next_power_of_two()?; + aligned_shard_len.checked_mul(high_rate_decoder_work_count.max(low_rate_decoder_work_count)) + } + + fn should_cache_workspace(&self, shard_len: usize) -> bool { + self.cache_workspaces + && self + .logical_shard_bytes_upper_bound(shard_len) + .is_some_and(|bytes| bytes <= LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE) + } + fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> { let mut shards_vec: Vec<&mut [u8]> = shards.into_vec(); if shards_vec.is_empty() { return Ok(()); } let shard_len = shards_vec[0].len(); + let cached_encoder = self + .encoder_cache + .write() + .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))? + .take(); let mut encoder = { - let mut cache_guard = self - .encoder_cache - .write() - .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?; - match cache_guard.take() { + match cached_encoder { Some(mut cached) => { if cached.reset(self.data_shards, self.parity_shards, shard_len).is_err() { reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len) @@ -204,10 +228,15 @@ impl LegacyReedSolomonEncoder { } } drop(result); - *self - .encoder_cache - .write() - .map_err(|_| io::Error::other("Failed to return encoder to cache"))? = Some(encoder); + if self.should_cache_workspace(shard_len) { + let mut cache = self + .encoder_cache + .write() + .map_err(|_| io::Error::other("Failed to return encoder to cache"))?; + if cache.is_none() { + *cache = Some(encoder); + } + } Ok(()) } @@ -221,13 +250,13 @@ impl LegacyReedSolomonEncoder { .find_map(|s| s.as_ref().map(|v| v.len())) .ok_or_else(|| io::Error::other("No valid shards found for reconstruction"))?; + let cached_decoder = self + .decoder_cache + .write() + .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))? + .take(); let mut decoder = { - let mut cache_guard = self - .decoder_cache - .write() - .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?; - - match cache_guard.take() { + match cached_decoder { Some(mut cached_decoder) => { if let Err(e) = cached_decoder.reset(self.data_shards, self.parity_shards, shard_len) { warn!("Failed to reset SIMD decoder: {:?}, creating new one", e); @@ -274,10 +303,15 @@ impl LegacyReedSolomonEncoder { drop(result); - *self - .decoder_cache - .write() - .map_err(|_| io::Error::other("Failed to return decoder to cache"))? = Some(decoder); + if self.should_cache_workspace(shard_len) { + let mut cache = self + .decoder_cache + .write() + .map_err(|_| io::Error::other("Failed to return decoder to cache"))?; + if cache.is_none() { + *cache = Some(decoder); + } + } Ok(()) } @@ -435,6 +469,39 @@ fn cached_modern_reed_solomon(data_shards: usize, parity_shards: usize) -> Resul Ok(encoder) } +fn cached_legacy_reed_solomon(data_shards: usize, parity_shards: usize) -> io::Result> { + let cache = LEGACY_REED_SOLOMON_CACHE.get_or_init(|| RwLock::new(HashMap::new())); + cached_legacy_reed_solomon_in(cache, data_shards, parity_shards) +} + +fn cached_legacy_reed_solomon_in( + cache: &LegacyReedSolomonCache, + data_shards: usize, + parity_shards: usize, +) -> io::Result> { + let key = (data_shards, parity_shards); + if let Some(encoder) = cache + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&key) + .cloned() + { + return Ok(encoder); + } + + let mut cache = cache.write().unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(existing) = cache.get(&key) { + return Ok(Arc::clone(existing)); + } + if cache.len() < LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES { + let encoder = Arc::new(LegacyReedSolomonEncoder::with_workspace_cache(data_shards, parity_shards, true)?); + cache.insert(key, Arc::clone(&encoder)); + return Ok(encoder); + } + drop(cache); + Ok(Arc::new(LegacyReedSolomonEncoder::new(data_shards, parity_shards)?)) +} + fn encode_parity_shards(shards: &mut [Option>], data_shards: usize, parity_shards: usize, encode: F) -> io::Result<()> where F: FnOnce(SmallVec<[&mut [u8]; 16]>) -> io::Result<()>, @@ -551,7 +618,7 @@ pub struct Erasure { pub data_shards: usize, pub parity_shards: usize, encoder: Option, - legacy_encoder: Option, + legacy_encoder: Option>, pub block_size: usize, uses_legacy: bool, _id: Uuid, @@ -687,7 +754,7 @@ impl Erasure { let legacy_encoder = if uses_legacy && parity_shards > 0 { Some( - LegacyReedSolomonEncoder::new(data_shards, parity_shards) + cached_legacy_reed_solomon(data_shards, parity_shards) .map_err(|source| ErasureConstructionError::LegacyEncoder { source })?, ) } else { @@ -1405,7 +1472,7 @@ mod tests { assert_eq!(cloned.block_size, legacy.block_size); assert!(cloned.uses_legacy); - let data = b"legacy clone should keep independent SIMD caches"; + let data = b"legacy clone should preserve SIMD codec behavior"; let encoded = cloned.encode_data(data).expect("legacy clone should encode"); let mut shards = optional_shards(&encoded); shards[0] = None; @@ -1413,6 +1480,93 @@ mod tests { assert_eq!(recover_data(&shards, cloned.data_shards, data.len()), data); } + #[test] + fn legacy_codecs_share_process_cache_across_erasure_instances() { + let first = Erasure::new_with_options(6, 3, 64, true) + .legacy_encoder + .expect("legacy codec should be initialized"); + let second = Erasure::new_with_options(6, 3, 128, true) + .legacy_encoder + .expect("same legacy shard layout should be initialized"); + + assert!(Arc::ptr_eq(&first, &second)); + } + + #[test] + fn legacy_workspace_cache_rejects_oversize_buffers_and_isolates_layouts() { + let four_plus_two = Erasure::new_with_options(4, 2, 64, true) + .legacy_encoder + .expect("legacy codec should be initialized"); + let four_plus_one = Erasure::new_with_options(4, 1, 64, true) + .legacy_encoder + .expect("distinct parity layout should be initialized"); + let three_plus_two = Erasure::new_with_options(3, 2, 64, true) + .legacy_encoder + .expect("distinct data layout should be initialized"); + + assert!(!Arc::ptr_eq(&four_plus_two, &four_plus_one)); + assert!(!Arc::ptr_eq(&four_plus_two, &three_plus_two)); + assert_eq!(four_plus_two.logical_shard_bytes_upper_bound(64 * 1024), Some(512 * 1024)); + assert!(four_plus_two.should_cache_workspace(64 * 1024)); + assert!(!four_plus_two.should_cache_workspace(64 * 1024 + 1)); + + let nine_plus_seven = + LegacyReedSolomonEncoder::with_workspace_cache(9, 7, true).expect("9+7 legacy codec should construct"); + assert_eq!(nine_plus_seven.logical_shard_bytes_upper_bound(16 * 1024), Some(512 * 1024)); + assert!(nine_plus_seven.should_cache_workspace(16 * 1024)); + assert!(!nine_plus_seven.should_cache_workspace(16 * 1024 + 1)); + + let uncached = LegacyReedSolomonEncoder::new(4, 2).expect("uncached legacy codec should construct"); + assert!(!uncached.should_cache_workspace(64)); + } + + #[test] + fn saturated_legacy_codec_cache_does_not_retain_more_workspaces() { + let cache = RwLock::new(HashMap::new()); + for parity_shards in 1..=LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES { + let cached = + cached_legacy_reed_solomon_in(&cache, 32, parity_shards).expect("cacheable legacy codec should construct"); + assert!(cached.cache_workspaces); + } + + let uncached = + cached_legacy_reed_solomon_in(&cache, 31, 1).expect("uncached legacy codec should construct after saturation"); + assert!(!uncached.cache_workspaces); + assert_eq!( + cache.read().expect("cache lock should remain healthy").len(), + LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES + ); + } + + #[test] + fn concurrent_legacy_codecs_preserve_byte_exact_results() { + let barrier = Arc::new(std::sync::Barrier::new(2)); + let payloads = [vec![0x35; 257], vec![0xca; 1025]]; + + std::thread::scope(|scope| { + let handles = payloads.each_ref().map(|payload| { + let barrier = Arc::clone(&barrier); + scope.spawn(move || { + let erasure = Erasure::new_with_options(6, 3, 2048, true); + barrier.wait(); + let encoded = erasure.encode_data(payload).expect("concurrent legacy encode should succeed"); + barrier.wait(); + + let mut shards = optional_shards(&encoded); + shards[0] = None; + erasure + .decode_data(&mut shards) + .expect("concurrent legacy decode should reconstruct the missing shard"); + recover_data(&shards, erasure.data_shards, payload.len()) + }) + }); + + for (handle, payload) in handles.into_iter().zip(payloads.iter()) { + assert_eq!(handle.join().expect("concurrent legacy codec worker should not panic"), *payload); + } + }); + } + #[test] fn legacy_verify_reports_invalid_empty_valid_and_corrupt_parity_sets() { let legacy = LegacyReedSolomonEncoder::new(2, 2).expect("legacy encoder should construct"); diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index 35aefe053..27178c17d 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/crates/ecstore/src/object_api/readers.rs @@ -15,6 +15,7 @@ use super::*; use crate::io_support::rio::Index; +use std::mem::MaybeUninit; #[cfg(feature = "rio-v2")] const DARE_PAYLOAD_SIZE: i64 = 64 * 1024; @@ -922,7 +923,7 @@ struct SkipReader { inner: R, bytes_to_skip: usize, bytes_skipped: usize, - scratch: Vec, + scratch: Box<[MaybeUninit]>, } impl SkipReader { @@ -931,7 +932,7 @@ impl SkipReader { inner, bytes_to_skip, bytes_skipped: 0, - scratch: vec![0u8; 8192], + scratch: Box::<[u8]>::new_uninit_slice(8192), } } } @@ -943,7 +944,7 @@ impl AsyncRead for SkipReader { while this.bytes_skipped < this.bytes_to_skip { let remaining = this.bytes_to_skip - this.bytes_skipped; let scratch_len = remaining.min(this.scratch.len()); - let mut scratch_buf = ReadBuf::new(&mut this.scratch[..scratch_len]); + let mut scratch_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]); match Pin::new(&mut this.inner).poll_read(cx, &mut scratch_buf) { Poll::Pending => return Poll::Pending, Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), @@ -974,7 +975,7 @@ pub struct RangedDecompressReader target_length: usize, current_offset: usize, bytes_returned: usize, - scratch: Vec, + scratch: Box<[MaybeUninit]>, drain_on_done: bool, drain_task: Option>, } @@ -1012,7 +1013,7 @@ impl RangedDecompressReader { target_length: actual_length, current_offset: 0, bytes_returned: 0, - scratch: vec![0u8; 8192], + scratch: Box::<[u8]>::new_uninit_slice(8192), drain_on_done, drain_task: None, }) @@ -1062,7 +1063,7 @@ impl AsyncRead for RangedDecompres } let scratch_len = std::cmp::min(this.scratch.len(), std::cmp::max(buf_capacity, 1)); - let mut temp_read_buf = ReadBuf::new(&mut this.scratch[..scratch_len]); + let mut temp_read_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]); let Some(inner) = this.inner.as_mut() else { return Poll::Ready(Ok(())); @@ -1114,7 +1115,8 @@ impl AsyncRead for RangedDecompres ); if bytes_to_return > 0 { - let data_slice = &this.scratch[data_start_in_buffer..data_start_in_buffer + bytes_to_return]; + let data_slice = + &temp_read_buf.filled()[data_start_in_buffer..data_start_in_buffer + bytes_to_return]; buf.put_slice(data_slice); this.bytes_returned += bytes_to_return; @@ -1133,7 +1135,7 @@ impl AsyncRead for RangedDecompres std::cmp::min(n, std::cmp::min(buf.remaining(), this.target_length - this.bytes_returned)); if bytes_to_return > 0 { - buf.put_slice(&this.scratch[..bytes_to_return]); + buf.put_slice(&temp_read_buf.filled()[..bytes_to_return]); this.bytes_returned += bytes_to_return; tracing::trace!("Returned {} bytes at offset {}", bytes_to_return, old_offset); @@ -1263,6 +1265,43 @@ mod tests { use temp_env::async_with_vars; use tokio::io::AsyncReadExt; + #[derive(Debug)] + struct PendingPartialReader { + data: &'static [u8], + position: usize, + pending: bool, + } + + impl PendingPartialReader { + fn new(data: &'static [u8]) -> Self { + Self { + data, + position: 0, + pending: true, + } + } + } + + impl AsyncRead for PendingPartialReader { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if self.pending { + self.pending = false; + cx.waker().wake_by_ref(); + return Poll::Pending; + } + if self.position == self.data.len() { + return Poll::Ready(Ok(())); + } + + let length = buf.remaining().min(3).min(self.data.len() - self.position); + let end = self.position + length; + buf.put_slice(&self.data[self.position..end]); + self.position = end; + self.pending = true; + Poll::Ready(Ok(())) + } + } + const TEST_DIRECT_KEY_HEADER: &str = "x-rustfs-test-direct-key"; const TEST_OBJECT_KEY_HEADER: &str = "x-rustfs-test-object-key"; const TEST_NONCE_HEADER: &str = "x-rustfs-test-nonce"; @@ -1400,6 +1439,36 @@ mod tests { assert_eq!(result, b"World"); } + #[tokio::test] + async fn uninitialized_scratch_preserves_partial_pending_and_eof_reads() { + let mut skipped = SkipReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5); + let mut skipped_output = Vec::new(); + skipped + .read_to_end(&mut skipped_output) + .await + .expect("skip reader should survive partial pending reads through EOF"); + assert_eq!(skipped_output, b"56789abcdef"); + + let mut ranged = RangedDecompressReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5, 7, 16) + .expect("valid range should construct"); + let mut ranged_output = Vec::new(); + ranged + .read_to_end(&mut ranged_output) + .await + .expect("range reader should survive partial pending reads through EOF"); + assert_eq!(ranged_output, b"56789ab"); + } + + #[tokio::test] + async fn uninitialized_skip_scratch_reports_early_eof() { + let mut reader = SkipReader::new(PendingPartialReader::new(b"short"), 6); + let error = reader + .read_to_end(&mut Vec::new()) + .await + .expect_err("EOF before the skip boundary must remain visible"); + assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof); + } + #[tokio::test] async fn test_ranged_decompress_reader_from_start() { let original_data = b"Hello, World! This is a test."; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4abbfe749..27e382a3e 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -2792,6 +2792,9 @@ pub struct SetDisks { get_object_metadata_cache: moka::future::Cache>, get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState, get_object_metadata_cache_generations: Arc<[AtomicU64]>, + /// GET codecs keyed by every persisted layout dimension that affects + /// decoding. Clones of a set share the memoized shells. + erasure_cache: Arc, pub lockers: Vec>, shared_lockers: Arc<[Arc]>, local_lock_manager: Arc, @@ -2814,6 +2817,137 @@ pub struct SetDisks { storage_class_config_override: Arc>>>, } +const ERASURE_CACHE_MAX_ENTRIES: usize = 32; + +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +struct ErasureCacheKey { + data_shards: usize, + parity_shards: usize, + block_size: usize, + uses_legacy: bool, +} + +struct ErasureCache { + entries: parking_lot::RwLock>>, +} + +impl std::fmt::Debug for ErasureCache { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("ErasureCache") + .field("entries", &self.entries.read().len()) + .finish() + } +} + +impl ErasureCache { + fn new() -> Self { + Self { + entries: parking_lot::RwLock::new(HashMap::new()), + } + } + + fn get_or_try_insert( + &self, + key: ErasureCacheKey, + ) -> std::result::Result, coding::ErasureConstructionError> { + if let Some(erasure) = self.entries.read().get(&key) { + return Ok(Arc::clone(erasure)); + } + + // Serialize first construction for a key so concurrent cold GETs still + // create exactly one shell. Codec construction never awaits. + let mut entries = self.entries.write(); + if let Some(erasure) = entries.get(&key) { + return Ok(Arc::clone(erasure)); + } + let erasure = Arc::new(coding::Erasure::try_new_with_options( + key.data_shards, + key.parity_shards, + key.block_size, + key.uses_legacy, + )?); + if entries.len() < ERASURE_CACHE_MAX_ENTRIES { + entries.insert(key, Arc::clone(&erasure)); + } + Ok(erasure) + } + + fn get_for_file_info(&self, fi: &FileInfo) -> Result> { + self.get_or_try_insert(ErasureCacheKey { + data_shards: fi.erasure.data_blocks, + parity_shards: fi.erasure.parity_blocks, + block_size: fi.erasure.block_size, + uses_legacy: fi.uses_legacy_checksum, + }) + .map_err(Error::from) + } +} + +#[cfg(test)] +mod erasure_cache_tests { + use super::*; + + #[test] + fn reuses_shells_and_keeps_every_layout_dimension_in_the_key() { + let cache = ErasureCache::new(); + let base = ErasureCacheKey { + data_shards: 4, + parity_shards: 2, + block_size: 1_048_576, + uses_legacy: false, + }; + let first = cache.get_or_try_insert(base).expect("modern shell should construct"); + let reused = cache.get_or_try_insert(base).expect("same modern shell should be cached"); + assert!(Arc::ptr_eq(&first, &reused)); + + for distinct in [ + ErasureCacheKey { data_shards: 3, ..base }, + ErasureCacheKey { + parity_shards: 1, + ..base + }, + ErasureCacheKey { + block_size: 524_288, + ..base + }, + ErasureCacheKey { + uses_legacy: true, + ..base + }, + ] { + let shell = cache.get_or_try_insert(distinct).expect("distinct shell should construct"); + assert!(!Arc::ptr_eq(&first, &shell)); + } + assert_eq!(cache.entries.read().len(), 5); + } + + #[test] + fn does_not_cache_invalid_layouts_or_grow_past_the_bound() { + let cache = ErasureCache::new(); + let invalid = ErasureCacheKey { + data_shards: 4, + parity_shards: 2, + block_size: 0, + uses_legacy: false, + }; + assert!(cache.get_or_try_insert(invalid).is_err()); + assert!(cache.entries.read().is_empty()); + + for block_size in 1..=(ERASURE_CACHE_MAX_ENTRIES + 1) { + cache + .get_or_try_insert(ErasureCacheKey { + data_shards: 4, + parity_shards: 2, + block_size, + uses_legacy: false, + }) + .expect("bounded cache fixture should construct"); + } + assert_eq!(cache.entries.read().len(), ERASURE_CACHE_MAX_ENTRIES); + } +} + #[derive(Clone, Debug, Eq, PartialEq)] struct GetObjectMetadataCacheKey { bucket: Arc, @@ -3212,6 +3346,7 @@ impl SetDisks { .map(|_| AtomicU64::new(0)) .collect::>(), ), + erasure_cache: Arc::new(ErasureCache::new()), lockers, shared_lockers, // Sourced from the instance context so each instance owns its lock @@ -9816,6 +9951,7 @@ mod tests { let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo( "bucket", "object", + Arc::new(ErasureCache::new()), &fi, &disk_files, &disks, @@ -9879,6 +10015,7 @@ mod tests { let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo( "bucket", "object", + Arc::new(ErasureCache::new()), &fi, &disk_files, &vec![Some(disk); erasure.total_shard_count()], @@ -9959,6 +10096,7 @@ mod tests { let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), &fi, &files, &disks, @@ -10044,6 +10182,7 @@ mod tests { SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), range_offset, range_length as i64, &mut writer, @@ -10155,6 +10294,7 @@ mod tests { SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, total_size as i64, &mut writer, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 46b25a3c7..2363319b3 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -660,7 +660,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } - let erasure = erasure_from_file_info(fi, fi.uses_legacy_checksum)?; + let erasure = self.erasure_cache.get_for_file_info(fi)?; let read_length = erasure.shard_file_offset(0, object_size, object_size); let total_shards = data_shards + fi.erasure.parity_blocks; let (_disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); @@ -829,6 +829,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { if let Some(body) = Self::try_get_object_direct_data_shards_with_fileinfo( bucket, object, + Arc::clone(&self.erasure_cache), fi, files, disks, @@ -864,6 +865,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { Self::get_object_with_fileinfo( bucket, object, + Arc::clone(&self.erasure_cache), 0, object_info.size, &mut output, @@ -907,6 +909,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { match Self::get_object_decode_reader_with_fileinfo( bucket, object, + Arc::clone(&self.erasure_cache), fi, files, disks, @@ -971,6 +974,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let set_index = self.set_index; let pool_index = self.pool_index; let skip_verify = opts.skip_verify_bitrot; + let erasure_cache = Arc::clone(&self.erasure_cache); let (fi, files, disks) = snapshot.into_owned(); tokio::spawn(async move { let _guard = read_lock_guard; @@ -982,6 +986,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { if let Err(e) = Self::get_object_with_fileinfo( &bucket, &object, + erasure_cache, offset, length, &mut writer, @@ -5009,11 +5014,13 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let pool_index = self.pool_index; let skip_verify = opts.skip_verify_bitrot; let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(cloned_fi.size); + let erasure_cache = Arc::clone(&self.erasure_cache); let producer = async move { let mut writer = TransitionUploadWriter::new(pw); Self::get_object_with_fileinfo( &cloned_bucket, &cloned_object, + erasure_cache, 0, cloned_fi.size, &mut writer, @@ -5989,6 +5996,37 @@ mod inline_put_commit_path_tests { assert_eq!(restored, payload); } + #[tokio::test] + async fn repeated_gets_reuse_the_set_erasure_shell() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "get-erasure-shell-cache"; + let object = "object.bin"; + let payload = vec![0x4d; 1024 * 1024]; + make_bucket(&disk_stores, bucket).await; + + let mut reader = PutObjReader::from_vec(payload.clone()); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("non-inline object should commit"); + assert!(set_disks.erasure_cache.entries.read().is_empty()); + + for _ in 0..2 { + let mut object_reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("cached-shell GET should succeed"); + let mut restored = Vec::new(); + object_reader + .stream + .read_to_end(&mut restored) + .await + .expect("cached-shell GET should stream"); + assert_eq!(restored, payload); + assert_eq!(set_disks.erasure_cache.entries.read().len(), 1); + } + } + #[tokio::test] async fn ec_8_4_default_budget_keeps_large_inline_candidate_out_of_xl_meta() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(12).await; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 3fd19b151..d0011cd11 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -482,6 +482,7 @@ impl SetDisks { pub(super) async fn try_get_object_direct_data_shards_with_fileinfo( bucket: &str, object: &str, + erasure_cache: Arc, fi: &FileInfo, files: &[FileInfo], disks: &[Option], @@ -502,13 +503,7 @@ impl SetDisks { return Ok(None); } - let erasure = coding::Erasure::try_new_with_options( - fi.erasure.data_blocks, - fi.erasure.parity_blocks, - fi.erasure.block_size, - fi.uses_legacy_checksum, - ) - .map_err(Error::from)?; + let erasure = erasure_cache.get_for_file_info(fi)?; let checksum_info = fi.erasure.get_checksum_info(part.number); let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { @@ -636,6 +631,7 @@ impl SetDisks { // &self, bucket: &str, object: &str, + erasure_cache: Arc, offset: usize, length: i64, writer: &mut W, @@ -730,13 +726,7 @@ impl SetDisks { object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds" ); - let erasure = coding::Erasure::try_new_with_options( - fi.erasure.data_blocks, - fi.erasure.parity_blocks, - fi.erasure.block_size, - fi.uses_legacy_checksum, - ) - .map_err(Error::from)?; + let erasure = erasure_cache.get_for_file_info(&fi)?; let part_indices: Vec = (part_index..=last_part_index).collect(); debug!(bucket, object, ?part_indices, "Multipart part indices to stream"); @@ -1170,6 +1160,7 @@ impl SetDisks { pub(super) async fn get_object_decode_reader_with_fileinfo( bucket: &str, object: &str, + erasure_cache: Arc, fi: &FileInfo, files: &[FileInfo], disks: &[Option], @@ -1180,14 +1171,7 @@ impl SetDisks { metrics_size_bucket: &'static str, prefer_data_blocks_first_reader_setup: bool, ) -> Result { - let erasure = coding::Erasure::try_new_with_options( - fi.erasure.data_blocks, - fi.erasure.parity_blocks, - fi.erasure.block_size, - fi.uses_legacy_checksum, - ) - .map_err(Error::from)?; - + let erasure = erasure_cache.get_for_file_info(fi)?; let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); if fi.parts.len() == 1 { @@ -1574,7 +1558,7 @@ struct LazyCodecPartContext { fi: FileInfo, files: Vec, disks: Vec>, - erasure: coding::Erasure, + erasure: Arc, skip_verify_bitrot: bool, metrics_object_class: &'static str, metrics_size_bucket: &'static str, @@ -2058,6 +2042,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( "bucket", "object", + Arc::new(ErasureCache::new()), 0, 1, &mut output, @@ -2088,6 +2073,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 2, 1, &mut output, @@ -2111,6 +2097,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), usize::MAX, 1, &mut output, @@ -2132,6 +2119,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 1, 1, &mut output, @@ -2155,6 +2143,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, 1, &mut output, @@ -2192,6 +2181,7 @@ mod metadata_cache_tests { SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, 0, &mut output, @@ -2224,6 +2214,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, 1, &mut output, @@ -4128,6 +4119,7 @@ mod tests { let result = SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &fi, &[], &[], @@ -4150,6 +4142,7 @@ mod tests { let invalid_size = SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &single_part, &[], &[], @@ -4170,6 +4163,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &multipart, &[], &[], @@ -4194,6 +4188,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &multipart, &[], &[], @@ -4222,6 +4217,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &multipart, &[], &[], @@ -4275,6 +4271,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &fi, &files, &disks, @@ -4328,6 +4325,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &fi, &files, &disks, @@ -4372,6 +4370,7 @@ mod tests { SetDisks::get_object_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), 0, part_data.len() as i64, &mut output,