gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-14 17:13:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-13 22:48:16 +08:00
parent e5a7b6f0d9
commit 78ac2aa3f8
4 changed files with 247 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/object_api/readers.rs
+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.";
crates/ecstore/src/set_disk/mod.rs
+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,
crates/ecstore/src/set_disk/ops/object.rs
+8 -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,
@@ -5008,11 +5013,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,
crates/ecstore/src/set_disk/read.rs
+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,