gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 11:56:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(get): add codec engine ab matrix (#3940)

Browse Source 
* upgrade version

* feat(get): add codec engine ab matrix

* chore(get): drop unrelated dependency drift

* upgrade version

* upgrade version

* fix cargo deny
This commit is contained in:
houseme
2026-06-27 13:27:16 +08:00
committed by GitHub
parent 996e58fd9a
commit 58b76a3d45
8 changed files with 440 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/diagnostics/get.rs
+4
View File
@@ -17,6 +17,8 @@ use crate::error::StorageError;
use std::io;
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_LEGACY_DUPLEX: &str = "legacy_duplex";
pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";
@@ -216,6 +218,8 @@ mod tests {
assert_eq!(GetObjectFailureReason::DecodeError.as_str(), "decode_error");
assert_eq!(GET_READER_BUFFER_OUTPUT, "output");
assert_eq!(GET_READER_BUFFER_PREFETCH, "prefetch");
assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, "codec_streaming_legacy_engine");
assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, "codec_streaming_rustfs_engine");
assert_eq!(GET_READER_PREFETCH_DIRECT, "direct");
assert_eq!(GET_READER_PREFETCH_STORED, "stored");
assert_eq!(GET_READER_PREFETCH_EOF, "eof");
crates/ecstore/src/erasure/coding/decode_reader.rs
+34 -49
View File
@@ -43,6 +43,7 @@ pub(crate) struct ErasureDecodeReader<S, E>
where
E: ErasureDecodeEngine,
{
metrics_path: &'static str,
source: Option<S>,
engine: E,
workspace: Option<E::Workspace>,
@@ -62,6 +63,15 @@ where
E: ErasureDecodeEngine + Clone + Send + Sync + 'static,
{
pub(crate) fn new(source: S, engine: E, total_length: usize) -> io::Result<Self> {
Self::new_with_metrics_path(source, engine, total_length, GET_OBJECT_PATH_CODEC_STREAMING)
}
pub(crate) fn new_with_metrics_path(
source: S,
engine: E,
total_length: usize,
metrics_path: &'static str,
) -> io::Result<Self> {
if engine.data_shards() == 0 {
return Err(io::Error::new(ErrorKind::InvalidInput, "erasure reader requires data shards"));
}
@@ -73,6 +83,7 @@ where
let workspace = engine.prepare_workspace(shard_len)?;
Ok(Self {
metrics_path,
source: Some(source),
engine,
workspace: Some(workspace),
@@ -97,25 +108,26 @@ where
};
let engine = self.engine.clone();
let metrics_path = self.metrics_path;
let remaining = self.remaining;
self.fill = Some(tokio::spawn(async move {
let fill_stage_start = Instant::now();
let stripe_read_stage_start = Instant::now();
let state = source.read_next_stripe().await;
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_STRIPE_READ,
stripe_read_stage_start.elapsed().as_secs_f64(),
);
let decode_stage_start = Instant::now();
let result = decode_stripe(&engine, &mut workspace, state, remaining);
let result = decode_stripe(metrics_path, &engine, &mut workspace, state, remaining);
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_DECODE,
decode_stage_start.elapsed().as_secs_f64(),
);
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_FILL,
fill_stage_start.elapsed().as_secs_f64(),
);
@@ -153,8 +165,8 @@ where
if buf.is_empty() && self.remaining > 0 {
return Poll::Ready(Err(DiskError::LessData.into()));
}
rustfs_io_metrics::record_get_object_reader_stripe(GET_OBJECT_PATH_CODEC_STREAMING);
rustfs_io_metrics::record_get_object_reader_bytes(GET_OBJECT_PATH_CODEC_STREAMING, buf.len());
rustfs_io_metrics::record_get_object_reader_stripe(self.metrics_path);
rustfs_io_metrics::record_get_object_reader_bytes(self.metrics_path, buf.len());
self.remaining -= buf.len();
Poll::Ready(Ok(Some(buf)))
}
@@ -182,7 +194,7 @@ where
Poll::Ready(result) => {
if let Some(started_at) = self.prefetch_wait_started_at.take() {
rustfs_io_metrics::record_get_object_reader_prefetch_wait(
GET_OBJECT_PATH_CODEC_STREAMING,
self.metrics_path,
started_at.elapsed().as_secs_f64(),
);
}
@@ -194,48 +206,28 @@ where
match fill {
Ok(Some(buf)) => {
if self.output_pos < self.output_buf.len() {
rustfs_io_metrics::record_get_object_reader_prefetch(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_READER_PREFETCH_STORED,
);
rustfs_io_metrics::record_get_object_reader_buffer(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_READER_BUFFER_PREFETCH,
buf.len(),
);
rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_STORED);
rustfs_io_metrics::record_get_object_reader_buffer(self.metrics_path, GET_READER_BUFFER_PREFETCH, buf.len());
self.prefetched_buf = Some(buf);
} else {
rustfs_io_metrics::record_get_object_reader_prefetch(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_READER_PREFETCH_DIRECT,
);
rustfs_io_metrics::record_get_object_reader_buffer(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_READER_BUFFER_OUTPUT,
buf.len(),
);
rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_DIRECT);
rustfs_io_metrics::record_get_object_reader_buffer(self.metrics_path, GET_READER_BUFFER_OUTPUT, buf.len());
self.output_buf = buf;
self.output_pos = 0;
}
Poll::Ready(Ok(()))
}
Ok(None) => {
rustfs_io_metrics::record_get_object_reader_prefetch(GET_OBJECT_PATH_CODEC_STREAMING, GET_READER_PREFETCH_EOF);
rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_EOF);
Poll::Ready(Ok(()))
}
Err(err) => {
if self.output_pos < self.output_buf.len() {
rustfs_io_metrics::record_get_object_reader_prefetch(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_READER_PREFETCH_ERROR_DEFERRED,
);
rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_ERROR_DEFERRED);
self.prefetch_error = Some(err);
Poll::Ready(Ok(()))
} else {
rustfs_io_metrics::record_get_object_reader_prefetch(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_READER_PREFETCH_ERROR_IMMEDIATE,
);
rustfs_io_metrics::record_get_object_reader_prefetch(self.metrics_path, GET_READER_PREFETCH_ERROR_IMMEDIATE);
Poll::Ready(Err(err))
}
}
@@ -281,7 +273,7 @@ where
self.output_pos += copy_len;
if copy_len > 0 {
rustfs_io_metrics::record_get_object_reader_copy(
GET_OBJECT_PATH_CODEC_STREAMING,
self.metrics_path,
copy_len,
read_buf_remaining_before,
output_remaining_before,
@@ -292,11 +284,7 @@ where
}
if let Some(next_buf) = self.prefetched_buf.take() {
rustfs_io_metrics::record_get_object_reader_buffer(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_READER_BUFFER_OUTPUT,
next_buf.len(),
);
rustfs_io_metrics::record_get_object_reader_buffer(self.metrics_path, GET_READER_BUFFER_OUTPUT, next_buf.len());
self.output_buf = next_buf;
self.output_pos = 0;
continue;
@@ -380,6 +368,7 @@ where
}
fn decode_stripe<E>(
metrics_path: &'static str,
engine: &E,
workspace: &mut E::Workspace,
state: StripeReadState,
@@ -398,14 +387,14 @@ where
let reconstruct_stage_start = Instant::now();
if state.data_shards_complete(engine.data_shards()) {
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start.elapsed().as_secs_f64(),
);
let emit_stage_start = Instant::now();
let output = emit_data_shards(&state, engine.data_shards(), engine.block_size(), remaining)?;
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_EMIT,
emit_stage_start.elapsed().as_secs_f64(),
);
@@ -415,14 +404,14 @@ where
let (mut shards, _errs) = state.into_parts();
if let Err(err) = engine.reconstruct_into(&mut shards, workspace) {
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start.elapsed().as_secs_f64(),
);
return Err(err);
}
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start.elapsed().as_secs_f64(),
);
@@ -446,11 +435,7 @@ where
let copy_len = shard.len().min(remaining - output.len());
output.extend_from_slice(&shard[..copy_len]);
}
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_STAGE_EMIT,
emit_stage_start.elapsed().as_secs_f64(),
);
rustfs_io_metrics::record_get_object_stage_duration(metrics_path, GET_STAGE_EMIT, emit_stage_start.elapsed().as_secs_f64());
Ok(Some(output))
}
crates/ecstore/src/set_disk/mod.rs
+11 -2
View File
@@ -24,8 +24,10 @@ use crate::bucket::versioning_sys::BucketVersioningSys;
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_EMPTY, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION,
GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error, record_get_object_pipeline_failure,
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,
record_get_object_pipeline_failure,
};
use crate::disk::error_reduce::{
BUCKET_OP_IGNORED_ERRS, OBJECT_OP_IGNORED_ERRS, build_write_quorum_failure_summary, count_errs, reduce_read_quorum_errs,
@@ -410,6 +412,13 @@ fn build_get_codec_streaming_decode_engine(erasure: coding::Erasure) -> std::io:
}
}
fn get_codec_streaming_metrics_path() -> &'static str {
match get_codec_streaming_engine() {
GetCodecStreamingEngine::Legacy => GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE,
GetCodecStreamingEngine::Rustfs => GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE,
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum GetCodecStreamingDecision {
Use,
crates/ecstore/src/set_disk/read.rs
+28 -9
View File
@@ -2244,8 +2244,9 @@ impl SetDisks {
BitrotReaderSetupMode::VerifyReconstruction,
)
.await;
let metrics_path = get_codec_streaming_metrics_path();
rustfs_io_metrics::record_get_object_stage_duration(
GET_OBJECT_PATH_CODEC_STREAMING,
metrics_path,
GET_STAGE_READER_SETUP,
reader_setup_stage_start.elapsed().as_secs_f64(),
);
@@ -2254,14 +2255,10 @@ impl SetDisks {
if available_shards < erasure.data_shards {
if let Some(read_err) = reduce_read_quorum_errs(&reader_setup.errors, OBJECT_OP_IGNORED_ERRS, erasure.data_shards) {
let reason = classify_disk_error(&read_err);
record_get_object_pipeline_failure_for_path(GET_OBJECT_PATH_CODEC_STREAMING, GET_STAGE_READER_SETUP, reason);
record_get_object_pipeline_failure_for_path(metrics_path, GET_STAGE_READER_SETUP, reason);
return Err(to_object_err(read_err.into(), vec![bucket, object]));
}
record_get_object_pipeline_failure_for_path(
GET_OBJECT_PATH_CODEC_STREAMING,
GET_STAGE_READER_SETUP,
GetObjectFailureReason::ReadQuorum,
);
record_get_object_pipeline_failure_for_path(metrics_path, GET_STAGE_READER_SETUP, GetObjectFailureReason::ReadQuorum);
return Err(Error::other(format!("not enough disks to read: {:?}", reader_setup.errors)));
}
@@ -2287,11 +2284,16 @@ impl SetDisks {
erasure.clone(),
0,
part_size,
Some(GET_OBJECT_PATH_CODEC_STREAMING),
Some(metrics_path),
read_costs,
);
let engine = build_get_codec_streaming_decode_engine(erasure)?;
let reader = crate::erasure::coding::decode_reader::ErasureDecodeReader::new(source, engine, part_length)?;
let reader = crate::erasure::coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(
source,
engine,
part_length,
metrics_path,
)?;
Ok(Box::new(crate::erasure::coding::decode_reader::SyncErasureDecodeReader::new(reader)))
}
}
@@ -3362,6 +3364,23 @@ 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
);
});
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
);
});
}
#[test]
fn codec_streaming_fallback_metric_labels_are_stable() {
assert_eq!(GetCodecStreamingFallbackReason::Disabled.as_str(), "disabled");