mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-20 03:22:18 +00:00
feat(get): Small-file GET performance optimization for 1KiB-1MiB objects (#4016)
* feat(get): SF01 - bucket validation cache Add 5s TTL cache for bucket validation to avoid repeated stat_volume() calls on every GET request. Changes: - Add BUCKET_VALIDATED_CACHE (OnceLock + RwLock + HashMap) - Add invalidate_bucket_validation_cache() for cache invalidation - Add invalidate_all_bucket_validation_cache() for bulk invalidation - Update get_validated_store() to use cache - Add cache invalidation in execute_delete_bucket() Expected impact: 3-5x improvement for small file GET latency. Closes rustfs/backlog#766 Co-Authored-By: heihutu <heihutu@gmail.com> * feat(get): SF03 - metadata cache TTL increase Increase metadata cache TTL from 250ms to 2s and capacity from 1024 to 4096 entries. Changes: - GET_OBJECT_METADATA_CACHE_TTL: 250ms -> 2s - GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: 1024 -> 4096 All mutation paths already call invalidate_get_object_metadata_cache, so the longer TTL is safe. Expected impact: 10-50x improvement for hot objects. Closes rustfs/backlog#768 Co-Authored-By: heihutu <heihutu@gmail.com> * feat(get): SF04 - remove unnecessary tokio::spawn in metadata fanout Replace tokio::spawn with direct async future in read_all_fileinfo_full_wait. join_all already provides concurrency, so tokio::spawn adds unnecessary task creation and scheduling overhead. Changes: - Remove tokio::spawn from metadata fanout futures - Update result handling for direct future results Expected impact: 16-32us reduction per GET request. Closes rustfs/backlog#769 Co-Authored-By: heihutu <heihutu@gmail.com> * feat(get): SF06 - conditional lifecycle check Only call resolve_put_object_expiration when the object has an x-amz-expiration metadata marker. This avoids unnecessary lifecycle configuration reads on every GET request. Expected impact: 50-100us reduction per GET request. Closes rustfs/backlog#771 Co-Authored-By: heihutu <heihutu@gmail.com> * feat(get): SF07 - conditional metrics recording Gate hot path metrics behind get_stage_metrics_enabled() to reduce overhead when metrics are not needed. Changes: - Conditional record_zero_copy_read - Conditional manager.record_disk_operation - Conditional manager.record_access - Conditional manager.record_transfer Expected impact: 20-50us reduction per GET request. Closes rustfs/backlog#772 Co-Authored-By: heihutu <heihutu@gmail.com> * refactor(get): SF01 - use moka instead of dashmap for bucket cache Replace OnceLock + RwLock + HashMap with moka::sync::Cache for bucket validation cache. moka provides built-in TTL support and is already available in the workspace. Changes: - Add moka dependency to rustfs crate - Replace manual TTL management with moka's time_to_live - Simplify cache operations Closes rustfs/backlog#766 Co-Authored-By: heihutu <heihutu@gmail.com> * feat(get): SF02 - inline data fast path Add fast path for small inline objects that bypasses duplex pipe, tokio::spawn, and bitrot reader creation when data is already in memory. Changes: - Add inline data detection before codec streaming gate - Direct in-memory erasure decode for inline objects <= 128KB - Add GET_OBJECT_PATH_INLINE_DIRECT metric path - Skip duplex pipe and background task for inline data Conditions for fast path: - Single part object - Inline data available - Size <= 128KB - Not encrypted/compressed/remote - No range request Expected impact: 2-3x improvement for small file GET latency. Closes rustfs/backlog#767 Co-Authored-By: heihutu <heihutu@gmail.com> * refactor: translate Chinese comments to English Translate all Chinese comments to English in modified files: - rustfs/src/storage/ecfs_extend.rs - rustfs/src/app/bucket_usecase.rs Co-Authored-By: heihutu <heihutu@gmail.com> * fix * add * fmt and improve import * fmt * feat(get): SF05 skip IO planning + refactor inline detection + adaptive bucket cache SF05: Skip disk I/O semaphore for inline data fast path - Reorder prepare_get_object_read_execution: read first, then decide semaphore - Inline objects skip acquire_disk_read_permit() entirely (saves 100-200us) - Add is_inline_fast_path field to GetObjectReadSetup Refactor: Unify inline detection logic - Add ObjectInfo::is_inline_fast_path_eligible() as single source of truth - Version-aware thresholds: non-versioned 128KB, versioned 16KB (matches PUT) - Eliminates divergent conditions between set_disk/mod.rs and object_usecase.rs Refactor: Restore fault tolerance in metadata fanout - Restore tokio::spawn + JoinError handling in read_all_fileinfo_full_wait - Prevents single disk read panic from unwinding the entire operation Refactor: Restore lifecycle check correctness - Remove incorrect SF06 conditional that skipped lifecycle for most objects - Always call resolve_put_object_expiration (original behavior) Fix: make_bucket cache invalidation - Invalidate bucket validation cache on create_bucket Fix: erasure decode written validation - Check decode() return value; error if 0 bytes written for non-empty object Adaptive bucket cache - Default: RwLock<HashMap> for < 100 buckets (low overhead) - Opt-in: starshard::ShardedHashMap via RUSTFS_BUCKET_CACHE_STARSHARD=1 - 5s TTL with manual timestamp checking Benchmark results (warp get, concurrency 32, 10s, 3 rounds): - 10KiB: 25.10 MiB/s (+28.2% vs SF01-07) - 100KiB: 221.81 MiB/s - 1MiB: 1972.78 MiB/s - vs main: -10% to -12% (inline path not triggered by warp) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(versioning): use read lock for versioning config query + five-expert analysis P0 fix: BucketVersioningSys::get() was using write lock on GLOBAL_BucketMetadataSys for a pure read operation. This serialized all concurrent GET requests (3 write-lock acquisitions per request). Changed to read lock — get_versioning_config() handles its own internal locking via metadata_map RwLock. Five-expert analysis identified top bottlenecks: 1. Versioning write lock (P0, fixed) 2. Inline fast path not triggered (P0, needs verification) 3. Metadata fanout no early-stop (P1, early-stop has bug, reverted) 4. Request-level versioning cache (P1, pending) 5. Duplex pipe for small objects (P2, pending) Benchmark (read-lock fix, warp concurrency 32): - 1KiB: 2.29 MiB/s (vs 2.53 before, within variance) - 10KiB: 25.00 MiB/s (same as before) - 100KiB: 246.72 MiB/s (+11% vs 221.81) - 1MiB: 2039.95 MiB/s (+3% vs 1972.78) Co-Authored-By: heihutu <heihutu@gmail.com> * chore: remove benchmark results from git, keep locally only Remove docs/benchmark/*.md from version control. Files remain on disk but are no longer tracked by git. Added docs/benchmark/*.md to .gitignore. Co-Authored-By: heihutu <heihutu@gmail.com> * fix(get): decode inline fast path through bitrot readers --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -75,8 +75,11 @@ impl BucketVersioningSys {
|
||||
return Ok(VersioningConfiguration::default());
|
||||
}
|
||||
|
||||
// Read lock is sufficient — get_versioning_config() handles its own
|
||||
// internal locking via metadata_map RwLock. The previous write lock
|
||||
// serialized all concurrent GET requests on this global lock.
|
||||
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
|
||||
let bucket_meta_sys = bucket_meta_sys_lock.write().await;
|
||||
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
|
||||
|
||||
let (cfg, _) = bucket_meta_sys.get_versioning_config(bucket).await?;
|
||||
|
||||
|
||||
@@ -21,6 +21,7 @@ pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming";
|
||||
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine";
|
||||
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_streaming_rustfs_engine";
|
||||
pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty";
|
||||
pub(crate) const GET_OBJECT_PATH_INLINE_DIRECT: &str = "inline_direct";
|
||||
pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex";
|
||||
pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";
|
||||
pub(crate) const GET_CODEC_STREAMING_DECISION_USE: &str = "use";
|
||||
|
||||
@@ -1,3 +1,17 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Static ECStore layout boundaries.
|
||||
//!
|
||||
//! This module owns read-only layout descriptors used to keep static set
|
||||
|
||||
@@ -1,3 +1,17 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::*;
|
||||
#[cfg(feature = "rio-v2")]
|
||||
use aes_gcm::aead::Payload;
|
||||
@@ -122,7 +136,7 @@ fn restore_request_active(opts: &ObjectOptions) -> bool {
|
||||
restore.type_.is_some() || restore.days.is_some() || restore.output_location.is_some() || restore.select_parameters.is_some()
|
||||
}
|
||||
|
||||
fn decode_compression_index(index: Option<&bytes::Bytes>) -> Option<Index> {
|
||||
fn decode_compression_index(index: Option<&Bytes>) -> Option<Index> {
|
||||
crate::io_support::rio::decode_compression_index_bytes(index?)
|
||||
}
|
||||
|
||||
@@ -833,11 +847,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> RangedDecompressReader<R> {
|
||||
}
|
||||
|
||||
impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompressReader<R> {
|
||||
fn poll_read(
|
||||
mut self: std::pin::Pin<&mut Self>,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
buf: &mut tokio::io::ReadBuf<'_>,
|
||||
) -> std::task::Poll<std::io::Result<()>> {
|
||||
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
|
||||
use std::pin::Pin;
|
||||
use std::task::Poll;
|
||||
use tokio::io::ReadBuf;
|
||||
@@ -985,11 +995,7 @@ impl<R: AsyncRead + Unpin + Send + 'static> StreamConsumer<R> {
|
||||
}
|
||||
|
||||
impl<R: AsyncRead + Unpin + Send + 'static> AsyncRead for StreamConsumer<R> {
|
||||
fn poll_read(
|
||||
mut self: std::pin::Pin<&mut Self>,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
buf: &mut tokio::io::ReadBuf<'_>,
|
||||
) -> std::task::Poll<std::io::Result<()>> {
|
||||
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
|
||||
use std::pin::Pin;
|
||||
use std::task::Poll;
|
||||
|
||||
@@ -1032,7 +1038,7 @@ fn encrypted_plaintext_size(oi: &ObjectInfo, is_multipart: bool, is_compressed:
|
||||
oi.decrypted_size().map_err(Into::into)
|
||||
}
|
||||
|
||||
fn is_multipart_encrypted_object(parts: &[rustfs_filemeta::ObjectPartInfo], etag: Option<&str>) -> bool {
|
||||
fn is_multipart_encrypted_object(parts: &[ObjectPartInfo], etag: Option<&str>) -> bool {
|
||||
if parts.len() > 1 {
|
||||
return true;
|
||||
}
|
||||
@@ -1040,13 +1046,13 @@ fn is_multipart_encrypted_object(parts: &[rustfs_filemeta::ObjectPartInfo], etag
|
||||
etag.map(|etag| etag.trim_matches('"').len() != 32).unwrap_or(false)
|
||||
}
|
||||
|
||||
fn multipart_plaintext_size(parts: &[rustfs_filemeta::ObjectPartInfo], fallback: i64) -> i64 {
|
||||
fn multipart_plaintext_size(parts: &[ObjectPartInfo], fallback: i64) -> i64 {
|
||||
let total: i64 = parts.iter().map(part_plaintext_size).sum();
|
||||
|
||||
if total > 0 { total } else { fallback }
|
||||
}
|
||||
|
||||
fn multipart_part_numbers(parts: &[rustfs_filemeta::ObjectPartInfo]) -> Vec<usize> {
|
||||
fn multipart_part_numbers(parts: &[ObjectPartInfo]) -> Vec<usize> {
|
||||
parts.iter().map(|part| part.number).collect()
|
||||
}
|
||||
|
||||
@@ -1605,13 +1611,13 @@ mod tests {
|
||||
|
||||
fn ssec_headers_from_key(key_bytes: [u8; 32]) -> HeaderMap<HeaderValue> {
|
||||
let mut headers = HeaderMap::new();
|
||||
headers.insert(rustfs_utils::http::SSEC_ALGORITHM_HEADER, HeaderValue::from_static("AES256"));
|
||||
headers.insert(SSEC_ALGORITHM_HEADER, HeaderValue::from_static("AES256"));
|
||||
headers.insert(
|
||||
rustfs_utils::http::SSEC_KEY_HEADER,
|
||||
SSEC_KEY_HEADER,
|
||||
HeaderValue::from_str(&BASE64_STANDARD.encode(key_bytes)).expect("valid base64 header"),
|
||||
);
|
||||
headers.insert(
|
||||
rustfs_utils::http::SSEC_KEY_MD5_HEADER,
|
||||
SSEC_KEY_MD5_HEADER,
|
||||
HeaderValue::from_str(&BASE64_STANDARD.encode(md5_bytes(key_bytes))).expect("valid md5 header"),
|
||||
);
|
||||
headers
|
||||
@@ -2616,7 +2622,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_object_reader_compressed_range_returns_physical_offset_from_index() {
|
||||
let mut index = crate::io_support::rio::Index::new();
|
||||
let mut index = Index::new();
|
||||
index.add(0, 0).unwrap();
|
||||
index.add(1_048_576, 2_097_152).unwrap();
|
||||
|
||||
@@ -2670,7 +2676,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_read_plan_compressed_range_tracks_storage_and_visible_offsets() {
|
||||
let mut index = crate::io_support::rio::Index::new();
|
||||
let mut index = Index::new();
|
||||
index.add(0, 0).unwrap();
|
||||
index.add(1_048_576, 2_097_152).unwrap();
|
||||
|
||||
@@ -2746,7 +2752,7 @@ mod tests {
|
||||
#[cfg(feature = "rio-v2")]
|
||||
#[tokio::test]
|
||||
async fn test_read_plan_accepts_minio_headerless_compression_index() {
|
||||
let mut index = crate::io_support::rio::Index::new();
|
||||
let mut index = Index::new();
|
||||
index.add(0, 0).unwrap();
|
||||
index.add(1_048_576, 2_097_152).unwrap();
|
||||
let headerless_index = crate::io_support::rio::compression_index_storage_bytes(&index);
|
||||
@@ -2800,7 +2806,7 @@ mod tests {
|
||||
#[cfg(feature = "rio-v2")]
|
||||
#[test]
|
||||
fn test_get_compressed_offsets_aligns_encrypted_ranges_to_dare_packages() {
|
||||
let mut index = crate::io_support::rio::Index::new();
|
||||
let mut index = Index::new();
|
||||
index.add(0, 0).unwrap();
|
||||
index.add(200_000, 2_097_152).unwrap();
|
||||
let stored_index = crate::io_support::rio::compression_index_storage_bytes(&index);
|
||||
|
||||
@@ -1,3 +1,17 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::*;
|
||||
use crate::storage_api_contracts::{
|
||||
list::VersionMarker,
|
||||
@@ -263,6 +277,46 @@ impl ObjectInfo {
|
||||
})
|
||||
}
|
||||
|
||||
/// Maximum inline size for non-versioned objects (128 KiB).
|
||||
/// Matches `DEFAULT_INLINE_BLOCK` in `storageclass.rs`.
|
||||
pub const INLINE_MAX_SIZE: i64 = 128 * 1024;
|
||||
|
||||
/// Maximum inline size for versioned objects (16 KiB).
|
||||
/// Matches `DEFAULT_INLINE_BLOCK / 8` in `storageclass.rs`.
|
||||
pub const INLINE_MAX_SIZE_VERSIONED: i64 = 16 * 1024;
|
||||
|
||||
/// Returns `true` when this object qualifies for the inline data fast path.
|
||||
///
|
||||
/// The inline fast path decodes erasure-coded data entirely in memory,
|
||||
/// bypassing disk I/O, duplex pipes, and the disk-read semaphore.
|
||||
///
|
||||
/// The `inlined` flag is the primary signal — it is set during PUT by
|
||||
/// `storage_class_should_inline()` which already applies the correct
|
||||
/// version-aware threshold (128 KiB non-versioned, 16 KiB versioned).
|
||||
/// The size check below is a safety net using the same thresholds.
|
||||
///
|
||||
/// Additional conditions:
|
||||
/// - Single part
|
||||
/// - Not encrypted
|
||||
/// - Not compressed
|
||||
/// - Not transitioned to remote tier
|
||||
pub fn is_inline_fast_path_eligible(&self) -> bool {
|
||||
if !self.inlined {
|
||||
return false;
|
||||
}
|
||||
// Apply the same version-aware threshold as PUT (storageclass.rs).
|
||||
let max_size = if self.version_id.is_some() {
|
||||
Self::INLINE_MAX_SIZE_VERSIONED
|
||||
} else {
|
||||
Self::INLINE_MAX_SIZE
|
||||
};
|
||||
self.parts.len() == 1
|
||||
&& self.size <= max_size
|
||||
&& !self.is_encrypted()
|
||||
&& !self.is_compressed()
|
||||
&& self.transitioned_object.tier.is_empty()
|
||||
}
|
||||
|
||||
pub fn encryption_original_size(&self) -> std::io::Result<Option<i64>> {
|
||||
let actual_size = rustfs_utils::http::get_str(&self.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE);
|
||||
if let Some(size_str) = self
|
||||
@@ -354,14 +408,14 @@ impl ObjectInfo {
|
||||
// Parse expires from metadata (HTTP date format RFC 7231 or ISO 8601)
|
||||
let expires = fi.metadata.get("expires").and_then(|s| {
|
||||
// Try parsing as ISO 8601 first
|
||||
time::OffsetDateTime::parse(s, &time::format_description::well_known::Iso8601::DEFAULT)
|
||||
OffsetDateTime::parse(s, &time::format_description::well_known::Iso8601::DEFAULT)
|
||||
.or_else(|_| {
|
||||
// Try RFC 2822 format
|
||||
time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc2822)
|
||||
OffsetDateTime::parse(s, &time::format_description::well_known::Rfc2822)
|
||||
})
|
||||
.or_else(|_| {
|
||||
// Try RFC 3339 format
|
||||
time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339)
|
||||
OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339)
|
||||
})
|
||||
.ok()
|
||||
});
|
||||
@@ -775,7 +829,7 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn versions_listing_applies_version_marker_only_to_first_entry() {
|
||||
let metadata = rustfs_filemeta::test_data::create_real_xlmeta().expect("test metadata should be valid");
|
||||
let entries = rustfs_filemeta::MetaCacheEntriesSorted {
|
||||
let entries = MetaCacheEntriesSorted {
|
||||
o: rustfs_filemeta::MetaCacheEntries(vec![
|
||||
Some(rustfs_filemeta::MetaCacheEntry {
|
||||
name: "obj-a".to_owned(),
|
||||
@@ -878,7 +932,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn from_file_info_preserves_replication_decision() {
|
||||
let fi = rustfs_filemeta::FileInfo {
|
||||
let fi = FileInfo {
|
||||
replication_state_internal: Some(ReplicationState {
|
||||
replicate_decision_str: "arn=true;false;arn:replication::1:dest;rule-id".to_string(),
|
||||
..Default::default()
|
||||
@@ -904,11 +958,11 @@ mod tests {
|
||||
actual_size: 0,
|
||||
user_defined: Arc::new(user_defined),
|
||||
parts: Arc::new(vec![
|
||||
rustfs_filemeta::ObjectPartInfo {
|
||||
ObjectPartInfo {
|
||||
actual_size: 4,
|
||||
..Default::default()
|
||||
},
|
||||
rustfs_filemeta::ObjectPartInfo {
|
||||
ObjectPartInfo {
|
||||
actual_size: 5,
|
||||
..Default::default()
|
||||
},
|
||||
@@ -1016,13 +1070,13 @@ mod tests {
|
||||
user_defined: Arc::new(ud),
|
||||
user_tags: Arc::new("env=prod&team=storage".to_string()),
|
||||
parts: Arc::new(vec![
|
||||
rustfs_filemeta::ObjectPartInfo {
|
||||
ObjectPartInfo {
|
||||
number: 1,
|
||||
size: 1024,
|
||||
actual_size: 1024,
|
||||
..Default::default()
|
||||
},
|
||||
rustfs_filemeta::ObjectPartInfo {
|
||||
ObjectPartInfo {
|
||||
number: 2,
|
||||
size: 512,
|
||||
actual_size: 512,
|
||||
|
||||
@@ -345,9 +345,7 @@ pub fn shutdown_background_services() {
|
||||
/// * `Ok(())` if successful
|
||||
/// * `Err(Arc<dyn LockClient>)` if setting fails (client already set)
|
||||
///
|
||||
pub fn set_global_lock_client(
|
||||
client: Arc<dyn rustfs_lock::client::LockClient>,
|
||||
) -> Result<(), Arc<dyn rustfs_lock::client::LockClient>> {
|
||||
pub fn set_global_lock_client(client: Arc<dyn LockClient>) -> Result<(), Arc<dyn LockClient>> {
|
||||
GLOBAL_LOCAL_LOCK_CLIENT.set(client)
|
||||
}
|
||||
|
||||
@@ -356,7 +354,7 @@ pub fn set_global_lock_client(
|
||||
/// # Returns
|
||||
/// * `Option<Arc<dyn LockClient>>` - The global lock client, if set
|
||||
///
|
||||
pub fn get_global_lock_client() -> Option<Arc<dyn rustfs_lock::client::LockClient>> {
|
||||
pub fn get_global_lock_client() -> Option<Arc<dyn LockClient>> {
|
||||
GLOBAL_LOCAL_LOCK_CLIENT.get().cloned()
|
||||
}
|
||||
|
||||
|
||||
@@ -237,9 +237,9 @@ pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option<usize>
|
||||
pub(crate) fn backend_storage_class_parities(default_standard_parity: usize) -> (Option<usize>, Option<usize>) {
|
||||
if let Some(sc) = get_global_storage_class() {
|
||||
let standard = sc
|
||||
.get_parity_for_sc(crate::config::storageclass::CLASS_STANDARD)
|
||||
.get_parity_for_sc(storageclass::CLASS_STANDARD)
|
||||
.or(Some(default_standard_parity));
|
||||
let reduced_redundancy = sc.get_parity_for_sc(crate::config::storageclass::RRS);
|
||||
let reduced_redundancy = sc.get_parity_for_sc(storageclass::RRS);
|
||||
(standard, reduced_redundancy)
|
||||
} else {
|
||||
(Some(default_standard_parity), None)
|
||||
|
||||
@@ -129,14 +129,14 @@ impl SetDisks {
|
||||
|
||||
let erasure = if !latest_meta.deleted && !latest_meta.is_remote() {
|
||||
// Initialize erasure coding; use legacy mode for old-version files
|
||||
crate::erasure::coding::Erasure::new_with_options(
|
||||
coding::Erasure::new_with_options(
|
||||
latest_meta.erasure.data_blocks,
|
||||
latest_meta.erasure.parity_blocks,
|
||||
latest_meta.erasure.block_size,
|
||||
latest_meta.uses_legacy_checksum,
|
||||
)
|
||||
} else {
|
||||
crate::erasure::coding::Erasure::default()
|
||||
coding::Erasure::default()
|
||||
};
|
||||
|
||||
result.object_size =
|
||||
@@ -385,9 +385,9 @@ impl SetDisks {
|
||||
if let (Some(disk), Some(metadata)) = (disk, ©_parts_metadata[index]) {
|
||||
let checksum_info = metadata.erasure.get_checksum_info(part.number);
|
||||
let checksum_algo = if metadata.uses_legacy_checksum
|
||||
&& checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S
|
||||
&& checksum_info.algorithm == HashAlgorithm::HighwayHash256S
|
||||
{
|
||||
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
|
||||
HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
@@ -498,7 +498,7 @@ impl SetDisks {
|
||||
// parts_metadata[index].data = Some(w.inline_data().to_vec());
|
||||
// }
|
||||
parts_metadata[index].data =
|
||||
Some(writer.into_inline_data().map(bytes::Bytes::from).unwrap_or_default());
|
||||
Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
|
||||
}
|
||||
parts_metadata[index].set_inline_data();
|
||||
} else {
|
||||
|
||||
@@ -131,7 +131,7 @@ impl SetDisks {
|
||||
|
||||
fn reprobe_runtime_candidates_once(&self, disks: &[DiskStore]) {
|
||||
for disk in disks {
|
||||
if disk.runtime_state() != crate::disk::health_state::RuntimeDriveHealthState::Online {
|
||||
if disk.runtime_state() != disk::health_state::RuntimeDriveHealthState::Online {
|
||||
disk.reset_health_for_store_init_retry();
|
||||
}
|
||||
}
|
||||
@@ -536,15 +536,15 @@ mod tests {
|
||||
all_disks[1]
|
||||
.as_ref()
|
||||
.expect("disk 1 should exist")
|
||||
.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Suspect);
|
||||
.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Suspect);
|
||||
all_disks[2]
|
||||
.as_ref()
|
||||
.expect("disk 2 should exist")
|
||||
.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Returning);
|
||||
.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Returning);
|
||||
all_disks[3]
|
||||
.as_ref()
|
||||
.expect("disk 3 should exist")
|
||||
.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Offline);
|
||||
.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Offline);
|
||||
|
||||
let snapshot = set_disks.drive_membership_snapshot().await;
|
||||
assert_eq!(snapshot.online.len(), 1);
|
||||
@@ -563,7 +563,7 @@ mod tests {
|
||||
assert!(
|
||||
online_disks
|
||||
.iter()
|
||||
.all(|disk| { disk.runtime_state() != crate::disk::health_state::RuntimeDriveHealthState::Offline }),
|
||||
.all(|disk| { disk.runtime_state() != disk::health_state::RuntimeDriveHealthState::Offline }),
|
||||
"offline disks should be filtered by membership snapshot"
|
||||
);
|
||||
|
||||
@@ -601,7 +601,7 @@ mod tests {
|
||||
|
||||
let all_disks = set_disks.get_disks_internal().await;
|
||||
for disk in all_disks.iter().flatten() {
|
||||
disk.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Returning);
|
||||
disk.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Returning);
|
||||
}
|
||||
|
||||
let (online_disks, infos, healing) = set_disks.get_online_disks_with_healing_and_info(false).await;
|
||||
|
||||
@@ -25,8 +25,8 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
|
||||
use crate::cluster::rpc::heal_bucket_local_on_disks;
|
||||
use crate::diagnostics::get::{
|
||||
GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE,
|
||||
GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_LEGACY_DUPLEX,
|
||||
GET_OBJECT_PATH_REMOTE_TRANSITION, GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error,
|
||||
GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT,
|
||||
GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error,
|
||||
record_get_object_pipeline_failure,
|
||||
};
|
||||
use crate::disk::error_reduce::{
|
||||
@@ -171,10 +171,10 @@ const EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY: &str = "set_disk_put_object_stage
|
||||
const SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS: u128 = 5_000;
|
||||
const ENV_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: &str = "RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES";
|
||||
const DEFAULT_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: usize = 64 * 1024 * 1024;
|
||||
static CACHED_PUT_LARGE_BATCH_MIN_SIZE_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
|
||||
static CACHED_PUT_LARGE_BATCH_MIN_SIZE_BYTES: OnceLock<usize> = OnceLock::new();
|
||||
const ENV_RUSTFS_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: &str = "RUSTFS_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES";
|
||||
const DEFAULT_RUSTFS_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: usize = 128 * 1024 * 1024;
|
||||
static CACHED_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
|
||||
static CACHED_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: OnceLock<usize> = OnceLock::new();
|
||||
|
||||
use crate::io_support::rio::{EtagResolvable, HashReader, HashReaderMut, TryGetIndex as _};
|
||||
|
||||
@@ -329,8 +329,8 @@ fn adaptive_duplex_buffer_size(object_size: i64) -> usize {
|
||||
|
||||
const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
|
||||
const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
|
||||
const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_millis(250);
|
||||
const GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 1024;
|
||||
const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_secs(2); // Increased from 250ms to 2s
|
||||
const GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 4096; // Increased from 1024 to 4096
|
||||
|
||||
// --- Codec Streaming Configuration ---
|
||||
|
||||
@@ -1616,6 +1616,86 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
return Ok(reader);
|
||||
}
|
||||
|
||||
// Inline data fast path: skip duplex pipe for small inline objects.
|
||||
// Uses the shared predicate from ObjectInfo; additionally checks that
|
||||
// inline data is actually present and no range request is in flight.
|
||||
if object_info.is_inline_fast_path_eligible() && fi.data.is_some() && range.is_none() {
|
||||
let data_shards = fi.erasure.data_blocks;
|
||||
let (_disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(&disks, &files, &fi);
|
||||
|
||||
// Check if we have enough inline data shards
|
||||
let inline_count = files
|
||||
.iter()
|
||||
.take(data_shards)
|
||||
.filter(|f| f.data.as_ref().is_some_and(|d| !d.is_empty()))
|
||||
.count();
|
||||
|
||||
if inline_count >= data_shards {
|
||||
// All data shards are inline - decode in memory
|
||||
let erasure = coding::Erasure::new_with_options(
|
||||
fi.erasure.data_blocks,
|
||||
fi.erasure.parity_blocks,
|
||||
fi.erasure.block_size,
|
||||
fi.uses_legacy_checksum,
|
||||
);
|
||||
let object_size = usize::try_from(fi.size)
|
||||
.map_err(|_| to_object_err(Error::other("inline fast path object size is invalid"), vec![bucket, object]))?;
|
||||
|
||||
let checksum_info = fi.erasure.get_checksum_info(fi.parts[0].number);
|
||||
let checksum_algo =
|
||||
if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S {
|
||||
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
let read_length = erasure.shard_file_offset(0, object_size, object_size);
|
||||
let mut readers: Vec<Option<coding::BitrotReader<Box<dyn tokio::io::AsyncRead + Send + Sync + Unpin>>>> =
|
||||
Vec::new();
|
||||
for file in files.iter().take(data_shards + fi.erasure.parity_blocks) {
|
||||
if let Some(data) = &file.data {
|
||||
readers.push(
|
||||
create_bitrot_reader(
|
||||
Some(data),
|
||||
None,
|
||||
bucket,
|
||||
object,
|
||||
0,
|
||||
read_length,
|
||||
erasure.shard_size(),
|
||||
checksum_algo.clone(),
|
||||
opts.skip_verify_bitrot,
|
||||
false,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
} else {
|
||||
readers.push(None);
|
||||
}
|
||||
}
|
||||
|
||||
// Decode directly
|
||||
let mut output = Cursor::new(Vec::with_capacity(object_size));
|
||||
let (written, err) = erasure.decode(&mut output, readers, 0, object_size, object_size).await;
|
||||
|
||||
if let Some(e) = err {
|
||||
return Err(to_object_err(e.into(), vec![bucket, object]));
|
||||
}
|
||||
if written == 0 && fi.size > 0 {
|
||||
return Err(to_object_err(
|
||||
Error::other("inline fast path: erasure decode returned 0 bytes"),
|
||||
vec![bucket, object],
|
||||
));
|
||||
}
|
||||
|
||||
rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_INLINE_DIRECT);
|
||||
let reader = GetObjectReader {
|
||||
stream: Box::new(Cursor::new(output.into_inner())),
|
||||
object_info,
|
||||
};
|
||||
return Ok(reader);
|
||||
}
|
||||
}
|
||||
|
||||
let codec_streaming_gate =
|
||||
get_codec_streaming_reader_gate(bucket, object, &range, &object_info, &fi, lock_optimization_enabled);
|
||||
|
||||
@@ -1827,8 +1907,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
|
||||
|
||||
let result: Result<ObjectInfo> = async {
|
||||
let erasure =
|
||||
crate::erasure::coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
|
||||
let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
|
||||
|
||||
let is_inline_buffer =
|
||||
runtime_sources::storage_class_should_inline(erasure.shard_file_size(data.size()), opts.versioned);
|
||||
@@ -2413,10 +2492,10 @@ impl SetDisks {
|
||||
}
|
||||
}
|
||||
Ok((_client_idx, Err(err))) => {
|
||||
tracing::warn!("late distributed delete lock batch request failed: {}", err);
|
||||
warn!("late distributed delete lock batch request failed: {}", err);
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::warn!("late distributed delete lock batch task join failed: {}", err);
|
||||
warn!("late distributed delete lock batch task join failed: {}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2428,7 +2507,7 @@ impl SetDisks {
|
||||
} else {
|
||||
Some(async move {
|
||||
if let Err(err) = client.release_locks_batch(&lock_ids).await {
|
||||
tracing::warn!(
|
||||
warn!(
|
||||
client_idx,
|
||||
lock_count = lock_ids.len(),
|
||||
"failed to cleanup late distributed delete locks in batch: {}",
|
||||
@@ -2490,7 +2569,7 @@ impl SetDisks {
|
||||
} else {
|
||||
Some(async move {
|
||||
if let Err(err) = client.release_locks_batch(&lock_ids).await {
|
||||
tracing::warn!(
|
||||
warn!(
|
||||
client_idx,
|
||||
lock_count = lock_ids.len(),
|
||||
"failed to release distributed delete locks in batch: {}",
|
||||
@@ -3590,7 +3669,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
if let Err(err) = dest_obj {
|
||||
return Err(to_object_err(err, vec![]));
|
||||
}
|
||||
let dest_obj = dest_obj.unwrap();
|
||||
let dest_obj = dest_obj?;
|
||||
|
||||
let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
|
||||
let mut transition_meta = (*oi.user_defined).clone();
|
||||
@@ -3723,7 +3802,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
if let Err(err) = fi {
|
||||
return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await;
|
||||
}
|
||||
let (actual_fi, _, _) = fi.unwrap();
|
||||
let (actual_fi, _, _) = fi?;
|
||||
|
||||
oi = ObjectInfo::from_file_info(&actual_fi, bucket, object, opts.versioned || opts.version_suspended);
|
||||
let ropts = put_restore_opts(bucket, object, &opts.transition.restore_request, &oi).await?;
|
||||
@@ -3735,7 +3814,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
if let Err(err) = gr {
|
||||
return set_restore_header_fn(&mut oi, Some(to_object_err(err.into(), vec![bucket, object]))).await;
|
||||
}
|
||||
let gr = gr.unwrap();
|
||||
let gr = gr?;
|
||||
let reader = BufReader::new(gr.stream);
|
||||
let hash_reader = HashReader::from_stream(reader, gr.object_info.size, gr.object_info.size, None, None, false)?;
|
||||
let mut p_reader = PutObjReader::new(hash_reader);
|
||||
@@ -4160,8 +4239,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
let tmp_part = format!("{}x{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp());
|
||||
let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}"));
|
||||
|
||||
let erasure =
|
||||
crate::erasure::coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
|
||||
let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
|
||||
let writer_setup_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
|
||||
|
||||
let mut writers = Vec::with_capacity(shuffle_disks.len());
|
||||
@@ -5854,9 +5932,7 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
|
||||
let runtime_state = disk.runtime_state();
|
||||
let offline_duration_seconds = disk.offline_duration_secs();
|
||||
let capacity_snapshot = disk.last_capacity_snapshot();
|
||||
if runtime_state.should_probe_for_admin()
|
||||
|| runtime_state == crate::disk::health_state::RuntimeDriveHealthState::Suspect
|
||||
{
|
||||
if runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect {
|
||||
match disk.disk_info(&DiskInfoOptions::default()).await {
|
||||
Ok(res) => {
|
||||
disk.record_capacity_probe(res.total, res.used, res.free);
|
||||
@@ -5948,7 +6024,7 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
|
||||
|
||||
fn build_runtime_snapshot_disk(
|
||||
endpoint: &Endpoint,
|
||||
runtime_state: crate::disk::health_state::RuntimeDriveHealthState,
|
||||
runtime_state: disk::health_state::RuntimeDriveHealthState,
|
||||
offline_duration_seconds: Option<u64>,
|
||||
capacity_snapshot: Option<(u64, u64, u64, u64)>,
|
||||
) -> rustfs_madmin::Disk {
|
||||
@@ -6811,7 +6887,7 @@ mod tests {
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let result = tokio::time::timeout(
|
||||
let result = timeout(
|
||||
Duration::from_secs(1),
|
||||
set_disks.copy_object(
|
||||
"bucket",
|
||||
@@ -6883,7 +6959,7 @@ mod tests {
|
||||
.await
|
||||
.expect("outer write lock should be acquired");
|
||||
|
||||
let result = tokio::time::timeout(
|
||||
let result = timeout(
|
||||
Duration::from_secs(1),
|
||||
set_disks.delete_object(
|
||||
"bucket",
|
||||
@@ -6921,7 +6997,7 @@ mod tests {
|
||||
.await
|
||||
.expect("outer write lock should be acquired");
|
||||
|
||||
tokio::time::timeout(
|
||||
timeout(
|
||||
Duration::from_secs(1),
|
||||
set_disks.delete_object(
|
||||
"bucket",
|
||||
@@ -6954,7 +7030,7 @@ mod tests {
|
||||
.await
|
||||
.expect("outer write lock should be acquired");
|
||||
|
||||
tokio::time::timeout(
|
||||
timeout(
|
||||
Duration::from_secs(1),
|
||||
set_disks.delete_object(
|
||||
"bucket",
|
||||
@@ -6989,7 +7065,7 @@ mod tests {
|
||||
.await
|
||||
.expect("outer write lock should be acquired");
|
||||
|
||||
let result = tokio::time::timeout(
|
||||
let result = timeout(
|
||||
Duration::from_millis(50),
|
||||
set_disks.delete_object(
|
||||
"bucket",
|
||||
@@ -7619,11 +7695,11 @@ mod tests {
|
||||
disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
|
||||
}
|
||||
|
||||
let (tx, _rx) = tokio::sync::mpsc::channel(1);
|
||||
let (tx, _rx) = mpsc::channel(1);
|
||||
let err = set_disks
|
||||
.list_path(
|
||||
CancellationToken::new(),
|
||||
crate::store::list_objects::ListPathOptions {
|
||||
ListPathOptions {
|
||||
bucket: "bucket".to_string(),
|
||||
recursive: true,
|
||||
..Default::default()
|
||||
@@ -7677,7 +7753,7 @@ mod tests {
|
||||
|
||||
disk.make_volume(bucket).await.expect("bucket should be created");
|
||||
let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
|
||||
disk.write_all(bucket, &metadata_path, bytes::Bytes::from_static(b"not-xl-meta"))
|
||||
disk.write_all(bucket, &metadata_path, Bytes::from_static(b"not-xl-meta"))
|
||||
.await
|
||||
.expect("corrupt metadata file should be written");
|
||||
|
||||
@@ -7732,7 +7808,7 @@ mod tests {
|
||||
|
||||
disk.make_volume(bucket).await.expect("bucket should be created");
|
||||
let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
|
||||
disk.write_all(bucket, &metadata_path, bytes::Bytes::from_static(b"not-an-xl-meta"))
|
||||
disk.write_all(bucket, &metadata_path, Bytes::from_static(b"not-an-xl-meta"))
|
||||
.await
|
||||
.expect("metadata file should be created");
|
||||
|
||||
@@ -7770,7 +7846,7 @@ mod tests {
|
||||
assert_eq!(walk_err, DiskError::Timeout);
|
||||
assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
|
||||
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<MetaCacheEntry>(4);
|
||||
let (tx, mut rx) = mpsc::channel::<MetaCacheEntry>(4);
|
||||
set_disks
|
||||
.list_path(
|
||||
CancellationToken::new(),
|
||||
@@ -7821,7 +7897,7 @@ mod tests {
|
||||
let object = "config/iam/sts/test/identity.json";
|
||||
|
||||
let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
|
||||
disk.write_all(RUSTFS_META_BUCKET, &metadata_path, bytes::Bytes::from_static(b"not-an-xl-meta"))
|
||||
disk.write_all(RUSTFS_META_BUCKET, &metadata_path, Bytes::from_static(b"not-an-xl-meta"))
|
||||
.await
|
||||
.expect("system path metadata file should be created");
|
||||
|
||||
@@ -7860,7 +7936,7 @@ mod tests {
|
||||
assert_eq!(walk_err, DiskError::Timeout);
|
||||
assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
|
||||
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<MetaCacheEntry>(4);
|
||||
let (tx, mut rx) = mpsc::channel::<MetaCacheEntry>(4);
|
||||
set_disks
|
||||
.list_path(
|
||||
CancellationToken::new(),
|
||||
@@ -8304,7 +8380,7 @@ mod tests {
|
||||
fi.size = payload.len() as i64;
|
||||
fi.add_object_part(1, String::new(), payload.len(), None, payload.len() as i64, None, None);
|
||||
|
||||
let erasure = crate::erasure::coding::Erasure::new_with_options(
|
||||
let erasure = coding::Erasure::new_with_options(
|
||||
fi.erasure.data_blocks,
|
||||
fi.erasure.parity_blocks,
|
||||
fi.erasure.block_size,
|
||||
@@ -8589,7 +8665,7 @@ mod tests {
|
||||
.await
|
||||
.expect("format should be saved");
|
||||
|
||||
std::mem::forget(dir);
|
||||
mem::forget(dir);
|
||||
endpoints.push(endpoint);
|
||||
disks.push(Some(disk));
|
||||
}
|
||||
@@ -8639,7 +8715,7 @@ mod tests {
|
||||
.expect("format should be saved");
|
||||
}
|
||||
|
||||
std::mem::forget(dir);
|
||||
mem::forget(dir);
|
||||
endpoints.push(endpoint);
|
||||
disks.push(Some(disk));
|
||||
}
|
||||
|
||||
@@ -221,7 +221,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn collect_list_parts_results_fails_early_when_quorum_is_impossible() {
|
||||
let started = std::time::Instant::now();
|
||||
let started = Instant::now();
|
||||
let tasks: Vec<_> = vec![
|
||||
(10_u64, Err(DiskError::DiskNotFound)),
|
||||
(15, Err(DiskError::DiskNotFound)),
|
||||
@@ -285,7 +285,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn collect_list_parts_results_fails_early_when_file_not_found_fallback_is_impossible() {
|
||||
let started = std::time::Instant::now();
|
||||
let started = Instant::now();
|
||||
let tasks: Vec<_> = vec![
|
||||
(5_u64, Err(DiskError::FileNotFound)),
|
||||
(10, Err(DiskError::FileCorrupt)),
|
||||
|
||||
@@ -1275,11 +1275,11 @@ impl SetDisks {
|
||||
})
|
||||
});
|
||||
|
||||
// Wait for all tasks to complete
|
||||
// Wait for all futures to complete
|
||||
let results = join_all(futures).await;
|
||||
|
||||
for result in results {
|
||||
match result {
|
||||
for join_result in results {
|
||||
match join_result {
|
||||
Ok((res, elapsed)) => match res {
|
||||
Ok(file_info) => {
|
||||
if let (Some(observations), Some(elapsed)) = (&mut observations, elapsed) {
|
||||
@@ -1296,13 +1296,13 @@ impl SetDisks {
|
||||
errors.push(Some(e));
|
||||
}
|
||||
},
|
||||
Err(_) => {
|
||||
let err = DiskError::Unexpected;
|
||||
if let (Some(observations), Some(fanout_start)) = (&mut observations, fanout_start) {
|
||||
observations.push(MetadataFanoutObservation::from_error(&err, fanout_start.elapsed()));
|
||||
Err(_join_err) => {
|
||||
// A spawned task panicked — treat as unexpected disk error
|
||||
if let Some(observations) = &mut observations {
|
||||
observations.push(MetadataFanoutObservation::from_error(&DiskError::Unexpected, Duration::ZERO));
|
||||
}
|
||||
ress.push(FileInfo::default());
|
||||
errors.push(Some(err));
|
||||
errors.push(Some(DiskError::Unexpected));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1946,7 +1946,7 @@ impl SetDisks {
|
||||
object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds"
|
||||
);
|
||||
|
||||
let erasure = crate::erasure::coding::Erasure::new_with_options(
|
||||
let erasure = coding::Erasure::new_with_options(
|
||||
fi.erasure.data_blocks,
|
||||
fi.erasure.parity_blocks,
|
||||
fi.erasure.block_size,
|
||||
@@ -1997,12 +1997,11 @@ impl SetDisks {
|
||||
);
|
||||
|
||||
let checksum_info = fi.erasure.get_checksum_info(part_number);
|
||||
let checksum_algo =
|
||||
if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S {
|
||||
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
|
||||
HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
let read_length = till_offset.saturating_sub(read_offset);
|
||||
|
||||
// Read zero-copy configuration from environment variable
|
||||
@@ -2288,7 +2287,7 @@ impl SetDisks {
|
||||
) -> Result<GetCodecStreamingReaderBuildOutcome> {
|
||||
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
|
||||
|
||||
let erasure = crate::erasure::coding::Erasure::new_with_options(
|
||||
let erasure = coding::Erasure::new_with_options(
|
||||
fi.erasure.data_blocks,
|
||||
fi.erasure.parity_blocks,
|
||||
fi.erasure.block_size,
|
||||
@@ -2372,7 +2371,7 @@ impl SetDisks {
|
||||
fi: &FileInfo,
|
||||
files: &[FileInfo],
|
||||
disks: &[Option<DiskStore>],
|
||||
erasure: &crate::erasure::coding::Erasure,
|
||||
erasure: &coding::Erasure,
|
||||
part_number: usize,
|
||||
part_offset: usize,
|
||||
part_length: usize,
|
||||
@@ -2383,9 +2382,8 @@ impl SetDisks {
|
||||
return Err(Error::other("codec streaming reader part length exceeds part size"));
|
||||
}
|
||||
let checksum_info = fi.erasure.get_checksum_info(part_number);
|
||||
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S
|
||||
{
|
||||
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
|
||||
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
|
||||
HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
@@ -2437,24 +2435,19 @@ impl SetDisks {
|
||||
}
|
||||
|
||||
let readers = reader_setup.readers;
|
||||
let source =
|
||||
crate::erasure::coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification(
|
||||
readers,
|
||||
erasure.clone(),
|
||||
part_offset,
|
||||
part_size,
|
||||
Some(metrics_path),
|
||||
read_costs,
|
||||
);
|
||||
let source = coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification(
|
||||
readers,
|
||||
erasure.clone(),
|
||||
part_offset,
|
||||
part_size,
|
||||
Some(metrics_path),
|
||||
read_costs,
|
||||
);
|
||||
let engine = build_get_codec_streaming_decode_engine(erasure.clone())?;
|
||||
let reader = crate::erasure::coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(
|
||||
source,
|
||||
engine,
|
||||
part_length,
|
||||
metrics_path,
|
||||
)?;
|
||||
let reader =
|
||||
coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?;
|
||||
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(
|
||||
crate::erasure::coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path),
|
||||
coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path),
|
||||
)))
|
||||
}
|
||||
}
|
||||
@@ -2637,7 +2630,7 @@ mod metadata_cache_tests {
|
||||
"read-repair submission should not wait for admission response"
|
||||
);
|
||||
|
||||
tokio::time::timeout(Duration::from_secs(1), async {
|
||||
timeout(Duration::from_secs(1), async {
|
||||
while SLOW_READ_REPAIR_SUBMITTER_CALLS.load(Ordering::Relaxed) == 0 {
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
}
|
||||
@@ -2666,7 +2659,7 @@ mod metadata_cache_tests {
|
||||
)
|
||||
.await;
|
||||
|
||||
let released_key = tokio::time::timeout(Duration::from_secs(1), async {
|
||||
let released_key = timeout(Duration::from_secs(1), async {
|
||||
loop {
|
||||
if let Some(key) = reserve_read_repair_heal(&bucket, "object", None, 0, 0).await {
|
||||
break key;
|
||||
@@ -3717,7 +3710,7 @@ mod tests {
|
||||
);
|
||||
|
||||
let mut remote_fi = fi;
|
||||
remote_fi.transition_status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string();
|
||||
remote_fi.transition_status = TRANSITION_COMPLETE.to_string();
|
||||
let remote = codec_streaming_test_object_info(&remote_fi);
|
||||
assert_eq!(
|
||||
codec_streaming_reader_gate_for_test(&None, &remote, &remote_fi, true).decision,
|
||||
@@ -3841,7 +3834,7 @@ mod tests {
|
||||
#[test]
|
||||
fn codec_streaming_decode_engine_builder_selects_rustfs() {
|
||||
temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || {
|
||||
let erasure = crate::erasure::coding::Erasure::new(4, 2, 32);
|
||||
let erasure = coding::Erasure::new(4, 2, 32);
|
||||
let engine = build_get_codec_streaming_decode_engine(erasure).expect("engine should be built");
|
||||
|
||||
assert!(matches!(engine, CodecStreamingDecodeEngine::Rustfs(_)));
|
||||
@@ -3851,17 +3844,11 @@ mod tests {
|
||||
#[test]
|
||||
fn codec_streaming_metrics_path_matches_selected_engine() {
|
||||
temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, None::<&str>, || {
|
||||
assert_eq!(
|
||||
get_codec_streaming_metrics_path(),
|
||||
crate::diagnostics::get::GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE
|
||||
);
|
||||
assert_eq!(get_codec_streaming_metrics_path(), GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE);
|
||||
});
|
||||
|
||||
temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || {
|
||||
assert_eq!(
|
||||
get_codec_streaming_metrics_path(),
|
||||
crate::diagnostics::get::GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE
|
||||
);
|
||||
assert_eq!(get_codec_streaming_metrics_path(), GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -4077,7 +4064,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn collect_read_multiple_results_fails_early_when_quorum_is_impossible() {
|
||||
let started = std::time::Instant::now();
|
||||
let started = Instant::now();
|
||||
let resp = ReadMultipleResp {
|
||||
bucket: "bucket".to_string(),
|
||||
prefix: "prefix".to_string(),
|
||||
@@ -4095,14 +4082,14 @@ mod tests {
|
||||
]
|
||||
.into_iter()
|
||||
.map(|(delay_ms, outcome)| async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
|
||||
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
|
||||
outcome
|
||||
})
|
||||
.collect();
|
||||
|
||||
let result = collect_read_multiple_results(tasks, 2).await;
|
||||
assert!(result.is_err(), "quorum should become impossible before slow tail completes");
|
||||
assert!(started.elapsed() < std::time::Duration::from_millis(120));
|
||||
assert!(started.elapsed() < Duration::from_millis(120));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -4124,7 +4111,7 @@ mod tests {
|
||||
]
|
||||
.into_iter()
|
||||
.map(|(delay_ms, outcome)| async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
|
||||
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
|
||||
outcome
|
||||
})
|
||||
.collect();
|
||||
@@ -4152,7 +4139,7 @@ mod tests {
|
||||
.map(|(delay_ms, should_panic)| {
|
||||
let resp = resp.clone();
|
||||
async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
|
||||
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
|
||||
if should_panic {
|
||||
panic!("simulated task panic");
|
||||
}
|
||||
@@ -4170,7 +4157,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn collect_read_parts_results_fails_early_when_quorum_is_impossible() {
|
||||
let started = std::time::Instant::now();
|
||||
let started = Instant::now();
|
||||
let part = ObjectPartInfo {
|
||||
number: 1,
|
||||
etag: "etag".to_string(),
|
||||
@@ -4184,14 +4171,14 @@ mod tests {
|
||||
]
|
||||
.into_iter()
|
||||
.map(|(delay_ms, outcome)| async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
|
||||
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
|
||||
outcome
|
||||
})
|
||||
.collect();
|
||||
|
||||
let result = collect_read_parts_results(tasks, 2).await;
|
||||
assert!(result.is_err(), "quorum should become impossible before slow tail completes");
|
||||
assert!(started.elapsed() < std::time::Duration::from_millis(120));
|
||||
assert!(started.elapsed() < Duration::from_millis(120));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -4209,7 +4196,7 @@ mod tests {
|
||||
]
|
||||
.into_iter()
|
||||
.map(|(delay_ms, outcome)| async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
|
||||
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
|
||||
outcome
|
||||
})
|
||||
.collect();
|
||||
@@ -4232,7 +4219,7 @@ mod tests {
|
||||
.map(|(delay_ms, should_panic)| {
|
||||
let part = part.clone();
|
||||
async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
|
||||
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
|
||||
if should_panic {
|
||||
panic!("simulated task panic");
|
||||
}
|
||||
|
||||
@@ -24,9 +24,7 @@ impl SetDisks {
|
||||
) -> Result<()> {
|
||||
let mut oi = obj_info.clone();
|
||||
oi.metadata_only = true;
|
||||
|
||||
Arc::make_mut(&mut oi.user_defined).remove(X_AMZ_RESTORE.as_str());
|
||||
|
||||
let version_id = oi.version_id.map(|v| v.to_string());
|
||||
let _obj = self
|
||||
.copy_object(
|
||||
|
||||
Reference in New Issue
Block a user