From 05182b0dc3055ba2fc0c450b740e62d3d38112a8 Mon Sep 17 00:00:00 2001
From: George Miao <gm@miao.dev>
Date: Wed, 5 Jun 2024 22:44:10 +0900
Subject: [PATCH 1/6] basic impl

---
 core/src/services/compfs/backend.rs |  70 ++++++++++++++-
 core/src/services/compfs/core.rs    | 133 ++++++++--------------------
 core/src/services/compfs/lister.rs  |  65 ++++++++++++++
 core/src/services/compfs/reader.rs  |  39 ++++++++
 core/src/services/compfs/writer.rs  |  46 ++++++++++
 5 files changed, 254 insertions(+), 99 deletions(-)

diff --git a/core/src/services/compfs/backend.rs b/core/src/services/compfs/backend.rs
index fcfa7ba67f09..69beb8407165 100644
--- a/core/src/services/compfs/backend.rs
+++ b/core/src/services/compfs/backend.rs
@@ -60,13 +60,13 @@ impl Builder for CompfsBuilder {
 
 #[derive(Debug)]
 pub struct CompfsBackend {
-    rt: CompioThread,
+    core: Arc<CompfsCore>,
 }
 
 impl Access for CompfsBackend {
-    type Reader = ();
-    type Writer = ();
-    type Lister = ();
+    type Reader = CompfsReader;
+    type Writer = CompfsWriter;
+    type Lister = Option<CompfsLister>;
     type BlockingReader = ();
     type BlockingWriter = ();
     type BlockingLister = ();
@@ -74,4 +74,66 @@ impl Access for CompfsBackend {
     fn info(&self) -> AccessorInfo {
         todo!()
     }
+
+    async fn read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
+        let path = self.core.root.join(path.trim_end_matches('/'));
+
+        let file = self
+            .core
+            .exec(|| async move { compio::fs::OpenOptions::new().read(true).open(&path).await })
+            .await?;
+
+        let r = CompfsReader::new(self.core.clone(), file);
+        Ok((RpRead::new(), r))
+    }
+
+    async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let path = self.core.root.join(path.trim_end_matches('/'));
+        let append = args.append();
+        let mut file = self
+            .core
+            .exec(move || async move {
+                compio::fs::OpenOptions::new()
+                    .create(true)
+                    .write(true)
+                    .truncate(!append)
+                    .open(path)
+                    .await
+            })
+            .await
+            .map(Cursor::new)?;
+
+        if append {
+            let f = file.clone();
+            let metadata = self
+                .core
+                .exec(|| async move { f.get_ref().metadata().await })
+                .await?;
+            file.set_position(metadata.len());
+        }
+
+        let w = CompfsWriter::new(self.core.clone(), file);
+        Ok((RpWrite::new(), w))
+    }
+
+    async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
+        let path = self.core.root.join(path.trim_end_matches('/'));
+        let read_dir = match self
+            .core
+            .exec_blocking(move || std::fs::read_dir(&path))
+            .await?
+        {
+            Ok(rd) => rd,
+            Err(e) => {
+                return if e.kind() == std::io::ErrorKind::NotFound {
+                    Ok((RpList::default(), None))
+                } else {
+                    Err(new_std_io_error(e))
+                };
+            }
+        };
+
+        let lister = CompfsLister::new(self.core.clone(), read_dir);
+        Ok((RpList::default(), Some(lister)))
+    }
 }
diff --git a/core/src/services/compfs/core.rs b/core/src/services/compfs/core.rs
index e6a462402cb9..60c68106f154 100644
--- a/core/src/services/compfs/core.rs
+++ b/core/src/services/compfs/core.rs
@@ -15,116 +15,59 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::future::Future;
-use std::thread::JoinHandle;
-
-use compio::buf::IoBuf;
-use compio::runtime::RuntimeBuilder;
-use futures::channel::mpsc;
-use futures::channel::mpsc::SendError;
-use futures::channel::oneshot;
-use futures::future::LocalBoxFuture;
-use futures::SinkExt;
-use futures::StreamExt;
-
-use crate::Buffer;
-
-/// This is arbitrary, but since all tasks are spawned instantly, we shouldn't need a too big buffer.
-const CHANNEL_SIZE: usize = 4;
-
-fn task<F, Fut, T>(func: F) -> (Task<T>, SpawnTask)
-where
-    F: (FnOnce() -> Fut) + Send + 'static,
-    Fut: Future<Output = T>,
-    T: Send + 'static,
-{
-    let (tx, recv) = oneshot::channel();
-
-    let boxed = Box::new(|| {
-        Box::pin(async move {
-            let res = func().await;
-            tx.send(res).ok();
-        }) as _
-    });
-
-    (Task(recv), SpawnTask(boxed))
-}
+use compio::{buf::IoBuf, dispatcher::Dispatcher};
+use std::{future::Future, path::PathBuf};
 
-/// A task handle that can be used to retrieve result spawned into [`CompioThread`].
-pub struct Task<T>(oneshot::Receiver<T>);
+use crate::raw::*;
+use crate::*;
 
-/// Type erased task that can be spawned into a [`CompioThread`].
-struct SpawnTask(Box<dyn (FnOnce() -> LocalBoxFuture<'static, ()>) + Send>);
+unsafe impl IoBuf for Buffer {
+    fn as_buf_ptr(&self) -> *const u8 {
+        self.current().as_ptr()
+    }
 
-impl SpawnTask {
-    fn call(self) -> LocalBoxFuture<'static, ()> {
-        (self.0)()
+    fn buf_len(&self) -> usize {
+        self.current().len()
+    }
+
+    fn buf_capacity(&self) -> usize {
+        // `Bytes` doesn't expose uninitialized capacity, so treat it as the same as `len`
+        self.current().len()
     }
 }
 
 #[derive(Debug)]
-pub struct CompioThread {
-    thread: JoinHandle<()>,
-    handle: SpawnHandle,
+pub(super) struct CompfsCore {
+    pub root: PathBuf,
+    pub dispatcher: Dispatcher,
+    pub buf_pool: oio::PooledBuf,
 }
 
-impl CompioThread {
-    pub fn new(builder: RuntimeBuilder) -> Self {
-        let (send, mut recv) = mpsc::channel(CHANNEL_SIZE);
-        let handle = SpawnHandle(send);
-        let thread = std::thread::spawn(move || {
-            let rt = builder.build().expect("failed to create runtime");
-            rt.block_on(async {
-                while let Some(task) = recv.next().await {
-                    rt.spawn(task.call()).detach();
-                }
-            });
-        });
-        Self { thread, handle }
-    }
-
-    pub async fn spawn<F, Fut, T>(&self, func: F) -> Result<Task<T>, SendError>
+impl CompfsCore {
+    pub async fn exec<Fn, Fut, R>(&self, f: Fn) -> crate::Result<R>
     where
-        F: (FnOnce() -> Fut) + Send + 'static,
-        Fut: Future<Output = T>,
-        T: Send + 'static,
+        Fn: FnOnce() -> Fut + Send + 'static,
+        Fut: Future<Output = std::io::Result<R>> + 'static,
+        R: Send + 'static,
     {
-        self.handle.clone().spawn(func).await
-    }
-
-    pub fn handle(&self) -> SpawnHandle {
-        self.handle.clone()
+        self.dispatcher
+            .dispatch(f)
+            .map_err(|_| Error::new(ErrorKind::Unexpected, "compio spawn io task failed"))?
+            .await
+            .map_err(|_| Error::new(ErrorKind::Unexpected, "compio task cancelled"))?
+            .map_err(new_std_io_error)
     }
-}
 
-#[derive(Debug, Clone)]
-pub struct SpawnHandle(mpsc::Sender<SpawnTask>);
-
-impl SpawnHandle {
-    pub async fn spawn<F, Fut, T>(&mut self, func: F) -> Result<Task<T>, SendError>
+    pub async fn exec_blocking<Fn, R>(&self, f: Fn) -> Result<R>
     where
-        F: (FnOnce() -> Fut) + Send + 'static,
-        Fut: Future<Output = T>,
-        T: Send + 'static,
+        Fn: FnOnce() -> R + Send + 'static,
+        R: Send + 'static,
     {
-        let (task, spawn) = task(func);
-        self.0.send(spawn).await?;
-        Ok(task)
-    }
-}
-
-unsafe impl IoBuf for Buffer {
-    fn as_buf_ptr(&self) -> *const u8 {
-        self.current().as_ptr()
-    }
-
-    fn buf_len(&self) -> usize {
-        self.current().len()
-    }
-
-    fn buf_capacity(&self) -> usize {
-        // `Bytes` doesn't expose uninitialized capacity, so treat it as the same as `len`
-        self.current().len()
+        self.dispatcher
+            .dispatch_blocking(f)
+            .map_err(|_| Error::new(ErrorKind::Unexpected, "compio spawn blocking task failed"))?
+            .await
+            .map_err(|_| Error::new(ErrorKind::Unexpected, "compio task cancelled"))
     }
 }
 
diff --git a/core/src/services/compfs/lister.rs b/core/src/services/compfs/lister.rs
index b248758bc120..8370d9869174 100644
--- a/core/src/services/compfs/lister.rs
+++ b/core/src/services/compfs/lister.rs
@@ -14,3 +14,68 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
+
+use std::{fs::ReadDir, path::Path, sync::Arc};
+
+use super::core::CompfsCore;
+use crate::raw::*;
+use crate::*;
+
+#[derive(Debug)]
+pub struct CompfsLister {
+    core: Arc<CompfsCore>,
+    read_dir: Option<ReadDir>,
+}
+
+impl CompfsLister {
+    pub fn new(core: Arc<CompfsCore>, read_dir: ReadDir) -> Self {
+        Self {
+            core,
+            read_dir: Some(read_dir),
+        }
+    }
+}
+
+fn next_entry(read_dir: &mut ReadDir, root: &Path) -> std::io::Result<Option<oio::Entry>> {
+    let Some(entry) = read_dir.next().transpose()? else {
+        return Ok(None);
+    };
+    let path = entry.path();
+    let rel_path = normalize_path(
+        &path
+            .strip_prefix(&root)
+            .expect("cannot fail because the prefix is iterated")
+            .to_string_lossy()
+            .replace('\\', "/"),
+    );
+
+    let file_type = entry.file_type()?;
+
+    let entry = if file_type.is_file() {
+        oio::Entry::new(&rel_path, Metadata::new(EntryMode::FILE))
+    } else if file_type.is_dir() {
+        oio::Entry::new(&format!("{rel_path}/"), Metadata::new(EntryMode::DIR))
+    } else {
+        oio::Entry::new(&rel_path, Metadata::new(EntryMode::Unknown))
+    };
+
+    Ok(Some(entry))
+}
+
+impl oio::List for CompfsLister {
+    async fn next(&mut self) -> Result<Option<oio::Entry>> {
+        let Some(mut read_dir) = self.read_dir.take() else {
+            return Ok(None);
+        };
+        let root = self.core.root.clone();
+        let (entry, read_dir) = self
+            .core
+            .exec_blocking(move || {
+                let entry = next_entry(&mut read_dir, &root).map_err(new_std_io_error);
+                (entry, read_dir)
+            })
+            .await?;
+        self.read_dir = Some(read_dir);
+        entry
+    }
+}
diff --git a/core/src/services/compfs/reader.rs b/core/src/services/compfs/reader.rs
index b248758bc120..8800aa764f7d 100644
--- a/core/src/services/compfs/reader.rs
+++ b/core/src/services/compfs/reader.rs
@@ -14,3 +14,42 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
+
+use std::sync::Arc;
+
+use compio::{buf::buf_try, io::AsyncReadAt};
+
+use super::core::CompfsCore;
+use crate::raw::*;
+use crate::*;
+
+#[derive(Debug)]
+pub struct CompfsReader {
+    core: Arc<CompfsCore>,
+    f: compio::fs::File,
+}
+
+impl CompfsReader {
+    pub fn new(core: Arc<CompfsCore>, f: compio::fs::File) -> Self {
+        Self { core, f }
+    }
+}
+
+impl oio::Read for CompfsReader {
+    async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
+        let mut bs = self.core.buf_pool.get();
+        bs.reserve(limit);
+        let f = self.f.clone();
+        let mut bs = self
+            .core
+            .exec(move || async move {
+                let (_, bs) = buf_try!(@try f.read_at(bs, offset).await);
+
+                Ok(bs)
+            })
+            .await?;
+        let frozen = bs.split().freeze();
+        self.core.buf_pool.put(bs);
+        Ok(Buffer::from(frozen))
+    }
+}
diff --git a/core/src/services/compfs/writer.rs b/core/src/services/compfs/writer.rs
index b248758bc120..95a97519104c 100644
--- a/core/src/services/compfs/writer.rs
+++ b/core/src/services/compfs/writer.rs
@@ -14,3 +14,49 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
+
+use std::{io::Cursor, sync::Arc};
+
+use compio::{buf::buf_try, fs::File, io::AsyncWrite};
+
+use super::core::CompfsCore;
+use crate::raw::*;
+use crate::*;
+
+#[derive(Debug)]
+pub struct CompfsWriter {
+    core: Arc<CompfsCore>,
+    file: Cursor<File>,
+}
+
+impl CompfsWriter {
+    pub fn new(core: Arc<CompfsCore>, file: Cursor<File>) -> Self {
+        Self { core, file }
+    }
+}
+
+impl oio::Write for CompfsWriter {
+    async fn write(&mut self, bs: Buffer) -> Result<usize> {
+        let mut file = self.file.clone();
+
+        let n = self
+            .core
+            .exec(move || async move {
+                let (n, _) = buf_try!(@try file.write(bs).await);
+                Ok(n)
+            })
+            .await?;
+        Ok(n)
+    }
+
+    async fn close(&mut self) -> Result<()> {
+        let f = self.file.clone();
+        self.core
+            .exec(move || async move { f.get_ref().sync_all().await })
+            .await
+    }
+
+    async fn abort(&mut self) -> Result<()> {
+        Ok(())
+    }
+}

From 6c356f4a6430434cfb36973cac3643959ee5ba93 Mon Sep 17 00:00:00 2001
From: George Miao <gm@miao.dev>
Date: Wed, 5 Jun 2024 23:23:19 +0900
Subject: [PATCH 2/6] impl info and remove append

---
 core/src/services/compfs/backend.rs | 36 ++++++++++++++++++++---------
 1 file changed, 25 insertions(+), 11 deletions(-)

diff --git a/core/src/services/compfs/backend.rs b/core/src/services/compfs/backend.rs
index 69beb8407165..f8562b926a7b 100644
--- a/core/src/services/compfs/backend.rs
+++ b/core/src/services/compfs/backend.rs
@@ -72,7 +72,30 @@ impl Access for CompfsBackend {
     type BlockingLister = ();
 
     fn info(&self) -> AccessorInfo {
-        todo!()
+        let mut am = AccessorInfo::default();
+        am.set_scheme(Scheme::Compfs)
+            .set_root(&self.core.root.to_string_lossy())
+            .set_native_capability(Capability {
+                stat: true,
+
+                read: true,
+
+                write: true,
+                write_can_empty: true,
+                write_can_multi: true,
+                create_dir: true,
+                delete: true,
+
+                list: true,
+
+                copy: true,
+                rename: true,
+                blocking: true,
+
+                ..Default::default()
+            });
+
+        am
     }
 
     async fn read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
@@ -90,7 +113,7 @@ impl Access for CompfsBackend {
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
         let path = self.core.root.join(path.trim_end_matches('/'));
         let append = args.append();
-        let mut file = self
+        let file = self
             .core
             .exec(move || async move {
                 compio::fs::OpenOptions::new()
@@ -103,15 +126,6 @@ impl Access for CompfsBackend {
             .await
             .map(Cursor::new)?;
 
-        if append {
-            let f = file.clone();
-            let metadata = self
-                .core
-                .exec(|| async move { f.get_ref().metadata().await })
-                .await?;
-            file.set_position(metadata.len());
-        }
-
         let w = CompfsWriter::new(self.core.clone(), file);
         Ok((RpWrite::new(), w))
     }

From 208918d1ffa0cc838cb57ec55f73e5a0acf42ae2 Mon Sep 17 00:00:00 2001
From: George Miao <gm@miao.dev>
Date: Wed, 5 Jun 2024 23:26:31 +0900
Subject: [PATCH 3/6] git dependency

---
 core/Cargo.lock                     | 41 +++++++++++++++++------------
 core/Cargo.toml                     |  7 ++---
 core/src/services/compfs/backend.rs |  7 +++--
 3 files changed, 29 insertions(+), 26 deletions(-)

diff --git a/core/Cargo.lock b/core/Cargo.lock
index 950f4e124fb0..a2a247a54a04 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -1653,10 +1653,10 @@ dependencies = [
 [[package]]
 name = "compio"
 version = "0.10.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "76871d1ab0619ccb72372cf8d87951ac2bf112e6e1ede1085f2f50ace4436ad3"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "compio-buf",
+ "compio-dispatcher",
  "compio-driver",
  "compio-fs",
  "compio-io",
@@ -1668,19 +1668,28 @@ dependencies = [
 [[package]]
 name = "compio-buf"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "24d9ae28226834dc0f8cd9af2a129488528bcaa4426baaba16072e13cff8440b"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "arrayvec",
  "bytes",
  "libc",
 ]
 
+[[package]]
+name = "compio-dispatcher"
+version = "0.2.0"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+dependencies = [
+ "compio-driver",
+ "compio-runtime",
+ "flume",
+ "futures-channel",
+]
+
 [[package]]
 name = "compio-driver"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a7b23ddd21203f5d4bee6e5a9f53e4f09f98229590a903ee1b5b4c88047a440d"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "aligned-array",
  "cfg-if",
@@ -1688,13 +1697,13 @@ dependencies = [
  "compio-log",
  "crossbeam-channel",
  "crossbeam-queue",
+ "futures-util",
  "io-uring",
  "libc",
  "once_cell",
  "os_pipe",
  "paste",
  "polling 3.7.1",
- "slab",
  "socket2 0.5.7",
  "windows-sys 0.52.0",
 ]
@@ -1702,8 +1711,7 @@ dependencies = [
 [[package]]
 name = "compio-fs"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "770d9c79a1174adc527b6ce880aa6a50d9f5fb630339024069785ebbd6848c2f"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "cfg-if",
  "compio-buf",
@@ -1719,8 +1727,7 @@ dependencies = [
 [[package]]
 name = "compio-io"
 version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ed6f47ce0edef794248555268b9e0cfbbbaaacaa3634fca4acc199818272a415"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "compio-buf",
  "futures-util",
@@ -1730,8 +1737,7 @@ dependencies = [
 [[package]]
 name = "compio-log"
 version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc4e560213c1996b618da369b7c9109564b41af9033802ae534465c4ee4e132f"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "tracing",
 ]
@@ -1739,8 +1745,7 @@ dependencies = [
 [[package]]
 name = "compio-net"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ae469c47794087f9110a8858394b6c8112317b66c5d41d44d874ebf4218bc3ee"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "cfg-if",
  "compio-buf",
@@ -1757,8 +1762,7 @@ dependencies = [
 [[package]]
 name = "compio-runtime"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c314b56ebb29d21e6f0dc3749ebbc83ddcb7d7a150f86b0c6d0e5b6ca228bae7"
+source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
 dependencies = [
  "async-task",
  "cfg-if",
@@ -1770,6 +1774,8 @@ dependencies = [
  "libc",
  "once_cell",
  "os_pipe",
+ "scoped-tls",
+ "slab",
  "smallvec",
  "socket2 0.5.7",
  "windows-sys 0.52.0",
@@ -4766,6 +4772,7 @@ dependencies = [
  "cacache",
  "chrono",
  "compio",
+ "compio-driver",
  "crc32c",
  "criterion",
  "dashmap",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index ec920365380e..a10acc522a43 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -344,11 +344,8 @@ hdfs-native = { version = "0.9.4", optional = true }
 # for services-surrealdb
 surrealdb = { version = "1.3.0", optional = true, features = ["protocol-http"] }
 # for services-compfs
-compio = { version = "0.10.0", optional = true, features = [
-  "runtime",
-  "bytes",
-  "polling",
-] }
+compio = { git = "https://github.com/compio-rs/compio.git", version = "0.10.0", optional = true, features = ["runtime", "bytes", "polling", "dispatcher"] }
+compio-driver = { git = "https://github.com/compio-rs/compio.git", package = "compio-driver" }
 # for services-s3
 crc32c = { version = "0.6.6", optional = true }
 
diff --git a/core/src/services/compfs/backend.rs b/core/src/services/compfs/backend.rs
index f8562b926a7b..a930de69989f 100644
--- a/core/src/services/compfs/backend.rs
+++ b/core/src/services/compfs/backend.rs
@@ -15,13 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::HashMap;
-use std::path::PathBuf;
-
-use super::core::CompioThread;
+use super::{core::CompfsCore, lister::CompfsLister, reader::CompfsReader, writer::CompfsWriter};
 use crate::raw::*;
 use crate::*;
 
+use std::{collections::HashMap, io::Cursor, path::PathBuf, sync::Arc};
+
 /// [`compio`]-based file system support.
 #[derive(Debug, Clone, Default)]
 pub struct CompfsBuilder {

From 37bbcbedfcdc14d5a61855130a535357489c1ed1 Mon Sep 17 00:00:00 2001
From: George Miao <gm@miao.dev>
Date: Thu, 6 Jun 2024 00:59:31 +0900
Subject: [PATCH 4/6] adapt to new `oio::Read`

---
 core/src/services/compfs/backend.rs |  4 ++--
 core/src/services/compfs/reader.rs  | 18 ++++++++++--------
 2 files changed, 12 insertions(+), 10 deletions(-)

diff --git a/core/src/services/compfs/backend.rs b/core/src/services/compfs/backend.rs
index a930de69989f..7105cbffa5d9 100644
--- a/core/src/services/compfs/backend.rs
+++ b/core/src/services/compfs/backend.rs
@@ -97,7 +97,7 @@ impl Access for CompfsBackend {
         am
     }
 
-    async fn read(&self, path: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
+    async fn read(&self, path: &str, op: OpRead) -> Result<(RpRead, Self::Reader)> {
         let path = self.core.root.join(path.trim_end_matches('/'));
 
         let file = self
@@ -105,7 +105,7 @@ impl Access for CompfsBackend {
             .exec(|| async move { compio::fs::OpenOptions::new().read(true).open(&path).await })
             .await?;
 
-        let r = CompfsReader::new(self.core.clone(), file);
+        let r = CompfsReader::new(self.core.clone(), file, op.range());
         Ok((RpRead::new(), r))
     }
 
diff --git a/core/src/services/compfs/reader.rs b/core/src/services/compfs/reader.rs
index 8800aa764f7d..758c110894db 100644
--- a/core/src/services/compfs/reader.rs
+++ b/core/src/services/compfs/reader.rs
@@ -26,25 +26,27 @@ use crate::*;
 #[derive(Debug)]
 pub struct CompfsReader {
     core: Arc<CompfsCore>,
-    f: compio::fs::File,
+    file: compio::fs::File,
+    range: BytesRange,
 }
 
 impl CompfsReader {
-    pub fn new(core: Arc<CompfsCore>, f: compio::fs::File) -> Self {
-        Self { core, f }
+    pub fn new(core: Arc<CompfsCore>, file: compio::fs::File, range: BytesRange) -> Self {
+        Self { core, file, range }
     }
 }
 
 impl oio::Read for CompfsReader {
-    async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
+    async fn read(&mut self) -> Result<Buffer> {
         let mut bs = self.core.buf_pool.get();
-        bs.reserve(limit);
-        let f = self.f.clone();
+
+        let len = self.range.size().expect("range size is always Some");
+        bs.reserve(len as _);
+        let f = self.file.clone();
         let mut bs = self
             .core
             .exec(move || async move {
-                let (_, bs) = buf_try!(@try f.read_at(bs, offset).await);
-
+                let (_, bs) = buf_try!(@try f.read_at(bs, len).await);
                 Ok(bs)
             })
             .await?;

From 72a96c5c32969a3a97d49d8e61ededd2be771d89 Mon Sep 17 00:00:00 2001
From: George Miao <gm@miao.dev>
Date: Thu, 6 Jun 2024 01:22:25 +0900
Subject: [PATCH 5/6] fix clippy

---
 core/src/services/compfs/backend.rs | 2 +-
 core/src/services/compfs/lister.rs  | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/core/src/services/compfs/backend.rs b/core/src/services/compfs/backend.rs
index 7105cbffa5d9..65bd189b683a 100644
--- a/core/src/services/compfs/backend.rs
+++ b/core/src/services/compfs/backend.rs
@@ -133,7 +133,7 @@ impl Access for CompfsBackend {
         let path = self.core.root.join(path.trim_end_matches('/'));
         let read_dir = match self
             .core
-            .exec_blocking(move || std::fs::read_dir(&path))
+            .exec_blocking(move || std::fs::read_dir(path))
             .await?
         {
             Ok(rd) => rd,
diff --git a/core/src/services/compfs/lister.rs b/core/src/services/compfs/lister.rs
index 8370d9869174..7a380d48b71b 100644
--- a/core/src/services/compfs/lister.rs
+++ b/core/src/services/compfs/lister.rs
@@ -43,7 +43,7 @@ fn next_entry(read_dir: &mut ReadDir, root: &Path) -> std::io::Result<Option<oio
     let path = entry.path();
     let rel_path = normalize_path(
         &path
-            .strip_prefix(&root)
+            .strip_prefix(root)
             .expect("cannot fail because the prefix is iterated")
             .to_string_lossy()
             .replace('\\', "/"),

From ce40456da6626582cb388ad18d25c50705e59895 Mon Sep 17 00:00:00 2001
From: George Miao <gm@miao.dev>
Date: Wed, 19 Jun 2024 11:22:37 +0900
Subject: [PATCH 6/6] use crates.io dep

---
 core/Cargo.lock | 44 ++++++++++++++++++++++++++------------------
 core/Cargo.toml |  3 +--
 2 files changed, 27 insertions(+), 20 deletions(-)

diff --git a/core/Cargo.lock b/core/Cargo.lock
index a2a247a54a04..7a365c345c96 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -1652,8 +1652,9 @@ dependencies = [
 
 [[package]]
 name = "compio"
-version = "0.10.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1704a939a6548e001b8502663fa4d90a87675569662ee4746d608c5ee62246ac"
 dependencies = [
  "compio-buf",
  "compio-dispatcher",
@@ -1667,8 +1668,9 @@ dependencies = [
 
 [[package]]
 name = "compio-buf"
-version = "0.3.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69e280904a879019a48c3a9b0bbf6e158c097b9c21555be0608602c5b101755b"
 dependencies = [
  "arrayvec",
  "bytes",
@@ -1677,8 +1679,9 @@ dependencies = [
 
 [[package]]
 name = "compio-dispatcher"
-version = "0.2.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "409f0680497022856f64e342ec9e0fd7384ed56bdaf82b3ef24ccab03eecd768"
 dependencies = [
  "compio-driver",
  "compio-runtime",
@@ -1688,8 +1691,9 @@ dependencies = [
 
 [[package]]
 name = "compio-driver"
-version = "0.3.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7ef034f0dafcead1e4738b1a18da070ff26abaf53ec9d9c4ba4b217b5f7a7c4a"
 dependencies = [
  "aligned-array",
  "cfg-if",
@@ -1710,8 +1714,9 @@ dependencies = [
 
 [[package]]
 name = "compio-fs"
-version = "0.3.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a51ed24575379a16f8aa23e04378b0fc291d7f608ba2b9928d0fffe24fb3100c"
 dependencies = [
  "cfg-if",
  "compio-buf",
@@ -1726,8 +1731,9 @@ dependencies = [
 
 [[package]]
 name = "compio-io"
-version = "0.2.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c5c6b524ae8226c007c41342a5803381f544f4f4066808d4c4ff146e75869ebb"
 dependencies = [
  "compio-buf",
  "futures-util",
@@ -1737,15 +1743,17 @@ dependencies = [
 [[package]]
 name = "compio-log"
 version = "0.1.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc4e560213c1996b618da369b7c9109564b41af9033802ae534465c4ee4e132f"
 dependencies = [
  "tracing",
 ]
 
 [[package]]
 name = "compio-net"
-version = "0.3.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1c6c4fcf2d8c86ac0f13b54c57d35ce351b81b2221a75f3cee286514634eef6e"
 dependencies = [
  "cfg-if",
  "compio-buf",
@@ -1761,8 +1769,9 @@ dependencies = [
 
 [[package]]
 name = "compio-runtime"
-version = "0.3.0"
-source = "git+https://github.com/compio-rs/compio.git#9939253f42a171d701d94ff6e2ca9ed4dd5865d4"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8181cc373318910ccc4610dd54683d63d7933a96d21c6c69e61e750c01cd2016"
 dependencies = [
  "async-task",
  "cfg-if",
@@ -4772,7 +4781,6 @@ dependencies = [
  "cacache",
  "chrono",
  "compio",
- "compio-driver",
  "crc32c",
  "criterion",
  "dashmap",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index a10acc522a43..26678b7aea47 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -344,8 +344,7 @@ hdfs-native = { version = "0.9.4", optional = true }
 # for services-surrealdb
 surrealdb = { version = "1.3.0", optional = true, features = ["protocol-http"] }
 # for services-compfs
-compio = { git = "https://github.com/compio-rs/compio.git", version = "0.10.0", optional = true, features = ["runtime", "bytes", "polling", "dispatcher"] }
-compio-driver = { git = "https://github.com/compio-rs/compio.git", package = "compio-driver" }
+compio = { version = "0.11.0", optional = true, features = ["runtime", "bytes", "polling", "dispatcher"] }
 # for services-s3
 crc32c = { version = "0.6.6", optional = true }