feat(disk): isolate fsync blocking pool from main runtime

Add a dedicated tokio blocking runtime for fsync/fdatasync operations,
controlled by RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS (default 0 = off).

When configured with >1 threads, fsync operations are dispatched to a
separate blocking pool so device-bound fsync does not starve read
operations (pread/stat/open) on the main blocking pool.

Changed call sites:
- run_file_sync_blocking (batch fdatasync)
- fsync_dir (directory fsync)
- fsync_open_dst_dir_group (dst dir group fsync)
- run_blocking_namespace_file_sync_operation_with_global (namespace+fsync)

Unchanged (stay on main pool):
- run_blocking_namespace_operation (rename/metadata)
- local.rs spawn_blocking calls (pread/stat/open)

Default behavior is unchanged (0 = no isolation).

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/config/src/constants/runtime.rs

@@ -57,6 +57,13 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
 pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
 pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
 
+/// Dedicated blocking thread pool for fsync/fdatasync operations.
+/// When > 1, fsync operations are isolated from the main blocking pool to
+/// prevent device-bound fsync from starving read operations (pread/stat/open).
+/// Default 0 means auto (no isolation, use main runtime).
+pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
+pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
+
 // Dial9 Tokio Telemetry Default values
 pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
 pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";

crates/ecstore/src/disk/os.rs

@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
     #[cfg(unix)]
     {
         let dir = dir.as_ref().to_path_buf();
-        tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
+        fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
     }
 
     #[cfg(not(unix))]
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
     #[cfg(test)]
     let dir = group.dir.clone();
     let dir_file = group.dir_file.clone();
-    tokio::task::spawn_blocking(move || {
+    fsync_spawn_blocking(move || {
         #[cfg(test)]
         {
             if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -1080,6 +1080,44 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
 
 static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
 static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+
+/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
+/// configured with >1 threads, isolates device-bound fsync from the main
+/// blocking pool so reads (pread/stat/open) are not starved. `None` means
+/// fall back to the main runtime (zero behavior change).
+static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
+    let threads =
+        rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
+    if threads <= 1 {
+        return None;
+    }
+    let mut builder = tokio::runtime::Builder::new_multi_thread();
+    builder
+        .worker_threads(2)
+        .max_blocking_threads(threads)
+        .thread_name("rustfs-fsync")
+        .thread_stack_size(512 * 1024)
+        .enable_all();
+    match builder.build() {
+        Ok(rt) => {
+            tracing::info!(threads, "fsync dedicated blocking pool enabled");
+            Some(rt)
+        }
+        Err(err) => {
+            tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
+            None
+        }
+    }
+});
+
+/// Spawn a blocking task on the fsync-dedicated runtime if configured,
+/// otherwise fall back to the main tokio blocking pool.
+fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
+    match FSYNC_RUNTIME.as_ref() {
+        Some(rt) => rt.spawn_blocking(f),
+        None => tokio::task::spawn_blocking(f),
+    }
+}
 static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
 type NamespaceMutationLock = AsyncMutex<()>;
@@ -1217,7 +1255,7 @@ where
     F: FnOnce() -> io::Result<T> + Send + 'static,
 {
     let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
-    let result = tokio::task::spawn_blocking(move || {
+    let result = fsync_spawn_blocking(move || {
         let _disk_permit = disk_permit;
         work()
     })
@@ -2146,7 +2184,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
         wait_started,
     );
     let disk_permit = admission.disk_permit.clone();
-    let result = tokio::task::spawn_blocking(move || {
+    let result = fsync_spawn_blocking(move || {
         let _lease = lease;
         let _disk_permit = disk_permit;
         operation()

crates/madmin/src/trace.rs

@@ -12,16 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::{collections::HashMap, time::Duration};
+
+use jiff::Timestamp;
 use serde::{Deserialize, Serialize};
 
-/// Bitflag helper for service trace categories.
+use crate::heal_commands::HealResultItem;
-///
+
-/// Each variant occupies a single bit so that a `TraceType` value can represent
-/// an arbitrary combination of categories via bitwise OR.
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
 pub struct TraceType(u64);
 
 impl TraceType {
+    // Define some constants
     pub const OS: TraceType = TraceType(1 << 0);
     pub const STORAGE: TraceType = TraceType(1 << 1);
     pub const S3: TraceType = TraceType(1 << 2);
@@ -38,13 +40,15 @@ impl TraceType {
     pub const FTP: TraceType = TraceType(1 << 13);
     pub const ILM: TraceType = TraceType(1 << 14);
 
-    /// All trace categories combined. Must be updated when adding new variants.
+    // MetricsAll must be last.
     pub const ALL: TraceType = TraceType((1 << 15) - 1);
 
     pub fn new(t: u64) -> Self {
         Self(t)
     }
+}
 
+impl TraceType {
     pub fn contains(&self, x: &TraceType) -> bool {
         (self.0 & x.0) == x.0
     }
@@ -72,38 +76,140 @@ impl TraceType {
     }
 }
 
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct TraceInfo {
+    #[serde(rename = "type")]
+    trace_type: u64,
+    #[serde(rename = "nodename")]
+    node_name: String,
+    #[serde(rename = "funcname")]
+    func_name: String,
+    #[serde(rename = "time")]
+    time: Timestamp,
+    #[serde(rename = "path")]
+    path: String,
+    #[serde(rename = "dur")]
+    duration: Duration,
+    #[serde(rename = "bytes", skip_serializing_if = "Option::is_none")]
+    bytes: Option<i64>,
+    #[serde(rename = "msg", skip_serializing_if = "Option::is_none")]
+    message: Option<String>,
+    #[serde(rename = "error", skip_serializing_if = "Option::is_none")]
+    error: Option<String>,
+    #[serde(rename = "custom", skip_serializing_if = "Option::is_none")]
+    custom: Option<HashMap<String, String>>,
+    #[serde(rename = "http", skip_serializing_if = "Option::is_none")]
+    http: Option<TraceHTTPStats>,
+    #[serde(rename = "healResult", skip_serializing_if = "Option::is_none")]
+    heal_result: Option<HealResultItem>,
+}
+
+impl TraceInfo {
+    pub fn mask(&self) -> u64 {
+        TraceType::new(self.trace_type).mask()
+    }
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct TraceInfoLegacy {
+    trace_info: TraceInfo,
+    #[serde(rename = "request")]
+    req_info: Option<TraceRequestInfo>,
+    #[serde(rename = "response")]
+    resp_info: Option<TraceResponseInfo>,
+    #[serde(rename = "stats")]
+    call_stats: Option<TraceCallStats>,
+    #[serde(rename = "storageStats")]
+    storage_stats: Option<StorageStats>,
+    #[serde(rename = "osStats")]
+    os_stats: Option<OSStats>,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct StorageStats {
+    path: String,
+    duration: Duration,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct OSStats {
+    path: String,
+    duration: Duration,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct TraceHTTPStats {
+    req_info: TraceRequestInfo,
+    resp_info: TraceResponseInfo,
+    call_stats: TraceCallStats,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct TraceCallStats {
+    input_bytes: i32,
+    output_bytes: i32,
+    latency: Duration,
+    time_to_first_byte: Duration,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct TraceRequestInfo {
+    time: Timestamp,
+    proto: String,
+    method: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    path: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    raw_query: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    headers: Option<HashMap<String, String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    body: Option<Vec<u8>>,
+    client: String,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct TraceResponseInfo {
+    time: Timestamp,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    headers: Option<HashMap<String, String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    body: Option<Vec<u8>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    status_code: Option<i32>,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
     #[test]
-    fn trace_type_contains_and_overlaps() {
+    fn trace_timestamps_serialize_as_rfc3339_utc() {
-        let mut combined = TraceType::default();
+        let timestamp = Timestamp::constant(1_700_000_000, 123_456_000);
-        combined.merge(&TraceType::S3);
+        let trace = TraceInfo {
-        combined.merge(&TraceType::HEALING);
+            time: timestamp,
+            http: Some(TraceHTTPStats {
+                req_info: TraceRequestInfo {
+                    time: timestamp,
+                    ..Default::default()
+                },
+                resp_info: TraceResponseInfo {
+                    time: timestamp,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
 
-        assert!(combined.contains(&TraceType::S3));
+        let value = serde_json::to_value(trace).expect("trace should serialize");
-        assert!(combined.contains(&TraceType::HEALING));
+        assert_eq!(value["time"], "2023-11-14T22:13:20.123456Z");
-        assert!(!combined.contains(&TraceType::SCANNER));
+        assert_eq!(value["http"]["req_info"]["time"], "2023-11-14T22:13:20.123456Z");
-        assert!(combined.overlaps(&TraceType::S3));
+        assert_eq!(value["http"]["resp_info"]["time"], "2023-11-14T22:13:20.123456Z");
-        assert!(combined.overlaps(&TraceType::HEALING));
+        let trace: TraceInfo = serde_json::from_value(value).expect("trace should deserialize");
-        assert!(!combined.overlaps(&TraceType::SCANNER));
+        assert_eq!(trace.time, timestamp);
-    }
+        let http = trace.http.expect("http trace should deserialize");
-
+        assert_eq!(http.req_info.time, timestamp);
-    #[test]
+        assert_eq!(http.resp_info.time, timestamp);
-    fn trace_type_set_if() {
-        let mut tt = TraceType::default();
-        tt.set_if(true, &TraceType::OS);
-        tt.set_if(false, &TraceType::S3);
-        assert!(tt.contains(&TraceType::OS));
-        assert!(!tt.contains(&TraceType::S3));
-    }
-
-    #[test]
-    fn trace_type_single_type() {
-        assert!(TraceType::S3.single_type());
-        let mut combined = TraceType::S3;
-        combined.merge(&TraceType::HEALING);
-        assert!(!combined.single_type());
     }
 }