From 92c8c6db757f5aee69e4c620ca96f72f020bbb0f Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 8 Jul 2026 16:39:43 +0800 Subject: [PATCH] perf(ecstore): Linux O_DIRECT shard-write path off the commit critical section (#4411) `create_file` writes erasure shard / multipart part data through the page cache, so the commit-point `sync_dir_files` fdatasync in `rename_data` has to flush the whole shard's dirty pages (profiling: create_file avg 67.52ms vs 1.51ms nosync; BitrotWriter::write p95 40.86ms), and concurrent PUTs on one device stall each other's writeback inside the rename critical section. Add an opt-in Linux O_DIRECT streaming writer that reuses the direct-io read infrastructure (statx DIOALIGN probe, aligned bounce buffer, supported latch, one-time fallback log). Incoming bytes are staged into an aligned buffer; full aligned batches and the shutdown tail are flushed on the blocking pool (same offloading posture as the buffered writer it replaces). The sub-alignment tail is written buffered after clearing O_DIRECT via fcntl (MinIO's recipe). Shard bytes reach the device during the write phase, so the unchanged commit-point `sync_dir_files` fdatasync degrades to a cheap metadata/device FLUSH instead of flushing ~2MiB of dirty pages. Gated by `RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE` (default false), Linux-only, and durability-neutral: `sync_dir_files` still fdatasyncs every shard at the commit point, so the strict/relaxed/none durability tiers keep their exact guarantees. Filesystems that reject O_DIRECT (tmpfs, overlayfs, 9p) latch the path off and fall back to buffered writes; an O_DIRECT open/write EINVAL is never surfaced as `InvalidInput` (which `to_file_error` maps to `FileNotFound` and would masquerade as a missing shard, triggering a spurious EC rebuild). Unit tests cover the env gate, the alignment/staging-capacity and tail-split math, an end-to-end `create_file` round trip across sizes crossing the alignment and multi-batch boundaries (buffered fallback on macOS / O_DIRECT on a block device), and the writer state machine over a plain file on Linux. Real O_DIRECT throughput on Linux NVMe (ext4/xfs) is deferred to the azure 4-node bench per the issue's GO/NO-GO gate; the development host is macOS and cannot exercise O_DIRECT. Refs: rustfs/backlog#927, rustfs/backlog#936 Co-authored-by: heihutu --- crates/ecstore/src/disk/local.rs | 562 +++++++++++++++++++++++++++++++ 1 file changed, 562 insertions(+) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index b0dcd04db..8facafc6a 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -227,6 +227,20 @@ const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: bool = false; /// Default: 4MB. const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD"; const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024; + +/// Enable O_DIRECT for erasure shard / multipart part data writes (Linux only). +/// When enabled, `create_file` streams shard bytes straight to the device with +/// O_DIRECT, so the commit-point `sync_dir_files` fdatasync no longer flushes +/// ~2 MiB of dirty pages inside the `rename_data` critical section (it degrades +/// to a cheap metadata/device FLUSH). Aligned whole blocks are written direct; +/// the trailing sub-alignment remainder falls back to a buffered write after +/// clearing O_DIRECT (MinIO's recipe). Durability is unchanged: the file is +/// still fdatasynced at the commit point by the unchanged `sync_dir_files`. +/// EINVAL/EOPNOTSUPP (tmpfs, overlayfs, 9p, ...) latch the path off and fall +/// back to buffered writes for the whole disk. Non-Linux always falls back. +/// Default: false (buffered writes via the page cache, as before). +const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE"; +const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false; const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE"; const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false; const ENV_RUSTFS_OBJECT_MMAP_READ_METHOD: &str = "RUSTFS_OBJECT_MMAP_READ_METHOD"; @@ -257,6 +271,11 @@ fn is_direct_io_read_enabled() -> bool { rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE) } +/// Check if O_DIRECT shard/part data writes are enabled. +fn is_direct_io_write_enabled() -> bool { + rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE) +} + const EVENT_DISK_LOCAL_DURABILITY_MODE: &str = "disk_local_durability_mode"; /// Process-wide durability tier for commit-point fsync work on the local disk. @@ -711,6 +730,376 @@ fn pread_direct_aligned(file_path: &Path, offset: u64, length: usize, state: &Di Ok(Bytes::copy_from_slice(&buf.as_slice()[logical_start..logical_end])) } +// `AlignedBuf` uniquely owns a single heap allocation reached only through +// `&self`/`&mut self`; there is no interior mutability and no aliasing, so it +// is sound to move it across threads (into a `spawn_blocking` flush closure) +// and to share `&AlignedBuf` between threads. +#[cfg(target_os = "linux")] +#[allow(unsafe_code)] +// SAFETY: exclusive heap ownership, no aliasing (see the note above). +unsafe impl Send for AlignedBuf {} +#[cfg(target_os = "linux")] +#[allow(unsafe_code)] +// SAFETY: `&AlignedBuf` only exposes read-only access to an immutable buffer. +unsafe impl Sync for AlignedBuf {} + +/// Runtime state for the true O_DIRECT write path (Linux only), mirroring +/// [`DirectIoReadState`]. +/// +/// `supported` starts true and latches false on the first EINVAL/EOPNOTSUPP +/// from an O_DIRECT open (tmpfs, overlayfs, and 9p commonly reject the flag); +/// `create_file` then permanently opens shard files buffered for this disk. +/// `align` caches the DIO alignment probed from the backing filesystem. As on +/// the read path, an O_DIRECT open error must never surface to callers: EINVAL +/// maps to `FileNotFound` in `to_file_error`, which would masquerade as a +/// missing shard and trigger spurious EC rebuilds. +#[cfg(target_os = "linux")] +#[derive(Debug)] +struct DirectIoWriteState { + supported: AtomicBool, + align: OnceLock, + fallback_logged: AtomicBool, +} + +#[cfg(target_os = "linux")] +impl DirectIoWriteState { + fn new() -> Self { + Self { + supported: AtomicBool::new(true), + align: OnceLock::new(), + fallback_logged: AtomicBool::new(false), + } + } +} + +/// Target staging size for O_DIRECT writes, rounded up to the DIO alignment. +/// Bounds the per-writer aligned bounce buffer and batches many shard blocks +/// into one positioned write to keep the syscall count low. +const DIRECT_WRITE_STAGING_BYTES: usize = 1024 * 1024; + +/// Aligned bounce-buffer capacity for a given DIO alignment: the target staging +/// size rounded up to a whole multiple of `align` so the buffer address, every +/// flushed batch length, and every write offset stay alignment-correct. +/// Platform-independent (no O_DIRECT), so it is unit-tested on any host. +fn direct_write_staging_capacity(align: usize) -> usize { + debug_assert!(align.is_power_of_two() && align >= 512); + DIRECT_WRITE_STAGING_BYTES.div_ceil(align) * align +} + +/// Split `filled` staged bytes into the alignment-sized prefix written with +/// O_DIRECT and the sub-alignment tail written buffered. Platform-independent, +/// so the tail-boundary math is unit-tested on any host. +fn direct_write_tail_split(filled: usize, align: usize) -> (usize, usize) { + let aligned = filled - (filled % align); + (aligned, filled - aligned) +} + +/// Positioned write-all helper: retries short writes at increasing offsets. +/// +/// Under O_DIRECT the buffer address, `offset`, and length must all be aligned; +/// callers guarantee that. After O_DIRECT has been cleared (tail path) there is +/// no alignment requirement. +#[cfg(target_os = "linux")] +fn pwrite_all(file: &std::fs::File, mut buf: &[u8], mut offset: u64) -> std::io::Result<()> { + use std::os::unix::fs::FileExt; + + while !buf.is_empty() { + let n = file.write_at(buf, offset)?; + if n == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "O_DIRECT positioned write wrote 0 bytes", + )); + } + buf = &buf[n..]; + offset += n as u64; + } + Ok(()) +} + +/// Never let an O_DIRECT write error reach `to_file_error` as `InvalidInput`: +/// that maps to `FileNotFound` and would masquerade as a missing shard, +/// triggering a spurious EC rebuild (backlog#897 / issue correction #2). Any +/// EINVAL/EOPNOTSUPP surfacing from a flush is remapped to a generic error so +/// the write-quorum machinery treats it as the real write failure it is. +#[cfg(target_os = "linux")] +fn sanitize_direct_write_error(err: std::io::Error) -> std::io::Error { + if is_direct_io_unsupported(&err) { + std::io::Error::other(format!("O_DIRECT shard write failed: {err}")) + } else { + err + } +} + +/// Owned O_DIRECT write state moved in and out of the `spawn_blocking` flush +/// closures so the reactor is never blocked on synchronous device I/O. +#[cfg(target_os = "linux")] +struct DirectWriteInner { + file: std::fs::File, + /// Aligned bounce buffer; its capacity (`buf.len`) is a whole multiple of + /// `align`. + buf: AlignedBuf, + /// Bytes currently staged in `buf` and not yet written to the device. + filled: usize, + /// Next file offset for an O_DIRECT positioned write; always a multiple of + /// `align` because every batch flushed before the tail is a whole multiple. + write_offset: u64, + align: usize, + direct_cleared: bool, +} + +#[cfg(target_os = "linux")] +impl DirectWriteInner { + /// Flush a full staging batch (`filled == buf capacity`, a multiple of + /// `align`) straight to the device with O_DIRECT. + fn flush_batch(&mut self) -> std::io::Result<()> { + if self.filled == 0 { + return Ok(()); + } + debug_assert_eq!(self.filled % self.align, 0, "batch flush must be alignment-sized"); + pwrite_all(&self.file, &self.buf.as_slice()[..self.filled], self.write_offset).map_err(sanitize_direct_write_error)?; + self.write_offset += self.filled as u64; + self.filled = 0; + Ok(()) + } + + /// Final flush at shutdown: write the aligned prefix with O_DIRECT, then the + /// sub-alignment tail buffered after clearing O_DIRECT (MinIO's recipe; the + /// tail is not separately fsynced — the commit-point `sync_dir_files` + /// fdatasync covers the whole file, issue correction #5). + fn finish(&mut self) -> std::io::Result<()> { + let (aligned, remainder) = direct_write_tail_split(self.filled, self.align); + if aligned > 0 { + pwrite_all(&self.file, &self.buf.as_slice()[..aligned], self.write_offset).map_err(sanitize_direct_write_error)?; + self.write_offset += aligned as u64; + } + + if remainder > 0 { + self.clear_direct()?; + // Snapshot the slice bounds first to avoid borrowing `self.buf` + // while `self.file` is borrowed immutably below. + let start = aligned; + let end = self.filled; + pwrite_all(&self.file, &self.buf.as_slice()[start..end], self.write_offset)?; + self.write_offset += remainder as u64; + } + self.filled = 0; + Ok(()) + } + + /// Drop the O_DIRECT flag from the open file so the unaligned tail can be + /// written through the page cache without an alignment fault. + fn clear_direct(&mut self) -> std::io::Result<()> { + if self.direct_cleared { + return Ok(()); + } + let flags = rustix::fs::fcntl_getfl(&self.file).map_err(std::io::Error::from)?; + rustix::fs::fcntl_setfl(&self.file, flags - rustix::fs::OFlags::DIRECT).map_err(std::io::Error::from)?; + self.direct_cleared = true; + Ok(()) + } +} + +#[cfg(target_os = "linux")] +type DirectFlushHandle = tokio::task::JoinHandle<(DirectWriteInner, std::io::Result<()>)>; + +#[cfg(target_os = "linux")] +enum DirectWriteState { + Idle(Option), + Busy(DirectFlushHandle), +} + +/// Streaming O_DIRECT writer returned by `create_file` on Linux when the path +/// is enabled and supported. +/// +/// Incoming bytes are memcpy'd into an aligned bounce buffer (cheap, on the +/// reactor); each full aligned batch and the shutdown tail are flushed on the +/// blocking pool so the reactor never stalls on synchronous device I/O — the +/// same offloading posture as the buffered `tokio::fs::File` writer it +/// replaces. Durability is unchanged: no fsync happens here; the commit-point +/// `sync_dir_files` fdatasync persists the file. +#[cfg(target_os = "linux")] +struct DirectWriter { + state: DirectWriteState, + shutdown_started: bool, + shutdown_done: bool, +} + +#[cfg(target_os = "linux")] +impl DirectWriter { + fn new(inner: DirectWriteInner) -> Self { + Self { + state: DirectWriteState::Idle(Some(inner)), + shutdown_started: false, + shutdown_done: false, + } + } + + /// Build a writer over an already-open plain file with a caller-chosen + /// alignment and staging capacity. Exercises the streaming/tail state + /// machine deterministically on CI filesystems that reject O_DIRECT + /// (tmpfs/overlayfs), where the production `open_direct_writer` would latch + /// off; `write_at`/`fcntl` behave identically on a buffered file. + #[cfg(test)] + fn from_std_file_for_test(file: std::fs::File, align: usize, capacity: usize) -> Self { + assert_eq!(capacity % align, 0, "test capacity must be an alignment multiple"); + let buf = AlignedBuf::new(capacity, align).expect("aligned buffer allocation"); + Self::new(DirectWriteInner { + file, + buf, + filled: 0, + write_offset: 0, + align, + direct_cleared: false, + }) + } + + /// Drive an in-flight flush to completion, returning the recovered inner + /// state. Returns `Pending`/errors verbatim; on success the state is left + /// `Idle`. + fn poll_drive_busy(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll> { + if let DirectWriteState::Busy(handle) = &mut self.state { + let (inner, res) = match std::task::ready!(std::pin::Pin::new(handle).poll(cx)) { + Ok(pair) => pair, + Err(join_err) => { + return std::task::Poll::Ready(Err(std::io::Error::other(format!("O_DIRECT flush task failed: {join_err}")))); + } + }; + self.state = DirectWriteState::Idle(Some(inner)); + if self.shutdown_started { + self.shutdown_done = true; + } + res?; + } + std::task::Poll::Ready(Ok(())) + } +} + +#[cfg(target_os = "linux")] +impl AsyncWrite for DirectWriter { + fn poll_write( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let this = self.get_mut(); + loop { + match &mut this.state { + DirectWriteState::Busy(_) => { + std::task::ready!(this.poll_drive_busy(cx))?; + } + DirectWriteState::Idle(inner_opt) => { + if buf.is_empty() { + return std::task::Poll::Ready(Ok(0)); + } + let inner = inner_opt.as_mut().expect("idle direct writer must hold inner state"); + let capacity = inner.buf.len; + let space = capacity - inner.filled; + let n = space.min(buf.len()); + let start = inner.filled; + inner.buf.as_mut_slice()[start..start + n].copy_from_slice(&buf[..n]); + inner.filled += n; + + if inner.filled == capacity { + let mut inner = inner_opt.take().expect("idle direct writer must hold inner state"); + let handle = tokio::task::spawn_blocking(move || { + let res = inner.flush_batch(); + (inner, res) + }); + this.state = DirectWriteState::Busy(handle); + } + return std::task::Poll::Ready(Ok(n)); + } + } + } + } + + fn poll_flush(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll> { + // Only drive an in-flight batch to completion. Sub-alignment staged + // bytes cannot be flushed mid-stream (they would misalign the next + // O_DIRECT offset); they are written by `poll_shutdown`. + self.get_mut().poll_drive_busy(cx) + } + + fn poll_shutdown(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll> { + let this = self.get_mut(); + loop { + match &mut this.state { + DirectWriteState::Busy(_) => { + std::task::ready!(this.poll_drive_busy(cx))?; + } + DirectWriteState::Idle(inner_opt) => { + if this.shutdown_done { + return std::task::Poll::Ready(Ok(())); + } + let mut inner = inner_opt.take().expect("idle direct writer must hold inner state"); + this.shutdown_started = true; + let handle = tokio::task::spawn_blocking(move || { + let res = inner.finish(); + (inner, res) + }); + this.state = DirectWriteState::Busy(handle); + } + } + } + } +} + +/// Open a shard file for an O_DIRECT streaming write, probing DIO alignment on +/// the freshly created file. Returns `Ok(None)` (with the state latched off and +/// a one-time warning) when the filesystem rejects O_DIRECT, so the caller can +/// fall back to the buffered writer without ever surfacing EINVAL. +#[cfg(target_os = "linux")] +fn open_direct_writer(file_path: &Path, state: &DirectIoWriteState) -> Result> { + use std::os::unix::fs::OpenOptionsExt; + + let open_result = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .custom_flags(libc::O_DIRECT) + .open(file_path); + + let file = match open_result { + Ok(file) => file, + Err(err) => { + if is_direct_io_unsupported(&err) { + state.supported.store(false, Ordering::Relaxed); + if !state.fallback_logged.swap(true, Ordering::Relaxed) { + warn!( + event = EVENT_DISK_LOCAL_DIRECT_IO_FALLBACK, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = %file_path.display(), + error = ?err, + "O_DIRECT write unavailable; falling back to buffered writes" + ); + } + return Ok(None); + } + // A genuine open failure (permissions, disk full, ...): map it as a + // normal file error so callers see the real cause. + return Err(to_file_error(err).into()); + } + }; + + let align = *state.align.get_or_init(|| probe_direct_io_align(&file)); + let capacity = direct_write_staging_capacity(align); + let buf = match AlignedBuf::new(capacity, align) { + Ok(buf) => buf, + Err(err) => return Err(to_file_error(err).into()), + }; + + Ok(Some(DirectWriter::new(DirectWriteInner { + file, + buf, + filled: 0, + write_offset: 0, + align, + direct_cleared: false, + }))) +} + #[cfg(unix)] #[allow(unsafe_code)] fn mmap_page_size() -> Result { @@ -1202,6 +1591,8 @@ pub(crate) struct StdBackend { root: PathBuf, #[cfg(target_os = "linux")] direct_io: Arc, + #[cfg(target_os = "linux")] + direct_io_write: Arc, } impl StdBackend { @@ -1210,6 +1601,8 @@ impl StdBackend { root, #[cfg(target_os = "linux")] direct_io: Arc::new(DirectIoReadState::new()), + #[cfg(target_os = "linux")] + direct_io_write: Arc::new(DirectIoWriteState::new()), } } @@ -1752,6 +2145,25 @@ impl LocalIoBackend for StdBackend { if let Some(parent) = file_path.parent() { os::make_dir_all(parent, &volume_dir).await?; } + + // O_DIRECT streaming write (Linux, opt-in): shard bytes stream + // straight to the device so the commit-point fdatasync no longer + // flushes the whole shard's dirty pages inside the rename_data + // critical section. Latches off and falls back to the buffered + // writer on filesystems that reject O_DIRECT; never surfaces the + // EINVAL (which would masquerade as a missing shard). + #[cfg(target_os = "linux")] + if is_direct_io_write_enabled() && self.direct_io_write.supported.load(Ordering::Relaxed) { + let write_state = self.direct_io_write.clone(); + let direct_path = file_path.clone(); + let direct = tokio::task::spawn_blocking(move || open_direct_writer(&direct_path, &write_state)) + .await + .map_err(|err| DiskError::other(format!("O_DIRECT open task failed: {err}")))??; + if let Some(writer) = direct { + return Ok(Box::new(writer)); + } + } + // O_TRUNC: if a file already exists at this path, stale trailing bytes past // the new content would otherwise survive and mismatch the metadata size. let f = super::fs::open_file(&file_path, O_CREATE | O_WRONLY | O_TRUNC) @@ -7994,6 +8406,156 @@ mod test { assert_eq!(result.results, vec![CHECK_PART_FILE_CORRUPT]); } + // ----- HP-6: O_DIRECT shard write ----- + + #[test] + fn direct_io_write_env_gate_defaults_off() { + temp_env::with_var_unset(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, || { + assert!(!is_direct_io_write_enabled(), "O_DIRECT write must default to off"); + }); + temp_env::with_var(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, Some("true"), || { + assert!(is_direct_io_write_enabled()); + }); + temp_env::with_var(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, Some("false"), || { + assert!(!is_direct_io_write_enabled()); + }); + } + + #[test] + fn direct_write_staging_capacity_is_smallest_alignment_multiple_covering_target() { + for align in [512usize, 1024, 4096, 8192, 65536] { + let cap = direct_write_staging_capacity(align); + assert_eq!(cap % align, 0, "capacity must be an alignment multiple (align={align})"); + assert!( + cap >= DIRECT_WRITE_STAGING_BYTES, + "capacity must cover the target staging size (align={align})" + ); + assert!( + cap - DIRECT_WRITE_STAGING_BYTES < align, + "capacity must be the smallest covering multiple (align={align})" + ); + } + } + + #[test] + fn direct_write_tail_split_separates_aligned_prefix_from_remainder() { + let align = 4096usize; + assert_eq!(direct_write_tail_split(0, align), (0, 0)); + assert_eq!(direct_write_tail_split(align, align), (align, 0)); + assert_eq!(direct_write_tail_split(3 * align, align), (3 * align, 0)); + assert_eq!(direct_write_tail_split(100, align), (0, 100)); + let (prefix, remainder) = direct_write_tail_split(2 * align + 123, align); + assert_eq!((prefix, remainder), (2 * align, 123)); + assert_eq!(prefix + remainder, 2 * align + 123, "split must reconstruct the input length"); + assert_eq!(prefix % align, 0, "prefix must be alignment-sized"); + assert!(remainder < align, "remainder must be sub-alignment"); + } + + /// End-to-end round trip through `create_file`'s writer with O_DIRECT writes + /// enabled. On a block-backed Linux filesystem this drives the true + /// O_DIRECT path; on macOS and on CI filesystems that reject O_DIRECT it + /// exercises the buffered fallback. Either way, enabling the gate must not + /// change the bytes read back, across sizes crossing the alignment and + /// multi-batch staging boundaries (zero-breakage contract). + #[cfg(unix)] + #[tokio::test] + async fn create_file_direct_write_round_trips_all_sizes() { + use tempfile::tempdir; + + let root_dir = tempdir().expect("tempdir"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint"); + let disk = LocalDisk::new(&endpoint, false).await.expect("disk"); + disk.make_volume("test-volume").await.expect("make_volume"); + + let sizes = [ + 1usize, + 511, + 512, + 4095, + 4096, + 4097, + 8192, + 12345, + DIRECT_WRITE_STAGING_BYTES - 1, + DIRECT_WRITE_STAGING_BYTES, + DIRECT_WRITE_STAGING_BYTES + 4096, + 2 * DIRECT_WRITE_STAGING_BYTES + 777, + ]; + + for (i, size) in sizes.into_iter().enumerate() { + let mut state = 0x1234_5678u64.wrapping_add(i as u64); + let content: Vec = (0..size) + .map(|_| { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + (state >> 33) as u8 + }) + .collect(); + let path = format!("obj-{i}.bin"); + + temp_env::async_with_vars([(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, Some("true"))], async { + let mut writer = disk + .create_file("", "test-volume", &path, size as i64) + .await + .expect("create_file"); + // Odd-sized chunks exercise partial staging fills and flushes. + let mut off = 0; + while off < content.len() { + let end = (off + 777).min(content.len()); + writer.write_all(&content[off..end]).await.expect("write_all"); + off = end; + } + writer.shutdown().await.expect("shutdown"); + }) + .await; + + let got = disk.read_file_mmap_copy("test-volume", &path, 0, size).await.expect("read"); + assert_eq!(got.as_ref(), content.as_slice(), "round-trip mismatch at size={size}"); + } + } + + /// Deterministic coverage of the O_DIRECT writer's aligned-batch + tail + /// state machine on Linux, independent of whether the CI filesystem + /// supports O_DIRECT: the writer is driven over a plain file with a small + /// alignment and staging capacity so both a full-batch flush and the tail + /// split are exercised for every size class. + #[cfg(target_os = "linux")] + #[tokio::test] + async fn direct_writer_state_machine_round_trips_over_plain_file() { + use std::io::Read; + use tempfile::tempdir; + + let dir = tempdir().expect("tempdir"); + let align = 512usize; + let capacity = 4 * align; // forces multi-batch flushes for larger sizes + for &size in &[0usize, 1, 511, 512, 513, 1024, 2048, 2049, 5000] { + let path = dir.path().join(format!("s{size}")); + let file = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .read(true) + .open(&path) + .expect("open plain file"); + let content: Vec = (0..size).map(|i| (i.wrapping_mul(7).wrapping_add(3)) as u8).collect(); + + let mut writer = DirectWriter::from_std_file_for_test(file, align, capacity); + let mut off = 0; + while off < content.len() { + let end = (off + 300).min(content.len()); + writer.write_all(&content[off..end]).await.expect("write_all"); + off = end; + } + writer.shutdown().await.expect("shutdown"); + + let mut got = Vec::new(); + std::fs::File::open(&path) + .expect("reopen") + .read_to_end(&mut got) + .expect("read_to_end"); + assert_eq!(got, content, "state-machine round-trip mismatch at size={size}"); + } + } + /// Differential test for the LocalIoBackend refactor: every read shape /// (pread via mmap_copy, pread via direct_read_copy, bounded stream, full /// read) must return byte-identical data for the same (offset, length),