feat(services/compfs): compio runtime and compfs structure

core/src/services/compfs/backend.rs

@@ -0,0 +1,78 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::core::CompioThread;
+use crate::raw::*;
+use crate::*;
+use async_trait::async_trait;
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+/// [`compio`]-based file system support.
+#[derive(Debug, Clone, Default)]
+pub struct CompFSBuilder {
+    root: Option<PathBuf>,
+}
+
+impl CompFSBuilder {
+    /// Set root for CompFS
+    pub fn root(&mut self, root: &str) -> &mut Self {
+        self.root = if root.is_empty() {
+            None
+        } else {
+            Some(PathBuf::from(root))
+        };
+
+        self
+    }
+}
+
+impl Builder for CompFSBuilder {
+    const SCHEME: Scheme = Scheme::CompFS;
+    type Accessor = ();
+
+    fn from_map(map: HashMap<String, String>) -> Self {
+        let mut builder = CompFSBuilder::default();
+
+        map.get("root").map(|v| builder.root(v));
+
+        builder
+    }
+
+    fn build(&mut self) -> Result<Self::Accessor> {
+        todo!()
+    }
+}
+
+#[derive(Debug)]
+pub struct CompFSBackend {
+    rt: CompioThread,
+}
+
+#[async_trait]
+impl Accessor for CompFSBackend {
+    type Reader = ();
+    type Writer = ();
+    type Lister = ();
+    type BlockingReader = ();
+    type BlockingWriter = ();
+    type BlockingLister = ();
+
+    fn info(&self) -> AccessorInfo {
+        todo!()
+    }
+}

core/src/services/compfs/core.rs

@@ -16,9 +16,100 @@
 // under the License.
 
 use compio::buf::IoBuf;
+use compio::runtime::RuntimeBuilder;
+use futures::channel::mpsc::SendError;
+use futures::channel::{mpsc, oneshot};
+use futures::future::LocalBoxFuture;
+use futures::{SinkExt, StreamExt};
+use std::future::Future;
+use std::thread::JoinHandle;
 
 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))
+}
+
+/// A task handle that can be used to retrieve result spawned into [`CompioThread`].
+pub struct Task<T>(oneshot::Receiver<T>);
+
+/// Type erased task that can be spawned into a [`CompioThread`].
+struct SpawnTask(Box<dyn (FnOnce() -> LocalBoxFuture<'static, ()>) + Send>);
+
+impl SpawnTask {
+    fn call(self) -> LocalBoxFuture<'static, ()> {
+        (self.0)()
+    }
+}
+
+#[derive(Debug)]
+pub struct CompioThread {
+    thread: JoinHandle<()>,
+    handle: SpawnHandle,
+}
+
+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>
+    where
+        F: (FnOnce() -> Fut) + Send + 'static,
+        Fut: Future<Output = T>,
+        T: Send + 'static,
+    {
+        self.handle.clone().spawn(func).await
+    }
+
+    pub fn handle(&self) -> SpawnHandle {
+        self.handle.clone()
+    }
+}
+
+#[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>
+    where
+        F: (FnOnce() -> Fut) + Send + 'static,
+        Fut: Future<Output = T>,
+        T: 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()

core/src/services/compfs/lister.rs

@@ -0,0 +1,16 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.

core/src/services/compfs/mod.rs

@@ -15,4 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#![allow(dead_code)] // TODO: Remove this after backend is implemented
+
+mod backend;
 mod core;
+mod lister;
+mod reader;
+mod writer;
+
+pub use backend::CompFSBuilder as CompFS;

core/src/services/compfs/reader.rs

@@ -0,0 +1,16 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.

core/src/services/compfs/writer.rs

@@ -0,0 +1,16 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.

core/src/services/mod.rs

@@ -409,3 +409,5 @@ pub use surrealdb::SurrealdbConfig;
 
 #[cfg(feature = "services-compfs")]
 mod compfs;
+#[cfg(feature = "services-compfs")]
+pub use compfs::CompFS;

core/src/types/scheme.rs

@@ -40,6 +40,8 @@ pub enum Scheme {
     Azdls,
     /// [B2][crate::services::B2]: Backblaze B2 Services.
     B2,
+    /// [CompFS][crate::services::CompFS]: Compio fs Services.
+    CompFS,
     /// [Seafile][crate::services::Seafile]: Seafile Services.
     Seafile,
     /// [Upyun][crate::services::Upyun]: Upyun Services.
@@ -205,6 +207,8 @@ impl Scheme {
             Scheme::Cacache,
             #[cfg(feature = "services-cos")]
             Scheme::Cos,
+            #[cfg(feature = "services-compfs")]
+            Scheme::CompFS,
             #[cfg(feature = "services-dashmap")]
             Scheme::Dashmap,
             #[cfg(feature = "services-dropbox")]
@@ -332,6 +336,7 @@ impl FromStr for Scheme {
             "b2" => Ok(Scheme::B2),
             "chainsafe" => Ok(Scheme::Chainsafe),
             "cacache" => Ok(Scheme::Cacache),
+            "compfs" => Ok(Scheme::CompFS),
             "cloudflare_kv" => Ok(Scheme::CloudflareKv),
             "cos" => Ok(Scheme::Cos),
             "d1" => Ok(Scheme::D1),
@@ -402,6 +407,7 @@ impl From<Scheme> for &'static str {
             Scheme::Cacache => "cacache",
             Scheme::CloudflareKv => "cloudflare_kv",
             Scheme::Cos => "cos",
+            Scheme::CompFS => "compfs",
             Scheme::D1 => "d1",
             Scheme::Dashmap => "dashmap",
             Scheme::Etcd => "etcd",