Revert "replace with pollster::block_on"

This reverts commit 64362a0.

core/Cargo.toml

@@ -251,7 +251,6 @@ reqwest = { version = "0.11.18", features = [
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1.27", features = ["sync"] }
-pollster = "0.3.0"
 uuid = { version = "1", features = ["serde", "v4"] }
 
 # Test only dependencies

core/src/layers/blocking.rs

@@ -19,6 +19,7 @@ use async_trait::async_trait;
 use bytes;
 use bytes::Bytes;
 use futures::future::poll_fn;
+use tokio::runtime::Handle;
 
 use crate::raw::oio::ReadExt;
 use crate::raw::*;
@@ -137,26 +138,36 @@ use crate::*;
 /// }
 /// ```
 #[derive(Debug, Clone)]
-pub struct BlockingLayer;
+pub struct BlockingLayer {
+    handle: Handle,
+}
 
 impl BlockingLayer {
     /// Create a new `BlockingLayer` with the current runtime's handle
     pub fn create() -> Result<Self> {
-        Ok(BlockingLayer)
+        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 }
+        BlockingAccessor {
+            inner,
+            handle: self.handle.clone(),
+        }
     }
 }
 
 #[derive(Clone, Debug)]
 pub struct BlockingAccessor<A: Accessor> {
     inner: A,
+
+    handle: Handle,
 }
 
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
@@ -221,103 +232,129 @@ impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
     }
 
     fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
-        pollster::block_on(self.inner.create_dir(path, args))
+        self.handle.block_on(self.inner.create_dir(path, args))
     }
 
     fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
-        pollster::block_on(async {
+        self.handle.block_on(async {
             let (rp, reader) = self.inner.read(path, args).await?;
-            let blocking_reader = Self::BlockingReader::new(reader);
+            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)> {
-        pollster::block_on(async {
+        self.handle.block_on(async {
             let (rp, writer) = self.inner.write(path, args).await?;
-            let blocking_writer = Self::BlockingWriter::new(writer);
+            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> {
-        pollster::block_on(self.inner.copy(from, to, args))
+        self.handle.block_on(self.inner.copy(from, to, args))
     }
 
     fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
-        pollster::block_on(self.inner.rename(from, to, args))
+        self.handle.block_on(self.inner.rename(from, to, args))
     }
 
     fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
-        pollster::block_on(self.inner.stat(path, args))
+        self.handle.block_on(self.inner.stat(path, args))
     }
 
     fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
-        pollster::block_on(self.inner.delete(path, args))
+        self.handle.block_on(self.inner.delete(path, args))
     }
 
     fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
-        pollster::block_on(async {
+        self.handle.block_on(async {
             let (rp, lister) = self.inner.list(path, args).await?;
-            let blocking_lister = Self::BlockingLister::new(lister);
+            let blocking_lister = Self::BlockingLister::new(self.handle.clone(), lister);
             Ok((rp, blocking_lister))
         })
     }
 }
 
 pub struct BlockingWrapper<I> {
+    handle: Handle,
     inner: I,
 }
 
 impl<I> BlockingWrapper<I> {
-    fn new(inner: I) -> Self {
-        Self { inner }
+    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> {
-        pollster::block_on(self.inner.read(buf))
+        self.handle.block_on(self.inner.read(buf))
     }
 
     fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64> {
-        pollster::block_on(self.inner.seek(pos))
+        self.handle.block_on(self.inner.seek(pos))
     }
 
     fn next(&mut self) -> Option<Result<Bytes>> {
-        pollster::block_on(self.inner.next())
+        self.handle.block_on(self.inner.next())
     }
 }
 
 impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
     fn write(&mut self, bs: &dyn oio::WriteBuf) -> Result<usize> {
-        pollster::block_on(poll_fn(|cx| self.inner.poll_write(cx, bs)))
+        self.handle
+            .block_on(poll_fn(|cx| self.inner.poll_write(cx, bs)))
     }
 
     fn close(&mut self) -> Result<()> {
-        pollster::block_on(poll_fn(|cx| self.inner.poll_close(cx)))
+        self.handle
+            .block_on(poll_fn(|cx| self.inner.poll_close(cx)))
     }
 }
 
 impl<I: oio::List> oio::BlockingList for BlockingWrapper<I> {
     fn next(&mut self) -> Result<Option<oio::Entry>> {
-        pollster::block_on(poll_fn(|cx| self.inner.poll_next(cx)))
+        self.handle.block_on(poll_fn(|cx| self.inner.poll_next(cx)))
     }
 }
 
 #[cfg(test)]
 mod tests {
+    use once_cell::sync::Lazy;
+
     use super::*;
+    use crate::types::Result;
+
+    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());
     }