diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index da8487fc9..ba97f67c5 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -23,7 +23,7 @@ use std::{ io, path::{Component, Path, PathBuf}, sync::{Arc, LazyLock, Weak}, - time::Instant, + time::{Duration, Instant}, }; use tokio::fs; use tokio::sync::{ @@ -328,6 +328,9 @@ const ENV_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: &str = "RUSTFS_EXPERIMENTAL_DST_DIR const DEFAULT_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: bool = false; const ENV_FILE_FDATASYNC_GROUP_COMMIT_ENABLE: &str = "RUSTFS_EXPERIMENTAL_FILE_FDATASYNC_GROUP_COMMIT_ENABLE"; const DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_ENABLE: bool = false; +const ENV_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: &str = "RUSTFS_EXPERIMENTAL_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS"; +const DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: u64 = 0; +const MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: u64 = 1_000; #[cfg(not(test))] const MAX_DST_DIR_FSYNC_GROUPS: usize = 1024; #[cfg(test)] @@ -354,6 +357,16 @@ static DST_DIR_FSYNC_GROUP_COMMIT_ENABLED: LazyLock = LazyLock::new(|| { static FILE_FDATASYNC_GROUP_COMMIT_ENABLED: LazyLock = LazyLock::new(|| { rustfs_utils::get_env_bool(ENV_FILE_FDATASYNC_GROUP_COMMIT_ENABLE, DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_ENABLE) }); +fn file_fdatasync_group_commit_wait_duration(wait_micros: u64) -> Duration { + Duration::from_micros(wait_micros.min(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS)) +} + +static FILE_FDATASYNC_GROUP_COMMIT_WAIT: LazyLock = LazyLock::new(|| { + file_fdatasync_group_commit_wait_duration(rustfs_utils::get_env_u64( + ENV_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS, + DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS, + )) +}); #[cfg(test)] mod dst_dir_fsync_group_commit_override { @@ -402,6 +415,7 @@ mod file_fdatasync_group_commit_override { use std::sync::{Mutex, MutexGuard, PoisonError, RwLock}; static OVERRIDE: RwLock> = RwLock::new(None); + static WAIT_OVERRIDE_MICROS: RwLock> = RwLock::new(None); static SERIAL: Mutex<()> = Mutex::new(()); pub(crate) fn get() -> Option { @@ -415,6 +429,7 @@ mod file_fdatasync_group_commit_override { impl Drop for OverrideGuard { fn drop(&mut self) { *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = None; + *WAIT_OVERRIDE_MICROS.write().unwrap_or_else(PoisonError::into_inner) = None; } } @@ -423,6 +438,14 @@ mod file_fdatasync_group_commit_override { *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = Some(enabled); OverrideGuard { _serial: serial } } + + pub(crate) fn set_wait_micros(wait_micros: u64) { + *WAIT_OVERRIDE_MICROS.write().unwrap_or_else(PoisonError::into_inner) = Some(wait_micros); + } + + pub(crate) fn wait_micros() -> Option { + *WAIT_OVERRIDE_MICROS.read().unwrap_or_else(PoisonError::into_inner) + } } #[cfg(test)] @@ -430,6 +453,11 @@ pub(crate) fn set_file_fdatasync_group_commit_for_test(enabled: bool) -> file_fd file_fdatasync_group_commit_override::set(enabled) } +#[cfg(test)] +fn set_file_fdatasync_group_commit_wait_for_test(wait_micros: u64) { + file_fdatasync_group_commit_override::set_wait_micros(wait_micros); +} + fn file_fdatasync_group_commit_enabled() -> bool { #[cfg(test)] if let Some(enabled) = file_fdatasync_group_commit_override::get() { @@ -439,6 +467,15 @@ fn file_fdatasync_group_commit_enabled() -> bool { *FILE_FDATASYNC_GROUP_COMMIT_ENABLED } +fn file_fdatasync_group_commit_wait() -> Duration { + #[cfg(test)] + if let Some(wait_micros) = file_fdatasync_group_commit_override::wait_micros() { + return file_fdatasync_group_commit_wait_duration(wait_micros); + } + + *FILE_FDATASYNC_GROUP_COMMIT_WAIT +} + #[derive(Clone, Eq, Hash, PartialEq)] struct DstDirFsyncGroupKey { canonical_path: PathBuf, @@ -934,6 +971,10 @@ async fn run_file_fdatasync_group_worker(group: Arc) { #[cfg(test)] file_sync_probe::run_before_group_batch(); tokio::task::yield_now().await; + let wait = file_fdatasync_group_commit_wait(); + if !wait.is_zero() { + tokio::time::sleep(wait).await; + } let (batch, batch_file_count): (Vec, usize) = { let mut group_state = group.inner.lock(); let batch_file_count = group_state.pending_files; @@ -6075,6 +6116,7 @@ mod tests { use std::sync::mpsc; let _group_commit = set_file_fdatasync_group_commit_for_test(true); + set_file_fdatasync_group_commit_wait_for_test(0); clear_file_fdatasync_group_commit_for_test(); let temp_dir = tempdir().expect("create temp dir"); let first_dir = temp_dir.path().join("first"); @@ -6141,12 +6183,105 @@ mod tests { assert_eq!(file_fdatasync_group_commit_counts_for_test(), (0, 0, 0)); } + #[test] + fn file_fdatasync_group_commit_wait_duration_uses_default_and_cap() { + assert_eq!(DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS, 0); + assert_eq!( + file_fdatasync_group_commit_wait_duration(DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS), + Duration::ZERO + ); + assert_eq!(file_fdatasync_group_commit_wait_duration(250), Duration::from_micros(250)); + assert_eq!( + file_fdatasync_group_commit_wait_duration(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS), + Duration::from_micros(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS) + ); + assert_eq!( + file_fdatasync_group_commit_wait_duration(u64::MAX), + Duration::from_micros(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS) + ); + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + #[serial_test::serial(file_sync_probe)] + async fn file_fdatasync_group_commit_wait_budget_batches_late_follower() { + use std::sync::mpsc; + + let _group_commit = set_file_fdatasync_group_commit_for_test(true); + let wait_budget_micros = 1_000; + let wait_budget = file_fdatasync_group_commit_wait_duration(wait_budget_micros); + set_file_fdatasync_group_commit_wait_for_test(wait_budget_micros); + clear_file_fdatasync_group_commit_for_test(); + let temp_dir = tempdir().expect("create temp dir"); + let first_dir = temp_dir.path().join("first"); + let second_dir = temp_dir.path().join("second"); + std::fs::create_dir(&first_dir).expect("create first dir"); + std::fs::create_dir(&second_dir).expect("create second dir"); + std::fs::write(first_dir.join("part.1"), b"first").expect("write first part"); + std::fs::write(second_dir.join("part.1"), b"second").expect("write second part"); + let _probe = file_sync_probe::set_blocking(temp_dir.path()); + let (entered_tx, entered_rx) = mpsc::channel(); + file_sync_probe::set_before_group_batch(move || { + entered_tx.send(()).expect("signal first file fdatasync group worker"); + }); + + let limiter = file_sync_limiter(); + let first_limiter = limiter.clone(); + let first_path = first_dir.clone(); + let first = tokio::spawn(async move { sync_dir_files_with_limiter(first_path, first_limiter).await }); + tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(30))) + .await + .expect("group worker hook waiter should run") + .expect("first file fdatasync group worker should start"); + + let second_limiter = limiter.clone(); + let second_path = second_dir.clone(); + let second = tokio::spawn(async move { sync_dir_files_with_limiter(second_path, second_limiter).await }); + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if file_fdatasync_group_commit_counts_for_test().1 == 2 { + return; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("second waiter should enqueue during the configured wait budget"); + tokio::time::advance(wait_budget).await; + tokio::task::yield_now().await; + file_sync_probe::wait_for_active(1).await; + + assert_eq!( + file_sync_probe::group_batches(), + vec![2], + "configured wait budget should let a follower join the leader's batch" + ); + file_sync_probe::release(); + first + .await + .expect("join first wait-budget file sync") + .expect("first wait-budget file sync must succeed"); + second + .await + .expect("join second wait-budget file sync") + .expect("second wait-budget file sync must succeed"); + assert!( + fsync_dir_recorder::was_fsynced(&first_dir), + "first source directory must still be fsynced" + ); + assert!( + fsync_dir_recorder::was_fsynced(&second_dir), + "second source directory must still be fsynced" + ); + assert_eq!(file_fdatasync_group_commit_counts_for_test(), (0, 0, 0)); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial_test::serial(file_sync_probe)] async fn file_fdatasync_group_commit_failure_fails_all_waiters_before_dir_fsync() { use std::sync::mpsc; let _group_commit = set_file_fdatasync_group_commit_for_test(true); + set_file_fdatasync_group_commit_wait_for_test(0); clear_file_fdatasync_group_commit_for_test(); let temp_dir = tempdir().expect("create temp dir"); let first_dir = temp_dir.path().join("first");