feat(layers/blocking): add blocking layer (#2780)

* Add blocking layer

Signed-off-by: yah01 <[email protected]>

* Add blocking layer

Signed-off-by: yah01 <[email protected]>

* Fix build

Signed-off-by: Xuanwo <[email protected]>

* Fix build

Signed-off-by: Xuanwo <[email protected]>

* Add blocking layer

Signed-off-by: yah01 <[email protected]>

* Fix behavior test

Signed-off-by: yah01 <[email protected]>

* add unit tests for blocking layer

Signed-off-by: yah01 <[email protected]>

* Fix blocking layer unit tests

Signed-off-by: yah01 <[email protected]>

* bump openssh-sftp-client to 0.13.8

Signed-off-by: yah01 <[email protected]>

---------

Signed-off-by: yah01 <[email protected]>
Signed-off-by: Xuanwo <[email protected]>
Signed-off-by: yah01 <[email protected]>
Co-authored-by: Xuanwo <[email protected]>

core/Cargo.toml

@@ -223,7 +223,7 @@ minitrace = { version = "0.5", optional = true }
 moka = { version = "0.10", optional = true, features = ["future"] }
 once_cell = "1"
 openssh = { version = "0.9.9", optional = true }
-openssh-sftp-client = { version = "0.13.7", optional = true, features = [
+openssh-sftp-client = { version = "0.13.8", optional = true, features = [
   "openssh",
   "tracing",
 ] }

core/src/layers/blocking.rs

@@ -0,0 +1,260 @@
+// 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 async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime::Handle;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+    handle: Handle,
+}
+
+impl BlockingLayer {
+    /// Create a new `BlockingLayer` with the current runtime's handle
+    pub fn create() -> Result<Self> {
+        Ok(Self {
+            handle: Handle::try_current()
+                .map_err(|_| Error::new(ErrorKind::Unexpected, "failed to get current handle"))?,
+        })
+    }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+    type LayeredAccessor = BlockingAccessor<A>;
+
+    fn layer(&self, inner: A) -> Self::LayeredAccessor {
+        BlockingAccessor {
+            inner,
+            handle: self.handle.clone(),
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+    inner: A,
+
+    handle: Handle,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+    type Inner = A;
+    type Reader = A::Reader;
+    type BlockingReader = BlockingWrapper<A::Reader>;
+    type Writer = A::Writer;
+    type BlockingWriter = BlockingWrapper<A::Writer>;
+    type Appender = A::Appender;
+    type Pager = A::Pager;
+    type BlockingPager = BlockingWrapper<A::Pager>;
+
+    fn inner(&self) -> &Self::Inner {
+        &self.inner
+    }
+
+    fn metadata(&self) -> AccessorInfo {
+        let mut info = self.inner.info();
+        let cap = info.capability_mut();
+        cap.blocking = true;
+        info
+    }
+
+    async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+        self.inner.create_dir(path, args).await
+    }
+
+    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+        self.inner.read(path, args).await
+    }
+
+    async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        self.inner.write(path, args).await
+    }
+
+    async fn append(&self, path: &str, args: OpAppend) -> Result<(RpAppend, Self::Appender)> {
+        self.inner.append(path, args).await
+    }
+
+    async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        self.inner.copy(from, to, args).await
+    }
+
+    async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+        self.inner.rename(from, to, args).await
+    }
+
+    async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+        self.inner.stat(path, args).await
+    }
+
+    async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+        self.inner.delete(path, args).await
+    }
+
+    async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Pager)> {
+        self.inner.list(path, args).await
+    }
+
+    async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+        self.inner.presign(path, args).await
+    }
+
+    async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
+        self.inner.batch(args).await
+    }
+
+    fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+        self.handle.block_on(self.inner.create_dir(path, args))
+    }
+
+    fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+        self.handle.block_on(async {
+            let (rp, reader) = self.inner.read(path, args).await?;
+            let blocking_reader = Self::BlockingReader::new(self.handle.clone(), reader);
+
+            Ok((rp, blocking_reader))
+        })
+    }
+
+    fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+        self.handle.block_on(async {
+            let (rp, writer) = self.inner.write(path, args).await?;
+            let blocking_writer = Self::BlockingWriter::new(self.handle.clone(), writer);
+            Ok((rp, blocking_writer))
+        })
+    }
+
+    fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        self.handle.block_on(self.inner.copy(from, to, args))
+    }
+
+    fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+        self.handle.block_on(self.inner.rename(from, to, args))
+    }
+
+    fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+        self.handle.block_on(self.inner.stat(path, args))
+    }
+
+    fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+        self.handle.block_on(self.inner.delete(path, args))
+    }
+
+    fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingPager)> {
+        self.handle.block_on(async {
+            let (rp, pager) = self.inner.list(path, args).await?;
+            let blocking_pager = Self::BlockingPager::new(self.handle.clone(), pager);
+            Ok((rp, blocking_pager))
+        })
+    }
+}
+
+pub struct BlockingWrapper<I> {
+    handle: Handle,
+    inner: I,
+}
+
+impl<I> BlockingWrapper<I> {
+    fn new(handle: Handle, inner: I) -> Self {
+        Self { handle, inner }
+    }
+}
+
+impl<I: oio::Read + 'static> oio::BlockingRead for BlockingWrapper<I> {
+    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+        self.handle
+            .block_on(oio::ReadExt::read(&mut self.inner, buf))
+    }
+
+    fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64> {
+        self.handle.block_on(self.inner.seek(pos))
+    }
+
+    fn next(&mut self) -> Option<Result<Bytes>> {
+        self.handle.block_on(self.inner.next())
+    }
+}
+
+impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
+    fn write(&mut self, bs: Bytes) -> Result<()> {
+        self.handle.block_on(oio::Write::write(&mut self.inner, bs))
+    }
+
+    fn close(&mut self) -> Result<()> {
+        self.handle.block_on(oio::Write::close(&mut self.inner))
+    }
+}
+
+impl<I: oio::Page> oio::BlockingPage for BlockingWrapper<I> {
+    fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+        self.handle.block_on(self.inner.next())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use once_cell::sync::Lazy;
+
+    use crate::types::Result;
+
+    use super::*;
+
+    static RUNTIME: Lazy<tokio::runtime::Runtime> = Lazy::new(|| {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+    });
+
+    fn create_blocking_layer() -> Result<BlockingLayer> {
+        let _guard = RUNTIME.enter();
+        BlockingLayer::create()
+    }
+
+    #[test]
+    fn test_blocking_layer_in_blocking_context() {
+        // create in a blocking context should fail
+        let layer = BlockingLayer::create();
+        assert!(layer.is_err());
+
+        // create in an async context and drop in a blocking context
+        let layer = create_blocking_layer();
+        assert!(layer.is_ok())
+    }
+
+    #[test]
+    fn test_blocking_layer_in_async_context() {
+        // create and drop in an async context
+        let _guard = RUNTIME.enter();
+
+        let layer = BlockingLayer::create();
+        assert!(layer.is_ok());
+    }
+}

core/src/layers/mod.rs

@@ -29,6 +29,9 @@ pub use logging::LoggingLayer;
 mod timeout;
 pub use timeout::TimeoutLayer;
 
+mod blocking;
+pub use blocking::BlockingLayer;
+
 #[cfg(feature = "layers-chaos")]
 mod chaos;
 #[cfg(feature = "layers-chaos")]

core/src/types/operator/builder.rs

@@ -370,7 +370,6 @@ impl<A: Accessor> OperatorBuilder<A> {
     /// Finish the building to construct an Operator.
     pub fn finish(self) -> Operator {
         let ob = self.layer(TypeEraseLayer);
-
         Operator::from_inner(Arc::new(ob.accessor) as FusedAccessor)
     }
 }

core/src/types/operator/operator.rs

@@ -608,7 +608,7 @@ impl Operator {
                         .with_context("path", path));
                     }
 
-                    Reader::create_dir(inner.clone(), &path, args).await
+                    Reader::create(inner.clone(), &path, args).await
                 };
 
                 Box::pin(fut)

core/src/types/reader.rs

@@ -66,7 +66,7 @@ impl Reader {
     ///
     /// We don't want to expose those details to users so keep this function
     /// in crate only.
-    pub(crate) async fn create_dir(acc: FusedAccessor, path: &str, op: OpRead) -> Result<Self> {
+    pub(crate) async fn create(acc: FusedAccessor, path: &str, op: OpRead) -> Result<Self> {
         let (_, r) = acc.read(path, op).await?;
 
         Ok(Reader {
@@ -192,24 +192,7 @@ impl BlockingReader {
     /// We don't want to expose those details to users so keep this function
     /// in crate only.
     pub(crate) fn create(acc: FusedAccessor, path: &str, op: OpRead) -> Result<Self> {
-        let acc_meta = acc.info();
-
-        let r = if acc_meta.capability().read_can_seek {
-            let (_, r) = acc.blocking_read(path, op)?;
-            r
-        } else {
-            return Err(Error::new(
-                ErrorKind::Unsupported,
-                "non seekable blocking reader is not supported",
-            ));
-        };
-
-        let r = if acc_meta.capability().read_can_next {
-            r
-        } else {
-            // Make this capacity configurable.
-            Box::new(oio::into_streamable_read(r, 256 * 1024))
-        };
+        let (_, r) = acc.blocking_read(path, op)?;
 
         Ok(BlockingReader { inner: r })
     }

core/tests/behavior/utils.rs

@@ -28,9 +28,9 @@ use futures::Future;
 use libtest_mimic::Failed;
 use libtest_mimic::Trial;
 use log::debug;
-use opendal::layers::LoggingLayer;
 use opendal::layers::RetryLayer;
 use opendal::layers::TimeoutLayer;
+use opendal::layers::{BlockingLayer, LoggingLayer};
 use opendal::*;
 use rand::distributions::uniform::SampleRange;
 use rand::prelude::*;
@@ -81,7 +81,9 @@ pub fn init_service<B: Builder>() -> Option<Operator> {
         op.layer(ChaosLayer::new(0.1))
     };
 
+    let _guard = RUNTIME.enter();
     let op = op
+        .layer(BlockingLayer::create().expect("blocking layer must be created"))
         .layer(LoggingLayer::default())
         .layer(TimeoutLayer::new())
         .layer(RetryLayer::new())