perf(ecstore): reuse erasure codecs on GET paths (#6074)

* perf(ecstore): share legacy SIMD workspaces

Reuse legacy Reed-Solomon encoder and decoder workspaces across Erasure instances with the same shard layout while keeping active codecs request-exclusive.

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

* perf(ecstore): reuse GET erasure shells and scratch buffers

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

* test(ecstore): satisfy concurrent codec lint

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

* perf(ecstore): bound cached legacy workspaces

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

* perf(ecstore): cap retained legacy codec memory

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-14 01:51:42 +08:00
committed by GitHub
parent e16c07b9cd
commit 6178083985
5 changed files with 472 additions and 72 deletions
+194 -40
View File
@@ -71,10 +71,16 @@ impl EncodedBlock {
const MODERN_MAX_TOTAL_SHARDS: usize = <reed_solomon_erasure::galois_8::Field as reed_solomon_erasure::Field>::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<HashMap<(usize, usize), Arc<ReedSolomon>>>;
type LegacyReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<LegacyReedSolomonEncoder>>>;
static MODERN_REED_SOLOMON_CACHE: OnceLock<ModernReedSolomonCache> = OnceLock::new();
static LEGACY_REED_SOLOMON_CACHE: OnceLock<LegacyReedSolomonCache> = 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<Option<reed_solomon_simd::ReedSolomonEncoder>>,
decoder_cache: std::sync::RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
}
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<Option<reed_solomon_simd::ReedSolomonEncoder>>,
decoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
}
impl LegacyReedSolomonEncoder {
fn new(_data_shards: usize, _parity_shards: usize) -> io::Result<Self> {
fn new(data_shards: usize, parity_shards: usize) -> io::Result<Self> {
Self::with_workspace_cache(data_shards, parity_shards, false)
}
fn with_workspace_cache(data_shards: usize, parity_shards: usize, cache_workspaces: bool) -> io::Result<Self> {
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<usize> {
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<Arc<LegacyReedSolomonEncoder>> {
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<Arc<LegacyReedSolomonEncoder>> {
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<F>(shards: &mut [Option<Vec<u8>>], 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<ReedSolomonEncoder>,
legacy_encoder: Option<LegacyReedSolomonEncoder>,
legacy_encoder: Option<Arc<LegacyReedSolomonEncoder>>,
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");
+77 -8
View File
@@ -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<R> {
inner: R,
bytes_to_skip: usize,
bytes_skipped: usize,
scratch: Vec<u8>,
scratch: Box<[MaybeUninit<u8>]>,
}
impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
@@ -931,7 +932,7 @@ impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
inner,
bytes_to_skip,
bytes_skipped: 0,
scratch: vec![0u8; 8192],
scratch: Box::<[u8]>::new_uninit_slice(8192),
}
}
}
@@ -943,7 +944,7 @@ impl<R: AsyncRead + Unpin + Send + Sync> AsyncRead for SkipReader<R> {
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<R: AsyncRead + Unpin + Send + Sync + 'static>
target_length: usize,
current_offset: usize,
bytes_returned: usize,
scratch: Vec<u8>,
scratch: Box<[MaybeUninit<u8>]>,
drain_on_done: bool,
drain_task: Option<tokio::task::JoinHandle<()>>,
}
@@ -1012,7 +1013,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> RangedDecompressReader<R> {
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<R: AsyncRead + Unpin + Send + Sync + 'static> 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<R: AsyncRead + Unpin + Send + Sync + 'static> 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<R: AsyncRead + Unpin + Send + Sync + 'static> 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<std::io::Result<()>> {
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.";
+140
View File
@@ -2792,6 +2792,9 @@ pub struct SetDisks {
get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
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<ErasureCache>,
pub lockers: Vec<Arc<dyn LockClient>>,
shared_lockers: Arc<[Arc<dyn LockClient>]>,
local_lock_manager: Arc<rustfs_lock::GlobalLockManager>,
@@ -2814,6 +2817,137 @@ pub struct SetDisks {
storage_class_config_override: Arc<std::sync::RwLock<Option<Arc<storageclass::Config>>>>,
}
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<HashMap<ErasureCacheKey, Arc<coding::Erasure>>>,
}
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<Arc<coding::Erasure>, 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<Arc<coding::Erasure>> {
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<str>,
@@ -3212,6 +3346,7 @@ impl SetDisks {
.map(|_| AtomicU64::new(0))
.collect::<Vec<_>>(),
),
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,
+39 -1
View File
@@ -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;
+22 -23
View File
@@ -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<ErasureCache>,
fi: &FileInfo,
files: &[FileInfo],
disks: &[Option<DiskStore>],
@@ -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<ErasureCache>,
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<usize> = (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<ErasureCache>,
fi: &FileInfo,
files: &[FileInfo],
disks: &[Option<DiskStore>],
@@ -1180,14 +1171,7 @@ impl SetDisks {
metrics_size_bucket: &'static str,
prefer_data_blocks_first_reader_setup: bool,
) -> Result<GetCodecStreamingReaderBuildOutcome> {
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<FileInfo>,
disks: Vec<Option<DiskStore>>,
erasure: coding::Erasure,
erasure: Arc<coding::Erasure>,
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,