fusedev: support splice to handle FUSE requests.

1. Support enable splice_read/splice_write on FuseChannel.
2. Add splice interface for ZeroCopyReader and ZeroCopyWriter.
3. Add unit-test cases for splice interface.

Signed-off-by: henry.hj <henry.hj@antgroup.com>

Author: henry.hj

src/api/filesystem/mod.rs

@@ -26,6 +26,8 @@ use crate::abi::fuse_abi::{ino64_t, stat64};
 mod async_io;
 #[cfg(feature = "async-io")]
 pub use async_io::{AsyncFileSystem, AsyncZeroCopyReader, AsyncZeroCopyWriter};
+#[cfg(all(target_os = "linux", feature = "fusedev"))]
+use std::os::unix::io::{AsRawFd, FromRawFd};
 
 mod sync_io;
 pub use sync_io::FileSystem;
@@ -208,6 +210,16 @@ pub trait ZeroCopyReader: io::Read {
         off: u64,
     ) -> io::Result<usize>;
 
+    /// Copies at most `count` bytes from `self` directly into `f` at offset `off` with less data copy
+    /// `f` could be local file description or tcp socket
+    #[cfg(all(target_os = "linux", feature = "fusedev"))]
+    fn splice_to(&mut self, f: &dyn AsRawFd, count: usize, off: u64) -> io::Result<usize> {
+        let mut file = unsafe { std::fs::File::from_raw_fd(f.as_raw_fd()) };
+        let res = self.read_to(&mut file, count, off);
+        std::mem::forget(file);
+        res
+    }
+
     /// Copies exactly `count` bytes of data from `self` into `f` at offset `off`. `off + count`
     /// must be less than `u64::MAX`.
     ///
@@ -251,6 +263,46 @@ pub trait ZeroCopyReader: io::Read {
         Ok(())
     }
 
+    /// Copies exactly `count` bytes of data from `self` into `f` at offset `off`. `off + count`
+    /// must be less than `u64::MAX`.
+    /// `f` could be local file description or tcp socket
+    #[cfg(all(target_os = "linux", feature = "fusedev"))]
+    fn splice_exact_to(
+        &mut self,
+        f: &mut dyn AsRawFd,
+        mut count: usize,
+        mut off: u64,
+    ) -> io::Result<()> {
+        let c = count
+            .try_into()
+            .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
+        if off.checked_add(c).is_none() {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "`off` + `count` must be less than u64::MAX",
+            ));
+        }
+
+        while count > 0 {
+            match self.splice_to(f, count, off) {
+                Ok(0) => {
+                    return Err(io::Error::new(
+                        io::ErrorKind::WriteZero,
+                        "failed to fill whole buffer",
+                    ))
+                }
+                Ok(n) => {
+                    count -= n;
+                    off += n as u64;
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
+                Err(e) => return Err(e),
+            }
+        }
+
+        Ok(())
+    }
+
     /// Copies all remaining bytes from `self` into `f` at offset `off`. Equivalent to repeatedly
     /// calling `read_to` until it returns either `Ok(0)` or a non-`ErrorKind::Interrupted` error.
     ///
@@ -275,6 +327,24 @@ pub trait ZeroCopyReader: io::Read {
             }
         }
     }
+
+    /// Copies all remaining bytes from `self` into `f` at offset `off`.
+    /// `f` could be local file description or tcp socket
+    #[cfg(all(target_os = "linux", feature = "fusedev"))]
+    fn splice_to_end(&mut self, f: &mut dyn AsRawFd, mut off: u64) -> io::Result<usize> {
+        let mut out = 0;
+        loop {
+            match self.splice_to(f, ::std::usize::MAX, off) {
+                Ok(0) => return Ok(out),
+                Ok(n) => {
+                    off = off.saturating_add(n as u64);
+                    out += n;
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
+                Err(e) => return Err(e),
+            }
+        }
+    }
 }
 
 /// A trait for directly copying data from a `File` into the fuse transport without first storing
@@ -300,6 +370,25 @@ pub trait ZeroCopyWriter: io::Write {
         off: u64,
     ) -> io::Result<usize>;
 
+    /// Append `count` bytes data from fd `f` at offset `off`
+    /// `f` should be file description or socket
+    /// This data always at the end of all data, only available for bufferd writer.
+    /// For example:
+    ///     We already write "aaa to writer. Then we append fd buf witch contains "bbb".
+    /// Finally we write "ccc" to writer. The final data is "aaacccbbb".
+    ///
+    /// # Errors
+    /// EINVAL: writer doesn't support this operation, should fallback to use `write_from`.
+    #[cfg(all(target_os = "linux", feature = "fusedev"))]
+    fn append_fd_buf(
+        &mut self,
+        _f: &dyn AsRawFd,
+        _count: usize,
+        _off: Option<u64>,
+    ) -> io::Result<usize> {
+        Err(io::Error::from_raw_os_error(libc::EINVAL))
+    }
+
     /// Copies exactly `count` bytes of data from `f` at offset `off` into `self`. `off + count`
     /// must be less than `u64::MAX`.
     ///

src/api/server/async_io.rs

@@ -60,15 +60,15 @@ impl<'a, S: BitmapSlice> io::Read for AsyncZcReader<'a, S> {
     }
 }
 
-struct AsyncZcWriter<'a, S: BitmapSlice = ()>(Writer<'a, S>);
+struct AsyncZcWriter<'a, 'b, S: BitmapSlice = ()>(Writer<'a, 'b, S>);
 
 // The underlying VolatileSlice contains "*mut u8", which is just a pointer to a u8 array.
 // Actually we rely on the AsyncExecutor is a single-threaded worker, and we do not really send
 // 'Reader' to other threads.
-unsafe impl<'a, S: BitmapSlice> Send for AsyncZcWriter<'a, S> {}
+unsafe impl<'a, 'b, S: BitmapSlice> Send for AsyncZcWriter<'a, 'b, S> {}
 
 #[async_trait(?Send)]
-impl<'a, S: BitmapSlice> AsyncZeroCopyWriter for AsyncZcWriter<'a, S> {
+impl<'a, 'b, S: BitmapSlice> AsyncZeroCopyWriter for AsyncZcWriter<'a, 'b, S> {
     async fn async_write_from(
         &mut self,
         f: Arc<dyn AsyncFileReadWriteVolatile>,
@@ -79,7 +79,7 @@ impl<'a, S: BitmapSlice> AsyncZeroCopyWriter for AsyncZcWriter<'a, S> {
     }
 }
 
-impl<'a, S: BitmapSlice> ZeroCopyWriter for AsyncZcWriter<'a, S> {
+impl<'a, 'b, S: BitmapSlice> ZeroCopyWriter for AsyncZcWriter<'a, 'b, S> {
     fn write_from(
         &mut self,
         f: &mut dyn FileReadWriteVolatile,
@@ -90,7 +90,7 @@ impl<'a, S: BitmapSlice> ZeroCopyWriter for AsyncZcWriter<'a, S> {
     }
 }
 
-impl<'a, S: BitmapSlice> io::Write for AsyncZcWriter<'a, S> {
+impl<'a, 'b, S: BitmapSlice> io::Write for AsyncZcWriter<'a, 'b, S> {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         self.0.write(buf)
     }
@@ -116,7 +116,7 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
     pub async unsafe fn async_handle_message<S: BitmapSlice>(
         &self,
         mut r: Reader<'_, S>,
-        w: Writer<'_, S>,
+        w: Writer<'_, '_, S>,
         vu_req: Option<&mut dyn FsCacheReqHandler>,
         hook: Option<&dyn MetricsHook>,
     ) -> Result<usize> {
@@ -210,10 +210,13 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         res
     }
 
-    async fn async_lookup<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_lookup<S: BitmapSlice>(
+        &self,
+        mut ctx: SrvContext<'_, '_, F, S>,
+    ) -> Result<usize> {
         let buf = ServerUtil::get_message_body(&mut ctx.r, &ctx.in_header, 0)?;
         let name = bytes_to_cstr(buf.as_ref())?;
-        let version = self.vers.load();
+        let version = &self.meta.load().version;
         let result = self
             .fs
             .async_lookup(ctx.context(), ctx.nodeid(), name)
@@ -236,7 +239,10 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         }
     }
 
-    async fn async_getattr<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_getattr<S: BitmapSlice>(
+        &self,
+        mut ctx: SrvContext<'_, '_, F, S>,
+    ) -> Result<usize> {
         let GetattrIn { flags, fh, .. } = ctx.r.read_obj().map_err(Error::DecodeMessage)?;
         let handle = if (flags & GETATTR_FH) != 0 {
             Some(fh.into())
@@ -251,7 +257,10 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         ctx.async_handle_attr_result(result).await
     }
 
-    async fn async_setattr<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_setattr<S: BitmapSlice>(
+        &self,
+        mut ctx: SrvContext<'_, '_, F, S>,
+    ) -> Result<usize> {
         let setattr_in: SetattrIn = ctx.r.read_obj().map_err(Error::DecodeMessage)?;
         let handle = if setattr_in.valid & FATTR_FH != 0 {
             Some(setattr_in.fh.into())
@@ -268,7 +277,7 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         ctx.async_handle_attr_result(result).await
     }
 
-    async fn async_open<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_open<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, '_, F, S>) -> Result<usize> {
         let OpenIn { flags, fuse_flags } = ctx.r.read_obj().map_err(Error::DecodeMessage)?;
         let result = self
             .fs
@@ -289,7 +298,7 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         }
     }
 
-    async fn async_read<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_read<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, '_, F, S>) -> Result<usize> {
         let ReadIn {
             fh,
             offset,
@@ -347,7 +356,7 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
                     .await
                     .map_err(Error::EncodeMessage)?;
                 ctx.w
-                    .async_commit(Some(&data_writer.0))
+                    .async_commit(Some(&mut data_writer.0))
                     .await
                     .map_err(Error::EncodeMessage)?;
                 Ok(out.len as usize)
@@ -356,7 +365,10 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         }
     }
 
-    async fn async_write<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_write<S: BitmapSlice>(
+        &self,
+        mut ctx: SrvContext<'_, '_, F, S>,
+    ) -> Result<usize> {
         let WriteIn {
             fh,
             offset,
@@ -408,7 +420,10 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         }
     }
 
-    async fn async_fsync<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_fsync<S: BitmapSlice>(
+        &self,
+        mut ctx: SrvContext<'_, '_, F, S>,
+    ) -> Result<usize> {
         let FsyncIn {
             fh, fsync_flags, ..
         } = ctx.r.read_obj().map_err(Error::DecodeMessage)?;
@@ -424,7 +439,10 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         }
     }
 
-    async fn async_fsyncdir<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_fsyncdir<S: BitmapSlice>(
+        &self,
+        mut ctx: SrvContext<'_, '_, F, S>,
+    ) -> Result<usize> {
         let FsyncIn {
             fh, fsync_flags, ..
         } = ctx.r.read_obj().map_err(Error::DecodeMessage)?;
@@ -440,7 +458,10 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
         }
     }
 
-    async fn async_create<S: BitmapSlice>(&self, mut ctx: SrvContext<'_, F, S>) -> Result<usize> {
+    async fn async_create<S: BitmapSlice>(
+        &self,
+        mut ctx: SrvContext<'_, '_, F, S>,
+    ) -> Result<usize> {
         let args: CreateIn = ctx.r.read_obj().map_err(Error::DecodeMessage)?;
         let buf = ServerUtil::get_message_body(&mut ctx.r, &ctx.in_header, size_of::<CreateIn>())?;
         let name = bytes_to_cstr(&buf)?;
@@ -476,7 +497,7 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
 
     async fn async_fallocate<S: BitmapSlice>(
         &self,
-        mut ctx: SrvContext<'_, F, S>,
+        mut ctx: SrvContext<'_, '_, F, S>,
     ) -> Result<usize> {
         let FallocateIn {
             fh,
@@ -497,7 +518,7 @@ impl<F: AsyncFileSystem + Sync> Server<F> {
     }
 }
 
-impl<'a, F: AsyncFileSystem, S: BitmapSlice> SrvContext<'a, F, S> {
+impl<'a, 'b, F: AsyncFileSystem, S: BitmapSlice> SrvContext<'a, 'b, F, S> {
     async fn async_reply_ok<T: ByteValued>(
         &mut self,
         out: Option<T>,