feat(read): add zero copy interfaces for object read

Cargo.toml

@@ -23,7 +23,6 @@ prometheus-client = { git = "https://github.com/iomesh/client_rust.git", tag = "
 strum = "0.26.2"
 strum_macros = "0.26.4"
 crossbeam-utils = "0.8.20"
-kanal = "0.1.0-pre8"
 
 [dev-dependencies]
 tokio-test = "0.4.2"

src/bin/bench/main.rs

@@ -1,4 +1,5 @@
 use std::{sync::Arc, time::Instant};
+
 use uzfs::*;
 
 #[derive(Clone, Copy, Debug)]
@@ -16,12 +17,19 @@ async fn worker(obj: u64, ds: Arc<Dataset>, blksize: u64, file_size: u64, sync:
             match op {
                 BenchOp::Write => {
                     let data = vec![1; blksize as usize];
-                    ds.write_object(&ino_hdl, offset, sync, vec![&data])
-                        .await
-                        .unwrap();
+                    ds.write_object(
+                        &ino_hdl,
+                        WriteMode::OverwriteFrom(offset),
+                        sync,
+                        vec![&data],
+                    )
+                    .await
+                    .unwrap();
                 }
                 BenchOp::Read => {
-                    ds.read_object(&ino_hdl, offset, blksize).await.unwrap();
+                    ds.read_object(&ino_hdl, offset, blksize)
+                        .await
+                        .unwrap();
                 }
             }
             ino_hdl
@@ -55,14 +63,14 @@ async fn bench(
     println!("{op:?} throughput: {throughput}MB/s");
 }
 
-#[tokio::main]
-async fn main() {
+async fn bench_main() {
     uzfs_env_init().await;
     let dev_path = std::env::args().nth(1).unwrap();
     let sync: bool = std::env::args().nth(2).unwrap().parse().unwrap();
-    let concurrency = 64;
+    let concurrency = 10;
     let blksize = 1 << 20;
-    let file_size = 1 << 30;
+    let file_size = 10 << 30;
+    config_uzfs(8 << 30, 10, true);
 
     let ds = Arc::new(
         Dataset::init("testzp/ds", &dev_path, DatasetType::Data, 0, false)
@@ -73,6 +81,7 @@ async fn main() {
     let objs = ds.create_objects(concurrency).await.unwrap().0;
 
     bench(&objs, ds.clone(), blksize, file_size, sync, BenchOp::Write).await;
+    ds.wait_synced().await;
     bench(&objs, ds.clone(), blksize, file_size, sync, BenchOp::Read).await;
 
     for obj in objs {
@@ -84,3 +93,12 @@ async fn main() {
     ds.close().await.unwrap();
     uzfs_env_fini().await;
 }
+
+fn main() {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .worker_threads(9)
+        .build()
+        .unwrap();
+    rt.block_on(bench_main());
+}

src/bindings/async_sys.rs

@@ -80,11 +80,9 @@ pub(crate) unsafe fn set_libuzfs_ops(log_func: Option<unsafe extern "C" fn(*cons
     };
 
     let aio_ops = aio_ops {
-        register_aio_fd: Some(register_fd),
-        unregister_aio_fd: Some(unregister_fd),
-        submit_aio_read: Some(submit_read),
-        submit_aio_write: Some(submit_write),
-        submit_aio_fsync: Some(submit_fsync),
+        aio_init: Some(aio_init),
+        aio_fini: Some(aio_fini),
+        submit_aio: Some(submit_aio),
     };
 
     let thread_ops = thread_ops {
@@ -532,6 +530,28 @@ pub unsafe extern "C" fn libuzfs_read_object_c(arg: *mut c_void) {
     }
 }
 
+pub struct ReadObjectZeroCopyArg {
+    pub ihp: *mut libuzfs_inode_handle_t,
+    pub offset: u64,
+    pub size: u64,
+
+    pub err: i32,
+    pub data: libuzfs_read_buf_t,
+}
+
+unsafe impl Send for ReadObjectZeroCopyArg {}
+unsafe impl Sync for ReadObjectZeroCopyArg {}
+
+pub unsafe extern "C" fn libuzfs_read_object_zero_copy_c(args: *mut c_void) {
+    let args = &mut *(args as *mut ReadObjectZeroCopyArg);
+    args.err = libuzfs_object_read_zero_copy(args.ihp, args.offset, args.size, &mut args.data);
+}
+
+pub unsafe extern "C" fn libuzfs_read_buf_rele_c(args: *mut c_void) {
+    let read_buf = args as *mut libuzfs_read_buf_t;
+    libuzfs_read_buf_rele(read_buf);
+}
+
 pub struct LibuzfsWriteObjectArg {
     pub ihp: *mut libuzfs_inode_handle_t,
     pub offset: u64,

src/context/coroutine_c.rs

@@ -1,10 +1,13 @@
-use super::coroutine::CoroutineFuture;
-use crate::bindings::sys;
+use std::time::Duration;
+
 use dashmap::DashMap;
 use libc::c_void;
 use once_cell::sync::OnceCell;
-use std::time::Duration;
-use tokio::runtime::{EnterGuard, Handle, Runtime};
+use tokio::runtime::Handle;
+
+use crate::bindings::sys;
+
+use super::coroutine::CoroutineFuture;
 
 #[allow(clippy::missing_safety_doc)]
 pub unsafe extern "C" fn co_sched_yield() {
@@ -54,9 +57,9 @@ const TS_JOINABLE: i32 = 0x00000004;
 const TS_BLOCKING: i32 = 0x00000008;
 
 static ID_TASK_HANDLE_MAP: OnceCell<DashMap<u64, tokio::task::JoinHandle<()>>> = OnceCell::new();
-static BACKGROUND_RT: OnceCell<Runtime> = OnceCell::new();
+static BACKGROUND_RT: OnceCell<tokio::runtime::Runtime> = OnceCell::new();
 
-fn enter_background_rt<'a>() -> EnterGuard<'a> {
+pub(crate) fn enter_background_rt<'a>() -> tokio::runtime::EnterGuard<'a> {
     let rt = BACKGROUND_RT.get_or_init(|| {
         tokio::runtime::Builder::new_multi_thread()
             .enable_all()