You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by er...@apache.org on 2025/06/19 20:26:09 UTC
(opendal) branch poc-services-fs-direct-io created (now 8c2dbc0c3)
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a change to branch poc-services-fs-direct-io
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 8c2dbc0c3 Attempt 1
This branch includes the following new commits:
new 8c2dbc0c3 Attempt 1
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/01: Attempt 1
Posted by er...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a commit to branch poc-services-fs-direct-io
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 8c2dbc0c3a7f63654085b119df7c6f0a1aaea28a
Author: Erick Guan <29...@users.noreply.github.com>
AuthorDate: Thu Jun 19 21:02:27 2025 +0200
Attempt 1
---
core/Cargo.toml | 3 ++
core/src/services/fs/backend.rs | 90 ++++++++++++++++++++++++-----------------
core/src/services/fs/config.rs | 3 ++
core/src/services/fs/core.rs | 60 +++++++++++++++++++++++++++
core/src/services/fs/docs.md | 41 ++++++++++++++++++-
5 files changed, 160 insertions(+), 37 deletions(-)
diff --git a/core/Cargo.toml b/core/Cargo.toml
index b82f00311..e43b820f5 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -49,6 +49,9 @@ version = "0.53.0"
[features]
default = ["reqwest/rustls-tls", "executors-tokio", "services-memory"]
+# Enable direct IO support for fs service
+services-fs-direct-io = []
+
# Enable opendal's blocking support.
blocking = ["internal-tokio-rt"]
diff --git a/core/src/services/fs/backend.rs b/core/src/services/fs/backend.rs
index 593a00d25..eba0dc428 100644
--- a/core/src/services/fs/backend.rs
+++ b/core/src/services/fs/backend.rs
@@ -71,6 +71,20 @@ impl FsBuilder {
self
}
+
+ /// Enable direct IO mode.
+ ///
+ /// # Notes
+ ///
+ /// - Direct IO bypasses page cache for better control over memory usage
+ /// - Useful for database management systems and applications that need
+ /// precise control over buffer cache
+ /// - Requires aligned memory access and may have performance implications
+ /// - Not supported on all filesystems and platforms
+ pub fn direct_io(mut self, enabled: bool) -> Self {
+ self.config.direct_io = enabled;
+ self
+ }
}
impl Builder for FsBuilder {
@@ -117,6 +131,17 @@ impl Builder for FsBuilder {
}
}
+ // Check direct IO support
+ if self.config.direct_io {
+ #[cfg(not(all(feature = "services-fs-direct-io", target_family = "unix")))]
+ {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "Direct IO requires the 'services-fs-direct-io' feature and Unix platform",
+ ));
+ }
+ }
+
// Canonicalize the root directory. This should work since we already know that we can
// get the metadata of the path.
let root = root.canonicalize().map_err(|e| {
@@ -180,6 +205,7 @@ impl Builder for FsBuilder {
root,
atomic_write_dir,
buf_pool: oio::PooledBuf::new(16).with_initial_capacity(256 * 1024),
+ direct_io: self.config.direct_io,
}),
})
}
@@ -246,11 +272,7 @@ impl Access for FsBackend {
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let p = self.core.root.join(path.trim_end_matches('/'));
- let mut f = tokio::fs::OpenOptions::new()
- .read(true)
- .open(&p)
- .await
- .map_err(new_std_io_error)?;
+ let mut f = self.core.open_read(&p).await?;
if args.range().offset() != 0 {
use tokio::io::AsyncSeekExt;
@@ -298,37 +320,15 @@ impl Access for FsBackend {
(p, None)
};
- let mut open_options = tokio::fs::OpenOptions::new();
- if op.if_not_exists() {
- open_options.create_new(true);
- } else {
- open_options.create(true);
- }
-
- open_options.write(true);
-
- if op.append() {
- open_options.append(true);
- } else {
- open_options.truncate(true);
- }
-
- let f = open_options
- .open(tmp_path.as_ref().unwrap_or(&target_path))
- .await
- .map_err(|e| {
- match e.kind() {
- std::io::ErrorKind::AlreadyExists => {
- // Map io AlreadyExists to opendal ConditionNotMatch
- Error::new(
- ErrorKind::ConditionNotMatch,
- "The file already exists in the filesystem",
- )
- .set_source(e)
- }
- _ => new_std_io_error(e),
- }
- })?;
+ let f = self
+ .core
+ .open_write(
+ tmp_path.as_ref().unwrap_or(&target_path),
+ true, // create
+ op.append(),
+ op.if_not_exists(),
+ )
+ .await?;
let w = FsWriter::new(target_path, tmp_path, f);
@@ -446,4 +446,22 @@ mod tests {
assert!(tmp_file.starts_with(expected_prefix));
}
}
+
+ #[test]
+ fn test_fs_builder_direct_io() {
+ let builder = FsBuilder::default().root("/tmp").direct_io(true);
+ assert_eq!(builder.config.direct_io, true);
+
+ let builder = FsBuilder::default().root("/tmp").direct_io(false);
+ assert_eq!(builder.config.direct_io, false);
+ }
+
+ #[test]
+ #[cfg(not(all(feature = "services-fs-direct-io", target_family = "unix")))]
+ fn test_fs_builder_direct_io_unsupported() {
+ let builder = FsBuilder::default().root("/tmp").direct_io(true);
+ let result = builder.build();
+ assert!(result.is_err());
+ assert_eq!(result.unwrap_err().kind(), ErrorKind::Unsupported);
+ }
}
diff --git a/core/src/services/fs/config.rs b/core/src/services/fs/config.rs
index eb501ee8c..bbfbb5fb9 100644
--- a/core/src/services/fs/config.rs
+++ b/core/src/services/fs/config.rs
@@ -30,4 +30,7 @@ pub struct FsConfig {
/// tmp dir for atomic write
pub atomic_write_dir: Option<String>,
+
+ /// enable direct IO mode
+ pub direct_io: bool,
}
diff --git a/core/src/services/fs/core.rs b/core/src/services/fs/core.rs
index 20a85e233..b24237c3a 100644
--- a/core/src/services/fs/core.rs
+++ b/core/src/services/fs/core.rs
@@ -24,12 +24,16 @@ use uuid::Uuid;
use crate::raw::*;
use crate::*;
+#[cfg(all(feature = "services-fs-direct-io", target_family = "unix"))]
+const O_DIRECT: i32 = 0x4000;
+
#[derive(Debug)]
pub struct FsCore {
pub info: Arc<AccessorInfo>,
pub root: PathBuf,
pub atomic_write_dir: Option<PathBuf>,
pub buf_pool: oio::PooledBuf,
+ pub direct_io: bool,
}
impl FsCore {
@@ -60,6 +64,62 @@ impl FsCore {
Ok(p)
}
+
+ /// Open file for reading with optional direct IO
+ pub async fn open_read(&self, path: &Path) -> Result<tokio::fs::File> {
+ let mut open_options = tokio::fs::OpenOptions::new();
+ open_options.read(true);
+
+ self.apply_direct_io_flags(&mut open_options);
+
+ open_options.open(path).await.map_err(new_std_io_error)
+ }
+
+ /// Open file for writing with optional direct IO
+ pub async fn open_write(
+ &self,
+ path: &Path,
+ create: bool,
+ append: bool,
+ if_not_exists: bool,
+ ) -> Result<tokio::fs::File> {
+ let mut open_options = tokio::fs::OpenOptions::new();
+ open_options.write(true);
+
+ if if_not_exists {
+ open_options.create_new(true);
+ } else if create {
+ open_options.create(true);
+ }
+
+ if append {
+ open_options.append(true);
+ } else {
+ open_options.truncate(true);
+ }
+
+ self.apply_direct_io_flags(&mut open_options);
+
+ open_options.open(path).await.map_err(|e| match e.kind() {
+ std::io::ErrorKind::AlreadyExists => Error::new(
+ ErrorKind::ConditionNotMatch,
+ "The file already exists in the filesystem",
+ )
+ .set_source(e),
+ _ => new_std_io_error(e),
+ })
+ }
+
+ /// Apply direct IO flags to OpenOptions if enabled and supported
+ fn apply_direct_io_flags(&self, open_options: &mut tokio::fs::OpenOptions) {
+ if self.direct_io {
+ #[cfg(all(feature = "services-fs-direct-io", target_family = "unix"))]
+ {
+ use std::os::unix::fs::OpenOptionsExt;
+ open_options.custom_flags(O_DIRECT);
+ }
+ }
+ }
}
#[inline]
diff --git a/core/src/services/fs/docs.md b/core/src/services/fs/docs.md
index 10dcf9c26..183a0c813 100644
--- a/core/src/services/fs/docs.md
+++ b/core/src/services/fs/docs.md
@@ -17,9 +17,25 @@ This service can be used to:
## Configuration
- `root`: Set the work dir for backend.
--
+- `atomic_write_dir`: Set temp dir for atomic write operations.
+- `direct_io`: Enable direct IO mode to bypass page cache (requires `services-fs-direct-io` feature and Unix platform).
+
You can refer to [`FsBuilder`]'s docs for more information
+### Direct IO Support
+
+Direct IO mode bypasses the operating system's page cache, providing:
+
+- Better control over memory usage
+- Predictable I/O performance
+- Useful for database management systems
+
+**Requirements:**
+- Must enable the `services-fs-direct-io` feature flag
+- Only supported on Unix platforms (Linux, macOS, etc.)
+- Requires proper memory alignment for optimal performance
+- Not all filesystems support direct IO operations
+
## Example
### Via Builder
@@ -47,3 +63,26 @@ async fn main() -> Result<()> {
Ok(())
}
```
+
+### Direct IO Example
+
+```rust,no_run
+use std::sync::Arc;
+
+use anyhow::Result;
+use opendal::services::Fs;
+use opendal::Operator;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+ // Create fs backend builder with direct IO enabled.
+ let mut builder = Fs::default()
+ .root("/tmp")
+ // Enable direct IO mode (requires services-fs-direct-io feature)
+ .direct_io(true);
+
+ let op: Operator = Operator::new(builder)?.finish();
+
+ Ok(())
+}
+```