This is an automated email from the ASF dual-hosted git repository. erickguan pushed a commit to branch services-fs in repository https://gitbox.apache.org/repos/asf/opendal.git
commit e598ca27631d46533001d56f6ef8f325a2e78e9a Author: Erick Guan <[email protected]> AuthorDate: Thu Jun 19 21:49:37 2025 +0200 refactor(services/fs): extract implementation to core --- core/src/services/fs/backend.rs | 202 +++------------------------ core/src/services/fs/core.rs | 166 ++++++++++++++++++++++ core/src/services/fs/delete.rs | 6 +- core/src/services/fs/{delete.rs => error.rs} | 51 +++---- core/src/services/fs/mod.rs | 2 + core/src/services/fs/reader.rs | 36 +---- 6 files changed, 214 insertions(+), 249 deletions(-) diff --git a/core/src/services/fs/backend.rs b/core/src/services/fs/backend.rs index 593a00d25..a06ab0408 100644 --- a/core/src/services/fs/backend.rs +++ b/core/src/services/fs/backend.rs @@ -15,11 +15,9 @@ // specific language governing permissions and limitations // under the License. -use std::io::SeekFrom; use std::path::PathBuf; use std::sync::Arc; -use chrono::DateTime; use log::debug; use super::core::*; @@ -120,13 +118,7 @@ impl Builder for FsBuilder { // 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| { - Error::new( - ErrorKind::Unexpected, - "canonicalize of root directory failed", - ) - .with_operation("Builder::build") - .with_context("root", root.to_string_lossy()) - .set_source(e) + Error::new(ErrorKind::Unexpected,"canonicalize of root directory failed").set_source(e) })?; // Canonicalize the atomic_write_dir directory. This should work since we already know that @@ -134,10 +126,7 @@ impl Builder for FsBuilder { let atomic_write_dir = atomic_write_dir .map(|p| { p.canonicalize().map(Some).map_err(|e| { - Error::new( - ErrorKind::Unexpected, - "canonicalize of atomic_write_dir directory failed", - ) + Error::new(ErrorKind::Unexpected, "canonicalize of atomic_write_dir directory failed") .with_operation("Builder::build") .with_context("root", root.to_string_lossy()) .set_source(e) @@ -202,35 +191,12 @@ impl Access for FsBackend { } async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> { - let p = self.core.root.join(path.trim_end_matches('/')); - - tokio::fs::create_dir_all(&p) - .await - .map_err(new_std_io_error)?; - + self.core.fs_create_dir(path).await?; Ok(RpCreateDir::default()) } async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> { - let p = self.core.root.join(path.trim_end_matches('/')); - - let meta = tokio::fs::metadata(&p).await.map_err(new_std_io_error)?; - - let mode = if meta.is_dir() { - EntryMode::DIR - } else if meta.is_file() { - EntryMode::FILE - } else { - EntryMode::Unknown - }; - let m = Metadata::new(mode) - .with_content_length(meta.len()) - .with_last_modified( - meta.modified() - .map(DateTime::from) - .map_err(new_std_io_error)?, - ); - + let m = self.core.fs_stat(path).await?; Ok(RpStat::new(m)) } @@ -244,22 +210,7 @@ impl Access for FsBackend { /// /// Benchmark could be found [here](https://gist.github.com/Xuanwo/48f9cfbc3022ea5f865388bb62e1a70f) 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)?; - - if args.range().offset() != 0 { - use tokio::io::AsyncSeekExt; - - f.seek(SeekFrom::Start(args.range().offset())) - .await - .map_err(new_std_io_error)?; - } - + let f = self.core.fs_read(path, &args).await?; let r = FsReader::new( self.core.clone(), f, @@ -269,80 +220,22 @@ impl Access for FsBackend { } async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> { - let (target_path, tmp_path) = if let Some(atomic_write_dir) = &self.core.atomic_write_dir { - let target_path = self - .core - .ensure_write_abs_path(&self.core.root, path) - .await?; - let tmp_path = self - .core - .ensure_write_abs_path(atomic_write_dir, &tmp_file_of(path)) - .await?; - - // If the target file exists, we should append to the end of it directly. - if op.append() - && tokio::fs::try_exists(&target_path) - .await - .map_err(new_std_io_error)? - { - (target_path, None) - } else { - (target_path, Some(tmp_path)) - } - } else { - let p = self - .core - .ensure_write_abs_path(&self.core.root, path) - .await?; - - (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); - } + let (target_path, tmp_path) = self.core.prepare_write(path, &op).await?; + let file = self.core.fs_write(&target_path, tmp_path.as_ref(), &op).await?; - open_options.write(true); + let writer = FsWriter::new(target_path, tmp_path, file); - 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 w = FsWriter::new(target_path, tmp_path, f); - - let w = if op.append() { - FsWriters::One(w) + let writer = if op.append() { + FsWriters::One(writer) } else { FsWriters::Two(oio::PositionWriter::new( self.info().clone(), - w, + writer, op.concurrent(), )) }; - Ok((RpWrite::default(), w)) + Ok((RpWrite::default(), writer)) } async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> { @@ -353,77 +246,22 @@ impl Access for FsBackend { } async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> { - let p = self.core.root.join(path.trim_end_matches('/')); - - let f = match tokio::fs::read_dir(&p).await { - Ok(rd) => rd, - Err(e) => { - return match e.kind() { - // Return empty list if the directory not found - std::io::ErrorKind::NotFound => Ok((RpList::default(), None)), - // TODO: enable after our MSRV has been raised to 1.83 - // - // If the path is not a directory, return an empty list - // - // The path could be a file or a symbolic link in this case. - // Returning a NotADirectory error to the user isn't helpful; instead, - // providing an empty directory is a more user-friendly. In fact, the dir - // `path/` does not exist. - // std::io::ErrorKind::NotADirectory => Ok((RpList::default(), None)), - _ => { - // TODO: remove this after we have MSRV 1.83 - #[cfg(unix)] - if e.raw_os_error() == Some(20) { - // On unix 20: Not a directory - return Ok((RpList::default(), None)); - } - #[cfg(windows)] - if e.raw_os_error() == Some(267) { - // On windows 267: DIRECTORY - return Ok((RpList::default(), None)); - } - - Err(new_std_io_error(e)) - } - }; + match self.core.fs_list(path).await? { + Some(f) => { + let rd = FsLister::new(&self.core.root, path, f); + Ok((RpList::default(), Some(rd))) } - }; - - let rd = FsLister::new(&self.core.root, path, f); - Ok((RpList::default(), Some(rd))) + None => Ok((RpList::default(), None)), + } } async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> { - let from = self.core.root.join(from.trim_end_matches('/')); - - // try to get the metadata of the source file to ensure it exists - tokio::fs::metadata(&from).await.map_err(new_std_io_error)?; - - let to = self - .core - .ensure_write_abs_path(&self.core.root, to.trim_end_matches('/')) - .await?; - - tokio::fs::copy(from, to).await.map_err(new_std_io_error)?; - + self.core.fs_copy(from, to).await?; Ok(RpCopy::default()) } async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> { - let from = self.core.root.join(from.trim_end_matches('/')); - - // try to get the metadata of the source file to ensure it exists - tokio::fs::metadata(&from).await.map_err(new_std_io_error)?; - - let to = self - .core - .ensure_write_abs_path(&self.core.root, to.trim_end_matches('/')) - .await?; - - tokio::fs::rename(from, to) - .await - .map_err(new_std_io_error)?; - + self.core.fs_rename(from, to).await?; Ok(RpRename::default()) } } diff --git a/core/src/services/fs/core.rs b/core/src/services/fs/core.rs index 20a85e233..2fe67fa75 100644 --- a/core/src/services/fs/core.rs +++ b/core/src/services/fs/core.rs @@ -15,12 +15,15 @@ // specific language governing permissions and limitations // under the License. +use std::io::SeekFrom; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use chrono::DateTime; use uuid::Uuid; +use super::error::*; use crate::raw::*; use crate::*; @@ -60,6 +63,169 @@ impl FsCore { Ok(p) } + + pub async fn fs_create_dir(&self, path: &str) -> Result<()> { + let p = self.root.join(path.trim_end_matches('/')); + tokio::fs::create_dir_all(&p) + .await + .map_err(new_std_io_error)?; + Ok(()) + } + + pub async fn fs_stat(&self, path: &str) -> Result<Metadata> { + let p = self.root.join(path.trim_end_matches('/')); + let meta = tokio::fs::metadata(&p).await.map_err(new_std_io_error)?; + + let mode = if meta.is_dir() { + EntryMode::DIR + } else if meta.is_file() { + EntryMode::FILE + } else { + EntryMode::Unknown + }; + let m = Metadata::new(mode) + .with_content_length(meta.len()) + .with_last_modified( + meta.modified() + .map(DateTime::from) + .map_err(new_std_io_error)?, + ); + + Ok(m) + } + + pub async fn fs_read(&self, path: &str, args: &OpRead) -> Result<tokio::fs::File> { + let p = self.root.join(path.trim_end_matches('/')); + + let mut f = tokio::fs::OpenOptions::new() + .read(true) + .open(&p) + .await + .map_err(new_std_io_error)?; + + if args.range().offset() != 0 { + use tokio::io::AsyncSeekExt; + f.seek(SeekFrom::Start(args.range().offset())) + .await + .map_err(new_std_io_error)?; + } + + Ok(f) + } + + pub async fn prepare_write(&self, path: &str, op: &OpWrite) -> Result<(PathBuf, Option<PathBuf>)> { + let (target_path, tmp_path) = if let Some(atomic_write_dir) = &self.atomic_write_dir { + let target_path = self.ensure_write_abs_path(&self.root, path).await?; + let tmp_path = self + .ensure_write_abs_path(atomic_write_dir, &tmp_file_of(path)) + .await?; + + // If the target file exists, we should append to the end of it directly. + if op.append() + && tokio::fs::try_exists(&target_path) + .await + .map_err(new_std_io_error)? + { + (target_path, None) + } else { + (target_path, Some(tmp_path)) + } + } else { + let p = self.ensure_write_abs_path(&self.root, path).await?; + (p, None) + }; + + Ok((target_path, tmp_path)) + } + + pub async fn fs_write(&self, target_path: &PathBuf, tmp_path: Option<&PathBuf>, op: &OpWrite) -> Result<tokio::fs::File> { + 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.unwrap_or(target_path)) + .await + .map_err(new_fs_io_error)?; + + Ok(f) + } + + pub async fn fs_list(&self, path: &str) -> Result<Option<tokio::fs::ReadDir>> { + let p = self.root.join(path.trim_end_matches('/')); + + match tokio::fs::read_dir(&p).await { + Ok(rd) => Ok(Some(rd)), + Err(e) => { + match e.kind() { + // Return empty list if the directory not found + std::io::ErrorKind::NotFound => Ok(None), + // TODO: enable after our MSRV has been raised to 1.83 + // + // If the path is not a directory, return an empty list + // + // The path could be a file or a symbolic link in this case. + // Returning a NotADirectory error to the user isn't helpful; instead, + // providing an empty directory is a more user-friendly. In fact, the dir + // `path/` does not exist. + // std::io::ErrorKind::NotADirectory => Ok((RpList::default(), None)), + _ => { + // TODO: remove this after we have MSRV 1.83 + #[cfg(unix)] + if e.raw_os_error() == Some(20) { + // On unix 20: Not a directory + return Ok(None); + } + #[cfg(windows)] + if e.raw_os_error() == Some(267) { + // On windows 267: DIRECTORY + return Ok(None); + } + + Err(new_std_io_error(e)) + } + } + } + } + } + + pub async fn fs_copy(&self, from: &str, to: &str) -> Result<()> { + let from = self.root.join(from.trim_end_matches('/')); + // try to get the metadata of the source file to ensure it exists + tokio::fs::metadata(&from).await.map_err(new_std_io_error)?; + + let to = self + .ensure_write_abs_path(&self.root, to.trim_end_matches('/')) + .await?; + + tokio::fs::copy(from, to).await.map_err(new_std_io_error)?; + Ok(()) + } + + pub async fn fs_rename(&self, from: &str, to: &str) -> Result<()> { + let from = self.root.join(from.trim_end_matches('/')); + tokio::fs::metadata(&from).await.map_err(new_std_io_error)?; + + let to = self + .ensure_write_abs_path(&self.root, to.trim_end_matches('/')) + .await?; + + tokio::fs::rename(from, to) + .await + .map_err(new_std_io_error)?; + Ok(()) + } } #[inline] diff --git a/core/src/services/fs/delete.rs b/core/src/services/fs/delete.rs index f14e6d13e..40040d984 100644 --- a/core/src/services/fs/delete.rs +++ b/core/src/services/fs/delete.rs @@ -18,6 +18,7 @@ use std::sync::Arc; use super::core::*; +use super::error::*; use crate::raw::*; use crate::*; @@ -40,11 +41,10 @@ impl oio::OneShotDelete for FsDeleter { match meta { Ok(meta) => { if meta.is_dir() { - tokio::fs::remove_dir(&p).await.map_err(new_std_io_error)?; + ignore_not_found(tokio::fs::remove_dir(&p).await, ())?; } else { - tokio::fs::remove_file(&p).await.map_err(new_std_io_error)?; + ignore_not_found(tokio::fs::remove_file(&p).await, ())?; } - Ok(()) } Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()), diff --git a/core/src/services/fs/delete.rs b/core/src/services/fs/error.rs similarity index 50% copy from core/src/services/fs/delete.rs copy to core/src/services/fs/error.rs index f14e6d13e..08f804b80 100644 --- a/core/src/services/fs/delete.rs +++ b/core/src/services/fs/error.rs @@ -15,40 +15,33 @@ // specific language governing permissions and limitations // under the License. -use std::sync::Arc; - -use super::core::*; use crate::raw::*; use crate::*; -pub struct FsDeleter { - core: Arc<FsCore>, -} - -impl FsDeleter { - pub fn new(core: Arc<FsCore>) -> Self { - Self { core } +#[cfg(unix)] +const UNIX_NOT_A_DIRECTORY: i32 = 20; +#[cfg(windows)] +const WINDOWS_NOT_A_DIRECTORY: i32 = 267; + +/// Enhanced filesystem-specific I/O error conversion with semantic mappings +pub fn new_fs_io_error(e: std::io::Error) -> Error { + 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), } } -impl oio::OneShotDelete for FsDeleter { - async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> { - let p = self.core.root.join(path.trim_end_matches('/')); - - let meta = tokio::fs::metadata(&p).await; - - match meta { - Ok(meta) => { - if meta.is_dir() { - tokio::fs::remove_dir(&p).await.map_err(new_std_io_error)?; - } else { - tokio::fs::remove_file(&p).await.map_err(new_std_io_error)?; - } - - Ok(()) - } - Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()), - Err(err) => Err(new_std_io_error(err)), - } +/// Convert "not found" errors to success for idempotent operations like delete +pub fn ignore_not_found<T>(result: std::io::Result<T>, default: T) -> Result<T> { + match result { + Ok(value) => Ok(value), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(default), + Err(e) => Err(new_std_io_error(e)), } } diff --git a/core/src/services/fs/mod.rs b/core/src/services/fs/mod.rs index caf858386..cbfefd3cf 100644 --- a/core/src/services/fs/mod.rs +++ b/core/src/services/fs/mod.rs @@ -20,6 +20,8 @@ mod core; #[cfg(feature = "services-fs")] mod delete; #[cfg(feature = "services-fs")] +mod error; +#[cfg(feature = "services-fs")] mod lister; #[cfg(feature = "services-fs")] mod reader; diff --git a/core/src/services/fs/reader.rs b/core/src/services/fs/reader.rs index 3bcca401e..9a7b41ce6 100644 --- a/core/src/services/fs/reader.rs +++ b/core/src/services/fs/reader.rs @@ -17,9 +17,6 @@ use std::sync::Arc; -use tokio::io::AsyncReadExt; -use tokio::io::ReadBuf; - use super::core::*; use crate::raw::*; use crate::*; @@ -47,37 +44,6 @@ impl<F> FsReader<F> { impl oio::Read for FsReader<tokio::fs::File> { async fn read(&mut self) -> Result<Buffer> { - if self.read >= self.size { - return Ok(Buffer::new()); - } - - let mut bs = self.core.buf_pool.get(); - bs.reserve(self.buf_size); - - let size = (self.size - self.read).min(self.buf_size); - let buf = &mut bs.spare_capacity_mut()[..size]; - let mut read_buf: ReadBuf = ReadBuf::uninit(buf); - - // SAFETY: Read at most `limit` bytes into `read_buf`. - unsafe { - read_buf.assume_init(size); - } - - let n = self - .f - .read_buf(&mut read_buf) - .await - .map_err(new_std_io_error)?; - self.read += n; - - // Safety: We make sure that bs contains `n` more bytes. - let filled = read_buf.filled().len(); - unsafe { bs.set_len(filled) } - - let frozen = bs.split().freeze(); - // Return the buffer to the pool. - self.core.buf_pool.put(bs); - - Ok(Buffer::from(frozen)) + self.core.fs_read_buffer(&mut self.f, &mut self.read, self.size, self.buf_size).await } }
