diff --git a/curvine-common/src/conf/fuse_conf.rs b/curvine-common/src/conf/fuse_conf.rs index f19a1a461..ff273008d 100644 --- a/curvine-common/src/conf/fuse_conf.rs +++ b/curvine-common/src/conf/fuse_conf.rs @@ -144,6 +144,18 @@ pub struct FuseConf { pub state_dir: String, + /// Optional override for the FUSE mount BDI `max_readahead_kb` (in KB). + /// + /// When set (recommended: 1024 = 1 MiB), curvine-fuse will, after each + /// successful mount, write the value to + /// `/sys/class/bdi/:/read_ahead_kb` and bump the FUSE + /// init `max_readahead` to at least `value * 1024` bytes, so the kernel + /// can issue larger sequential read requests. + /// + /// `None` (default) keeps current behavior. Linux only; on other platforms + /// the value is accepted but has no effect. + pub max_readahead_kb: Option, + /// The following are some time types, which are initialized only after init is called. #[serde(skip_serializing, skip_deserializing)] pub attr_ttl: Duration, @@ -188,6 +200,11 @@ impl FuseConf { if self.mnt_per_task == 0 { self.mnt_per_task = self.io_threads; } + + if let Some(0) = self.max_readahead_kb { + return err_box!("fuse.max_readahead_kb must be > 0 when set"); + } + Ok(()) } @@ -321,6 +338,7 @@ impl Default for FuseConf { state_dir: std::env::temp_dir().to_string_lossy().to_string(), + max_readahead_kb: None, attr_ttl: Default::default(), entry_ttl: Default::default(), negative_ttl: Default::default(), @@ -335,3 +353,54 @@ impl Default for FuseConf { conf } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_max_readahead_kb_is_none() { + let conf = FuseConf::default(); + assert!(conf.max_readahead_kb.is_none()); + } + + #[test] + fn init_rejects_zero_max_readahead_kb() { + let mut conf = FuseConf { + max_readahead_kb: Some(0), + ..Default::default() + }; + let err = conf.init().expect_err("zero must be rejected"); + assert!( + err.to_string().contains("max_readahead_kb"), + "error message should mention the field, got: {}", + err + ); + } + + #[test] + fn init_accepts_positive_max_readahead_kb() { + let mut conf = FuseConf { + max_readahead_kb: Some(1024), + ..Default::default() + }; + conf.init().expect("positive value must be accepted"); + assert_eq!(conf.max_readahead_kb, Some(1024)); + } + + #[test] + fn toml_round_trip_with_max_readahead_kb() { + let toml = r#" +max_readahead_kb = 1024 +"#; + let conf: FuseConf = toml::from_str(toml).expect("parse"); + assert_eq!(conf.max_readahead_kb, Some(1024)); + } + + #[test] + fn toml_round_trip_without_max_readahead_kb() { + // Backward compatibility: existing configs without the field load fine. + let conf: FuseConf = toml::from_str("").expect("parse empty"); + assert!(conf.max_readahead_kb.is_none()); + } +} diff --git a/curvine-fuse/src/fs/curvine_file_system.rs b/curvine-fuse/src/fs/curvine_file_system.rs index 96d89b68e..f3b268675 100644 --- a/curvine-fuse/src/fs/curvine_file_system.rs +++ b/curvine-fuse/src/fs/curvine_file_system.rs @@ -630,10 +630,18 @@ impl fs::FileSystem for CurvineFileSystem { out_flags &= !FUSE_WRITEBACK_CACHE; } + // If `fuse.max_readahead_kb` is configured, raise the negotiated + // `max_readahead` so the kernel cap (min(bdi.max_readahead_kb, + // fuse_conn.max_readahead)) does not silently shrink reads. + let max_readahead = match self.conf.max_readahead_kb { + Some(kb) => op.arg.max_readahead.max(kb.saturating_mul(1024)), + None => op.arg.max_readahead, + }; + let out = fuse_init_out { major: op.arg.major, minor: op.arg.minor, - max_readahead: op.arg.max_readahead, + max_readahead, flags: out_flags, max_background: self.conf.max_background, congestion_threshold: self.conf.congestion_threshold, diff --git a/curvine-fuse/src/session/bdi.rs b/curvine-fuse/src/session/bdi.rs new file mode 100644 index 000000000..f07a1416a --- /dev/null +++ b/curvine-fuse/src/session/bdi.rs @@ -0,0 +1,201 @@ +// Copyright 2025 OPPO. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Linux-only helpers to override the FUSE mount BDI's read-ahead size. +//! +//! After the mount syscall completes successfully, sysfs exposes +//! `/sys/class/bdi/:/read_ahead_kb` for the FUSE +//! superblock device. Writing the desired value here lets the FUSE kernel +//! issue larger read requests (e.g. 1 MiB) instead of the 128 KB default. +//! +//! Note: the user-facing config is named `fuse.max_readahead_kb` to mirror +//! the FUSE protocol field `max_readahead`. The kernel sysfs entry, however, +//! is `read_ahead_kb` (without the "max_" prefix), and that's what we write. + +use std::path::Path; + +use log::{info, warn}; + +#[cfg(target_os = "linux")] +const BDI_RETRY_COUNT: u32 = 10; + +#[cfg(target_os = "linux")] +const BDI_RETRY_DELAY_MS: u64 = 200; + +#[cfg(target_os = "linux")] +fn bdi_path_from_majmin(majmin: &str) -> String { + format!("/sys/class/bdi/{}/read_ahead_kb", majmin) +} + +#[cfg(target_os = "linux")] +fn mountinfo_bdi_path(mnt_path: &Path) -> std::io::Result> { + let mountinfo = std::fs::read_to_string("/proc/self/mountinfo")?; + Ok(mountinfo_bdi_path_from(&mountinfo, mnt_path)) +} + +#[cfg(target_os = "linux")] +fn mountinfo_bdi_path_from(mountinfo: &str, mnt_path: &Path) -> Option { + let target = mnt_path.to_string_lossy(); + + for line in mountinfo.lines() { + let mut fields = line.split_whitespace(); + let majmin = match fields.nth(2) { + Some(v) => v, + None => continue, + }; + // mountinfo fields are: id parent major:minor root mount_point ... + // We already consumed through major:minor, so skip root and read mount_point. + let mount_point = match fields.nth(1) { + Some(v) => v, + None => continue, + }; + if mount_point == target { + return Some(bdi_path_from_majmin(majmin)); + } + } + + None +} + +/// Write `kb` into the BDI sysfs entry that backs `mnt_path`. +/// +/// On any failure (mount path stat error, sysfs path missing, no write +/// permission, etc.) this function logs a warning and returns: the caller's +/// mount succeeds either way. On non-Linux platforms this is a no-op. +#[cfg(target_os = "linux")] +pub fn apply_max_readahead_kb(mnt_path: &Path, kb: u32) { + let bdi_path = match mountinfo_bdi_path(mnt_path) { + Ok(Some(path)) => path, + Ok(None) => { + warn!( + "bdi max_readahead_kb skip: mountinfo entry for {} not found (mount continues)", + mnt_path.display() + ); + return; + } + Err(e) => { + warn!( + "bdi max_readahead_kb skip: read /proc/self/mountinfo failed: {} (mount continues)", + e + ); + return; + } + }; + // The BDI sysfs entry may not be immediately available after the mount + // syscall. Retry a few times with a short delay to give the kernel time + // to create /sys/class/bdi/:/read_ahead_kb. + let mut tries = BDI_RETRY_COUNT; + while tries > 0 { + match std::fs::write(&bdi_path, kb.to_string()) { + Ok(()) => { + info!( + "bdi max_readahead_kb set: path={}, bdi={}, value={}", + mnt_path.display(), + bdi_path, + kb + ); + return; + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + tries -= 1; + if tries > 0 { + std::thread::sleep(std::time::Duration::from_millis(BDI_RETRY_DELAY_MS)); + } + } + Err(e) => { + warn!( + "bdi max_readahead_kb skip: write {} failed: {} (mount continues)", + bdi_path, e + ); + return; + } + } + } + warn!( + "bdi max_readahead_kb skip: {} not found after retries (mount continues)", + bdi_path + ); +} + +#[cfg(not(target_os = "linux"))] +pub fn apply_max_readahead_kb(_mnt_path: &Path, _kb: u32) { + // sysfs / BDI is Linux-only; intentional no-op elsewhere. +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::*; + use std::path::PathBuf; + + #[test] + fn bdi_path_from_majmin_formats_sysfs_path() { + assert_eq!( + bdi_path_from_majmin("8:1"), + "/sys/class/bdi/8:1/read_ahead_kb" + ); + } + + #[test] + fn mountinfo_bdi_path_from_matches_mount_point_field() { + let mountinfo = "126 32 0:114 / /curvine-fuse rw,relatime shared:98 - fuse curvinefs rw,user_id=0,group_id=0,allow_other\n"; + assert_eq!( + mountinfo_bdi_path_from(mountinfo, Path::new("/curvine-fuse")), + Some("/sys/class/bdi/0:114/read_ahead_kb".to_string()) + ); + } + + #[test] + fn mountinfo_bdi_path_from_returns_none_when_not_found() { + let mountinfo = "126 32 0:114 / /curvine-fuse rw,relatime shared:98 - fuse curvinefs rw\n"; + assert_eq!( + mountinfo_bdi_path_from(mountinfo, Path::new("/other/mount")), + None + ); + } + + #[test] + fn mountinfo_bdi_path_from_returns_none_on_empty() { + assert_eq!(mountinfo_bdi_path_from("", Path::new("/curvine-fuse")), None); + } + + #[test] + fn mountinfo_bdi_path_from_skips_malformed_lines() { + // A line with too few fields (no majmin, no mount point) must be skipped. + let mountinfo = "bad line\n126 32 0:114 / /curvine-fuse rw - fuse curvinefs rw\n"; + assert_eq!( + mountinfo_bdi_path_from(mountinfo, Path::new("/curvine-fuse")), + Some("/sys/class/bdi/0:114/read_ahead_kb".to_string()) + ); + } + + #[test] + fn mountinfo_bdi_path_from_picks_correct_entry_from_multiple() { + let mountinfo = concat!( + "126 32 0:114 / /curvine-fuse rw,relatime shared:98 - fuse curvinefs rw\n", + "127 32 8:1 / /other rw,relatime shared:99 - ext4 /dev/sda1 rw\n", + ); + assert_eq!( + mountinfo_bdi_path_from(mountinfo, Path::new("/other")), + Some("/sys/class/bdi/8:1/read_ahead_kb".to_string()) + ); + } + + #[test] + fn apply_does_not_panic_on_missing_path() { + // The function must remain best-effort: a non-existent mount path + // should produce a warning, not a panic / propagated error. + let bogus = PathBuf::from("/definitely/not/a/real/mount/point/curvine-bdi-test"); + apply_max_readahead_kb(&bogus, 1024); + } +} diff --git a/curvine-fuse/src/session/fuse_session.rs b/curvine-fuse/src/session/fuse_session.rs index 1276398f6..a423fa8eb 100644 --- a/curvine-fuse/src/session/fuse_session.rs +++ b/curvine-fuse/src/session/fuse_session.rs @@ -54,6 +54,12 @@ impl FuseSession { pub async fn new(rt: Arc, fs: T, conf: FuseConf) -> FuseResult { let mnts = Self::setup_mnts(&conf, &fs).await?; + if let Some(kb) = conf.max_readahead_kb { + for mnt in &mnts { + crate::session::bdi::apply_max_readahead_kb(&mnt.path, kb); + } + } + let fs = Arc::new(fs); let (shutdown_tx, _shutdown_rx) = watch::channel(false); diff --git a/curvine-fuse/src/session/mod.rs b/curvine-fuse/src/session/mod.rs index 608e06eb3..b41b43aa0 100644 --- a/curvine-fuse/src/session/mod.rs +++ b/curvine-fuse/src/session/mod.rs @@ -39,6 +39,8 @@ pub use fuse_buf::FuseBuf; mod fuse_notify_code; pub use self::fuse_notify_code::FuseNotifyCode; +pub mod bdi; + pub enum FuseTask { Reply(ResponseData), Request(FuseRequest), diff --git a/etc/curvine-cluster.toml b/etc/curvine-cluster.toml index b2f9cc79d..8dab598da 100644 --- a/etc/curvine-cluster.toml +++ b/etc/curvine-cluster.toml @@ -54,6 +54,11 @@ subnqn = "nqn.2024-01.io.curvine:subsystem2" # fuse configuration [fuse] +# Optional. Override the FUSE mount BDI `max_readahead_kb` after mount, and +# raise the FUSE init `max_readahead` accordingly. Recommended: 1024 (= 1 MiB) +# for sequential read workloads. Leave commented out to keep kernel default +# (typically 128 KB). Linux only; no effect on macOS. +# max_readahead_kb = 1024 # The log configuration of the customer service side, and the customer service side of rust, java, and fuse use this log file. [log]