From ba4cd694389e0e48c718da7f1267c21d1c124489 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 23 Aug 2026 15:41:25 +0800 Subject: [PATCH] fix(ecstore): default rename fanout to parallel early-ack path (#6443) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 --------- Co-authored-by: heihutu --- .../src/set_disk/core/io_primitives.rs | 98 +++++++++++-------- rustfs/src/server/http.rs | 8 ++ 2 files changed, 63 insertions(+), 43 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 1a77ba267..88125cbac 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -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 = 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] diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index ca421962f..20253d2df 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -1831,6 +1831,13 @@ fn handle_connection_error(peer_addr: Option<&str>, err: &(dyn std::error::Error } else if hyper_err.is_parse() { log_transport_failed(peer_addr, "parse_failure", &hyper_err.to_string()); } else if hyper_err.is_user() { + // is_user() = "error from user's Body stream": the application + // returned a streaming body that failed mid-flight. Log the full + // error source chain so the underlying cause (disk read failure, + // upstream RPC error, deleted object, etc.) is visible. + let cause = std::error::Error::source(hyper_err) + .map(|e| e.to_string()) + .unwrap_or_default(); error!( event = EVENT_HTTP_TRANSPORT_FAILED, component = LOG_COMPONENT_SERVER, @@ -1838,6 +1845,7 @@ fn handle_connection_error(peer_addr: Option<&str>, err: &(dyn std::error::Error peer_addr = %peer_addr, error_kind = "service_error", error = %hyper_err, + cause = %cause, result = "transport_error", "HTTP transport failed" );