mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-16 18:08:21 +00:00
perf(ecstore): borrow rename metadata during commit fanout (#6104)
* perf(ecstore): borrow rename metadata during commit fanout Co-Authored-By: heihutu <heihutu@gmail.com> * fix(ecstore): preserve rename_data API compatibility Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: Zhengchao An <anzhengchao@gmail.com>
This commit is contained in:
@@ -13,6 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use super::super::*;
|
||||
use crate::disk::disk_store::DiskStoreRenameDataExt;
|
||||
use crate::io_support::bitrot::object_mmap_read_enabled;
|
||||
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
|
||||
use tracing::trace;
|
||||
@@ -1164,10 +1165,10 @@ impl SetDisks {
|
||||
let rename_result = if should_fail_heal_rename(bucket, object, index) {
|
||||
Err(DiskError::Unexpected)
|
||||
} else {
|
||||
disk.rename_data(
|
||||
disk.rename_data_borrowed(
|
||||
RUSTFS_META_TMP_BUCKET,
|
||||
&tmp_id,
|
||||
parts_metadata[index].clone(),
|
||||
&parts_metadata[index],
|
||||
bucket,
|
||||
object,
|
||||
)
|
||||
|
||||
@@ -2527,11 +2527,12 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
// The trailing `_` drops the rename_data old-size backfill
|
||||
// (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
|
||||
// `get_object_info` lookup, so the backfill has no consumer here yet.
|
||||
let rename_result = SetDisks::rename_data(
|
||||
Self::assign_rename_data_indexes(&mut parts_metadatas);
|
||||
let rename_result = SetDisks::rename_data_owned(
|
||||
&commit_disks,
|
||||
RUSTFS_META_MULTIPART_BUCKET,
|
||||
&commit_upload_id_path,
|
||||
&parts_metadatas,
|
||||
parts_metadatas,
|
||||
&commit_bucket,
|
||||
&commit_object,
|
||||
write_quorum,
|
||||
@@ -2550,10 +2551,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
if rename_result.is_ok() {
|
||||
quota_reservation.commit().await;
|
||||
}
|
||||
let (online_disks, convergence, op_old_dir, cleanup_disks, _) = match rename_result {
|
||||
let rename_commit = match rename_result {
|
||||
Ok(result) => result,
|
||||
Err(err) => return Err(err.into()),
|
||||
};
|
||||
let online_disks = rename_commit.online_disks;
|
||||
let convergence = rename_commit.convergence;
|
||||
let op_old_dir = rename_commit.data_dir;
|
||||
let cleanup_disks = rename_commit.cleanup_disks;
|
||||
let committed_file_info = rename_commit.committed_file_info;
|
||||
|
||||
// Detach admission before any post-commit await: client cancellation
|
||||
// must not couple durable convergence repair to cleanup work.
|
||||
@@ -2598,9 +2604,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
return Err(StorageError::Unexpected);
|
||||
}
|
||||
|
||||
if let Some(committed_slot) = online_disks.iter().position(Option::is_some) {
|
||||
fi = parts_metadatas[committed_slot].clone();
|
||||
}
|
||||
fi = committed_file_info;
|
||||
let committed_dir = fi.data_dir.unwrap_or_default().to_string();
|
||||
|
||||
commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
|
||||
|
||||
@@ -97,10 +97,6 @@ fn duration_millis_f64(duration: std::time::Duration) -> f64 {
|
||||
duration.as_secs_f64() * 1000.0
|
||||
}
|
||||
|
||||
fn committed_response_metadata_slot<D>(committed_disks: &[Option<D>], fallback_slot: usize) -> usize {
|
||||
committed_disks.iter().position(Option::is_some).unwrap_or(fallback_slot)
|
||||
}
|
||||
|
||||
pub(in crate::set_disk::ops) fn assign_object_transaction_epoch(
|
||||
shuffle_disks: &[Option<DiskStore>],
|
||||
parts_metadatas: &mut [FileInfo],
|
||||
@@ -239,38 +235,6 @@ mod duration_metrics_tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod put_metadata_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn committed_file_info_follows_exact_quorum_success_slot() {
|
||||
let mut first_success = FileInfo::new("bucket/object", 2, 2);
|
||||
first_success.name = "first-success".to_string();
|
||||
let mut second_success = first_success.clone();
|
||||
second_success.name = "second-success".to_string();
|
||||
let mut parts_metadata = [FileInfo::default(), first_success, second_success, FileInfo::default()];
|
||||
let committed_disks = [None, Some(()), Some(()), None];
|
||||
|
||||
assert_eq!(
|
||||
committed_disks.iter().filter(|disk| disk.is_some()).count(),
|
||||
2,
|
||||
"fixture must meet exact quorum"
|
||||
);
|
||||
let selected_slot = committed_response_metadata_slot(&committed_disks, 3);
|
||||
let selected = std::mem::take(&mut parts_metadata[selected_slot]);
|
||||
|
||||
assert_eq!(selected.name, "first-success");
|
||||
assert_eq!(parts_metadata[1], FileInfo::default(), "selected metadata should move without cloning");
|
||||
assert_eq!(parts_metadata[2].name, "second-success", "other committed metadata must remain available");
|
||||
assert_eq!(
|
||||
committed_response_metadata_slot::<()>(&[None, None, None, None], 3),
|
||||
3,
|
||||
"a violated post-commit success-mask invariant must not turn a durable PUT into an error"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn is_restore_control_metadata(key: &str) -> bool {
|
||||
key.eq_ignore_ascii_case(X_AMZ_RESTORE.as_str())
|
||||
|| key.eq_ignore_ascii_case(rustfs_utils::http::headers::AMZ_RESTORE_EXPIRY_DAYS)
|
||||
@@ -2236,11 +2200,12 @@ impl SetDisks {
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
let rename_result = SetDisks::rename_data(
|
||||
Self::assign_rename_data_indexes(&mut parts_metadatas);
|
||||
let rename_result = SetDisks::rename_data_owned(
|
||||
&commit_disks,
|
||||
RUSTFS_META_TMP_BUCKET,
|
||||
commit_tmp_dir.as_str(),
|
||||
&parts_metadatas,
|
||||
parts_metadatas,
|
||||
&commit_bucket,
|
||||
&commit_object,
|
||||
write_quorum,
|
||||
@@ -2259,7 +2224,7 @@ impl SetDisks {
|
||||
if rename_result.is_ok() {
|
||||
quota_reservation.commit().await;
|
||||
}
|
||||
let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = match rename_result {
|
||||
let rename_commit = match rename_result {
|
||||
Ok(commit) => commit,
|
||||
Err(err) => {
|
||||
if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
|
||||
@@ -2276,6 +2241,12 @@ impl SetDisks {
|
||||
return Err(err.into());
|
||||
}
|
||||
};
|
||||
let online_disks = rename_commit.online_disks;
|
||||
let convergence = rename_commit.convergence;
|
||||
let op_old_dir = rename_commit.data_dir;
|
||||
let cleanup_disks = rename_commit.cleanup_disks;
|
||||
let old_current_size = rename_commit.old_current_size;
|
||||
let mut fi = rename_commit.committed_file_info;
|
||||
// Do this before any post-commit await so request cancellation cannot
|
||||
// bypass best-effort admission. A process crash before admission
|
||||
// remains subject to the existing scanner reconciliation path.
|
||||
@@ -2384,9 +2355,6 @@ impl SetDisks {
|
||||
}
|
||||
}
|
||||
|
||||
let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
|
||||
let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
|
||||
|
||||
if is_compressed {
|
||||
record_compression_total_memory(actual_size as u64, w_size as u64).await;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user