feat(layer): add madsim layer

core/Cargo.toml

@@ -53,13 +53,15 @@ native-tls = ["reqwest/native-tls", "ureq/native-tls"]
 native-tls-vendored = ["reqwest/native-tls-vendored", "ureq/native-tls"]
 
 # Enable all layers.
-layers-all = ["layers-chaos", "layers-metrics", "layers-tracing"]
+layers-all = ["layers-chaos", "layers-metrics", "layers-tracing","layers-madsim"]
 # Enable layers chaos support
 layers-chaos = ["dep:rand"]
 # Enable layers metrics support
 layers-metrics = ["dep:metrics"]
 # Enable layers tracing support.
 layers-tracing = ["dep:tracing"]
+# Enable layers madsim support
+layers-madsim = ["dep:madsim"]
 
 # Enable services dashmap support
 services-dashmap = ["dep:dashmap"]
@@ -104,6 +106,7 @@ http = "0.2.5"
 hyper = "0.14"
 lazy-regex = { version = "2.5.0", optional = true }
 log = "0.4"
+madsim = {version = "0.2.19", optional = true }
 md-5 = "0.10"
 metrics = { version = "0.20", optional = true }
 moka = { version = "0.10", optional = true, features = ["future"] }

core/src/docs/concepts.rs

@@ -19,7 +19,7 @@
 //!
 //! OpenDAL provides a unified abstraction for all storage services.
 //!
-//! There are three core concepts in OpenDAL:
+//! There are two core concepts in OpenDAL:
 //!
 //! - [`Builder`]: Build an instance of underlying services.
 //! - [`Operator`]: A bridge between underlying implementation detail and unified abstraction.

core/src/layers/madsim.rs

@@ -0,0 +1,355 @@
+// 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 crate::ops::{OpList, OpRead, OpScan, OpWrite};
+use crate::raw::oio::Entry;
+use crate::raw::{oio, Accessor, Layer, LayeredAccessor, RpList, RpRead, RpScan, RpWrite};
+use crate::EntryMode;
+use async_trait::async_trait;
+use bytes::Bytes;
+use madsim::net::Endpoint;
+use madsim::net::Payload;
+use std::collections::HashMap;
+use std::fmt::{Debug, Formatter};
+use std::io::Result;
+use std::io::SeekFrom;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::task::{Context, Poll};
+
+#[derive(Debug, Copy, Clone)]
+pub struct MadsimLayer {
+    sim_server_socket: SocketAddr,
+}
+
+impl MadsimLayer {
+    pub fn new(sim_server_socket: SocketAddr) -> Self {
+        Self { sim_server_socket }
+    }
+}
+
+impl<A: Accessor> Layer<A> for MadsimLayer {
+    type LayeredAccessor = MadsimAccessor;
+
+    fn layer(&self, _inner: A) -> Self::LayeredAccessor {
+        MadsimAccessor {
+            sim_server_socket: self.sim_server_socket,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct MadsimAccessor {
+    sim_server_socket: SocketAddr,
+}
+
+#[async_trait]
+impl LayeredAccessor for MadsimAccessor {
+    type Inner = ();
+    type Reader = MadsimReader;
+    type BlockingReader = MadsimReader;
+    type Writer = MadsimWriter;
+    type BlockingWriter = MadsimWriter;
+    type Pager = MadsimPager;
+    type BlockingPager = MadsimPager;
+
+    fn inner(&self) -> &Self::Inner {
+        &()
+    }
+
+    async fn read(&self, path: &str, args: OpRead) -> crate::Result<(RpRead, Self::Reader)> {
+        let req = Request::Read(path.to_string(), args);
+        let ep = Endpoint::connect(self.sim_server_socket)
+            .await
+            .expect("fail to connect to sim server");
+        let (tx, mut rx) = ep
+            .connect1(self.sim_server_socket)
+            .await
+            .expect("fail to connect1 to sim server");
+        tx.send(Box::new(req))
+            .await
+            .expect("fail to send request to sim server");
+        let resp = rx
+            .recv()
+            .await
+            .expect("fail to recv response from sim server");
+        let resp = resp
+            .downcast::<ReadResponse>()
+            .expect("fail to downcast response to ReadResponse");
+        let content_length = resp.data.as_ref().map(|b| b.len()).unwrap_or(0);
+        Ok((
+            RpRead::new(content_length as u64),
+            MadsimReader { data: resp.data },
+        ))
+    }
+
+    async fn write(&self, path: &str, args: OpWrite) -> crate::Result<(RpWrite, Self::Writer)> {
+        Ok((
+            RpWrite::default(),
+            MadsimWriter {
+                path: path.to_string(),
+                args,
+                sim_server_socket: self.sim_server_socket,
+            },
+        ))
+    }
+
+    async fn list(&self, path: &str, args: OpList) -> crate::Result<(RpList, Self::Pager)> {
+        todo!()
+    }
+
+    async fn scan(&self, path: &str, args: OpScan) -> crate::Result<(RpScan, Self::Pager)> {
+        todo!()
+    }
+
+    fn blocking_read(
+        &self,
+        path: &str,
+        args: OpRead,
+    ) -> crate::Result<(RpRead, Self::BlockingReader)> {
+        panic!("blocking_read is not supported in MadsimLayer");
+    }
+
+    fn blocking_write(
+        &self,
+        path: &str,
+        args: OpWrite,
+    ) -> crate::Result<(RpWrite, Self::BlockingWriter)> {
+        panic!("blocking_write is not supported in MadsimLayer");
+    }
+
+    fn blocking_list(
+        &self,
+        path: &str,
+        args: OpList,
+    ) -> crate::Result<(RpList, Self::BlockingPager)> {
+        panic!("blocking_list is not supported in MadsimLayer");
+    }
+
+    fn blocking_scan(
+        &self,
+        path: &str,
+        args: OpScan,
+    ) -> crate::Result<(RpScan, Self::BlockingPager)> {
+        panic!("blocking_scan is not supported in MadsimLayer");
+    }
+}
+
+pub struct MadsimReader {
+    data: Option<Bytes>,
+}
+
+impl oio::Read for MadsimReader {
+    fn poll_read(&mut self, _cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<crate::Result<usize>> {
+        if let Some(ref data) = self.data {
+            let len = data.len();
+            buf[..len].copy_from_slice(data);
+            Poll::Ready(Ok(len))
+        } else {
+            Poll::Ready(Ok(0))
+        }
+    }
+
+    fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<crate::Result<u64>> {
+        todo!()
+    }
+
+    fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<crate::Result<Bytes>>> {
+        todo!()
+    }
+}
+
+impl oio::BlockingRead for MadsimReader {
+    fn read(&mut self, buf: &mut [u8]) -> crate::Result<usize> {
+        todo!()
+    }
+
+    fn seek(&mut self, pos: SeekFrom) -> crate::Result<u64> {
+        todo!()
+    }
+
+    fn next(&mut self) -> Option<crate::Result<Bytes>> {
+        todo!()
+    }
+}
+
+pub struct MadsimWriter {
+    path: String,
+    args: OpWrite,
+    sim_server_socket: SocketAddr,
+}
+
+impl oio::BlockingWrite for MadsimWriter {
+    fn write(&mut self, bs: Bytes) -> crate::Result<()> {
+        todo!()
+    }
+
+    fn append(&mut self, bs: Bytes) -> crate::Result<()> {
+        todo!()
+    }
+
+    fn close(&mut self) -> crate::Result<()> {
+        todo!()
+    }
+}
+
+#[async_trait]
+impl oio::Write for MadsimWriter {
+    async fn write(&mut self, bs: Bytes) -> crate::Result<()> {
+        let req = Request::Write(self.path.to_string(), bs);
+        let ep = Endpoint::connect(self.sim_server_socket)
+            .await
+            .expect("fail to connect to sim server");
+        let (tx, mut rx) = ep
+            .connect1(self.sim_server_socket)
+            .await
+            .expect("fail to connect1 to sim server");
+        tx.send(Box::new(req))
+            .await
+            .expect("fail to send request to sim server");
+        rx.recv()
+            .await
+            .expect("fail to recv response from sim server");
+        Ok(())
+    }
+
+    async fn append(&mut self, bs: Bytes) -> crate::Result<()> {
+        todo!()
+    }
+
+    async fn close(&mut self) -> crate::Result<()> {
+        Ok(())
+    }
+}
+
+pub struct MadsimPager {}
+
+#[async_trait]
+impl oio::Page for MadsimPager {
+    async fn next(&mut self) -> crate::Result<Option<Vec<Entry>>> {
+        todo!()
+    }
+}
+
+impl oio::BlockingPage for MadsimPager {
+    fn next(&mut self) -> crate::Result<Option<Vec<Entry>>> {
+        todo!()
+    }
+}
+
+/// A simulated server.
+#[derive(Default, Clone)]
+pub struct SimServer;
+
+impl SimServer {
+    pub async fn serve(addr: SocketAddr) -> Result<()> {
+        let ep = Endpoint::bind(addr).await?;
+        let service = Arc::new(SimService::default());
+        loop {
+            let (tx, mut rx, _) = ep.accept1().await?;
+            let service = service.clone();
+            madsim::task::spawn(async move {
+                let request = *rx
+                    .recv()
+                    .await?
+                    .downcast::<Request>()
+                    .expect("invalid request");
+                let response = match request {
+                    Request::Read(path, args) => {
+                        Box::new(service.read(&path, args).await) as Payload
+                    }
+                    Request::Write(path, args) => {
+                        Box::new(service.write(&path, args).await) as Payload
+                    }
+                };
+                tx.send(response).await?;
+                Ok(()) as Result<()>
+            });
+        }
+    }
+}
+
+enum Request {
+    Read(String, OpRead),
+    Write(String, Bytes),
+}
+
+#[derive(Default)]
+pub struct SimService {
+    inner: Mutex<HashMap<String, Bytes>>,
+}
+
+impl SimService {
+    async fn read(&self, path: &str, args: OpRead) -> ReadResponse {
+        let inner = self.inner.lock().unwrap();
+        let data = inner.get(path);
+        ReadResponse {
+            data: data.cloned(),
+        }
+    }
+
+    async fn write(&self, path: &str, data: Bytes) -> WriteResponse {
+        let mut inner = self.inner.lock().unwrap();
+        inner.insert(path.to_string(), data);
+        WriteResponse {}
+    }
+}
+
+struct ReadResponse {
+    data: Option<Bytes>,
+}
+
+struct WriteResponse {}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::{services, EntryMode, Operator};
+    use madsim::{net::NetSim, runtime::Handle, time::sleep};
+    use std::time::Duration;
+
+    #[madsim::test]
+    async fn test_madsim_layer() {
+        let handle = Handle::current();
+        let ip1 = "10.0.0.1".parse().unwrap();
+        let ip2 = "10.0.0.2".parse().unwrap();
+        let sim_server_socket = "10.0.0.1:2379".parse().unwrap();
+        let server = handle.create_node().name("server").ip(ip1).build();
+        let client = handle.create_node().name("client").ip(ip2).build();
+
+        server.spawn(async move {
+            SimServer::serve(sim_server_socket).await.unwrap();
+        });
+        sleep(Duration::from_secs(1)).await;
+
+        let handle = client.spawn(async move {
+            let mut builder = services::Fs::default();
+            builder.root(".");
+            let op = Operator::new(builder)
+                .unwrap()
+                .layer(MadsimLayer::new(sim_server_socket))
+                .finish();
+
+            let path = "hello.txt";
+            let data = "Hello, World!";
+            op.write(path, data).await.unwrap();
+            assert_eq!(data.as_bytes(), op.read(path).await.unwrap());
+        });
+        handle.await.unwrap();
+    }
+}