mirror of
https://github.com/rustfs/rustfs.git
synced 2026-11-15 11:30:31 +00:00
feat(obs): split PUT commit and remote rename timing (#9416)
* feat(obs): split PUT commit and remote rename timing Measure the app commit-owner scheduling gap and split remote rename client preparation into compatibility encoding, client acquisition, and request digest stages. Keep the timers behind the existing detailed PUT diagnostics gate. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * bench(filemeta): measure inline rename RPC encoding Compare JSON compatibility, msgpack, and dual encoding for metadata-only and representative inline FileInfo payloads used by remote rename requests. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(bench): use exact inline rename shard fixture Model the 100 KiB EC12 inline data as one rounded data shard plus its HighwayHash header, and report metadata-only throughput as operations. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser Zhi
@@ -2699,9 +2699,28 @@ impl RemoteDisk {
|
||||
"rename_data",
|
||||
|| async {
|
||||
let prepare_started = rustfs_io_metrics::put_stage_timer();
|
||||
let file_info = compat_json(fi)?;
|
||||
let file_info_bin = encode_file_info_msgpack(fi)?;
|
||||
let mut client = self.get_client().await?;
|
||||
let collect_prepare_timing = prepare_started.is_some();
|
||||
let json_encode_started = collect_prepare_timing.then(Instant::now);
|
||||
let file_info = compat_json(fi);
|
||||
rustfs_io_metrics::record_put_object_stage_duration_from(
|
||||
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_JSON_ENCODE,
|
||||
json_encode_started,
|
||||
);
|
||||
let file_info = file_info?;
|
||||
let msgpack_encode_started = collect_prepare_timing.then(Instant::now);
|
||||
let file_info_bin = encode_file_info_msgpack(fi);
|
||||
rustfs_io_metrics::record_put_object_stage_duration_from(
|
||||
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_MSGPACK_ENCODE,
|
||||
msgpack_encode_started,
|
||||
);
|
||||
let file_info_bin = file_info_bin?;
|
||||
let get_client_started = collect_prepare_timing.then(Instant::now);
|
||||
let client = self.get_client().await;
|
||||
rustfs_io_metrics::record_put_object_stage_duration_from(
|
||||
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_GET_CLIENT,
|
||||
get_client_started,
|
||||
);
|
||||
let mut client = client?;
|
||||
let mut request = Request::new(RenameDataRequest {
|
||||
bucket_incarnation_id: crate::store::bucket_heal_scope_for_object(dst_volume, dst_path)
|
||||
.map(|scope| scope.incarnation.as_bytes().to_vec().into())
|
||||
@@ -2717,24 +2736,35 @@ impl RemoteDisk {
|
||||
.map(|token| token.as_bytes().to_vec().into())
|
||||
.unwrap_or_default(),
|
||||
});
|
||||
let canonical_body = rustfs_protos::canonical_rename_data_request_body(request.get_ref());
|
||||
let canonical_body_digest_started = collect_prepare_timing.then(Instant::now);
|
||||
let incarnation_bound = !request.get_ref().bucket_incarnation_id.is_empty();
|
||||
let quota_fenced = rustfs_utils::http::metadata_compat::contains_key_str(
|
||||
&fi.metadata,
|
||||
crate::disk::QUOTA_MUTATION_FENCE_METADATA_SUFFIX,
|
||||
);
|
||||
if pool_meta_cas {
|
||||
let body = canonical_body.map_err(|_| Error::other("pool metadata CAS body length cannot be represented"))?;
|
||||
crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &body).map_err(Error::other)?;
|
||||
} else if quota_fenced {
|
||||
attach_quota_mutation_body_digest(&mut request, canonical_body)?;
|
||||
} else if multipart_rename || scanner_publication_lease_token.is_some() || incarnation_bound {
|
||||
let canonical_body =
|
||||
canonical_body.map_err(|_| Error::other("rename_data request length cannot be represented"))?;
|
||||
crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &canonical_body).map_err(Error::other)?;
|
||||
} else {
|
||||
attach_mutation_body_digest(&mut request, canonical_body, "rename_data")?;
|
||||
}
|
||||
let canonical_body_digest = (|| -> Result<()> {
|
||||
let canonical_body = rustfs_protos::canonical_rename_data_request_body(request.get_ref());
|
||||
if pool_meta_cas {
|
||||
let body =
|
||||
canonical_body.map_err(|_| Error::other("pool metadata CAS body length cannot be represented"))?;
|
||||
crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &body).map_err(Error::other)?;
|
||||
} else if quota_fenced {
|
||||
attach_quota_mutation_body_digest(&mut request, canonical_body)?;
|
||||
} else if multipart_rename || scanner_publication_lease_token.is_some() || incarnation_bound {
|
||||
let canonical_body =
|
||||
canonical_body.map_err(|_| Error::other("rename_data request length cannot be represented"))?;
|
||||
crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &canonical_body)
|
||||
.map_err(Error::other)?;
|
||||
} else {
|
||||
attach_mutation_body_digest(&mut request, canonical_body, "rename_data")?;
|
||||
}
|
||||
Ok(())
|
||||
})();
|
||||
rustfs_io_metrics::record_put_object_stage_duration_from(
|
||||
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_CANONICAL_BODY_DIGEST,
|
||||
canonical_body_digest_started,
|
||||
);
|
||||
canonical_body_digest?;
|
||||
rustfs_io_metrics::record_put_object_stage_duration_from(
|
||||
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_PREPARE,
|
||||
prepare_started,
|
||||
|
||||
@@ -13,8 +13,9 @@
|
||||
// limitations under the License.
|
||||
|
||||
use bytes::Bytes;
|
||||
use criterion::{BatchSize, BenchmarkId, Criterion, SamplingMode, criterion_group, criterion_main};
|
||||
use criterion::{BatchSize, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group, criterion_main};
|
||||
use rustfs_filemeta::{ErasureAlgo, FileInfo, FileMeta, ObjectPartInfo};
|
||||
use serde::Serialize;
|
||||
use std::hint::black_box;
|
||||
use std::time::Duration;
|
||||
use time::OffsetDateTime;
|
||||
@@ -22,6 +23,8 @@ use uuid::Uuid;
|
||||
|
||||
const VERSION_COUNT_CASES: &[usize] = &[1, 8, 32, 64];
|
||||
const BENCH_BASE_TIME_UNIX_SECS: i64 = 1_700_000_000;
|
||||
// One EC12 data shard plus the 32-byte inline HighwayHash header.
|
||||
const PUT_100_KIB_EC12_INLINE_DATA_SIZE: usize = (100_usize * 1024).div_ceil(12) + 32;
|
||||
|
||||
fn make_file_info(version_id: Uuid, data_dir: Uuid, size: i64, mod_time: OffsetDateTime) -> FileInfo {
|
||||
FileInfo {
|
||||
@@ -189,12 +192,73 @@ fn bench_delete_version_batches(c: &mut Criterion) {
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn encode_file_info_msgpack(file_info: &FileInfo) -> Vec<u8> {
|
||||
let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(1024));
|
||||
file_info.serialize(&mut serializer).expect("serialize FileInfo as msgpack");
|
||||
serializer.into_inner()
|
||||
}
|
||||
|
||||
fn inline_rpc_file_info(data_size: usize) -> FileInfo {
|
||||
let mut info = make_file_info(
|
||||
Uuid::from_u128(1),
|
||||
Uuid::from_u128(2),
|
||||
100 * 1024,
|
||||
OffsetDateTime::from_unix_timestamp(BENCH_BASE_TIME_UNIX_SECS).expect("fixture timestamp"),
|
||||
);
|
||||
info.erasure.data_blocks = 12;
|
||||
info.erasure.parity_blocks = 4;
|
||||
info.erasure.index = 1;
|
||||
info.erasure.distribution = (1..=16).collect();
|
||||
info.parts.push(ObjectPartInfo {
|
||||
number: 1,
|
||||
size: 100 * 1024,
|
||||
actual_size: 100 * 1024,
|
||||
..Default::default()
|
||||
});
|
||||
if data_size > 0 {
|
||||
info.set_inline_data();
|
||||
info.data = Some(Bytes::from(vec![42; data_size]));
|
||||
}
|
||||
info
|
||||
}
|
||||
|
||||
fn bench_rpc_file_info_encoding(c: &mut Criterion) {
|
||||
for (kind, data_size) in [
|
||||
("metadata-only", 0_usize),
|
||||
("inline-100k-ec12", PUT_100_KIB_EC12_INLINE_DATA_SIZE),
|
||||
] {
|
||||
let file_info = inline_rpc_file_info(data_size);
|
||||
let json_bytes = serde_json::to_vec(&file_info).expect("fixture JSON").len();
|
||||
let msgpack_bytes = encode_file_info_msgpack(&file_info).len();
|
||||
let mut group = c.benchmark_group(format!("rpc_file_info_encode/{kind}"));
|
||||
if data_size == 0 {
|
||||
group.throughput(Throughput::Elements(1));
|
||||
} else {
|
||||
group.throughput(Throughput::Bytes(u64::try_from(data_size).expect("fixture size")));
|
||||
}
|
||||
group.bench_function(BenchmarkId::new("json", json_bytes), |b| {
|
||||
b.iter(|| black_box(serde_json::to_string(black_box(&file_info)).expect("encode JSON")))
|
||||
});
|
||||
group.bench_function(BenchmarkId::new("msgpack", msgpack_bytes), |b| {
|
||||
b.iter(|| black_box(encode_file_info_msgpack(black_box(&file_info))))
|
||||
});
|
||||
group.bench_function(BenchmarkId::new("dual", json_bytes + msgpack_bytes), |b| {
|
||||
b.iter(|| {
|
||||
let json = serde_json::to_string(black_box(&file_info)).expect("encode JSON");
|
||||
let msgpack = encode_file_info_msgpack(black_box(&file_info));
|
||||
black_box((json, msgpack))
|
||||
})
|
||||
});
|
||||
group.finish();
|
||||
}
|
||||
}
|
||||
|
||||
criterion_group! {
|
||||
name = benches;
|
||||
config = Criterion::default()
|
||||
.sample_size(20)
|
||||
.warm_up_time(Duration::from_millis(300))
|
||||
.measurement_time(Duration::from_secs(1));
|
||||
targets = bench_rename_data_meta_path, bench_delete_version_batches
|
||||
targets = bench_rename_data_meta_path, bench_delete_version_batches, bench_rpc_file_info_encoding
|
||||
}
|
||||
criterion_main!(benches);
|
||||
|
||||
@@ -260,6 +260,7 @@ pub const PUT_STAGE_PUT_OBJECT_QUOTA_BEGIN_SNAPSHOT_ADMISSION: &str = "put_objec
|
||||
pub const PUT_STAGE_PUT_OBJECT_QUOTA_BEGIN_OBJECT_STORE_LOOKUP: &str = "put_object_quota_begin_object_store_lookup";
|
||||
pub const PUT_STAGE_PUT_OBJECT_QUOTA_BEGIN_CONTEXT_BUILD: &str = "put_object_quota_begin_context_build";
|
||||
pub const PUT_STAGE_PUT_OBJECT_COMMIT_SUBMIT_TO_CLOSURE_ENTER: &str = "put_object_commit_submit_to_closure_enter";
|
||||
pub const PUT_STAGE_APP_PUT_COMMIT_OWNER_SUBMIT_TO_START: &str = "app_put_commit_owner_submit_to_start";
|
||||
pub const PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_ENTER_TO_FIRST_AWAIT: &str = "put_object_commit_closure_enter_to_first_await";
|
||||
pub const PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_TOTAL: &str = "put_object_commit_closure_total";
|
||||
pub const PUT_STAGE_PUT_OBJECT_COMMIT_BEFORE_RENAME: &str = "put_object_commit_before_rename";
|
||||
@@ -359,6 +360,11 @@ pub const PUT_STAGE_SET_DISK_RENAME_DISK_WAIT: &str = "set_disk_rename_disk_wait
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_DISK_WAIT_LOCAL: &str = "set_disk_rename_disk_wait_local";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_DISK_WAIT_REMOTE: &str = "set_disk_rename_disk_wait_remote";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_PREPARE: &str = "set_disk_rename_remote_client_prepare";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_JSON_ENCODE: &str = "set_disk_rename_remote_client_json_encode";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_MSGPACK_ENCODE: &str = "set_disk_rename_remote_client_msgpack_encode";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_GET_CLIENT: &str = "set_disk_rename_remote_client_get_client";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_CANONICAL_BODY_DIGEST: &str =
|
||||
"set_disk_rename_remote_client_canonical_body_digest";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_RPC: &str = "set_disk_rename_remote_client_rpc";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_RPC_AWAIT: &str = "set_disk_rename_remote_client_rpc_await";
|
||||
pub const PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_RPC_AWAIT_POLL_PENDING_COUNT: &str =
|
||||
@@ -4027,6 +4033,7 @@ mod tests {
|
||||
PUT_STAGE_PUT_OBJECT_QUOTA_BEGIN_SNAPSHOT_ADMISSION,
|
||||
PUT_STAGE_PUT_OBJECT_QUOTA_BEGIN_OBJECT_STORE_LOOKUP,
|
||||
PUT_STAGE_PUT_OBJECT_QUOTA_BEGIN_CONTEXT_BUILD,
|
||||
PUT_STAGE_APP_PUT_COMMIT_OWNER_SUBMIT_TO_START,
|
||||
PUT_STAGE_PUT_OBJECT_COMMIT_SUBMIT_TO_CLOSURE_ENTER,
|
||||
PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_ENTER_TO_FIRST_AWAIT,
|
||||
PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_TOTAL,
|
||||
@@ -4106,6 +4113,10 @@ mod tests {
|
||||
PUT_STAGE_SET_DISK_RENAME_DISK_WAIT_LOCAL,
|
||||
PUT_STAGE_SET_DISK_RENAME_DISK_WAIT_REMOTE,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_PREPARE,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_JSON_ENCODE,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_MSGPACK_ENCODE,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_GET_CLIENT,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_CANONICAL_BODY_DIGEST,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_RPC,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_RPC_AWAIT,
|
||||
PUT_STAGE_SET_DISK_RENAME_REMOTE_CLIENT_RPC_AWAIT_FIRST_PENDING_TO_READY,
|
||||
@@ -4164,6 +4175,7 @@ mod tests {
|
||||
|| *stage == PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_ACQUIRED_TO_COMMIT_READY
|
||||
|| stage.starts_with("put_object_commit_ready_")
|
||||
|| stage.starts_with("put_object_quota_begin_")
|
||||
|| *stage == PUT_STAGE_APP_PUT_COMMIT_OWNER_SUBMIT_TO_START
|
||||
|| *stage == PUT_STAGE_PUT_OBJECT_COMMIT_SUBMIT_TO_CLOSURE_ENTER
|
||||
|| *stage == PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_ENTER_TO_FIRST_AWAIT
|
||||
|| *stage == PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_TOTAL
|
||||
|
||||
@@ -97,7 +97,7 @@ Use one fixed binary and the four cells below to distinguish the two factors and
|
||||
|
||||
Before A1, record the acceptable baseline drift, minimum gain above measurement noise, and p99, RSS and error limits. Verify A/A stability, then compare each candidate with the default through A1/B1/B2/A2. Include sustained fresh 100 KiB PUT and short burst/idle/burst traffic. Track the dev-to-1.0.5 change and the beta.10 gap as separate outcomes.
|
||||
|
||||
Collect detailed diagnostics in dedicated rounds with `RUSTFS_OBS_PUT_STAGE_METRICS_ENABLED=true` and metrics export enabled. Include `disk_prepared_metadata_fdatasync` for inline `xl.meta`, `disk_fsync_worker_dispatch_wait`, `disk_fsync_worker_execute`, destination and ancestor directory sync stages, and actual worker/spin activity. Operation execution timers cover the job closure; idle-spin diagnostics cover the loop between jobs. Heavy syscall tracing and on/off-CPU profiling belong in these diagnostic rounds. Use the same lightweight observation settings for all acceptance rounds.
|
||||
Collect detailed diagnostics in dedicated rounds with `RUSTFS_OBS_PUT_STAGE_METRICS_ENABLED=true` and metrics export enabled. Include `app_put_commit_owner_submit_to_start`, the `set_disk_rename_remote_client_*` preparation stages, `disk_prepared_metadata_fdatasync` for inline `xl.meta`, `disk_fsync_worker_dispatch_wait`, `disk_fsync_worker_execute`, destination and ancestor directory sync stages, and actual worker/spin activity. The app owner timer includes submission and runtime scheduling before the outer spawned commit task starts. The lower `put_object_commit_submit_to_closure_enter` timer covers a different ECStore owner, so the two timers must not be added as one wait. Remote-client preparation is split into JSON compatibility encoding, msgpack encoding, client acquisition, and canonical request-body/digest preparation. Completed errors are included in those four sub-stage samples, while cancellation can drop an in-flight timer. Request construction and metric publication remain in the preparation total outside the four sub-stages, so their histograms can have different sample counts and must not be added or subtracted to reconstruct an unmeasured residual. Operation execution timers cover the job closure; idle-spin diagnostics cover the loop between jobs. Heavy syscall tracing and on/off-CPU profiling belong in these diagnostic rounds. Use the same lightweight observation settings for all acceptance rounds.
|
||||
|
||||
| Metric | Meaning |
|
||||
| --- | --- |
|
||||
|
||||
@@ -2316,7 +2316,12 @@ impl DefaultObjectUsecase {
|
||||
rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_context(),
|
||||
put_commit_future,
|
||||
);
|
||||
let commit_owner_submit_started = rustfs_io_metrics::put_stage_timer();
|
||||
let put_commit = spawn_traced_join(async move {
|
||||
rustfs_io_metrics::record_put_object_stage_duration_from(
|
||||
rustfs_io_metrics::PUT_STAGE_APP_PUT_COMMIT_OWNER_SUBMIT_TO_START,
|
||||
commit_owner_submit_started,
|
||||
);
|
||||
let commit_stage_start = put_stage_metrics_enabled.then(Instant::now);
|
||||
let result = std::panic::AssertUnwindSafe(put_commit_future).catch_unwind().await;
|
||||
rustfs_io_metrics::record_put_object_stage_duration_from("store_commit", commit_stage_start);
|
||||
|
||||
Reference in New Issue
Block a user