From da755eb66b2e76d6e95763ae1bcc8b4a2321f3c0 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 10 Jul 2026 00:57:17 +0800 Subject: [PATCH] feat(ecstore): runtime-probed io_uring read backend, gray-off by default (backlog#1104) (#4632) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit feat(ecstore): add runtime-probed io_uring read backend, gray-off by default (rustfs/backlog#1104) Wire the standalone rustfs-uring crate (https://github.com/rustfs/uring) into LocalIoBackend as an opt-in read backend. When RUSTFS_IO_URING_READ_ENABLE is set AND the per-disk io_uring probe succeeds, positioned reads go through the cancel-safe UringDriver; otherwise — default, or on a restricted host, or on any per-read driver error — reads fall back to StdBackend byte-for-byte, so the default build is unchanged. - Cargo.toml: rustfs-uring as a Linux-only git dependency (pinned rev). The guard scripts/check_no_tokio_io_uring.sh allows an explicit io-uring integration; only the tokio "io-uring" runtime feature is banned. - UringBackend mirrors StdBackend::pread_bytes's resolution/access/bounds preamble and only swaps the raw byte read; the other three trait methods delegate to the inner StdBackend. - Backend selection at LocalDisk construction via build_local_io_backend. - Differential test uring_backend_reads_match_std: with the flag set, every positioned read returns byte-identical data (driver-served when io_uring is available, StdBackend fallback otherwise). Verified: cargo check -p rustfs-ecstore on Linux; the differential test passes under real io_uring (seccomp=unconfined). The runtime errno degradation latch, per-disk probe cache, and productionization are tracked in rustfs/backlog#1101/#1102. Co-authored-by: heihutu --- .github/workflows/audit.yml | 5 + Cargo.lock | 22 +++ crates/ecstore/Cargo.toml | 7 + crates/ecstore/src/disk/local.rs | 251 ++++++++++++++++++++++++++++++- deny.toml | 3 + 5 files changed, 287 insertions(+), 1 deletion(-) diff --git a/.github/workflows/audit.yml b/.github/workflows/audit.yml index 99fe9d4b2..f5e539773 100644 --- a/.github/workflows/audit.yml +++ b/.github/workflows/audit.yml @@ -112,4 +112,9 @@ jobs: with: fail-on-severity: moderate allow-ghsas: GHSA-2f9f-gq7v-9h6m + # rustfs-uring is a first-party Apache-2.0 crate pulled as a git + # dependency (backlog#1104); its license cannot be read from the git + # source, so allow it explicitly to avoid a spurious unknown-license + # warning on every Cargo.lock change. + allow-dependencies-licenses: pkg:cargo/rustfs-uring comment-summary-in-pr: always diff --git a/Cargo.lock b/Cargo.lock index 46befc1ae..75436e0bc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5326,6 +5326,17 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "io-uring" +version = "0.7.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9080b15e63775b9a2ac7dca720f7050a8b955e092ea0f6020a4a80f69998cdc0" +dependencies = [ + "bitflags 2.13.0", + "cfg-if", + "libc", +] + [[package]] name = "ipconfig" version = "0.3.4" @@ -9169,6 +9180,7 @@ dependencies = [ "rustfs-signer", "rustfs-storage-api", "rustfs-tls-runtime", + "rustfs-uring", "rustfs-utils", "rustix 1.1.4", "rustls", @@ -10015,6 +10027,16 @@ dependencies = [ "tracing", ] +[[package]] +name = "rustfs-uring" +version = "0.1.0" +source = "git+https://github.com/rustfs/uring?rev=39018c09cde98b380658f1a1511349a3574daddf#39018c09cde98b380658f1a1511349a3574daddf" +dependencies = [ + "io-uring", + "libc", + "tokio", +] + [[package]] name = "rustfs-utils" version = "1.0.0-beta.8" diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 5fec84cde..94f3c25b1 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -140,6 +140,13 @@ aws-smithy-http-client.workspace = true # Observability and Metrics metrics = { workspace = true } +# Runtime-probed io_uring read backend (backlog#1104). Linux-only, git +# dependency until rustfs-uring is production-ready and published. The guard +# scripts/check_no_tokio_io_uring.sh allows an explicit io-uring integration; +# only the tokio "io-uring" runtime feature is banned. +[target.'cfg(target_os = "linux")'.dependencies] +rustfs-uring = { git = "https://github.com/rustfs/uring", rev = "39018c09cde98b380658f1a1511349a3574daddf" } + [dev-dependencies] tokio = { workspace = true, features = ["rt-multi-thread", "macros", "test-util"] } criterion = { workspace = true, features = ["html_reports"] } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 22acd18bc..c59f5979f 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -276,6 +276,27 @@ 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) } +/// Enable the runtime-probed io_uring read backend (backlog#1104). Default: +/// false (gray-off). The backend is used only when this is set AND the per-disk +/// probe succeeds; otherwise, and on any per-read driver error, reads fall back +/// to `StdBackend` byte-for-byte. +#[cfg(target_os = "linux")] +const ENV_RUSTFS_IO_URING_READ_ENABLE: &str = "RUSTFS_IO_URING_READ_ENABLE"; +#[cfg(target_os = "linux")] +const DEFAULT_RUSTFS_IO_URING_READ_ENABLE: bool = false; + +/// io_uring submission-queue depth used when probing a disk (backlog#1104). +/// Backpressure caps in-flight at this value, below CQ capacity (2×), so CQ +/// overflow is structurally unreachable. +#[cfg(target_os = "linux")] +const URING_QUEUE_DEPTH: u32 = 128; + +/// Check if the runtime-probed io_uring read backend is enabled. +#[cfg(target_os = "linux")] +fn is_io_uring_read_enabled() -> bool { + rustfs_utils::get_env_bool(ENV_RUSTFS_IO_URING_READ_ENABLE, DEFAULT_RUSTFS_IO_URING_READ_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. @@ -2253,6 +2274,176 @@ impl LocalIoBackend for StdBackend { } } +/// Runtime-probed io_uring read backend (backlog#1104). +/// +/// Wraps a [`StdBackend`] for everything except positioned reads, which go +/// through rustfs-uring's cancel-safe `UringDriver`. Constructed only when +/// `RUSTFS_IO_URING_READ_ENABLE` is set AND the per-disk probe succeeds; on any +/// per-read driver error a read falls back to the inner `StdBackend`, so +/// behavior never regresses. The read preamble (path resolution, access checks, +/// bounds) mirrors `StdBackend::pread_bytes` exactly — only the raw byte read +/// differs. +#[cfg(target_os = "linux")] +pub(crate) struct UringBackend { + root: PathBuf, + inner: StdBackend, + driver: Arc, + fallback_logged: std::sync::atomic::AtomicBool, +} + +#[cfg(target_os = "linux")] +impl std::fmt::Debug for UringBackend { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("UringBackend") + .field("root", &self.root) + .finish_non_exhaustive() + } +} + +#[cfg(target_os = "linux")] +impl UringBackend { + /// Probe io_uring on `root`; `Some(backend)` if usable, `None` to fall back + /// to `StdBackend`. A restricted-environment errno degrades quietly; an + /// unexpected errno is surfaced as a warning (both still fall back). + pub(crate) fn try_new(root: PathBuf) -> Option { + match rustfs_uring::UringDriver::probe_and_start(URING_QUEUE_DEPTH) { + Ok(driver) => { + info!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + root = %root.display(), + "io_uring read backend enabled" + ); + Some(Self { + inner: StdBackend::new(root.clone()), + root, + driver: Arc::new(driver), + fallback_logged: std::sync::atomic::AtomicBool::new(false), + }) + } + Err(err) => { + if err.is_expected_restriction() { + debug!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + root = %root.display(), + error = ?err, + "io_uring unavailable (restricted environment); using StdBackend" + ); + } else { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + root = %root.display(), + error = ?err, + "io_uring probe failed unexpectedly; using StdBackend" + ); + } + None + } + } + } + + /// Positioned read via io_uring. Mirrors `StdBackend::pread_bytes`'s + /// resolution/access/bounds preamble, then reads the range with the driver + /// (whole-range: the driver resubmits short reads for positioned reads). + async fn pread_uring(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { + let Some(end_offset) = offset.checked_add(length) else { + return Err(DiskError::FileCorrupt); + }; + let root = self.root.clone(); + let volume_owned = volume.to_owned(); + let path_owned = path.to_owned(); + + let (file, offset_u64) = tokio::task::spawn_blocking(move || -> Result<(std::fs::File, u64)> { + let volume_dir = local_disk_bucket_path(&root, &volume_owned)?; + if !skip_access_checks(&volume_owned) { + access_std(&volume_dir).map_err(|e| DiskError::from(to_access_error(e, DiskError::VolumeAccessDenied)))?; + } + let file_path = local_disk_object_path(&root, &volume_owned, &path_owned)?; + check_path_length(file_path.to_string_lossy().as_ref())?; + let file = std::fs::File::open(&file_path).map_err(DiskError::from)?; + let meta = file.metadata().map_err(DiskError::from)?; + let end_offset_u64 = u64::try_from(end_offset).map_err(|_| DiskError::FileCorrupt)?; + if meta.len() < end_offset_u64 { + return Err(DiskError::FileCorrupt); + } + let offset_u64 = u64::try_from(offset).map_err(|_| DiskError::FileCorrupt)?; + Ok((file, offset_u64)) + }) + .await + .map_err(|e| DiskError::other(format!("uring pread join error: {e}")))??; + + if length == 0 { + return Ok(Bytes::new()); + } + + let bytes = self + .driver + .read_at(Arc::new(file), offset_u64, length) + .await + .map_err(DiskError::from)?; + if bytes.len() != length { + return Err(DiskError::other("io_uring returned a short read")); + } + Ok(Bytes::from(bytes)) + } +} + +#[cfg(target_os = "linux")] +#[async_trait::async_trait] +impl LocalIoBackend for UringBackend { + async fn pread_bytes( + &self, + volume: &str, + path: &str, + offset: usize, + length: usize, + metrics: Option, + ) -> Result { + match self.pread_uring(volume, path, offset, length).await { + Ok(bytes) => Ok(bytes), + Err(err) => { + if !self.fallback_logged.swap(true, Ordering::Relaxed) { + debug!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + error = ?err, + "io_uring read fell back to StdBackend (logged once per disk)" + ); + } + self.inner.pread_bytes(volume, path, offset, length, metrics).await + } + } + } + + async fn open_read_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { + self.inner.open_read_stream(volume, path, offset, length).await + } + + async fn open_full_read(&self, volume: &str, path: &str) -> Result { + self.inner.open_full_read(volume, path).await + } + + async fn open_write(&self, volume: &str, path: &str, mode: WriteMode) -> Result { + self.inner.open_write(volume, path, mode).await + } +} + +/// Select the local read backend: the runtime-probed io_uring backend when +/// enabled and the per-disk probe succeeds, otherwise the default +/// [`StdBackend`] (backlog#1104). Enabling io_uring is opt-in and falls back +/// byte-for-byte, so the default build is unchanged. +fn build_local_io_backend(root: PathBuf) -> Arc { + #[cfg(target_os = "linux")] + if is_io_uring_read_enabled() + && let Some(backend) = UringBackend::try_new(root.clone()) + { + return Arc::new(backend); + } + Arc::new(StdBackend::new(root)) +} + pub struct LocalDisk { pub root: PathBuf, pub format_path: PathBuf, @@ -2513,7 +2704,7 @@ impl LocalDisk { startup_cleanup_ready, startup_cleanup_notify, exit_signal: None, - io_backend: Arc::new(StdBackend::new(root.clone())), + io_backend: build_local_io_backend(root.clone()), }; let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| { log_startup_disk_error("get_disk_info", &root, err); @@ -10914,6 +11105,64 @@ mod test { assert_eq!(Bytes::from(all), content, "open_full_read mismatch"); } + /// io_uring read backend (backlog#1104): with `RUSTFS_IO_URING_READ_ENABLE` + /// set, every positioned read returns byte-identical data (the driver serves + /// it when io_uring is available; on a restricted host the backend degrades + /// to `StdBackend` and the bytes still match). + #[cfg(target_os = "linux")] + #[tokio::test(flavor = "multi_thread")] + async fn uring_backend_reads_match_std() { + use tempfile::tempdir; + + const FILE_LEN: usize = 64 * 1024; + let mut state = 0x9e3779b97f4a7c15u64; + let content: Vec = (0..FILE_LEN) + .map(|_| { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + (state >> 33) as u8 + }) + .collect(); + let content = Bytes::from(content); + + let page = mmap_page_size().expect("page size should be available") as usize; + let ranges = [ + (0usize, FILE_LEN), + (1, 17), + (page - 1, 2), + (page, page), + (2 * page - 1, page + 2), + (FILE_LEN - 7, 7), + (0, 0), + ]; + + // The env var must be set before LocalDisk::new (the backend is chosen + // at construction). + let got_ranges = temp_env::async_with_vars([(ENV_RUSTFS_IO_URING_READ_ENABLE, Some("true"))], async { + let root_dir = tempdir().expect("operation should succeed"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("operation should succeed"); + let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed"); + disk.make_volume("test-volume").await.expect("operation should succeed"); + disk.write_all("test-volume", "blob.bin", content.clone()) + .await + .expect("operation should succeed"); + let mut out = Vec::new(); + for (offset, length) in ranges { + out.push( + disk.read_file_mmap_copy("test-volume", "blob.bin", offset, length) + .await + .expect("io_uring-enabled read must succeed (driver or fallback)"), + ); + } + out + }) + .await; + + for ((offset, length), got) in ranges.into_iter().zip(got_ranges) { + let expected = content.slice(offset..offset + length); + assert_eq!(got, expected, "uring read mismatch at offset={offset} length={length}"); + } + } + /// The O_DIRECT read path must return the same bytes as the buffered /// path for unaligned shard sizes and ranges, and must silently fall /// back (never error) on filesystems that reject O_DIRECT (e.g. tmpfs). diff --git a/deny.toml b/deny.toml index f36957da9..f2259871d 100644 --- a/deny.toml +++ b/deny.toml @@ -40,6 +40,9 @@ allow-git = [ # Pinned to a specific commit in workspace Cargo.toml. # "https://github.com/rustfs/s3s", "https://github.com/apache/datafusion.git", + # Runtime-probed io_uring read backend (backlog#1104), Linux-only. + # Pinned to a specific commit in crates/ecstore/Cargo.toml. + "https://github.com/rustfs/uring", ] [bans]