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

perf: avoid eager parity reader setup (#4133)

Browse Source
This commit is contained in:
houseme
2026-07-02 08:49:08 +08:00
committed by GitHub
parent 18d89d1e31
commit 7a075c91da
13 changed files with 3159 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/erasure/coding/decode.rs
+4
View File
@@ -50,6 +50,10 @@ fn get_shard_locality_preference_enabled() -> bool {
)
}
pub(crate) fn should_collect_shard_read_costs() -> bool {
rustfs_io_metrics::get_stage_metrics_enabled() || get_shard_locality_preference_enabled()
}
/// Number of stripes to prefetch in the legacy decode path.
/// When > 1, stripe reads are batched to overlap disk I/O with decode.
/// Default: 1 (no prefetch, current behavior).
crates/ecstore/src/set_disk/mod.rs
+131 -4
View File
@@ -347,6 +347,8 @@ const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE: bool = false; // Disabled until
const ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: usize = MI_B;
const ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE: usize = MI_B;
const ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = "RUSTFS_GET_CODEC_STREAMING_ENGINE";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = GET_CODEC_STREAMING_ENGINE_LEGACY;
@@ -366,6 +368,11 @@ const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE: bool = false;
const ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS: &str = "RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS: usize = 256;
const ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE: bool = false;
const ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE: usize = 512 * 1024;
const ENV_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY: &str = "RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY";
const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY: bool = false;
const ENV_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: &str = "RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD";
@@ -648,7 +655,57 @@ pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool {
}
fn get_codec_streaming_min_size() -> usize {
rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE)
if std::env::var_os(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE).is_some() {
return rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE);
}
match get_codec_streaming_engine() {
GetCodecStreamingEngine::Rustfs => rustfs_utils::get_env_usize(
ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE,
),
GetCodecStreamingEngine::Legacy => DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE,
}
}
fn is_get_codec_streaming_data_blocks_first_enabled() -> bool {
#[cfg(test)]
{
rustfs_utils::get_env_bool(
ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE,
)
}
#[cfg(not(test))]
{
static CACHED: OnceLock<bool> = OnceLock::new();
*CACHED.get_or_init(|| {
rustfs_utils::get_env_bool(
ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE,
)
})
}
}
fn get_codec_streaming_data_blocks_first_max_size() -> usize {
#[cfg(test)]
{
rustfs_utils::get_env_usize(
ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE,
)
}
#[cfg(not(test))]
{
static CACHED: OnceLock<usize> = OnceLock::new();
*CACHED.get_or_init(|| {
rustfs_utils::get_env_usize(
ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE,
)
})
}
}
fn get_object_metadata_cache_max_entries() -> usize {
@@ -840,6 +897,7 @@ impl GetCodecStreamingObjectClass {
struct GetCodecStreamingGate {
object_class: GetCodecStreamingObjectClass,
decision: GetCodecStreamingDecision,
prefer_data_blocks_first_reader_setup: bool,
}
fn record_get_codec_streaming_gate_decision(
@@ -946,6 +1004,24 @@ fn is_get_small_object_direct_memory_eligible(
)
}
fn should_prefer_codec_streaming_data_blocks_first_reader_setup(
object_class: GetCodecStreamingObjectClass,
object_size: i64,
) -> bool {
if !is_get_codec_streaming_data_blocks_first_enabled()
|| object_class != GetCodecStreamingObjectClass::PlainSinglePart
|| object_size <= 0
{
return false;
}
let Ok(object_size) = usize::try_from(object_size) else {
return false;
};
let max_size = get_codec_streaming_data_blocks_first_max_size();
max_size > 0 && object_size <= max_size
}
fn get_codec_streaming_reader_gate(
bucket: &str,
object: &str,
@@ -960,42 +1036,49 @@ fn get_codec_streaming_reader_gate(
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled),
prefer_data_blocks_first_reader_setup: false,
};
}
if !get_codec_streaming_rollout().is_opted_in() {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutNotOptedIn),
prefer_data_blocks_first_reader_setup: false,
};
}
if !should_use_codec_streaming(bucket, object) {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutPctNotSelected),
prefer_data_blocks_first_reader_setup: false,
};
}
if !is_get_codec_streaming_body_compat_confirmed() {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed),
prefer_data_blocks_first_reader_setup: false,
};
}
if !is_get_codec_streaming_header_compat_confirmed() {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::HeaderCompatibilityUnconfirmed),
prefer_data_blocks_first_reader_setup: false,
};
}
if object_class == GetCodecStreamingObjectClass::Range {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Range),
prefer_data_blocks_first_reader_setup: false,
};
}
if !lock_optimization_enabled {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::LockOptimizationDisabled),
prefer_data_blocks_first_reader_setup: false,
};
}
@@ -1003,30 +1086,35 @@ fn get_codec_streaming_reader_gate(
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize),
prefer_data_blocks_first_reader_setup: false,
};
};
if object_info.size < min_size {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize),
prefer_data_blocks_first_reader_setup: false,
};
}
if object_class == GetCodecStreamingObjectClass::Encrypted {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted),
prefer_data_blocks_first_reader_setup: false,
};
}
if object_class == GetCodecStreamingObjectClass::Compressed {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed),
prefer_data_blocks_first_reader_setup: false,
};
}
if object_class == GetCodecStreamingObjectClass::Remote {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote),
prefer_data_blocks_first_reader_setup: false,
};
}
if object_class == GetCodecStreamingObjectClass::Multipart {
@@ -1034,12 +1122,14 @@ fn get_codec_streaming_reader_gate(
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart),
prefer_data_blocks_first_reader_setup: false,
};
}
if fi.parts.len() > get_codec_streaming_multipart_max_parts() {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::MultipartPartLimit),
prefer_data_blocks_first_reader_setup: false,
};
}
}
@@ -1047,6 +1137,10 @@ fn get_codec_streaming_reader_gate(
GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Use,
prefer_data_blocks_first_reader_setup: should_prefer_codec_streaming_data_blocks_first_reader_setup(
object_class,
object_info.size,
),
}
}
@@ -1694,6 +1788,14 @@ fn classify_multipart_part_write_path(object_size: i64, block_size: usize) -> Sm
}
}
fn known_put_object_storage_size(data_size: i64) -> i64 {
if data_size >= 0 {
data_size
} else {
HashReader::SIZE_PRESERVE_LAYER
}
}
#[allow(clippy::too_many_arguments)]
async fn build_inline_bitrot_readers(
files: &[FileInfo],
@@ -2274,6 +2376,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
self.set_index,
self.pool_index,
opts.skip_verify_bitrot,
object_class.as_str(),
size_bucket,
codec_streaming_gate.prefer_data_blocks_first_reader_setup,
)
.await?
{
@@ -2450,10 +2555,11 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
let result: Result<ObjectInfo> = async {
let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let put_object_size = known_put_object_storage_size(data.size());
let is_inline_buffer =
runtime_sources::storage_class_should_inline(erasure.shard_file_size(data.size()), opts.versioned);
runtime_sources::storage_class_should_inline(erasure.shard_file_size(put_object_size), opts.versioned);
let shard_file_size = erasure.shard_file_size(data.size());
let shard_file_size = erasure.shard_file_size(put_object_size);
let shard_size = erasure.shard_size();
let writer_setup_stage_start = Instant::now();
let writer_futs: Vec<_> = shuffle_disks
@@ -2531,7 +2637,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
);
let write_path = classify_put_write_path(is_inline_buffer, data.size(), fi.erasure.block_size);
let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
rustfs_io_metrics::record_put_object_path(write_path.metric_label());
let encode_stage_start = Instant::now();
@@ -9894,12 +10000,33 @@ mod tests {
classify_put_write_path(false, 32 * 1024 * 1024, 1024 * 1024),
SmallWritePath::Pipeline
));
assert!(matches!(
classify_put_write_path(false, 31 * 1024 * 1024, 1024 * 1024),
SmallWritePath::Pipeline
));
assert!(matches!(
classify_put_write_path(true, 64 * 1024 * 1024, 1024 * 1024),
SmallWritePath::Pipeline
));
}
#[test]
fn put_object_classification_uses_only_known_storage_size() {
assert_eq!(known_put_object_storage_size(42), 42);
assert_eq!(
known_put_object_storage_size(HashReader::SIZE_PRESERVE_LAYER),
HashReader::SIZE_PRESERVE_LAYER
);
assert!(matches!(
classify_put_write_path(false, known_put_object_storage_size(HashReader::SIZE_PRESERVE_LAYER), 1024 * 1024),
SmallWritePath::Pipeline
));
assert!(matches!(
classify_put_write_path(false, known_put_object_storage_size(1024 * 1024), 1024 * 1024),
SmallWritePath::SingleBlockNonInline
));
}
#[test]
fn put_object_part_fast_path_selection_matches_single_block_non_inline_rules() {
assert!(should_use_single_block_non_inline_fast_path(false, 4096, 4096));
crates/ecstore/src/set_disk/read.rs
+481 -61
View File
@@ -843,6 +843,13 @@ struct BitrotReaderSetup {
deferred_count: usize,
}
#[derive(Clone, Copy)]
struct BitrotReaderSetupAttribution {
path: &'static str,
object_class: &'static str,
size_bucket: &'static str,
}
#[derive(Clone, Copy)]
enum BitrotReaderSetupMode {
ReadQuorum,
@@ -883,7 +890,8 @@ fn get_bitrot_reader_setup_strategy(mode: BitrotReaderSetupMode, prefer_data_blo
BitrotReaderSetupStrategy::DataBlocksFirst
}
BitrotReaderSetupMode::VerifyReconstruction
if rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP, false) =>
if prefer_data_blocks_first
|| rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP, false) =>
{
BitrotReaderSetupStrategy::DataBlocksFirst
}
@@ -955,6 +963,14 @@ impl BitrotReaderSetup {
}
}
fn scheduling_target(&self, data_shards: usize, parity_shards: usize, mode: BitrotReaderSetupMode) -> usize {
match mode {
BitrotReaderSetupMode::ReadQuorum => data_shards,
BitrotReaderSetupMode::VerifyReconstruction if !self.data_shards_attempted(data_shards) => data_shards,
BitrotReaderSetupMode::VerifyReconstruction => self.reconstruction_verification_target(data_shards, parity_shards),
}
}
fn deferred_shards(&self) -> usize {
self.deferred_count
}
@@ -1105,16 +1121,49 @@ fn record_bitrot_reader_setup_fanout(
strategy: BitrotReaderSetupStrategy,
mode: BitrotReaderSetupMode,
setup: &BitrotReaderSetup,
attribution: Option<BitrotReaderSetupAttribution>,
) {
rustfs_io_metrics::record_get_object_reader_setup_fanout(
strategy.as_str(),
mode.as_str(),
setup.scheduled_shards(),
setup.attempted_shards(),
setup.available_shards(),
setup.completed_failed_shards(),
setup.deferred_shards(),
);
let strategy = strategy.as_str();
let mode = mode.as_str();
let scheduled = setup.scheduled_shards();
let attempted = setup.attempted_shards();
let ready = setup.available_shards();
let failed = setup.completed_failed_shards();
let deferred = setup.deferred_shards();
rustfs_io_metrics::record_get_object_reader_setup_fanout(strategy, mode, scheduled, attempted, ready, failed, deferred);
if let Some(attribution) = attribution {
rustfs_io_metrics::record_get_object_reader_setup_fanout_by_size(
attribution.path,
strategy,
mode,
attribution.object_class,
attribution.size_bucket,
scheduled,
attempted,
ready,
failed,
deferred,
);
}
}
fn record_bitrot_reader_setup_strategy(
strategy: BitrotReaderSetupStrategy,
mode: BitrotReaderSetupMode,
attribution: Option<BitrotReaderSetupAttribution>,
) {
let strategy = strategy.as_str();
let mode = mode.as_str();
rustfs_io_metrics::record_get_object_reader_setup_strategy(strategy, mode);
if let Some(attribution) = attribution {
rustfs_io_metrics::record_get_object_reader_setup_strategy_by_size(
attribution.path,
strategy,
mode,
attribution.object_class,
attribution.size_bucket,
);
}
}
#[allow(clippy::too_many_arguments)]
@@ -1134,13 +1183,14 @@ async fn create_bitrot_readers_until_quorum_all_shards(
parity_shards: usize,
mode: BitrotReaderSetupMode,
stage_metrics: Option<BitrotReaderStageMetrics>,
attribution: Option<BitrotReaderSetupAttribution>,
) -> BitrotReaderSetup {
let strategy = BitrotReaderSetupStrategy::AllShards;
let mut setup = BitrotReaderSetup::new(disks.len());
let mut reader_tasks = FuturesUnordered::new();
let stage_metrics = stage_metrics.filter(|_| rustfs_io_metrics::get_stage_metrics_enabled());
rustfs_io_metrics::record_get_object_reader_setup_strategy(strategy.as_str(), mode.as_str());
record_bitrot_reader_setup_strategy(strategy, mode, attribution);
let schedule_stage_start = stage_metrics.map(|_| Instant::now());
for (idx, disk_op) in disks.iter().enumerate() {
@@ -1207,7 +1257,7 @@ async fn create_bitrot_readers_until_quorum_all_shards(
if let Some(stage_metrics) = stage_metrics {
record_get_stage_duration_if_enabled(stage_metrics.path, GET_STAGE_READER_SETUP_DROP_PENDING, drop_pending_stage_start);
}
record_bitrot_reader_setup_fanout(strategy, mode, &setup);
record_bitrot_reader_setup_fanout(strategy, mode, &setup, attribution);
setup
}
@@ -1229,6 +1279,7 @@ async fn create_bitrot_readers_until_quorum(
parity_shards: usize,
mode: BitrotReaderSetupMode,
stage_metrics: Option<BitrotReaderStageMetrics>,
attribution: Option<BitrotReaderSetupAttribution>,
) -> BitrotReaderSetup {
create_bitrot_readers_until_quorum_with_preference(
files,
@@ -1247,6 +1298,7 @@ async fn create_bitrot_readers_until_quorum(
mode,
false,
stage_metrics,
attribution,
)
.await
}
@@ -1269,6 +1321,7 @@ async fn create_bitrot_readers_until_quorum_with_preference(
mode: BitrotReaderSetupMode,
prefer_data_blocks_first: bool,
stage_metrics: Option<BitrotReaderStageMetrics>,
attribution: Option<BitrotReaderSetupAttribution>,
) -> BitrotReaderSetup {
let strategy = get_bitrot_reader_setup_strategy(mode, prefer_data_blocks_first);
if strategy == BitrotReaderSetupStrategy::AllShards {
@@ -1288,6 +1341,7 @@ async fn create_bitrot_readers_until_quorum_with_preference(
parity_shards,
mode,
stage_metrics,
attribution,
)
.await;
}
@@ -1297,10 +1351,11 @@ async fn create_bitrot_readers_until_quorum_with_preference(
let total_shards = disks.len();
let stage_metrics = stage_metrics.filter(|_| rustfs_io_metrics::get_stage_metrics_enabled());
rustfs_io_metrics::record_get_object_reader_setup_strategy(strategy.as_str(), mode.as_str());
record_bitrot_reader_setup_strategy(strategy, mode, attribution);
let schedule_stage_start = stage_metrics.map(|_| Instant::now());
for idx in 0..data_shards.min(total_shards) {
let initial_target = setup.setup_target(data_shards, parity_shards, mode);
for idx in 0..initial_target.min(data_shards).min(total_shards) {
schedule_bitrot_reader_task(
&mut reader_tasks,
&mut setup,
@@ -1319,26 +1374,6 @@ async fn create_bitrot_readers_until_quorum_with_preference(
stage_metrics,
);
}
if data_shards < total_shards {
schedule_bitrot_reader_task(
&mut reader_tasks,
&mut setup,
data_shards,
files,
disks,
bucket,
object,
part_number,
read_offset,
read_length,
shard_size,
checksum_algo.clone(),
skip_verify_bitrot,
use_mmap_read,
stage_metrics,
);
}
if let Some(stage_metrics) = stage_metrics {
record_get_stage_duration_if_enabled(stage_metrics.path, GET_STAGE_READER_SETUP_SCHEDULE, schedule_stage_start);
}
@@ -1351,7 +1386,7 @@ async fn create_bitrot_readers_until_quorum_with_preference(
break;
}
let target = setup.setup_target(data_shards, parity_shards, mode);
let target = setup.scheduling_target(data_shards, parity_shards, mode);
while setup.available_shards().saturating_add(setup.pending_scheduled_shards()) < target {
let Some(next_idx) = next_unscheduled_reader_index(&setup, total_shards, data_shards) else {
break;
@@ -1401,7 +1436,7 @@ async fn create_bitrot_readers_until_quorum_with_preference(
if let Some(stage_metrics) = stage_metrics {
record_get_stage_duration_if_enabled(stage_metrics.path, GET_STAGE_READER_SETUP_DROP_PENDING, drop_pending_stage_start);
}
record_bitrot_reader_setup_fanout(strategy, mode, &setup);
record_bitrot_reader_setup_fanout(strategy, mode, &setup, attribution);
setup
}
@@ -1475,7 +1510,7 @@ async fn create_data_block_bitrot_readers(
// The direct-memory path only consumes the data shard readers. If one of
// them is missing, the caller falls back to the regular GET path.
record_bitrot_reader_setup_fanout(strategy, BitrotReaderSetupMode::ReadQuorum, &setup);
record_bitrot_reader_setup_fanout(strategy, BitrotReaderSetupMode::ReadQuorum, &setup, None);
setup
}
@@ -2765,10 +2800,12 @@ impl SetDisks {
let use_mmap_read = object_mmap_read_enabled();
let reader_setup_stage_start = Instant::now();
let read_costs = disks
.iter()
.map(|disk| shard_read_cost_for_disk(disk.as_ref()))
.collect::<Vec<_>>();
let read_costs = coding::decode::should_collect_shard_read_costs().then(|| {
disks
.iter()
.map(|disk| shard_read_cost_for_disk(disk.as_ref()))
.collect::<Vec<_>>()
});
let reader_setup = create_bitrot_readers_until_quorum_with_preference(
&files,
&disks,
@@ -2786,6 +2823,11 @@ impl SetDisks {
BitrotReaderSetupMode::ReadQuorum,
prefer_data_blocks_first_reader_setup,
None,
Some(BitrotReaderSetupAttribution {
path: metrics_path,
object_class: metrics_object_class,
size_bucket: metrics_size_bucket,
}),
)
.await;
let reader_setup_elapsed = reader_setup_stage_start.elapsed();
@@ -2926,9 +2968,13 @@ impl SetDisks {
let decode_stage_start = Instant::now();
let unattempted_data_shards = !reader_setup.data_shards_attempted(erasure.data_shards);
let readers = reader_setup.readers;
let (written, err) = erasure
.decode_with_read_costs(writer, readers, part_offset, part_length, part_size, read_costs)
.await;
let (written, err) = if let Some(read_costs) = read_costs {
erasure
.decode_with_read_costs(writer, readers, part_offset, part_length, part_size, read_costs)
.await
} else {
erasure.decode(writer, readers, part_offset, part_length, part_size).await
};
let decode_elapsed = decode_stage_start.elapsed();
rustfs_io_metrics::record_get_object_decode_duration(decode_elapsed.as_secs_f64());
rustfs_io_metrics::record_get_object_stage_duration_by_size(
@@ -3056,6 +3102,9 @@ impl SetDisks {
_set_index: usize,
_pool_index: usize,
skip_verify_bitrot: bool,
metrics_object_class: &'static str,
metrics_size_bucket: &'static str,
prefer_data_blocks_first_reader_setup: bool,
) -> Result<GetCodecStreamingReaderBuildOutcome> {
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
@@ -3082,6 +3131,9 @@ impl SetDisks {
part_length,
part.size,
skip_verify_bitrot,
metrics_object_class,
metrics_size_bucket,
prefer_data_blocks_first_reader_setup,
)
.await;
}
@@ -3121,6 +3173,9 @@ impl SetDisks {
part.size,
part.size,
skip_verify_bitrot,
metrics_object_class,
metrics_size_bucket,
false,
)
.await?
{
@@ -3149,6 +3204,9 @@ impl SetDisks {
part_length: usize,
part_size: usize,
skip_verify_bitrot: bool,
metrics_object_class: &'static str,
metrics_size_bucket: &'static str,
prefer_data_blocks_first_reader_setup: bool,
) -> Result<GetCodecStreamingReaderBuildOutcome> {
if part_length > part_size {
return Err(Error::other("codec streaming reader part length exceeds part size"));
@@ -3173,11 +3231,13 @@ impl SetDisks {
bitrot_reader_init_stage: GET_STAGE_READER_TASK_BITROT_READER_INIT,
});
let reader_setup_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let read_costs = disks
.iter()
.map(|disk| shard_read_cost_for_disk(disk.as_ref()))
.collect::<Vec<_>>();
let reader_setup = create_bitrot_readers_until_quorum(
let read_costs = coding::decode::should_collect_shard_read_costs().then(|| {
disks
.iter()
.map(|disk| shard_read_cost_for_disk(disk.as_ref()))
.collect::<Vec<_>>()
});
let reader_setup = create_bitrot_readers_until_quorum_with_preference(
files,
disks,
bucket,
@@ -3192,7 +3252,13 @@ impl SetDisks {
erasure.data_shards,
erasure.parity_shards,
BitrotReaderSetupMode::VerifyReconstruction,
prefer_data_blocks_first_reader_setup,
reader_stage_metrics,
Some(BitrotReaderSetupAttribution {
path: metrics_path,
object_class: metrics_object_class,
size_bucket: metrics_size_bucket,
}),
)
.await;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READER_SETUP, reader_setup_stage_start);
@@ -3214,14 +3280,24 @@ impl SetDisks {
}
let readers = reader_setup.readers;
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 source = if let Some(read_costs) = read_costs {
coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification(
readers,
erasure.clone(),
part_offset,
part_size,
Some(metrics_path),
read_costs,
)
} else {
coding::decode::ParallelReader::new_with_metrics_path_and_reconstruction_verification(
readers,
erasure.clone(),
part_offset,
part_size,
Some(metrics_path),
)
};
let engine = build_get_codec_streaming_decode_engine(erasure.clone())?;
let reader =
coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?;
@@ -3735,7 +3811,8 @@ mod metadata_cache_tests {
#[cfg(test)]
mod tests {
use super::*;
use std::io::Cursor;
use crate::erasure::coding::BitrotWriter;
use std::io::{Cursor, ErrorKind};
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
@@ -4556,6 +4633,7 @@ mod tests {
parity_shards,
mode,
None,
None,
)
.await
},
@@ -4590,10 +4668,110 @@ mod tests {
mode,
prefer_data_blocks_first,
None,
None,
)
.await
}
fn encoded_reader_setup_fileinfo(data: Option<Vec<u8>>) -> FileInfo {
let mut fi = FileInfo::new("object", 2, 2);
fi.volume = "bucket".to_string();
fi.name = "object".to_string();
fi.size = data
.as_ref()
.map_or(0, |data| i64::try_from(data.len()).expect("test data length should fit i64"));
fi.data = data.map(Bytes::from);
fi
}
async fn setup_codec_data_blocks_first_encoded_bitrot_readers(
erasure: &coding::Erasure,
data: &[u8],
missing_indexes: &[usize],
bitrot_corrupt_indexes: &[usize],
inconsistent_source_indexes: &[usize],
hash_algo: HashAlgorithm,
) -> BitrotReaderSetup {
let shard_size = erasure.shard_size();
let encoded_shards = erasure.encode_data(data).expect("test stripe should encode");
let mut files = Vec::with_capacity(encoded_shards.len());
for (index, shard) in encoded_shards.into_iter().enumerate() {
if missing_indexes.contains(&index) {
files.push(encoded_reader_setup_fileinfo(None));
continue;
}
let mut shard = shard.to_vec();
if inconsistent_source_indexes.contains(&index)
&& let Some(byte) = shard.first_mut()
{
*byte ^= 0x80;
}
let mut writer = BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo.clone());
writer.write(&shard).await.expect("test shard should write with bitrot hash");
let mut encoded = writer.into_inner().into_inner();
if bitrot_corrupt_indexes.contains(&index) {
let data_offset = hash_algo.size();
let byte = encoded
.get_mut(data_offset)
.expect("encoded test shard should contain payload bytes");
*byte ^= 0x80;
}
files.push(encoded_reader_setup_fileinfo(Some(encoded)));
}
let disks = vec![None; files.len()];
temp_env::async_with_vars([(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP, Some("true"))], async {
create_bitrot_readers_until_quorum(
&files,
&disks,
"bucket",
"object",
1,
0,
shard_size,
shard_size,
hash_algo,
false,
false,
erasure.data_shards,
erasure.parity_shards,
BitrotReaderSetupMode::VerifyReconstruction,
None,
None,
)
.await
})
.await
}
async fn decode_codec_data_blocks_first_setup(
erasure: coding::Erasure,
data: &[u8],
readers: Vec<Option<ObjectBitrotReader>>,
) -> std::io::Result<Vec<u8>> {
let source = coding::decode::ParallelReader::new_with_metrics_path_and_reconstruction_verification(
readers,
erasure.clone(),
0,
data.len(),
Some(GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE),
);
let engine = CodecStreamingDecodeEngine::rustfs(&erasure).expect("rustfs codec engine should be created");
let mut reader = coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE,
)
.expect("codec streaming reader should be constructed");
let mut decoded = Vec::new();
reader.read_to_end(&mut decoded).await?;
Ok(decoded)
}
#[tokio::test]
async fn bitrot_reader_setup_stops_at_read_quorum() {
let setup = setup_inline_bitrot_readers(
@@ -4655,7 +4833,7 @@ mod tests {
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::ReadQuorum));
assert_eq!(setup.available_shards(), 2);
assert!(setup.scheduled_shards() < 4);
assert_eq!(setup.scheduled_shards(), 2);
assert_eq!(setup.readers.iter().filter(|reader| reader.is_some()).count(), 4);
let fallback_index = setup
@@ -4689,7 +4867,25 @@ mod tests {
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::ReadQuorum));
assert_eq!(setup.available_shards(), 2);
assert!(setup.scheduled_shards() < 4);
assert_eq!(setup.scheduled_shards(), 2);
assert_eq!(setup.deferred_shards(), 2);
}
#[tokio::test]
async fn bitrot_reader_setup_preference_can_apply_to_verify_mode() {
let setup = setup_inline_bitrot_readers_with_preference(
vec![Some(b"aaaa"), Some(b"bbbb"), Some(b"cccc"), Some(b"dddd")],
2,
2,
BitrotReaderSetupMode::VerifyReconstruction,
true,
)
.await;
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::VerifyReconstruction));
assert_eq!(setup.available_data_shards(2), 2);
assert_eq!(setup.scheduled_shards(), 2);
assert_eq!(setup.attempted_shards(), 2);
assert_eq!(setup.deferred_shards(), 2);
}
@@ -4707,6 +4903,7 @@ mod tests {
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::ReadQuorum));
assert_eq!(setup.available_shards(), 2);
assert_eq!(setup.completed_failed_shards(), 1);
assert_eq!(setup.scheduled_shards(), 3);
assert!(setup.ready.iter().skip(2).any(|ready| *ready));
}
@@ -4739,7 +4936,7 @@ mod tests {
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::VerifyReconstruction));
assert_eq!(setup.available_data_shards(2), 2);
assert!(setup.scheduled_shards() < 4);
assert_eq!(setup.scheduled_shards(), 2);
}
#[tokio::test]
@@ -4758,6 +4955,7 @@ mod tests {
assert_eq!(setup.available_data_shards(2), 1);
assert!(setup.data_shards_attempted(2));
assert_eq!(setup.completed_failed_shards(), 1);
assert_eq!(setup.scheduled_shards(), 4);
}
#[tokio::test]
@@ -4806,6 +5004,69 @@ mod tests {
assert_eq!(setup.completed_failed_shards(), 1);
}
#[tokio::test]
async fn codec_data_blocks_first_recovers_corrupt_data_with_deferred_parity() {
let erasure = coding::Erasure::new(2, 2, 64);
let data = (0..64u8).collect::<Vec<_>>();
let setup =
setup_codec_data_blocks_first_encoded_bitrot_readers(&erasure, &data, &[], &[0], &[], HashAlgorithm::HighwayHash256)
.await;
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::VerifyReconstruction));
assert_eq!(setup.scheduled_shards(), 2);
assert_eq!(setup.attempted_shards(), 2);
assert_eq!(setup.available_shards(), 2);
assert_eq!(setup.deferred_shards(), 2);
let decoded = decode_codec_data_blocks_first_setup(erasure, &data, setup.readers)
.await
.expect("deferred parity should recover the corrupt data shard");
assert_eq!(decoded, data);
}
#[tokio::test]
async fn codec_data_blocks_first_reconstructs_missing_data_shard() {
let erasure = coding::Erasure::new(2, 2, 64);
let data = (0..64u8).rev().collect::<Vec<_>>();
let setup =
setup_codec_data_blocks_first_encoded_bitrot_readers(&erasure, &data, &[0], &[], &[], HashAlgorithm::HighwayHash256)
.await;
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::VerifyReconstruction));
assert_eq!(setup.scheduled_shards(), 4);
assert_eq!(setup.available_shards(), 3);
assert_eq!(setup.available_data_shards(2), 1);
assert_eq!(setup.completed_failed_shards(), 1);
let decoded = decode_codec_data_blocks_first_setup(erasure, &data, setup.readers)
.await
.expect("parity shards should reconstruct the missing data shard");
assert_eq!(decoded, data);
}
#[tokio::test]
async fn codec_data_blocks_first_rejects_inconsistent_deferred_reconstruction_source() {
let erasure = coding::Erasure::new(2, 2, 64);
let data = (0..64u8).map(|value| value.wrapping_mul(3)).collect::<Vec<_>>();
let setup =
setup_codec_data_blocks_first_encoded_bitrot_readers(&erasure, &data, &[], &[0], &[2], HashAlgorithm::HighwayHash256)
.await;
assert!(setup.has_setup_quorum(2, 2, BitrotReaderSetupMode::VerifyReconstruction));
assert_eq!(setup.scheduled_shards(), 2);
assert_eq!(setup.attempted_shards(), 2);
assert_eq!(setup.deferred_shards(), 2);
let err = decode_codec_data_blocks_first_setup(erasure, &data, setup.readers)
.await
.expect_err("inconsistent deferred parity source must fail reconstruction verification");
assert_eq!(err.kind(), ErrorKind::InvalidData);
assert!(err.to_string().contains("inconsistent read source shards"));
}
#[test]
fn codec_streaming_reader_gate_is_conservative() {
temp_env::with_vars(
@@ -4897,6 +5158,165 @@ mod tests {
});
}
#[test]
fn rustfs_codec_streaming_uses_conservative_default_min_size() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
(ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, None::<&str>),
],
|| {
let below_threshold_fi = codec_streaming_test_fileinfo(512 * 1024, 1);
let below_threshold_object_info = codec_streaming_test_object_info(&below_threshold_fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &below_threshold_object_info, &below_threshold_fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize)
);
let threshold_fi = codec_streaming_test_fileinfo(1_048_576, 1);
let threshold_object_info = codec_streaming_test_object_info(&threshold_fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &threshold_object_info, &threshold_fi, true).decision,
GetCodecStreamingDecision::Use
);
},
);
}
#[test]
fn rustfs_codec_streaming_min_size_override_can_lower_threshold() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>),
(ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, Some("524288")),
],
|| {
let fi = codec_streaming_test_fileinfo(512 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Use
);
},
);
}
#[test]
fn codec_streaming_data_blocks_first_gate_defaults_to_off() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE, None::<&str>),
],
|| {
let fi = codec_streaming_test_fileinfo(512 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
let gate = codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true);
assert_eq!(gate.decision, GetCodecStreamingDecision::Use);
assert!(!gate.prefer_data_blocks_first_reader_setup);
},
);
}
#[test]
fn codec_streaming_data_blocks_first_gate_allows_small_plain_single_part() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE, Some("524288")),
],
|| {
let fi = codec_streaming_test_fileinfo(512 * 1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
let gate = codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true);
assert_eq!(gate.decision, GetCodecStreamingDecision::Use);
assert!(gate.prefer_data_blocks_first_reader_setup);
},
);
}
#[test]
fn codec_streaming_data_blocks_first_gate_rejects_large_and_non_plain_objects() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_MAX_SIZE, Some("524288")),
],
|| {
let large_fi = codec_streaming_test_fileinfo(768 * 1024, 1);
let large_object_info = codec_streaming_test_object_info(&large_fi);
let large_gate = codec_streaming_reader_gate_for_test(&None, &large_object_info, &large_fi, true);
assert_eq!(large_gate.decision, GetCodecStreamingDecision::Use);
assert!(!large_gate.prefer_data_blocks_first_reader_setup);
let multipart_fi = codec_streaming_test_fileinfo(512 * 1024, 2);
let multipart_object_info = codec_streaming_test_object_info(&multipart_fi);
let multipart_gate = codec_streaming_reader_gate_for_test(&None, &multipart_object_info, &multipart_fi, true);
assert!(!multipart_gate.prefer_data_blocks_first_reader_setup);
let range = Some(HTTPRangeSpec {
is_suffix_length: false,
start: 0,
end: 1,
});
let range_gate = codec_streaming_reader_gate_for_test(&range, &large_object_info, &large_fi, true);
assert!(!range_gate.prefer_data_blocks_first_reader_setup);
},
);
}
#[test]
fn generic_codec_streaming_min_size_override_remains_authoritative() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
(ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, Some("786432")),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Use
);
},
);
}
#[test]
fn codec_streaming_engine_env_is_ignored_when_streaming_is_disabled() {
temp_env::with_vars(
crates/io-metrics/src/lib.rs
+200 -2
View File
@@ -190,6 +190,9 @@ pub const GET_OBJECT_SIZE_BUCKET_LE_4_KIB: &str = "le_4kib";
pub const GET_OBJECT_SIZE_BUCKET_LE_16_KIB: &str = "le_16kib";
pub const GET_OBJECT_SIZE_BUCKET_LE_64_KIB: &str = "le_64kib";
pub const GET_OBJECT_SIZE_BUCKET_LE_128_KIB: &str = "le_128kib";
pub const GET_OBJECT_SIZE_BUCKET_LE_192_KIB: &str = "le_192kib";
pub const GET_OBJECT_SIZE_BUCKET_LE_256_KIB: &str = "le_256kib";
pub const GET_OBJECT_SIZE_BUCKET_LE_512_KIB: &str = "le_512kib";
pub const GET_OBJECT_SIZE_BUCKET_LE_1_MIB: &str = "le_1mib";
pub const GET_OBJECT_SIZE_BUCKET_GT_1_MIB: &str = "gt_1mib";
@@ -201,7 +204,10 @@ pub const fn get_object_size_bucket(size_bytes: i64) -> &'static str {
4_097..=16_384 => GET_OBJECT_SIZE_BUCKET_LE_16_KIB,
16_385..=65_536 => GET_OBJECT_SIZE_BUCKET_LE_64_KIB,
65_537..=131_072 => GET_OBJECT_SIZE_BUCKET_LE_128_KIB,
131_073..=1_048_576 => GET_OBJECT_SIZE_BUCKET_LE_1_MIB,
131_073..=196_608 => GET_OBJECT_SIZE_BUCKET_LE_192_KIB,
196_609..=262_144 => GET_OBJECT_SIZE_BUCKET_LE_256_KIB,
262_145..=524_288 => GET_OBJECT_SIZE_BUCKET_LE_512_KIB,
524_289..=1_048_576 => GET_OBJECT_SIZE_BUCKET_LE_1_MIB,
_ => GET_OBJECT_SIZE_BUCKET_GT_1_MIB,
}
}
@@ -1124,6 +1130,29 @@ pub fn record_get_object_reader_setup_strategy(strategy: &'static str, mode: &'s
.increment(1);
}
/// Record the bitrot reader setup scheduling strategy with bounded GET attribution labels.
#[inline(always)]
pub fn record_get_object_reader_setup_strategy_by_size(
path: &'static str,
strategy: &'static str,
mode: &'static str,
object_class: &'static str,
size_bucket: &'static str,
) {
if !get_stage_metrics_enabled() {
return;
}
counter!(
"rustfs_io_get_object_reader_setup_strategy_by_size_total",
"path" => path,
"strategy" => strategy,
"mode" => mode,
"object_class" => object_class,
"size_bucket" => size_bucket
)
.increment(1);
}
/// Record the final bitrot reader setup fanout shape for a GET read.
#[inline(always)]
pub fn record_get_object_reader_setup_fanout(
@@ -1170,6 +1199,71 @@ pub fn record_get_object_reader_setup_fanout(
.record(shard_read_fanout_to_f64(deferred));
}
/// Record the final bitrot reader setup fanout shape with bounded GET attribution labels.
#[inline(always)]
#[allow(clippy::too_many_arguments)]
pub fn record_get_object_reader_setup_fanout_by_size(
path: &'static str,
strategy: &'static str,
mode: &'static str,
object_class: &'static str,
size_bucket: &'static str,
scheduled: usize,
attempted: usize,
ready: usize,
failed: usize,
deferred: usize,
) {
if !get_stage_metrics_enabled() {
return;
}
histogram!(
"rustfs_io_get_object_reader_setup_scheduled_by_size",
"path" => path,
"strategy" => strategy,
"mode" => mode,
"object_class" => object_class,
"size_bucket" => size_bucket
)
.record(shard_read_fanout_to_f64(scheduled));
histogram!(
"rustfs_io_get_object_reader_setup_attempted_by_size",
"path" => path,
"strategy" => strategy,
"mode" => mode,
"object_class" => object_class,
"size_bucket" => size_bucket
)
.record(shard_read_fanout_to_f64(attempted));
histogram!(
"rustfs_io_get_object_reader_setup_ready_by_size",
"path" => path,
"strategy" => strategy,
"mode" => mode,
"object_class" => object_class,
"size_bucket" => size_bucket
)
.record(shard_read_fanout_to_f64(ready));
histogram!(
"rustfs_io_get_object_reader_setup_failed_by_size",
"path" => path,
"strategy" => strategy,
"mode" => mode,
"object_class" => object_class,
"size_bucket" => size_bucket
)
.record(shard_read_fanout_to_f64(failed));
histogram!(
"rustfs_io_get_object_reader_setup_deferred_by_size",
"path" => path,
"strategy" => strategy,
"mode" => mode,
"object_class" => object_class,
"size_bucket" => size_bucket
)
.record(shard_read_fanout_to_f64(deferred));
}
/// Record GetObject metadata resolution duration.
#[inline(always)]
pub fn record_get_object_metadata_phase_duration(duration_secs: f64) {
@@ -1458,6 +1552,72 @@ pub fn record_put_object_path(path: &'static str) {
counter!("rustfs_s3_put_object_path_total", "path" => path).increment(1);
}
#[inline(always)]
fn put_object_size_bucket(size_bytes: i64) -> &'static str {
const MI_B: i64 = 1024 * 1024;
match size_bytes {
i64::MIN..=0 => "unknown",
1..=MI_B => "le_1mib",
_ if size_bytes <= 10 * MI_B => "le_10mib",
_ if size_bytes <= 16 * MI_B => "le_16mib",
_ if size_bytes <= 32 * MI_B => "le_32mib",
_ if size_bytes <= 64 * MI_B => "le_64mib",
_ => "gt_64mib",
}
}
#[inline(always)]
fn put_object_buffer_bucket(buffer_size: usize) -> &'static str {
const KI_B: usize = 1024;
const MI_B: usize = 1024 * 1024;
match buffer_size {
0..=65536 => "le_64kib",
_ if buffer_size <= 128 * KI_B => "le_128kib",
_ if buffer_size <= 256 * KI_B => "le_256kib",
_ if buffer_size <= 512 * KI_B => "le_512kib",
_ if buffer_size <= MI_B => "le_1mib",
_ => "gt_1mib",
}
}
#[inline(always)]
fn bool_label(value: bool) -> &'static str {
if value { "true" } else { "false" }
}
#[inline(always)]
pub fn record_put_object_diagnostics(
path: &'static str,
eager_status: &'static str,
size_bytes: i64,
buffer_size: usize,
large_concurrency_tuning: bool,
) {
if !put_stage_metrics_enabled() {
return;
}
let size_bucket = put_object_size_bucket(size_bytes);
let buffer_bucket = put_object_buffer_bucket(buffer_size);
counter!(
"rustfs_s3_put_object_diagnostics_total",
"path" => path,
"eager_status" => eager_status,
"size_bucket" => size_bucket,
"buffer_bucket" => buffer_bucket,
"large_concurrency_tuning" => bool_label(large_concurrency_tuning),
)
.increment(1);
histogram!(
"rustfs_s3_put_object_selected_buffer_size_bytes",
"path" => path,
"size_bucket" => size_bucket,
)
.record(buffer_size as f64);
}
#[inline(always)]
pub fn record_put_object_stage_duration(stage: &'static str, duration_ms: f64) {
if !put_stage_metrics_enabled() {
@@ -1935,7 +2095,26 @@ mod tests {
record_get_object_shard_read_observation("codec_streaming", 0, "data", "local", "success", "none", 1024, 0.004, 0.001);
record_get_object_shard_read_cost_summary("codec_streaming", 3, 1, 2, 0, 4, 4, 4, true);
record_get_object_reader_setup_strategy("data_blocks_first", "read_quorum");
record_get_object_reader_setup_strategy_by_size(
"codec_streaming",
"data_blocks_first",
"read_quorum",
"plain_single_part",
"le_1mib",
);
record_get_object_reader_setup_fanout("data_blocks_first", "read_quorum", 3, 2, 2, 0, 2);
record_get_object_reader_setup_fanout_by_size(
"codec_streaming",
"data_blocks_first",
"read_quorum",
"plain_single_part",
"le_1mib",
3,
2,
2,
0,
2,
);
assert!(0.005_f64.is_sign_positive());
}
@@ -1977,9 +2156,22 @@ mod tests {
record_put_object_path("write_inline");
record_put_object_stage_duration("ingress_prepare", 12.5);
record_put_object_stage_duration("set_disk_encode", 8.0);
record_put_object_diagnostics("zero_copy_eager", "eligible", 32 * 1024 * 1024, 256 * 1024, true);
assert!(put_stage_metrics_enabled());
set_put_stage_metrics_enabled(false);
}
#[test]
fn test_put_object_diagnostic_buckets() {
assert_eq!(put_object_size_bucket(0), "unknown");
assert_eq!(put_object_size_bucket(10 * 1024 * 1024), "le_10mib");
assert_eq!(put_object_size_bucket(32 * 1024 * 1024), "le_32mib");
assert_eq!(put_object_size_bucket(32 * 1024 * 1024 + 1), "le_64mib");
assert_eq!(put_object_buffer_bucket(64 * 1024), "le_64kib");
assert_eq!(put_object_buffer_bucket(256 * 1024), "le_256kib");
assert_eq!(put_object_buffer_bucket(2 * 1024 * 1024), "gt_1mib");
}
#[test]
fn test_put_stage_metrics_disabled_by_default() {
let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
@@ -2098,7 +2290,13 @@ mod tests {
assert_eq!(get_object_size_bucket((16 * 1024) + 1), GET_OBJECT_SIZE_BUCKET_LE_64_KIB);
assert_eq!(get_object_size_bucket(100 * 1024), GET_OBJECT_SIZE_BUCKET_LE_128_KIB);
assert_eq!(get_object_size_bucket(128 * 1024), GET_OBJECT_SIZE_BUCKET_LE_128_KIB);
assert_eq!(get_object_size_bucket((128 * 1024) + 1), GET_OBJECT_SIZE_BUCKET_LE_1_MIB);
assert_eq!(get_object_size_bucket((128 * 1024) + 1), GET_OBJECT_SIZE_BUCKET_LE_192_KIB);
assert_eq!(get_object_size_bucket(192 * 1024), GET_OBJECT_SIZE_BUCKET_LE_192_KIB);
assert_eq!(get_object_size_bucket((192 * 1024) + 1), GET_OBJECT_SIZE_BUCKET_LE_256_KIB);
assert_eq!(get_object_size_bucket(256 * 1024), GET_OBJECT_SIZE_BUCKET_LE_256_KIB);
assert_eq!(get_object_size_bucket((256 * 1024) + 1), GET_OBJECT_SIZE_BUCKET_LE_512_KIB);
assert_eq!(get_object_size_bucket(512 * 1024), GET_OBJECT_SIZE_BUCKET_LE_512_KIB);
assert_eq!(get_object_size_bucket((512 * 1024) + 1), GET_OBJECT_SIZE_BUCKET_LE_1_MIB);
assert_eq!(get_object_size_bucket(1024 * 1024), GET_OBJECT_SIZE_BUCKET_LE_1_MIB);
assert_eq!(get_object_size_bucket((1024 * 1024) + 1), GET_OBJECT_SIZE_BUCKET_GT_1_MIB);
}