feat(services/monoiofs): impl read and write, add behavior test

.github/services/monoiofs/monoiofs/action.yml

@@ -0,0 +1,27 @@
+# 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.
+
+name: monoiofs
+description: 'Behavior test for monoiofs'
+
+runs:
+  using: "composite"
+  steps:
+    - name: Setup
+      shell: bash
+      run: |
+        echo "OPENDAL_MONOIOFS_ROOT=${{ runner.temp }}/" >> $GITHUB_ENV

bindings/java/Cargo.toml

@@ -134,6 +134,7 @@ services-memcached = ["opendal/services-memcached"]
 services-mini-moka = ["opendal/services-mini-moka"]
 services-moka = ["opendal/services-moka"]
 services-mongodb = ["opendal/services-mongodb"]
+services-monoiofs = ["opendal/services-monoiofs"]
 services-mysql = ["opendal/services-mysql"]
 services-onedrive = ["opendal/services-onedrive"]
 services-persy = ["opendal/services-persy"]

bindings/nodejs/Cargo.toml

@@ -132,6 +132,7 @@ services-memcached = ["opendal/services-memcached"]
 services-mini-moka = ["opendal/services-mini-moka"]
 services-moka = ["opendal/services-moka"]
 services-mongodb = ["opendal/services-mongodb"]
+services-monoiofs = ["opendal/services-monoiofs"]
 services-mysql = ["opendal/services-mysql"]
 services-onedrive = ["opendal/services-onedrive"]
 services-persy = ["opendal/services-persy"]

bindings/python/Cargo.toml

@@ -131,6 +131,7 @@ services-memcached = ["opendal/services-memcached"]
 services-mini-moka = ["opendal/services-mini-moka"]
 services-moka = ["opendal/services-moka"]
 services-mongodb = ["opendal/services-mongodb"]
+services-monoiofs = ["opendal/services-monoiofs"]
 services-mysql = ["opendal/services-mysql"]
 services-onedrive = ["opendal/services-onedrive"]
 services-persy = ["opendal/services-persy"]

core/src/services/monoiofs/backend.rs

@@ -20,10 +20,13 @@ use std::io;
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use chrono::DateTime;
 use serde::Deserialize;
 use serde::Serialize;
 
 use super::core::MonoiofsCore;
+use super::reader::MonoiofsReader;
+use super::writer::MonoiofsWriter;
 use crate::raw::*;
 use crate::*;
 
@@ -109,8 +112,8 @@ pub struct MonoiofsBackend {
 }
 
 impl Access for MonoiofsBackend {
-    type Reader = ();
-    type Writer = ();
+    type Reader = MonoiofsReader;
+    type Writer = MonoiofsWriter;
     type Lister = ();
     type BlockingReader = ();
     type BlockingWriter = ();
@@ -121,8 +124,72 @@ impl Access for MonoiofsBackend {
         am.set_scheme(Scheme::Monoiofs)
             .set_root(&self.core.root().to_string_lossy())
             .set_native_capability(Capability {
+                stat: true,
+                read: true,
+                write: true,
+                delete: true,
                 ..Default::default()
             });
         am.into()
     }
+
+    async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+        let path = self.core.prepare_path(path);
+        // TODO: borrowed from FsBackend because statx support for monoio
+        // is not released yet, but stat capability is required for read
+        // and write
+        let meta = tokio::fs::metadata(&path).await.map_err(new_std_io_error)?;
+        let mode = if meta.is_dir() {
+            EntryMode::DIR
+        } else if meta.is_file() {
+            EntryMode::FILE
+        } else {
+            EntryMode::Unknown
+        };
+        let m = Metadata::new(mode)
+            .with_content_length(meta.len())
+            .with_last_modified(
+                meta.modified()
+                    .map(DateTime::from)
+                    .map_err(new_std_io_error)?,
+            );
+        Ok(RpStat::new(m))
+    }
+
+    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+        let path = self.core.prepare_path(path);
+        let reader = MonoiofsReader::new(self.core.clone(), path, args.range()).await?;
+        Ok((RpRead::default(), reader))
+    }
+
+    async fn write(&self, path: &str, _args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        // TODO: create parent directory before write
+        let path = self.core.prepare_path(path);
+        let writer = MonoiofsWriter::new(self.core.clone(), path).await?;
+        Ok((RpWrite::default(), writer))
+    }
+
+    async fn delete(&self, path: &str, _args: OpDelete) -> Result<RpDelete> {
+        let path = self.core.prepare_path(path);
+        // TODO: borrowed from FsBackend because monoio doesn't support unlink,
+        // but delete capability is required for behavior tests
+        let meta = tokio::fs::metadata(&path).await;
+        match meta {
+            Ok(meta) => {
+                if meta.is_dir() {
+                    tokio::fs::remove_dir(&path)
+                        .await
+                        .map_err(new_std_io_error)?;
+                } else {
+                    tokio::fs::remove_file(&path)
+                        .await
+                        .map_err(new_std_io_error)?;
+                }
+
+                Ok(RpDelete::default())
+            }
+            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(RpDelete::default()),
+            Err(err) => Err(new_std_io_error(err)),
+        }
+    }
 }

core/src/services/monoiofs/core.rs

@@ -24,9 +24,15 @@ use flume::Receiver;
 use flume::Sender;
 use futures::channel::oneshot;
 use futures::Future;
+use monoio::task::JoinHandle;
 use monoio::FusionDriver;
 use monoio::RuntimeBuilder;
 
+use crate::raw::*;
+use crate::*;
+
+pub const BUFFER_SIZE: usize = 2 * 1024 * 1024; // 2 MiB
+
 /// a boxed function that spawns task in current monoio runtime
 type TaskSpawner = Box<dyn FnOnce() + Send>;
 
@@ -39,6 +45,7 @@ pub struct MonoiofsCore {
     #[allow(dead_code)]
     /// join handles of worker threads
     threads: Mutex<Vec<std::thread::JoinHandle<()>>>,
+    pub buf_pool: oio::PooledBuf,
 }
 
 impl MonoiofsCore {
@@ -59,13 +66,22 @@ impl MonoiofsCore {
             .collect();
         let threads = Mutex::new(threads);
 
-        Self { root, tx, threads }
+        Self {
+            root,
+            tx,
+            threads,
+            buf_pool: oio::PooledBuf::new(16).with_initial_capacity(BUFFER_SIZE),
+        }
     }
 
     pub fn root(&self) -> &PathBuf {
         &self.root
     }
 
+    pub fn prepare_path(&self, path: &str) -> PathBuf {
+        self.root.join(path.trim_end_matches('/'))
+    }
+
     /// entrypoint of each worker thread, sets up monoio runtimes and channels
     fn worker_entrypoint(rx: Receiver<TaskSpawner>, io_uring_entries: u32) {
         let mut rt = RuntimeBuilder::<FusionDriver>::new()
@@ -83,8 +99,8 @@ impl MonoiofsCore {
     }
 
     #[allow(dead_code)]
-    /// create a TaskSpawner, send it to the thread pool and wait
-    /// for its result
+    /// Create a TaskSpawner, send it to the thread pool and wait
+    /// for its result. Task panic will propagate.
     pub async fn dispatch<F, Fut, T>(&self, f: F) -> T
     where
         F: FnOnce() -> Fut + 'static + Send,
@@ -93,32 +109,54 @@ impl MonoiofsCore {
     {
         // oneshot channel to send result back
         let (tx, rx) = oneshot::channel();
-        self.tx
+        let result = self
+            .tx
             .send_async(Box::new(move || {
+                // task will be spawned on current thread, task panic
+                // will cause current worker thread panic
                 monoio::spawn(async move {
-                    tx.send(f().await)
-                        // discard result because it may be non-Debug and
-                        // we don't need it to appear in the panic message
-                        .map_err(|_| ())
-                        .expect("send result from worker thread should success");
+                    // discard the result if send failed due to
+                    // MonoiofsCore::dispatch cancelled
+                    let _ = tx.send(f().await);
                 });
             }))
-            .await
-            .expect("send new TaskSpawner to worker thread should success");
-        match rx.await {
-            Ok(result) => result,
-            // tx is dropped without sending result, probably the worker
-            // thread has panicked.
-            Err(_) => self.propagate_worker_panic(),
-        }
+            .await;
+        self.unwrap(result);
+        self.unwrap(rx.await)
+    }
+
+    /// Create a TaskSpawner, send it to the thread pool, spawn the task
+    /// and return its [`JoinHandle`]. Task panic cannot propagate
+    /// through the [`JoinHandle`] and should be handled elsewhere.
+    pub async fn spawn<F, Fut, T>(&self, f: F) -> JoinHandle<T>
+    where
+        F: FnOnce() -> Fut + 'static + Send,
+        Fut: Future<Output = T>,
+        T: 'static + Send,
+    {
+        // oneshot channel to send JoinHandle back
+        let (tx, rx) = oneshot::channel();
+        let result = self
+            .tx
+            .send_async(Box::new(move || {
+                // task will be spawned on current thread, task panic
+                // will cause current worker thread panic
+                let handle = monoio::spawn(async move { f().await });
+                // discard the result if send failed due to
+                // MonoiofsCore::spawn cancelled
+                let _ = tx.send(handle);
+            }))
+            .await;
+        self.unwrap(result);
+        self.unwrap(rx.await)
     }
 
     /// This method always panics. It is called only when at least a
     /// worker thread has panicked or meet a broken rx, which is
     /// unrecoverable. It propagates worker thread's panic if there
     /// is any and panics on normally exited thread.
-    fn propagate_worker_panic(&self) -> ! {
-        let mut guard = self.threads.lock().unwrap();
+    pub fn propagate_worker_panic(&self) -> ! {
+        let mut guard = self.threads.lock().expect("worker thread has panicked");
         // wait until the panicked thread exits
         std::thread::sleep(Duration::from_millis(100));
         let threads = mem::take(&mut *guard);
@@ -137,6 +175,17 @@ impl MonoiofsCore {
         }
         unreachable!("this method should panic")
     }
+
+    /// Unwrap result if result is Ok, otherwise propagates worker thread's
+    /// panic. This method facilitates panic propagation in situation where
+    /// Err returned by broken channel indicates that the worker thread has
+    /// panicked.
+    pub fn unwrap<T, E>(&self, result: Result<T, E>) -> T {
+        match result {
+            Ok(result) => result,
+            Err(_) => self.propagate_worker_panic(),
+        }
+    }
 }
 
 #[cfg(test)]

core/src/services/monoiofs/docs.md

@@ -3,8 +3,8 @@
 This service can be used to:
 
 - [ ] stat
-- [ ] read
-- [ ] write
+- [x] read
+- [x] write
 - [ ] append
 - [ ] create_dir
 - [ ] delete

core/src/services/monoiofs/mod.rs

@@ -20,3 +20,5 @@ pub use backend::MonoiofsBuilder as Monoiofs;
 pub use backend::MonoiofsConfig;
 
 mod core;
+mod reader;
+mod writer;