feat: fs add concurrent write

core/src/services/fs/backend.rs

@@ -28,6 +28,7 @@ use super::core::*;
 use super::lister::FsLister;
 use super::reader::FsReader;
 use super::writer::FsWriter;
+use super::writer::FsWriters;
 use crate::raw::*;
 use crate::*;
 
@@ -169,7 +170,7 @@ pub struct FsBackend {
 
 impl Access for FsBackend {
  type Reader = FsReader<tokio::fs::File>;
- type Writer = FsWriter<tokio::fs::File>;
+ type Writer = FsWriters;
  type Lister = Option<FsLister<tokio::fs::ReadDir>>;
  type BlockingReader = FsReader<std::fs::File>;
  type BlockingWriter = FsWriter<std::fs::File>;
@@ -313,7 +314,20 @@ impl Access for FsBackend {
  .await
  .map_err(new_std_io_error)?;
 
- Ok((RpWrite::new(), FsWriter::new(target_path, tmp_path, f)))
+ let w = FsWriter::new(target_path, tmp_path, f);
+
+ let w = if op.append() {
+ FsWriters::One(w)
+ } else {
+ debug!("write with concurrent: {}", op.concurrent());
+ FsWriters::Two(oio::PositionWriter::new(
+ w,
+ op.executor().cloned(),
+ op.concurrent(),
+ ))
+ };
+
+ Ok((RpWrite::default(), w))
  }
 
  async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {

core/src/services/fs/writer.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use std::io::Write;
+use std::os::unix::fs::FileExt;
 use std::path::PathBuf;
 
 use bytes::Buf;
@@ -24,6 +25,9 @@ use tokio::io::AsyncWriteExt;
 use crate::raw::*;
 use crate::*;
 
+pub type FsWriters =
+ TwoWays<FsWriter<tokio::fs::File>, oio::PositionWriter<FsWriter<tokio::fs::File>>>;
+
 pub struct FsWriter<F> {
  target_path: PathBuf,
  tmp_path: Option<PathBuf>,
@@ -101,3 +105,23 @@ impl oio::BlockingWrite for FsWriter<std::fs::File> {
  Ok(())
  }
 }
+
+impl oio::PositionWrite for FsWriter<tokio::fs::File> {
+ async fn write_all_at(&self, offset: u64, buf: Buffer) -> Result<()> {
+ let f = self.f.as_ref().expect("FsWriter must be initialized");
+
+ let f = f
+ .try_clone()
+ .await
+ .map_err(new_std_io_error)?
+ .into_std()
+ .await;
+
+ tokio::task::spawn_blocking(move || {
+ f.write_all_at(buf.chunk(), offset)
+ .map_err(new_std_io_error)
+ })
+ .await
+ .map_err(new_task_join_error)?
+ }
+}