mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-25 05:26:50 +00:00
fix(ecstore): default rename fanout to parallel early-ack path (#6443)
* feat(allocator): replace mimalloc/libmimalloc-sys with rustfs-mimalloc/rustfs-mimalloc-sys Replace the upstream xonatius/mimalloc_rust.git fork (mimalloc + libmimalloc-sys) with the published rustfs-mimalloc (v0.5.0) and rustfs-mimalloc-sys (v0.5.0) crates from crates.io. The new crates are based on mimalloc V3 (v3.5.0) and provide: - MiMalloc global allocator with safe API (collect, stats_json, process_info) - Heap management and arena operations (heap module) - Full FFI bindings to mimalloc V3 Changes: - Workspace deps: mimalloc + libmimalloc-sys (git) → rustfs-mimalloc + rustfs-mimalloc-sys (crates.io) - allocator_reclaim.rs: libmimalloc_sys::mi_collect → rustfs_mimalloc::MiMalloc::collect - memory_observability.rs: raw FFI mi_stats_get_json → MiMalloc::stats_json() - main.rs: heap ownership tests use Heap::contains() (V3 API) - deny.toml: remove xonatius/mimalloc_rust.git from allow-git Co-Authored-By: heihutu <heihutu@gmail.com> * fix(ecstore): default rename fanout to parallel early-ack path Switch the default rename_data commit fanout from serial join_all to the parallel JoinSet early-ack path. The serial path (#5987) was the primary cause of the 1MiB PUT regression (-71.7%) observed in rc.3 benchmarks. A/B verification on testing 4-node cluster (c=64, 1MiB PUT, 2min): - Serial (join_all): 96.99 MiB/s, P50=644ms - Early ack (JoinSet): 177.46 MiB/s, P50=407ms (+83%) Also: - Update rename_data_reclaims_synthetic_inline_rollback_dir_after_commit to use rename_data_owned and await tail_drain for proper cleanup. - Update rename_data_waits_for_tail_disk_after_write_quorum to explicitly test the serial path (now non-default) via env override. - Add error source chain to HTTP Body stream transport error log (backlog#2005) so the underlying cause is visible. Ref: rustfs/backlog#2005 Ref: rustfs/backlog#1792#issuecomment-5384346238 Ref: rustfs/backlog#1792#issuecomment-5384370938 Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -3472,7 +3472,7 @@ type RenameDataLegacyTuple = (
|
||||
);
|
||||
|
||||
fn put_rename_early_ack_enabled() -> bool {
|
||||
rustfs_utils::get_env_bool(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, false)
|
||||
rustfs_utils::get_env_bool(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, true)
|
||||
}
|
||||
|
||||
impl RenameDataCommit {
|
||||
@@ -8631,17 +8631,21 @@ mod tests {
|
||||
inline_fi.mod_time = Some(OffsetDateTime::now_utc());
|
||||
std::fs::create_dir_all(disk_root.join(RUSTFS_META_TMP_BUCKET).join("tmp-inline"))
|
||||
.expect("inline staging dir should be created");
|
||||
SetDisks::rename_data(
|
||||
// Use rename_data_owned so we can await the tail_drain for cleanup.
|
||||
let commit = SetDisks::rename_data_owned(
|
||||
std::slice::from_ref(&online_disk),
|
||||
RUSTFS_META_TMP_BUCKET,
|
||||
"tmp-inline",
|
||||
std::slice::from_ref(&inline_fi),
|
||||
vec![inline_fi],
|
||||
bucket,
|
||||
object,
|
||||
1,
|
||||
)
|
||||
.await
|
||||
.expect("inline version should commit");
|
||||
if let Some(td) = commit.tail_drain {
|
||||
td.await.expect("inline commit tail drain must succeed");
|
||||
}
|
||||
|
||||
// Overwrite the same (nil) version with a non-inline one.
|
||||
let new_data_dir = Uuid::new_v4();
|
||||
@@ -8654,17 +8658,20 @@ mod tests {
|
||||
.join(new_data_dir.to_string());
|
||||
std::fs::create_dir_all(&staged_data_dir).expect("streaming staging dir should be created");
|
||||
std::fs::write(staged_data_dir.join("part.1"), b"streamed-body").expect("staged part should be written");
|
||||
SetDisks::rename_data(
|
||||
let commit = SetDisks::rename_data_owned(
|
||||
std::slice::from_ref(&online_disk),
|
||||
RUSTFS_META_TMP_BUCKET,
|
||||
"tmp-streaming",
|
||||
std::slice::from_ref(&streaming_fi),
|
||||
vec![streaming_fi],
|
||||
bucket,
|
||||
object,
|
||||
1,
|
||||
)
|
||||
.await
|
||||
.expect("non-inline overwrite should commit");
|
||||
if let Some(td) = commit.tail_drain {
|
||||
td.await.expect("non-inline overwrite tail drain must succeed");
|
||||
}
|
||||
|
||||
let mut leftovers: Vec<String> = std::fs::read_dir(disk_root.join(bucket).join(object))
|
||||
.expect("committed object dir should be readable")
|
||||
@@ -8825,49 +8832,54 @@ mod tests {
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(rename_quorum_ack)]
|
||||
async fn rename_data_waits_for_tail_disk_after_write_quorum() {
|
||||
const DISKS: usize = 4;
|
||||
let bucket = "rename-tail-success-bucket";
|
||||
let object = "rename-tail-success-object";
|
||||
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
|
||||
prepare_rename_source_dirs(&dirs, &disks, "source").await;
|
||||
let file_infos = rename_commit_fileinfos(object, DISKS, "tail-success-etag");
|
||||
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
|
||||
// Explicitly test the serial (join_all) path: early ack is now the
|
||||
// default, so disable it to verify the legacy behaviour still works.
|
||||
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false"))], async {
|
||||
const DISKS: usize = 4;
|
||||
let bucket = "rename-tail-success-bucket";
|
||||
let object = "rename-tail-success-object";
|
||||
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
|
||||
prepare_rename_source_dirs(&dirs, &disks, "source").await;
|
||||
let file_infos = rename_commit_fileinfos(object, DISKS, "tail-success-etag");
|
||||
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
|
||||
|
||||
let rename = SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 3);
|
||||
tokio::pin!(rename);
|
||||
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
|
||||
tokio::select! {
|
||||
() = barrier.wait_until_paused() => {}
|
||||
result = &mut rename => panic!("rename_data returned before the armed fan-out barrier: {result:?}"),
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("paused disk must reach the armed rename barrier");
|
||||
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(50), &mut rename).await.is_err(),
|
||||
"current rename_data waits for the paused fan-out disk even after the other three disks can reach write quorum"
|
||||
);
|
||||
|
||||
barrier.release();
|
||||
rename
|
||||
let rename = SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 3);
|
||||
tokio::pin!(rename);
|
||||
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
|
||||
tokio::select! {
|
||||
() = barrier.wait_until_paused() => {}
|
||||
result = &mut rename => panic!("rename_data returned before the armed fan-out barrier: {result:?}"),
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("tail success must complete the rename after the barrier is released");
|
||||
.expect("paused disk must reach the armed rename barrier");
|
||||
|
||||
for (idx, dir) in dirs.iter().enumerate() {
|
||||
let reopened = reopen_local_disk(dir).await;
|
||||
let stored = reopened
|
||||
.read_version("", bucket, object, "", &ReadOptions::default())
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("disk {idx} must contain the tail-success commit after reopen: {err:?}"));
|
||||
assert_eq!(
|
||||
stored.metadata.get("etag").map(String::as_str),
|
||||
Some("tail-success-etag"),
|
||||
"disk {idx} must expose the same committed metadata after tail success and reopen"
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(50), &mut rename).await.is_err(),
|
||||
"serial rename_data waits for the paused fan-out disk even after write quorum"
|
||||
);
|
||||
}
|
||||
|
||||
drop(dirs);
|
||||
barrier.release();
|
||||
rename
|
||||
.await
|
||||
.expect("tail success must complete the rename after the barrier is released");
|
||||
|
||||
for (idx, dir) in dirs.iter().enumerate() {
|
||||
let reopened = reopen_local_disk(dir).await;
|
||||
let stored = reopened
|
||||
.read_version("", bucket, object, "", &ReadOptions::default())
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("disk {idx} must contain the tail-success commit after reopen: {err:?}"));
|
||||
assert_eq!(
|
||||
stored.metadata.get("etag").map(String::as_str),
|
||||
Some("tail-success-etag"),
|
||||
"disk {idx} must expose the same committed metadata after tail success and reopen"
|
||||
);
|
||||
}
|
||||
|
||||
drop(dirs);
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in New Issue
Block a user