You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/05/23 10:44:59 UTC
(opendal) 02/18: Save work
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch remove-blocking-api-in-oio-access
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 17ccd09956934d6933dbf1512fd664928022fcb6
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri May 9 21:51:31 2025 +0800
Save work
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/await_tree.rs | 44 ---
core/src/layers/blocking.rs | 362 ------------------
core/src/layers/capability_check.rs | 68 ----
core/src/layers/chaos.rs | 32 --
core/src/layers/complete.rs | 125 -------
core/src/layers/concurrent_limit.rs | 86 -----
core/src/layers/correctness_check.rs | 83 ----
core/src/layers/dtrace.rs | 130 +------
core/src/layers/error_context.rs | 188 ----------
core/src/layers/fastrace.rs | 112 ------
core/src/layers/immutable_index.rs | 37 --
core/src/layers/logging.rs | 520 --------------------------
core/src/layers/mime_guess.rs | 61 ---
core/src/layers/mod.rs | 5 -
core/src/layers/oteltrace.rs | 102 -----
core/src/layers/retry.rs | 183 ---------
core/src/layers/throttle.rs | 62 ---
core/src/layers/timeout.rs | 24 --
core/src/layers/tracing.rs | 112 ------
core/src/layers/type_eraser.rs | 28 --
core/src/raw/oio/list/mod.rs | 2 -
core/src/raw/oio/write/mod.rs | 2 -
core/src/raw/tests/utils.rs | 9 -
core/src/services/aliyun_drive/backend.rs | 4 -
core/src/services/alluxio/backend.rs | 4 -
core/src/services/azblob/backend.rs | 4 -
core/src/services/azdls/backend.rs | 4 -
core/src/services/azfile/backend.rs | 4 -
core/src/services/b2/backend.rs | 4 -
core/src/services/compfs/backend.rs | 4 -
core/src/services/cos/backend.rs | 4 -
core/src/services/dbfs/backend.rs | 4 -
core/src/services/dropbox/backend.rs | 4 -
core/src/services/fs/backend.rs | 183 ---------
core/src/services/fs/delete.rs | 22 --
core/src/services/fs/reader.rs | 37 --
core/src/services/ftp/backend.rs | 4 -
core/src/services/gcs/backend.rs | 4 -
core/src/services/gdrive/backend.rs | 4 -
core/src/services/ghac/backend.rs | 4 -
core/src/services/github/backend.rs | 4 -
core/src/services/hdfs/backend.rs | 187 ---------
core/src/services/hdfs/delete.rs | 29 --
core/src/services/hdfs/reader.rs | 33 --
core/src/services/hdfs_native/backend.rs | 4 -
core/src/services/http/backend.rs | 4 -
core/src/services/huggingface/backend.rs | 4 -
core/src/services/icloud/backend.rs | 4 -
core/src/services/ipfs/backend.rs | 4 -
core/src/services/ipmfs/backend.rs | 4 -
core/src/services/koofr/backend.rs | 4 -
core/src/services/lakefs/backend.rs | 4 -
core/src/services/memory/backend.rs | 37 --
core/src/services/mini_moka/backend.rs | 16 -
core/src/services/moka/backend.rs | 16 -
core/src/services/monoiofs/backend.rs | 4 -
core/src/services/obs/backend.rs | 4 -
core/src/services/onedrive/backend.rs | 4 -
core/src/services/oss/backend.rs | 4 -
core/src/services/pcloud/backend.rs | 4 -
core/src/services/persy/backend.rs | 30 --
core/src/services/redb/backend.rs | 31 --
core/src/services/rocksdb/backend.rs | 40 --
core/src/services/s3/backend.rs | 4 -
core/src/services/seafile/backend.rs | 4 -
core/src/services/sftp/backend.rs | 4 -
core/src/services/sled/backend.rs | 39 --
core/src/services/swift/backend.rs | 4 -
core/src/services/upyun/backend.rs | 4 -
core/src/services/vercel_artifacts/backend.rs | 4 -
core/src/services/vercel_blob/backend.rs | 4 -
core/src/services/webdav/backend.rs | 4 -
core/src/services/webhdfs/backend.rs | 4 -
core/src/services/yandex_disk/backend.rs | 4 -
74 files changed, 1 insertion(+), 3228 deletions(-)
diff --git a/core/src/layers/await_tree.rs b/core/src/layers/await_tree.rs
index b6ffa3198..02722fbfd 100644
--- a/core/src/layers/await_tree.rs
+++ b/core/src/layers/await_tree.rs
@@ -71,13 +71,9 @@ pub struct AwaitTreeAccessor<A: Access> {
impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
type Inner = A;
type Reader = AwaitTreeWrapper<A::Reader>;
- type BlockingReader = AwaitTreeWrapper<A::BlockingReader>;
type Writer = AwaitTreeWrapper<A::Writer>;
- type BlockingWriter = AwaitTreeWrapper<A::BlockingWriter>;
type Lister = AwaitTreeWrapper<A::Lister>;
- type BlockingLister = AwaitTreeWrapper<A::BlockingLister>;
type Deleter = AwaitTreeWrapper<A::Deleter>;
- type BlockingDeleter = AwaitTreeWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -142,30 +138,6 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
.instrument_await(format!("opendal::{}", Operation::Presign))
.await
}
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner
- .blocking_write(path, args)
- .map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner
- .blocking_list(path, args)
- .map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner
- .blocking_delete()
- .map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
- }
}
pub struct AwaitTreeWrapper<R> {
@@ -187,12 +159,6 @@ impl<R: oio::Read> oio::Read for AwaitTreeWrapper<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for AwaitTreeWrapper<R> {
- fn read(&mut self) -> Result<Buffer> {
- self.inner.read()
- }
-}
-
impl<R: oio::Write> oio::Write for AwaitTreeWrapper<R> {
fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner
@@ -250,13 +216,3 @@ impl<R: oio::Delete> oio::Delete for AwaitTreeWrapper<R> {
.await
}
}
-
-impl<R: oio::BlockingDelete> oio::BlockingDelete for AwaitTreeWrapper<R> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- self.inner.delete(path, args)
- }
-
- fn flush(&mut self) -> Result<usize> {
- self.inner.flush()
- }
-}
diff --git a/core/src/layers/blocking.rs b/core/src/layers/blocking.rs
deleted file mode 100644
index 63def0e3c..000000000
--- a/core/src/layers/blocking.rs
+++ /dev/null
@@ -1,362 +0,0 @@
-// 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 tokio::runtime::Handle;
-
-use crate::raw::*;
-use crate::*;
-
-/// Add blocking API support for non-blocking services.
-///
-/// # Notes
-///
-/// - Please only enable this layer when the underlying service does not support blocking.
-///
-/// # Examples
-///
-/// ## In async context
-///
-/// BlockingLayer will use current async context's runtime to handle the async calls.
-///
-/// ```rust,no_run
-/// # use opendal::layers::BlockingLayer;
-/// # use opendal::services;
-/// # use opendal::BlockingOperator;
-/// # use opendal::Operator;
-/// # use opendal::Result;
-///
-/// #[tokio::main]
-/// async fn main() -> Result<()> {
-/// // Create fs backend builder.
-/// let mut builder = services::S3::default().bucket("test").region("us-east-1");
-///
-/// // Build an `BlockingOperator` with blocking layer to start operating the storage.
-/// let _: BlockingOperator = Operator::new(builder)?
-/// .layer(BlockingLayer::create()?)
-/// .finish()
-/// .blocking();
-///
-/// Ok(())
-/// }
-/// ```
-///
-/// ## In async context with blocking functions
-///
-/// If `BlockingLayer` is called in blocking function, please fetch a [`tokio::runtime::EnterGuard`]
-/// first. You can use [`Handle::try_current`] first to get the handle and then call [`Handle::enter`].
-/// This often happens in the case that async function calls blocking function.
-///
-/// ```rust,no_run
-/// # use opendal::layers::BlockingLayer;
-/// # use opendal::services;
-/// # use opendal::BlockingOperator;
-/// # use opendal::Operator;
-/// # use opendal::Result;
-///
-/// #[tokio::main]
-/// async fn main() -> Result<()> {
-/// let _ = blocking_fn()?;
-/// Ok(())
-/// }
-///
-/// fn blocking_fn() -> Result<BlockingOperator> {
-/// // Create fs backend builder.
-/// let mut builder = services::S3::default().bucket("test").region("us-east-1");
-///
-/// let handle = tokio::runtime::Handle::try_current().unwrap();
-/// let _guard = handle.enter();
-/// // Build an `BlockingOperator` with blocking layer to start operating the storage.
-/// let op: BlockingOperator = Operator::new(builder)?
-/// .layer(BlockingLayer::create()?)
-/// .finish()
-/// .blocking();
-/// Ok(op)
-/// }
-/// ```
-///
-/// ## In blocking context
-///
-/// In a pure blocking context, we can create a runtime and use it to create the `BlockingLayer`.
-///
-/// > The following code uses a global statically created runtime as an example, please manage the
-/// > runtime on demand.
-///
-/// ```rust,no_run
-/// # use std::sync::LazyLock;
-/// # use opendal::layers::BlockingLayer;
-/// # use opendal::services;
-/// # use opendal::BlockingOperator;
-/// # use opendal::Operator;
-/// # use opendal::Result;
-///
-/// static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
-/// tokio::runtime::Builder::new_multi_thread()
-/// .enable_all()
-/// .build()
-/// .unwrap()
-/// });
-///
-/// fn main() -> Result<()> {
-/// // Create fs backend builder.
-/// let mut builder = services::S3::default().bucket("test").region("us-east-1");
-///
-/// // Fetch the `EnterGuard` from global runtime.
-/// let _guard = RUNTIME.enter();
-/// // Build an `BlockingOperator` with blocking layer to start operating the storage.
-/// let _: BlockingOperator = Operator::new(builder)?
-/// .layer(BlockingLayer::create()?)
-/// .finish()
-/// .blocking();
-///
-/// Ok(())
-/// }
-/// ```
-#[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: Access> Layer<A> for BlockingLayer {
- type LayeredAccess = BlockingAccessor<A>;
-
- fn layer(&self, inner: A) -> Self::LayeredAccess {
- let info = inner.info();
- info.update_full_capability(|mut cap| {
- cap.blocking = true;
- cap
- });
-
- BlockingAccessor {
- inner,
- handle: self.handle.clone(),
- }
- }
-}
-
-#[derive(Clone, Debug)]
-pub struct BlockingAccessor<A: Access> {
- inner: A,
-
- handle: Handle,
-}
-
-impl<A: Access> LayeredAccess 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 Lister = A::Lister;
- type BlockingLister = BlockingWrapper<A::Lister>;
- type Deleter = A::Deleter;
- type BlockingDeleter = BlockingWrapper<A::Deleter>;
-
- fn inner(&self) -> &Self::Inner {
- &self.inner
- }
-
- 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 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) -> Result<(RpDelete, Self::Deleter)> {
- self.inner.delete().await
- }
-
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- self.inner.list(path, args).await
- }
-
- async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
- self.inner.presign(path, 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) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.handle.block_on(async {
- let (rp, writer) = self.inner.delete().await?;
- let blocking_deleter = Self::BlockingDeleter::new(self.handle.clone(), writer);
- Ok((rp, blocking_deleter))
- })
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.handle.block_on(async {
- let (rp, lister) = self.inner.list(path, args).await?;
- let blocking_lister = Self::BlockingLister::new(self.handle.clone(), lister);
- Ok((rp, blocking_lister))
- })
- }
-}
-
-pub struct BlockingWrapper<I> {
- handle: Handle,
- inner: Option<I>,
-}
-
-impl<I> BlockingWrapper<I> {
- fn new(handle: Handle, inner: I) -> Self {
- Self {
- handle,
- inner: Some(inner),
- }
- }
-}
-
-impl<I: oio::Read + 'static> oio::BlockingRead for BlockingWrapper<I> {
- fn read(&mut self) -> Result<Buffer> {
- self.handle.block_on(self.inner.as_mut().unwrap().read())
- }
-}
-
-impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- self.handle.block_on(self.inner.as_mut().unwrap().write(bs))
- }
-
- fn close(&mut self) -> Result<Metadata> {
- self.handle.block_on(self.inner.as_mut().unwrap().close())
- }
-}
-
-impl<I> Drop for BlockingWrapper<I> {
- fn drop(&mut self) {
- if let Some(inner) = self.inner.take() {
- self.handle.block_on(async move {
- drop(inner);
- });
- }
- }
-}
-
-impl<I: oio::List> oio::BlockingList for BlockingWrapper<I> {
- fn next(&mut self) -> Result<Option<oio::Entry>> {
- self.handle.block_on(self.inner.as_mut().unwrap().next())
- }
-}
-
-impl<I: oio::Delete + 'static> oio::BlockingDelete for BlockingWrapper<I> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- self.inner.as_mut().unwrap().delete(path, args)
- }
-
- fn flush(&mut self) -> Result<usize> {
- self.handle.block_on(self.inner.as_mut().unwrap().flush())
- }
-}
-
-#[cfg(test)]
-mod tests {
- use std::sync::LazyLock;
-
- use super::*;
- use crate::types::Result;
-
- static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::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());
- }
-}
diff --git a/core/src/layers/capability_check.rs b/core/src/layers/capability_check.rs
index 2c9ce5849..6d5c3c9a6 100644
--- a/core/src/layers/capability_check.rs
+++ b/core/src/layers/capability_check.rs
@@ -84,10 +84,6 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = A::Deleter;
- type BlockingReader = A::BlockingReader;
- type BlockingWriter = A::BlockingWriter;
- type BlockingLister = A::BlockingLister;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -140,66 +136,6 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
self.inner.list(path, args).await
}
-
- fn blocking_read(
- &self,
- path: &str,
- args: OpRead,
- ) -> crate::Result<(RpRead, Self::BlockingReader)> {
- self.inner().blocking_read(path, args)
- }
-
- fn blocking_write(
- &self,
- path: &str,
- args: OpWrite,
- ) -> crate::Result<(RpWrite, Self::BlockingWriter)> {
- let capability = self.info.full_capability();
- if !capability.write_with_content_type && args.content_type().is_some() {
- return Err(new_unsupported_error(
- self.info.as_ref(),
- Operation::Write,
- "content_type",
- ));
- }
- if !capability.write_with_cache_control && args.cache_control().is_some() {
- return Err(new_unsupported_error(
- self.info.as_ref(),
- Operation::Write,
- "cache_control",
- ));
- }
- if !capability.write_with_content_disposition && args.content_disposition().is_some() {
- return Err(new_unsupported_error(
- self.info.as_ref(),
- Operation::Write,
- "content_disposition",
- ));
- }
-
- self.inner.blocking_write(path, args)
- }
-
- fn blocking_delete(&self) -> crate::Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete()
- }
-
- fn blocking_list(
- &self,
- path: &str,
- args: OpList,
- ) -> crate::Result<(RpList, Self::BlockingLister)> {
- let capability = self.info.full_capability();
- if !capability.list_with_versions && args.versions() {
- return Err(new_unsupported_error(
- self.info.as_ref(),
- Operation::List,
- "version",
- ));
- }
-
- self.inner.blocking_list(path, args)
- }
}
#[cfg(test)]
@@ -217,10 +153,6 @@ mod tests {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
- type BlockingReader = oio::BlockingReader;
- type BlockingWriter = oio::BlockingWriter;
- type BlockingLister = oio::BlockingLister;
- type BlockingDeleter = oio::BlockingDeleter;
fn info(&self) -> Arc<AccessorInfo> {
let info = AccessorInfo::default();
diff --git a/core/src/layers/chaos.rs b/core/src/layers/chaos.rs
index 342e56857..8a9424ec9 100644
--- a/core/src/layers/chaos.rs
+++ b/core/src/layers/chaos.rs
@@ -100,13 +100,9 @@ pub struct ChaosAccessor<A> {
impl<A: Access> LayeredAccess for ChaosAccessor<A> {
type Inner = A;
type Reader = ChaosReader<A::Reader>;
- type BlockingReader = ChaosReader<A::BlockingReader>;
type Writer = A::Writer;
- type BlockingWriter = A::BlockingWriter;
type Lister = A::Lister;
- type BlockingLister = A::BlockingLister;
type Deleter = A::Deleter;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -119,35 +115,17 @@ impl<A: Access> LayeredAccess for ChaosAccessor<A> {
.map(|(rp, r)| (rp, ChaosReader::new(r, self.rng.clone(), self.error_ratio)))
}
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, ChaosReader::new(r, self.rng.clone(), self.error_ratio)))
- }
-
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.inner.write(path, args).await
}
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner.blocking_write(path, args)
- }
-
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner.list(path, args).await
}
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner.blocking_list(path, args)
- }
-
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner.delete().await
}
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete()
- }
}
/// ChaosReader will inject error into read operations.
@@ -190,13 +168,3 @@ impl<R: oio::Read> oio::Read for ChaosReader<R> {
}
}
}
-
-impl<R: oio::BlockingRead> oio::BlockingRead for ChaosReader<R> {
- fn read(&mut self) -> Result<Buffer> {
- if self.i_feel_lucky() {
- self.inner.read()
- } else {
- Err(Self::unexpected_eof())
- }
- }
-}
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index ffc91d80f..0006b4ec1 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -288,63 +288,14 @@ impl<A: Access> CompleteAccessor<A> {
}
}
}
-
- fn complete_blocking_list(
- &self,
- path: &str,
- args: OpList,
- ) -> Result<(RpList, CompleteLister<A, A::BlockingLister>)> {
- let cap = self.info.native_capability();
-
- let recursive = args.recursive();
-
- match (recursive, cap.list_with_recursive) {
- // - If service can list_with_recursive, we can forward list to it directly.
- (_, true) => {
- let (rp, p) = self.inner.blocking_list(path, args)?;
- Ok((rp, CompleteLister::One(p)))
- }
- // If recursive is true but service can't list_with_recursive
- (true, false) => {
- // Forward path that ends with /
- if path.ends_with('/') {
- let p = FlatLister::new(self.inner.clone(), path);
- Ok((RpList::default(), CompleteLister::Two(p)))
- } else {
- let parent = get_parent(path);
- let p = FlatLister::new(self.inner.clone(), parent);
- let p = PrefixLister::new(p, path);
- Ok((RpList::default(), CompleteLister::Four(p)))
- }
- }
- // If recursive and service doesn't support list_with_recursive, we need to handle
- // list prefix by ourselves.
- (false, false) => {
- // Forward path that ends with /
- if path.ends_with('/') {
- let (rp, p) = self.inner.blocking_list(path, args)?;
- Ok((rp, CompleteLister::One(p)))
- } else {
- let parent = get_parent(path);
- let (rp, p) = self.inner.blocking_list(parent, args)?;
- let p = PrefixLister::new(p, path);
- Ok((rp, CompleteLister::Three(p)))
- }
- }
- }
- }
}
impl<A: Access> LayeredAccess for CompleteAccessor<A> {
type Inner = A;
type Reader = CompleteReader<A::Reader>;
- type BlockingReader = CompleteReader<A::BlockingReader>;
type Writer = CompleteWriter<A::Writer>;
- type BlockingWriter = CompleteWriter<A::BlockingWriter>;
type Lister = CompleteLister<A, A::Lister>;
- type BlockingLister = CompleteLister<A, A::BlockingLister>;
type Deleter = A::Deleter;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -387,36 +338,6 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
self.inner.presign(path, args).await
}
-
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- self.complete_blocking_create_dir(path, args)
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let size = args.range().size();
- self.inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, CompleteReader::new(r, size)))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let append = args.append();
- self.inner
- .blocking_write(path, args)
- .map(|(rp, w)| (rp, CompleteWriter::new(w, append)))
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.complete_blocking_stat(path, args)
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner().blocking_delete()
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.complete_blocking_list(path, args)
- }
}
pub type CompleteLister<A, P> =
@@ -472,20 +393,6 @@ impl<R: oio::Read> oio::Read for CompleteReader<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for CompleteReader<R> {
- fn read(&mut self) -> Result<Buffer> {
- let buf = self.inner.read()?;
-
- if buf.is_empty() {
- self.check()?;
- } else {
- self.read += buf.len() as u64;
- }
-
- Ok(buf)
- }
-}
-
pub struct CompleteWriter<W> {
inner: Option<W>,
append: bool,
@@ -577,35 +484,3 @@ where
Ok(())
}
}
-
-impl<W> oio::BlockingWrite for CompleteWriter<W>
-where
- W: oio::BlockingWrite,
-{
- fn write(&mut self, bs: Buffer) -> Result<()> {
- let w = self.inner.as_mut().ok_or_else(|| {
- Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
- })?;
-
- let len = bs.len();
- w.write(bs)?;
- self.size += len as u64;
-
- Ok(())
- }
-
- fn close(&mut self) -> Result<Metadata> {
- let w = self.inner.as_mut().ok_or_else(|| {
- Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
- })?;
-
- let mut ret = w.close()?;
- self.check(ret.content_length())?;
- if ret.content_length() == 0 {
- ret = ret.with_content_length(self.size);
- }
- self.inner = None;
-
- Ok(ret)
- }
-}
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index 738a1e034..64cabde58 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -184,13 +184,9 @@ pub struct ConcurrentLimitAccessor<A: Access> {
impl<A: Access> LayeredAccess for ConcurrentLimitAccessor<A> {
type Inner = A;
type Reader = ConcurrentLimitWrapper<A::Reader>;
- type BlockingReader = ConcurrentLimitWrapper<A::BlockingReader>;
type Writer = ConcurrentLimitWrapper<A::Writer>;
- type BlockingWriter = ConcurrentLimitWrapper<A::BlockingWriter>;
type Lister = ConcurrentLimitWrapper<A::Lister>;
- type BlockingLister = ConcurrentLimitWrapper<A::BlockingLister>;
type Deleter = ConcurrentLimitWrapper<A::Deleter>;
- type BlockingDeleter = ConcurrentLimitWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -271,72 +267,6 @@ impl<A: Access> LayeredAccess for ConcurrentLimitAccessor<A> {
.await
.map(|(rp, s)| (rp, ConcurrentLimitWrapper::new(s, permit)))
}
-
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- let _permit = self
- .semaphore
- .try_acquire()
- .expect("semaphore must be valid");
-
- self.inner.blocking_create_dir(path, args)
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let permit = self
- .semaphore
- .clone()
- .try_acquire_owned()
- .expect("semaphore must be valid");
-
- self.inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, ConcurrentLimitWrapper::new(r, permit)))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let permit = self
- .semaphore
- .clone()
- .try_acquire_owned()
- .expect("semaphore must be valid");
-
- self.inner
- .blocking_write(path, args)
- .map(|(rp, w)| (rp, ConcurrentLimitWrapper::new(w, permit)))
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let _permit = self
- .semaphore
- .try_acquire()
- .expect("semaphore must be valid");
-
- self.inner.blocking_stat(path, args)
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- let permit = self
- .semaphore
- .clone()
- .try_acquire_owned()
- .expect("semaphore must be valid");
-
- self.inner
- .blocking_delete()
- .map(|(rp, w)| (rp, ConcurrentLimitWrapper::new(w, permit)))
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let permit = self
- .semaphore
- .clone()
- .try_acquire_owned()
- .expect("semaphore must be valid");
-
- self.inner
- .blocking_list(path, args)
- .map(|(rp, it)| (rp, ConcurrentLimitWrapper::new(it, permit)))
- }
}
pub struct ConcurrentLimitWrapper<R> {
@@ -361,12 +291,6 @@ impl<R: oio::Read> oio::Read for ConcurrentLimitWrapper<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for ConcurrentLimitWrapper<R> {
- fn read(&mut self) -> Result<Buffer> {
- self.inner.read()
- }
-}
-
impl<R: oio::Write> oio::Write for ConcurrentLimitWrapper<R> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
self.inner.write(bs).await
@@ -412,13 +336,3 @@ impl<R: oio::Delete> oio::Delete for ConcurrentLimitWrapper<R> {
self.inner.flush().await
}
}
-
-impl<R: oio::BlockingDelete> oio::BlockingDelete for ConcurrentLimitWrapper<R> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- self.inner.delete(path, args)
- }
-
- fn flush(&mut self) -> Result<usize> {
- self.inner.flush()
- }
-}
diff --git a/core/src/layers/correctness_check.rs b/core/src/layers/correctness_check.rs
index 8a7658630..47e8a0716 100644
--- a/core/src/layers/correctness_check.rs
+++ b/core/src/layers/correctness_check.rs
@@ -80,10 +80,6 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = CheckWrapper<A::Deleter>;
- type BlockingReader = A::BlockingReader;
- type BlockingWriter = A::BlockingWriter;
- type BlockingLister = A::BlockingLister;
- type BlockingDeleter = CheckWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -216,70 +212,6 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner.list(path, args).await
}
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let capability = self.info.full_capability();
- if !capability.read_with_version && args.version().is_some() {
- return Err(new_unsupported_error(
- self.info.as_ref(),
- Operation::Read,
- "version",
- ));
- }
-
- self.inner.blocking_read(path, args)
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let capability = self.info.full_capability();
- if args.append() && !capability.write_can_append {
- return Err(new_unsupported_error(
- &self.info,
- Operation::Write,
- "append",
- ));
- }
- if args.if_not_exists() && !capability.write_with_if_not_exists {
- return Err(new_unsupported_error(
- &self.info,
- Operation::Write,
- "if_not_exists",
- ));
- }
- if args.if_none_match().is_some() && !capability.write_with_if_none_match {
- return Err(new_unsupported_error(
- self.info.as_ref(),
- Operation::Write,
- "if_none_match",
- ));
- }
-
- self.inner.blocking_write(path, args)
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let capability = self.info.full_capability();
- if !capability.stat_with_version && args.version().is_some() {
- return Err(new_unsupported_error(
- self.info.as_ref(),
- Operation::Stat,
- "version",
- ));
- }
-
- self.inner.blocking_stat(path, args)
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete().map(|(rp, deleter)| {
- let deleter = CheckWrapper::new(deleter, self.info.clone());
- (rp, deleter)
- })
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner.blocking_list(path, args)
- }
}
pub struct CheckWrapper<T> {
@@ -316,17 +248,6 @@ impl<T: oio::Delete> oio::Delete for CheckWrapper<T> {
}
}
-impl<T: oio::BlockingDelete> oio::BlockingDelete for CheckWrapper<T> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- self.check_delete(&args)?;
- self.inner.delete(path, args)
- }
-
- fn flush(&mut self) -> Result<usize> {
- self.inner.flush()
- }
-}
-
#[cfg(test)]
mod tests {
use super::*;
@@ -343,10 +264,6 @@ mod tests {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
- type BlockingReader = oio::BlockingReader;
- type BlockingWriter = oio::BlockingWriter;
- type BlockingLister = oio::BlockingLister;
- type BlockingDeleter = oio::BlockingDeleter;
fn info(&self) -> Arc<AccessorInfo> {
let info = AccessorInfo::default();
diff --git a/core/src/layers/dtrace.rs b/core/src/layers/dtrace.rs
index e51e65f64..0f822530b 100644
--- a/core/src/layers/dtrace.rs
+++ b/core/src/layers/dtrace.rs
@@ -42,12 +42,7 @@ use crate::*;
/// 5. delete
/// 6. list
/// 7. presign
-/// 8. blocking_create_dir
-/// 9. blocking_read
-/// 10. blocking_write
-/// 11. blocking_stat
-/// 12. blocking_delete
-/// 13. blocking_list
+///
/// 2. ${operation}_end, arguments: path
/// 1. create_dir
/// 2. read
@@ -56,12 +51,6 @@ use crate::*;
/// 5. delete
/// 6. list
/// 7. presign
-/// 8. blocking_create_dir
-/// 9. blocking_read
-/// 10. blocking_write
-/// 11. blocking_stat
-/// 12. blocking_delete
-/// 13. blocking_list
///
/// ### For Reader
///
@@ -69,12 +58,6 @@ use crate::*;
/// 2. reader_read_ok, arguments: path, length
/// 3. reader_read_error, arguments: path
///
-/// ### For BlockingReader
-///
-/// 1. blocking_reader_read_start, arguments: path
-/// 2. blocking_reader_read_ok, arguments: path, length
-/// 3. blocking_reader_read_error, arguments: path
-///
/// ### For Writer
///
/// 1. writer_write_start, arguments: path
@@ -87,15 +70,6 @@ use crate::*;
/// 8. writer_close_ok, arguments: path
/// 9. writer_close_error, arguments: path
///
-/// ### For BlockingWriter
-///
-/// 1. blocking_writer_write_start, arguments: path
-/// 2. blocking_writer_write_ok, arguments: path, length
-/// 3. blocking_writer_write_error, arguments: path
-/// 4. blocking_writer_close_start, arguments: path
-/// 5. blocking_writer_close_ok, arguments: path
-/// 6. blocking_writer_close_error, arguments: path
-///
/// Example:
///
/// ```no_run
@@ -174,13 +148,9 @@ impl<A: Access> Debug for DTraceAccessor<A> {
impl<A: Access> LayeredAccess for DTraceAccessor<A> {
type Inner = A;
type Reader = DtraceLayerWrapper<A::Reader>;
- type BlockingReader = DtraceLayerWrapper<A::BlockingReader>;
type Writer = DtraceLayerWrapper<A::Writer>;
- type BlockingWriter = DtraceLayerWrapper<A::BlockingWriter>;
type Lister = A::Lister;
- type BlockingLister = A::BlockingLister;
type Deleter = A::Deleter;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -246,56 +216,6 @@ impl<A: Access> LayeredAccess for DTraceAccessor<A> {
probe_lazy!(opendal, presign_end, c_path.as_ptr());
result
}
-
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- let c_path = CString::new(path).unwrap();
- probe_lazy!(opendal, blocking_create_dir_start, c_path.as_ptr());
- let result = self.inner.blocking_create_dir(path, args);
- probe_lazy!(opendal, blocking_create_dir_end, c_path.as_ptr());
- result
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let c_path = CString::new(path).unwrap();
- probe_lazy!(opendal, blocking_read_start, c_path.as_ptr());
- let result = self
- .inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, DtraceLayerWrapper::new(r, &path.to_string())));
- probe_lazy!(opendal, blocking_read_end, c_path.as_ptr());
- result
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let c_path = CString::new(path).unwrap();
- probe_lazy!(opendal, blocking_write_start, c_path.as_ptr());
- let result = self
- .inner
- .blocking_write(path, args)
- .map(|(rp, r)| (rp, DtraceLayerWrapper::new(r, &path.to_string())));
- probe_lazy!(opendal, blocking_write_end, c_path.as_ptr());
- result
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let c_path = CString::new(path).unwrap();
- probe_lazy!(opendal, blocking_stat_start, c_path.as_ptr());
- let result = self.inner.blocking_stat(path, args);
- probe_lazy!(opendal, blocking_stat_end, c_path.as_ptr());
- result
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete()
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let c_path = CString::new(path).unwrap();
- probe_lazy!(opendal, blocking_list_start, c_path.as_ptr());
- let result = self.inner.blocking_list(path, args);
- probe_lazy!(opendal, blocking_list_end, c_path.as_ptr());
- result
- }
}
pub struct DtraceLayerWrapper<R> {
@@ -329,26 +249,6 @@ impl<R: oio::Read> oio::Read for DtraceLayerWrapper<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for DtraceLayerWrapper<R> {
- fn read(&mut self) -> Result<Buffer> {
- let c_path = CString::new(self.path.clone()).unwrap();
- probe_lazy!(opendal, blocking_reader_read_start, c_path.as_ptr());
- self.inner
- .read()
- .inspect(|bs| {
- probe_lazy!(
- opendal,
- blocking_reader_read_ok,
- c_path.as_ptr(),
- bs.remaining()
- );
- })
- .inspect_err(|_| {
- probe_lazy!(opendal, blocking_reader_read_error, c_path.as_ptr());
- })
- }
-}
-
impl<R: oio::Write> oio::Write for DtraceLayerWrapper<R> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
let c_path = CString::new(self.path.clone()).unwrap();
@@ -392,31 +292,3 @@ impl<R: oio::Write> oio::Write for DtraceLayerWrapper<R> {
})
}
}
-
-impl<R: oio::BlockingWrite> oio::BlockingWrite for DtraceLayerWrapper<R> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- let c_path = CString::new(self.path.clone()).unwrap();
- probe_lazy!(opendal, blocking_writer_write_start, c_path.as_ptr());
- self.inner
- .write(bs)
- .map(|_| {
- probe_lazy!(opendal, blocking_writer_write_ok, c_path.as_ptr());
- })
- .inspect_err(|_| {
- probe_lazy!(opendal, blocking_writer_write_error, c_path.as_ptr());
- })
- }
-
- fn close(&mut self) -> Result<Metadata> {
- let c_path = CString::new(self.path.clone()).unwrap();
- probe_lazy!(opendal, blocking_writer_close_start, c_path.as_ptr());
- self.inner
- .close()
- .inspect(|_| {
- probe_lazy!(opendal, blocking_writer_close_ok, c_path.as_ptr());
- })
- .inspect_err(|_| {
- probe_lazy!(opendal, blocking_writer_close_error, c_path.as_ptr());
- })
- }
-}
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index a820bd07a..f82a75152 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -66,13 +66,9 @@ impl<A: Access> Debug for ErrorContextAccessor<A> {
impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
type Inner = A;
type Reader = ErrorContextWrapper<A::Reader>;
- type BlockingReader = ErrorContextWrapper<A::BlockingReader>;
type Writer = ErrorContextWrapper<A::Writer>;
- type BlockingWriter = ErrorContextWrapper<A::BlockingWriter>;
type Lister = ErrorContextWrapper<A::Lister>;
- type BlockingLister = ErrorContextWrapper<A::BlockingLister>;
type Deleter = ErrorContextWrapper<A::Deleter>;
- type BlockingDeleter = ErrorContextWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -189,106 +185,6 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
.with_context("path", path)
})
}
-
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- self.inner.blocking_create_dir(path, args).map_err(|err| {
- err.with_operation(Operation::CreateDir)
- .with_context("service", self.info.scheme())
- .with_context("path", path)
- })
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let range = args.range();
- self.inner
- .blocking_read(path, args)
- .map(|(rp, os)| {
- (
- rp,
- ErrorContextWrapper::new(self.info.scheme(), path.to_string(), os)
- .with_range(range),
- )
- })
- .map_err(|err| {
- err.with_operation(Operation::Read)
- .with_context("service", self.info.scheme())
- .with_context("path", path)
- .with_context("range", range.to_string())
- })
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner
- .blocking_write(path, args)
- .map(|(rp, os)| {
- (
- rp,
- ErrorContextWrapper::new(self.info.scheme(), path.to_string(), os),
- )
- })
- .map_err(|err| {
- err.with_operation(Operation::Write)
- .with_context("service", self.info.scheme())
- .with_context("path", path)
- })
- }
-
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner.blocking_copy(from, to, args).map_err(|err| {
- err.with_operation(Operation::Copy)
- .with_context("service", self.info.scheme())
- .with_context("from", from)
- .with_context("to", to)
- })
- }
-
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- self.inner.blocking_rename(from, to, args).map_err(|err| {
- err.with_operation(Operation::Rename)
- .with_context("service", self.info.scheme())
- .with_context("from", from)
- .with_context("to", to)
- })
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.inner.blocking_stat(path, args).map_err(|err| {
- err.with_operation(Operation::Stat)
- .with_context("service", self.info.scheme())
- .with_context("path", path)
- })
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner
- .blocking_delete()
- .map(|(rp, w)| {
- (
- rp,
- ErrorContextWrapper::new(self.info.scheme(), "".to_string(), w),
- )
- })
- .map_err(|err| {
- err.with_operation(Operation::Delete)
- .with_context("service", self.info.scheme())
- })
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner
- .blocking_list(path, args)
- .map(|(rp, os)| {
- (
- rp,
- ErrorContextWrapper::new(self.info.scheme(), path.to_string(), os),
- )
- })
- .map_err(|err| {
- err.with_operation(Operation::List)
- .with_context("service", self.info.scheme())
- .with_context("path", path)
- })
- }
}
pub struct ErrorContextWrapper<T> {
@@ -334,23 +230,6 @@ impl<T: oio::Read> oio::Read for ErrorContextWrapper<T> {
}
}
-impl<T: oio::BlockingRead> oio::BlockingRead for ErrorContextWrapper<T> {
- fn read(&mut self) -> Result<Buffer> {
- self.inner
- .read()
- .inspect(|bs| {
- self.processed += bs.len() as u64;
- })
- .map_err(|err| {
- err.with_operation(Operation::Read)
- .with_context("service", self.scheme)
- .with_context("path", &self.path)
- .with_context("range", self.range.to_string())
- .with_context("read", self.processed.to_string())
- })
- }
-}
-
impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
let size = bs.len();
@@ -388,33 +267,6 @@ impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
}
}
-impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- let size = bs.len();
- self.inner
- .write(bs)
- .map(|_| {
- self.processed += size as u64;
- })
- .map_err(|err| {
- err.with_operation(Operation::Write)
- .with_context("service", self.scheme)
- .with_context("path", &self.path)
- .with_context("size", size.to_string())
- .with_context("written", self.processed.to_string())
- })
- }
-
- fn close(&mut self) -> Result<Metadata> {
- self.inner.close().map_err(|err| {
- err.with_operation(Operation::Write)
- .with_context("service", self.scheme)
- .with_context("path", &self.path)
- .with_context("written", self.processed.to_string())
- })
- }
-}
-
impl<T: oio::List> oio::List for ErrorContextWrapper<T> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
self.inner
@@ -432,22 +284,6 @@ impl<T: oio::List> oio::List for ErrorContextWrapper<T> {
}
}
-impl<T: oio::BlockingList> oio::BlockingList for ErrorContextWrapper<T> {
- fn next(&mut self) -> Result<Option<oio::Entry>> {
- self.inner
- .next()
- .inspect(|bs| {
- self.processed += bs.is_some() as u64;
- })
- .map_err(|err| {
- err.with_operation(Operation::List)
- .with_context("service", self.scheme)
- .with_context("path", &self.path)
- .with_context("listed", self.processed.to_string())
- })
- }
-}
-
impl<T: oio::Delete> oio::Delete for ErrorContextWrapper<T> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
self.inner.delete(path, args).map_err(|err| {
@@ -472,27 +308,3 @@ impl<T: oio::Delete> oio::Delete for ErrorContextWrapper<T> {
})
}
}
-
-impl<T: oio::BlockingDelete> oio::BlockingDelete for ErrorContextWrapper<T> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- self.inner.delete(path, args).map_err(|err| {
- err.with_operation(Operation::Delete)
- .with_context("service", self.scheme)
- .with_context("path", path)
- .with_context("deleted", self.processed.to_string())
- })
- }
-
- fn flush(&mut self) -> Result<usize> {
- self.inner
- .flush()
- .inspect(|&n| {
- self.processed += n as u64;
- })
- .map_err(|err| {
- err.with_operation(Operation::Delete)
- .with_context("service", self.scheme)
- .with_context("deleted", self.processed.to_string())
- })
- }
-}
diff --git a/core/src/layers/fastrace.rs b/core/src/layers/fastrace.rs
index f01f4d2bc..57eaf257e 100644
--- a/core/src/layers/fastrace.rs
+++ b/core/src/layers/fastrace.rs
@@ -121,13 +121,9 @@ pub struct FastraceAccessor<A> {
impl<A: Access> LayeredAccess for FastraceAccessor<A> {
type Inner = A;
type Reader = FastraceWrapper<A::Reader>;
- type BlockingReader = FastraceWrapper<A::BlockingReader>;
type Writer = FastraceWrapper<A::Writer>;
- type BlockingWriter = FastraceWrapper<A::BlockingWriter>;
type Lister = FastraceWrapper<A::Lister>;
- type BlockingLister = FastraceWrapper<A::BlockingLister>;
type Deleter = FastraceWrapper<A::Deleter>;
- type BlockingDeleter = FastraceWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -214,78 +210,6 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
self.inner.presign(path, args).await
}
-
- #[trace]
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- self.inner.blocking_create_dir(path, args)
- }
-
- #[trace]
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner.blocking_read(path, args).map(|(rp, r)| {
- (
- rp,
- FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::Read.into_static()),
- r,
- ),
- )
- })
- }
-
- #[trace]
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner.blocking_write(path, args).map(|(rp, r)| {
- (
- rp,
- FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::Write.into_static()),
- r,
- ),
- )
- })
- }
-
- #[trace]
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner().blocking_copy(from, to, args)
- }
-
- #[trace]
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- self.inner().blocking_rename(from, to, args)
- }
-
- #[trace]
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.inner.blocking_stat(path, args)
- }
-
- #[trace]
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete().map(|(rp, r)| {
- (
- rp,
- FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::Delete.into_static()),
- r,
- ),
- )
- })
- }
-
- #[trace]
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner.blocking_list(path, args).map(|(rp, it)| {
- (
- rp,
- FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::List.into_static()),
- it,
- ),
- )
- })
- }
}
pub struct FastraceWrapper<R> {
@@ -306,14 +230,6 @@ impl<R: oio::Read> oio::Read for FastraceWrapper<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for FastraceWrapper<R> {
- fn read(&mut self) -> Result<Buffer> {
- let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::Read.into_static());
- self.inner.read()
- }
-}
-
impl<R: oio::Write> oio::Write for FastraceWrapper<R> {
fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<()>> + MaybeSend {
let _g = self.span.set_local_parent();
@@ -334,20 +250,6 @@ impl<R: oio::Write> oio::Write for FastraceWrapper<R> {
}
}
-impl<R: oio::BlockingWrite> oio::BlockingWrite for FastraceWrapper<R> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::Write.into_static());
- self.inner.write(bs)
- }
-
- fn close(&mut self) -> Result<Metadata> {
- let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::Write.into_static());
- self.inner.close()
- }
-}
-
impl<R: oio::List> oio::List for FastraceWrapper<R> {
#[trace(enter_on_poll = true)]
async fn next(&mut self) -> Result<Option<oio::Entry>> {
@@ -375,17 +277,3 @@ impl<R: oio::Delete> oio::Delete for FastraceWrapper<R> {
self.inner.flush().await
}
}
-
-impl<R: oio::BlockingDelete> oio::BlockingDelete for FastraceWrapper<R> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::Delete.into_static());
- self.inner.delete(path, args)
- }
-
- fn flush(&mut self) -> Result<usize> {
- let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::Delete.into_static());
- self.inner.flush()
- }
-}
diff --git a/core/src/layers/immutable_index.rs b/core/src/layers/immutable_index.rs
index 44b6c96da..28f67ef81 100644
--- a/core/src/layers/immutable_index.rs
+++ b/core/src/layers/immutable_index.rs
@@ -147,10 +147,6 @@ impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
type Writer = A::Writer;
type Lister = ImmutableDir;
type Deleter = A::Deleter;
- type BlockingReader = A::BlockingReader;
- type BlockingWriter = A::BlockingWriter;
- type BlockingLister = ImmutableDir;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -182,33 +178,6 @@ impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner.delete().await
}
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner.blocking_read(path, args)
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner.blocking_write(path, args)
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let mut path = path;
- if path == "/" {
- path = ""
- }
-
- let idx = if args.recursive() {
- self.children_flat(path)
- } else {
- self.children_hierarchy(path)
- };
-
- Ok((RpList::default(), ImmutableDir::new(idx)))
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete()
- }
}
pub struct ImmutableDir {
@@ -241,12 +210,6 @@ impl oio::List for ImmutableDir {
}
}
-impl oio::BlockingList for ImmutableDir {
- fn next(&mut self) -> Result<Option<oio::Entry>> {
- Ok(self.inner_next())
- }
-}
-
#[cfg(test)]
#[cfg(feature = "services-http")]
mod tests {
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index c679b2e01..208d19ad2 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -245,13 +245,9 @@ static LOGGING_TARGET: &str = "opendal::services";
impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
type Inner = A;
type Reader = LoggingReader<A::Reader, I>;
- type BlockingReader = LoggingReader<A::BlockingReader, I>;
type Writer = LoggingWriter<A::Writer, I>;
- type BlockingWriter = LoggingWriter<A::BlockingWriter, I>;
type Lister = LoggingLister<A::Lister, I>;
- type BlockingLister = LoggingLister<A::BlockingLister, I>;
type Deleter = LoggingDeleter<A::Deleter, I>;
- type BlockingDeleter = LoggingDeleter<A::BlockingDeleter, I>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -543,247 +539,6 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
);
})
}
-
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- self.logger.log(
- &self.info,
- Operation::CreateDir,
- &[("path", path)],
- "started",
- None,
- );
-
- self.inner
- .blocking_create_dir(path, args)
- .inspect(|_| {
- self.logger.log(
- &self.info,
- Operation::CreateDir,
- &[("path", path)],
- "finished",
- None,
- );
- })
- .inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::CreateDir,
- &[("path", path)],
- "failed",
- Some(err),
- );
- })
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.logger.log(
- &self.info,
- Operation::Read,
- &[("path", path)],
- "started",
- None,
- );
-
- self.inner
- .blocking_read(path, args.clone())
- .map(|(rp, r)| {
- self.logger.log(
- &self.info,
- Operation::Read,
- &[("path", path)],
- "created reader",
- None,
- );
- let r = LoggingReader::new(self.info.clone(), self.logger.clone(), path, r);
- (rp, r)
- })
- .inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::Read,
- &[("path", path)],
- "failed",
- Some(err),
- );
- })
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[("path", path)],
- "started",
- None,
- );
-
- self.inner
- .blocking_write(path, args)
- .map(|(rp, w)| {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[("path", path)],
- "created writer",
- None,
- );
- let w = LoggingWriter::new(self.info.clone(), self.logger.clone(), path, w);
- (rp, w)
- })
- .inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[("path", path)],
- "failed",
- Some(err),
- );
- })
- }
-
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.logger.log(
- &self.info,
- Operation::Copy,
- &[("from", from), ("to", to)],
- "started",
- None,
- );
-
- self.inner
- .blocking_copy(from, to, args)
- .inspect(|_| {
- self.logger.log(
- &self.info,
- Operation::Copy,
- &[("from", from), ("to", to)],
- "finished",
- None,
- );
- })
- .inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::Copy,
- &[("from", from), ("to", to)],
- "",
- Some(err),
- );
- })
- }
-
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- self.logger.log(
- &self.info,
- Operation::Rename,
- &[("from", from), ("to", to)],
- "started",
- None,
- );
-
- self.inner
- .blocking_rename(from, to, args)
- .inspect(|_| {
- self.logger.log(
- &self.info,
- Operation::Rename,
- &[("from", from), ("to", to)],
- "finished",
- None,
- );
- })
- .inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::Rename,
- &[("from", from), ("to", to)],
- "failed",
- Some(err),
- );
- })
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.logger.log(
- &self.info,
- Operation::Stat,
- &[("path", path)],
- "started",
- None,
- );
-
- self.inner
- .blocking_stat(path, args)
- .inspect(|_| {
- self.logger.log(
- &self.info,
- Operation::Stat,
- &[("path", path)],
- "finished",
- None,
- );
- })
- .inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::Stat,
- &[("path", path)],
- "failed",
- Some(err),
- );
- })
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.logger
- .log(&self.info, Operation::Delete, &[], "started", None);
-
- self.inner
- .blocking_delete()
- .map(|(rp, d)| {
- self.logger
- .log(&self.info, Operation::Delete, &[], "finished", None);
- let d = LoggingDeleter::new(self.info.clone(), self.logger.clone(), d);
- (rp, d)
- })
- .inspect_err(|err| {
- self.logger
- .log(&self.info, Operation::Delete, &[], "failed", Some(err));
- })
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.logger.log(
- &self.info,
- Operation::List,
- &[("path", path)],
- "started",
- None,
- );
-
- self.inner
- .blocking_list(path, args)
- .map(|(rp, v)| {
- self.logger.log(
- &self.info,
- Operation::List,
- &[("path", path)],
- "created lister",
- None,
- );
- let li = LoggingLister::new(self.info.clone(), self.logger.clone(), path, v);
- (rp, li)
- })
- .inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::List,
- &[("path", path)],
- "",
- Some(err),
- );
- })
- }
}
/// `LoggingReader` is a wrapper of `BytesReader`, with logging functionality.
@@ -853,50 +608,6 @@ impl<R: oio::Read, I: LoggingInterceptor> oio::Read for LoggingReader<R, I> {
}
}
-impl<R: oio::BlockingRead, I: LoggingInterceptor> oio::BlockingRead for LoggingReader<R, I> {
- fn read(&mut self) -> Result<Buffer> {
- self.logger.log(
- &self.info,
- Operation::Read,
- &[("path", &self.path), ("read", &self.read.to_string())],
- "started",
- None,
- );
-
- match self.inner.read() {
- Ok(bs) => {
- self.read += bs.len() as u64;
- self.logger.log(
- &self.info,
- Operation::Read,
- &[
- ("path", &self.path),
- ("read", &self.read.to_string()),
- ("size", &bs.len().to_string()),
- ],
- if bs.is_empty() {
- "finished"
- } else {
- "succeeded"
- },
- None,
- );
- Ok(bs)
- }
- Err(err) => {
- self.logger.log(
- &self.info,
- Operation::Read,
- &[("path", &self.path), ("read", &self.read.to_string())],
- "failed",
- Some(&err),
- );
- Err(err)
- }
- }
- }
-}
-
pub struct LoggingWriter<W, I> {
info: Arc<AccessorInfo>,
logger: I,
@@ -1035,88 +746,6 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
}
}
-impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for LoggingWriter<W, I> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- let size = bs.len();
-
- self.logger.log(
- &self.info,
- Operation::Write,
- &[
- ("path", &self.path),
- ("written", &self.written.to_string()),
- ("size", &size.to_string()),
- ],
- "started",
- None,
- );
-
- match self.inner.write(bs) {
- Ok(_) => {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[
- ("path", &self.path),
- ("written", &self.written.to_string()),
- ("size", &size.to_string()),
- ],
- "succeeded",
- None,
- );
- Ok(())
- }
- Err(err) => {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[
- ("path", &self.path),
- ("written", &self.written.to_string()),
- ("size", &size.to_string()),
- ],
- "failed",
- Some(&err),
- );
- Err(err)
- }
- }
- }
-
- fn close(&mut self) -> Result<Metadata> {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[("path", &self.path), ("written", &self.written.to_string())],
- "started",
- None,
- );
-
- match self.inner.close() {
- Ok(meta) => {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[("path", &self.path), ("written", &self.written.to_string())],
- "succeeded",
- None,
- );
- Ok(meta)
- }
- Err(err) => {
- self.logger.log(
- &self.info,
- Operation::Write,
- &[("path", &self.path), ("written", &self.written.to_string())],
- "failed",
- Some(&err),
- );
- Err(err)
- }
- }
- }
-}
-
pub struct LoggingLister<P, I: LoggingInterceptor> {
info: Arc<AccessorInfo>,
logger: I,
@@ -1190,56 +819,6 @@ impl<P: oio::List, I: LoggingInterceptor> oio::List for LoggingLister<P, I> {
}
}
-impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingLister<P, I> {
- fn next(&mut self) -> Result<Option<oio::Entry>> {
- self.logger.log(
- &self.info,
- Operation::List,
- &[("path", &self.path), ("listed", &self.listed.to_string())],
- "started",
- None,
- );
-
- let res = self.inner.next();
- match &res {
- Ok(Some(de)) => {
- self.listed += 1;
- self.logger.log(
- &self.info,
- Operation::List,
- &[
- ("path", &self.path),
- ("listed", &self.listed.to_string()),
- ("entry", de.path()),
- ],
- "succeeded",
- None,
- );
- }
- Ok(None) => {
- self.logger.log(
- &self.info,
- Operation::List,
- &[("path", &self.path), ("listed", &self.listed.to_string())],
- "finished",
- None,
- );
- }
- Err(err) => {
- self.logger.log(
- &self.info,
- Operation::List,
- &[("path", &self.path), ("listed", &self.listed.to_string())],
- "failed",
- Some(err),
- );
- }
- };
-
- res
- }
-}
-
pub struct LoggingDeleter<D, I: LoggingInterceptor> {
info: Arc<AccessorInfo>,
logger: I,
@@ -1360,102 +939,3 @@ impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I>
res
}
}
-
-impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for LoggingDeleter<D, I> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- let version = args
- .version()
- .map(|v| v.to_string())
- .unwrap_or_else(|| "<latest>".to_string());
-
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[("path", path), ("version", &version)],
- "started",
- None,
- );
-
- let res = self.inner.delete(path, args);
-
- match &res {
- Ok(_) => {
- self.queued += 1;
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[
- ("path", path),
- ("version", &version),
- ("queued", &self.queued.to_string()),
- ("deleted", &self.deleted.to_string()),
- ],
- "succeeded",
- None,
- );
- }
- Err(err) => {
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[
- ("path", path),
- ("version", &version),
- ("queued", &self.queued.to_string()),
- ("deleted", &self.deleted.to_string()),
- ],
- "failed",
- Some(err),
- );
- }
- };
-
- res
- }
-
- fn flush(&mut self) -> Result<usize> {
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[
- ("queued", &self.queued.to_string()),
- ("deleted", &self.deleted.to_string()),
- ],
- "started",
- None,
- );
-
- let res = self.inner.flush();
-
- match &res {
- Ok(flushed) => {
- self.queued -= flushed;
- self.deleted += flushed;
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[
- ("queued", &self.queued.to_string()),
- ("deleted", &self.deleted.to_string()),
- ],
- "succeeded",
- None,
- );
- }
- Err(err) => {
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[
- ("queued", &self.queued.to_string()),
- ("deleted", &self.deleted.to_string()),
- ],
- "failed",
- Some(err),
- );
- }
- };
-
- res
- }
-}
diff --git a/core/src/layers/mime_guess.rs b/core/src/layers/mime_guess.rs
index 245aecf55..10f5762d5 100644
--- a/core/src/layers/mime_guess.rs
+++ b/core/src/layers/mime_guess.rs
@@ -109,10 +109,6 @@ impl<A: Access> LayeredAccess for MimeGuessAccessor<A> {
type Writer = A::Writer;
type Lister = A::Lister;
type Deleter = A::Deleter;
- type BlockingReader = A::BlockingReader;
- type BlockingWriter = A::BlockingWriter;
- type BlockingLister = A::BlockingLister;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.0
@@ -142,29 +138,6 @@ impl<A: Access> LayeredAccess for MimeGuessAccessor<A> {
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner().list(path, args).await
}
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner().blocking_read(path, args)
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner()
- .blocking_write(path, opwrite_with_mime(path, args))
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.inner()
- .blocking_stat(path, args)
- .map(|rp| rpstat_with_mime(path, rp))
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner().blocking_delete()
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner().blocking_list(path, args)
- }
}
#[cfg(test)]
@@ -222,38 +195,4 @@ mod tests {
assert_eq!(entries[1].content_type(), None);
assert_eq!(entries[2].content_type(), Some(CUSTOM));
}
-
- #[test]
- fn test_blocking() {
- let op = Operator::new(Memory::default())
- .unwrap()
- .layer(MimeGuessLayer::default())
- .finish()
- .blocking();
-
- op.write("test0.html", DATA).unwrap();
- assert_eq!(op.stat("test0.html").unwrap().content_type(), Some(HTML));
-
- op.write("test1.asdfghjkl", DATA).unwrap();
- assert_eq!(op.stat("test1.asdfghjkl").unwrap().content_type(), None);
-
- op.write_with("test2.html", DATA)
- .content_type(CUSTOM)
- .call()
- .unwrap();
- assert_eq!(op.stat("test2.html").unwrap().content_type(), Some(CUSTOM));
-
- let entries: Vec<Metadata> = op
- .lister_with("")
- .call()
- .unwrap()
- .map(|entry| {
- let op = op.clone();
- op.stat(entry.unwrap().path()).unwrap()
- })
- .collect();
- assert_eq!(entries[0].content_type(), Some(HTML));
- assert_eq!(entries[1].content_type(), None);
- assert_eq!(entries[2].content_type(), Some(CUSTOM));
- }
}
diff --git a/core/src/layers/mod.rs b/core/src/layers/mod.rs
index 0c7657c57..a4adf9161 100644
--- a/core/src/layers/mod.rs
+++ b/core/src/layers/mod.rs
@@ -39,11 +39,6 @@ pub use logging::LoggingLayer;
mod timeout;
pub use timeout::TimeoutLayer;
-#[cfg(feature = "layers-blocking")]
-mod blocking;
-#[cfg(feature = "layers-blocking")]
-pub use blocking::BlockingLayer;
-
#[cfg(feature = "layers-chaos")]
mod chaos;
#[cfg(feature = "layers-chaos")]
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index 1dd15c0ef..4b41ebece 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -67,13 +67,9 @@ pub struct OtelTraceAccessor<A> {
impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
type Inner = A;
type Reader = OtelTraceWrapper<A::Reader>;
- type BlockingReader = OtelTraceWrapper<A::BlockingReader>;
type Writer = OtelTraceWrapper<A::Writer>;
- type BlockingWriter = OtelTraceWrapper<A::BlockingWriter>;
type Lister = OtelTraceWrapper<A::Lister>;
- type BlockingLister = OtelTraceWrapper<A::BlockingLister>;
type Deleter = A::Deleter;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -167,82 +163,6 @@ impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
let cx = TraceContext::current_with_span(span);
self.inner().presign(path, args).with_context(cx).await
}
-
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- let tracer = global::tracer("opendal");
- tracer.in_span("blocking_create_dir", |cx| {
- let span = cx.span(); // let mut span = cx.();
- span.set_attribute(KeyValue::new("path", path.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner().blocking_create_dir(path, args)
- })
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let tracer = global::tracer("opendal");
- let mut span = tracer.start("blocking_read");
- span.set_attribute(KeyValue::new("path", path.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, OtelTraceWrapper::new(span, r)))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let tracer = global::tracer("opendal");
- let mut span = tracer.start("blocking_write");
- span.set_attribute(KeyValue::new("path", path.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner
- .blocking_write(path, args)
- .map(|(rp, r)| (rp, OtelTraceWrapper::new(span, r)))
- }
-
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- let tracer = global::tracer("opendal");
- tracer.in_span("blocking_copy", |cx| {
- let span = cx.span();
- span.set_attribute(KeyValue::new("from", from.to_string()));
- span.set_attribute(KeyValue::new("to", to.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner().blocking_copy(from, to, args)
- })
- }
-
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- let tracer = global::tracer("opendal");
- tracer.in_span("blocking_rename", |cx| {
- let span = cx.span();
- span.set_attribute(KeyValue::new("from", from.to_string()));
- span.set_attribute(KeyValue::new("to", to.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner().blocking_rename(from, to, args)
- })
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let tracer = global::tracer("opendal");
- tracer.in_span("blocking_stat", |cx| {
- let span = cx.span();
- span.set_attribute(KeyValue::new("path", path.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner().blocking_stat(path, args)
- })
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner().blocking_delete()
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let tracer = global::tracer("opendal");
- let mut span = tracer.start("blocking_list");
- span.set_attribute(KeyValue::new("path", path.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner
- .blocking_list(path, args)
- .map(|(rp, it)| (rp, OtelTraceWrapper::new(span, it)))
- }
}
pub struct OtelTraceWrapper<R> {
@@ -262,12 +182,6 @@ impl<R: oio::Read> oio::Read for OtelTraceWrapper<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for OtelTraceWrapper<R> {
- fn read(&mut self) -> Result<Buffer> {
- self.inner.read()
- }
-}
-
impl<R: oio::Write> oio::Write for OtelTraceWrapper<R> {
fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner.write(bs)
@@ -282,24 +196,8 @@ impl<R: oio::Write> oio::Write for OtelTraceWrapper<R> {
}
}
-impl<R: oio::BlockingWrite> oio::BlockingWrite for OtelTraceWrapper<R> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- self.inner.write(bs)
- }
-
- fn close(&mut self) -> Result<Metadata> {
- self.inner.close()
- }
-}
-
impl<R: oio::List> oio::List for OtelTraceWrapper<R> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
self.inner.next().await
}
}
-
-impl<R: oio::BlockingList> oio::BlockingList for OtelTraceWrapper<R> {
- fn next(&mut self) -> Result<Option<oio::Entry>> {
- self.inner.next()
- }
-}
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index b0ce9742e..3781e0157 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -293,13 +293,9 @@ impl<A: Access, I: RetryInterceptor> Debug for RetryAccessor<A, I> {
impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
type Inner = A;
type Reader = RetryWrapper<RetryReader<A, A::Reader>, I>;
- type BlockingReader = RetryWrapper<RetryReader<A, A::BlockingReader>, I>;
type Writer = RetryWrapper<A::Writer, I>;
- type BlockingWriter = RetryWrapper<A::BlockingWriter, I>;
type Lister = RetryWrapper<A::Lister, I>;
- type BlockingLister = RetryWrapper<A::BlockingLister, I>;
type Deleter = RetryWrapper<A::Deleter, I>;
- type BlockingDeleter = RetryWrapper<A::BlockingDeleter, I>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -384,89 +380,6 @@ impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
.map(|(rp, r)| (rp, RetryWrapper::new(r, self.notify.clone(), self.builder)))
.map_err(|e| e.set_persistent())
}
-
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- { || self.inner.blocking_create_dir(path, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map_err(|e| e.set_persistent())
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let (rp, reader) = { || self.inner.blocking_read(path, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map_err(|e| e.set_persistent())?;
-
- let retry_reader = RetryReader::new(self.inner.clone(), path.to_string(), args, reader);
- let retry_wrapper = RetryWrapper::new(retry_reader, self.notify.clone(), self.builder);
-
- Ok((rp, retry_wrapper))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- { || self.inner.blocking_write(path, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map(|(rp, r)| (rp, RetryWrapper::new(r, self.notify.clone(), self.builder)))
- .map_err(|e| e.set_persistent())
- }
-
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- { || self.inner.blocking_stat(path, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map_err(|e| e.set_persistent())
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- { || self.inner.blocking_delete() }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map(|(rp, r)| (rp, RetryWrapper::new(r, self.notify.clone(), self.builder)))
- .map_err(|e| e.set_persistent())
- }
-
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- { || self.inner.blocking_copy(from, to, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map_err(|e| e.set_persistent())
- }
-
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- { || self.inner.blocking_rename(from, to, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map_err(|e| e.set_persistent())
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- { || self.inner.blocking_list(path, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call()
- .map(|(rp, p)| {
- let p = RetryWrapper::new(p, self.notify.clone(), self.builder);
- (rp, p)
- })
- .map_err(|e| e.set_persistent())
- }
}
pub struct RetryReader<A, R> {
@@ -509,26 +422,6 @@ impl<A: Access> oio::Read for RetryReader<A, A::Reader> {
}
}
-impl<A: Access> oio::BlockingRead for RetryReader<A, A::BlockingReader> {
- fn read(&mut self) -> Result<Buffer> {
- loop {
- match self.reader.take() {
- None => {
- let (_, r) = self.inner.blocking_read(&self.path, self.args.clone())?;
- self.reader = Some(r);
- continue;
- }
- Some(mut reader) => {
- let buf = reader.read()?;
- self.reader = Some(reader);
- self.args.range_mut().advance(buf.len() as u64);
- return Ok(buf);
- }
- }
- }
- }
-}
-
pub struct RetryWrapper<R, I> {
inner: Option<R>,
notify: Arc<I>,
@@ -579,30 +472,6 @@ impl<R: oio::Read, I: RetryInterceptor> oio::Read for RetryWrapper<R, I> {
}
}
-impl<R: oio::BlockingRead, I: RetryInterceptor> oio::BlockingRead for RetryWrapper<R, I> {
- fn read(&mut self) -> Result<Buffer> {
- use backon::BlockingRetryableWithContext;
-
- let inner = self.take_inner()?;
-
- let (inner, res) = {
- |mut r: R| {
- let res = r.read();
-
- (r, res)
- }
- }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .context(inner)
- .notify(|err, dur| self.notify.intercept(err, dur))
- .call();
-
- self.inner = Some(inner);
- res.map_err(|err| err.set_persistent())
- }
-}
-
impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
use backon::RetryableWithContext;
@@ -671,30 +540,6 @@ impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
}
}
-impl<R: oio::BlockingWrite, I: RetryInterceptor> oio::BlockingWrite for RetryWrapper<R, I> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- { || self.inner.as_mut().unwrap().write(bs.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| {
- self.notify.intercept(err, dur);
- })
- .call()
- .map_err(|e| e.set_persistent())
- }
-
- fn close(&mut self) -> Result<Metadata> {
- { || self.inner.as_mut().unwrap().close() }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| {
- self.notify.intercept(err, dur);
- })
- .call()
- .map_err(|e| e.set_persistent())
- }
-}
-
impl<P: oio::List, I: RetryInterceptor> oio::List for RetryWrapper<P, I> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
use backon::RetryableWithContext;
@@ -767,30 +612,6 @@ impl<P: oio::Delete, I: RetryInterceptor> oio::Delete for RetryWrapper<P, I> {
}
}
-impl<P: oio::BlockingDelete, I: RetryInterceptor> oio::BlockingDelete for RetryWrapper<P, I> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- { || self.inner.as_mut().unwrap().delete(path, args.clone()) }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| {
- self.notify.intercept(err, dur);
- })
- .call()
- .map_err(|e| e.set_persistent())
- }
-
- fn flush(&mut self) -> Result<usize> {
- { || self.inner.as_mut().unwrap().flush() }
- .retry(self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| {
- self.notify.intercept(err, dur);
- })
- .call()
- .map_err(|e| e.set_persistent())
- }
-}
-
#[cfg(test)]
mod tests {
use std::mem;
@@ -830,10 +651,6 @@ mod tests {
type Writer = MockWriter;
type Lister = MockLister;
type Deleter = MockDeleter;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/src/layers/throttle.rs b/core/src/layers/throttle.rs
index 4a5e674e1..70eb588c6 100644
--- a/core/src/layers/throttle.rs
+++ b/core/src/layers/throttle.rs
@@ -117,10 +117,6 @@ impl<A: Access> LayeredAccess for ThrottleAccessor<A> {
type Writer = ThrottleWrapper<A::Writer>;
type Lister = A::Lister;
type Deleter = A::Deleter;
- type BlockingReader = ThrottleWrapper<A::BlockingReader>;
- type BlockingWriter = ThrottleWrapper<A::BlockingWriter>;
- type BlockingLister = A::BlockingLister;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -151,30 +147,6 @@ impl<A: Access> LayeredAccess for ThrottleAccessor<A> {
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner.list(path, args).await
}
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let limiter = self.rate_limiter.clone();
-
- self.inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, ThrottleWrapper::new(r, limiter)))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let limiter = self.rate_limiter.clone();
-
- self.inner
- .blocking_write(path, args)
- .map(|(rp, w)| (rp, ThrottleWrapper::new(w, limiter)))
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete()
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner.blocking_list(path, args)
- }
}
pub struct ThrottleWrapper<R> {
@@ -197,12 +169,6 @@ impl<R: oio::Read> oio::Read for ThrottleWrapper<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for ThrottleWrapper<R> {
- fn read(&mut self) -> Result<Buffer> {
- self.inner.read()
- }
-}
-
impl<R: oio::Write> oio::Write for ThrottleWrapper<R> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
let buf_length = NonZeroU32::new(bs.len() as u32).unwrap();
@@ -237,31 +203,3 @@ impl<R: oio::Write> oio::Write for ThrottleWrapper<R> {
self.inner.close().await
}
}
-
-impl<R: oio::BlockingWrite> oio::BlockingWrite for ThrottleWrapper<R> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- let buf_length = NonZeroU32::new(bs.len() as u32).unwrap();
-
- loop {
- match self.limiter.check_n(buf_length) {
- Ok(res) => match res {
- Ok(_) => return self.inner.write(bs),
- // the query is valid but the Decider can not accommodate them.
- Err(not_until) => {
- let wait_time = not_until.wait_time_from(DefaultClock::default().now());
- thread::sleep(wait_time);
- }
- },
- // the query was invalid as the rate limit parameters can "never" accommodate the number of cells queried for.
- Err(_) => return Err(Error::new(
- ErrorKind::RateLimited,
- "InsufficientCapacity due to burst size being smaller than the request size",
- )),
- }
- }
- }
-
- fn close(&mut self) -> Result<Metadata> {
- self.inner.close()
- }
-}
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index b89e13912..932794a94 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -220,13 +220,9 @@ impl<A: Access> TimeoutAccessor<A> {
impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
type Inner = A;
type Reader = TimeoutWrapper<A::Reader>;
- type BlockingReader = A::BlockingReader;
type Writer = TimeoutWrapper<A::Writer>;
- type BlockingWriter = A::BlockingWriter;
type Lister = TimeoutWrapper<A::Lister>;
- type BlockingLister = A::BlockingLister;
type Deleter = TimeoutWrapper<A::Deleter>;
- type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -280,22 +276,6 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
self.timeout(Operation::Presign, self.inner.presign(path, args))
.await
}
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner.blocking_read(path, args)
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner.blocking_write(path, args)
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner.blocking_list(path, args)
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner.blocking_delete()
- }
}
pub struct TimeoutExecutor {
@@ -410,11 +390,7 @@ mod tests {
type Reader = MockReader;
type Writer = ();
type Lister = MockLister;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
type Deleter = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index 5084a09dc..a432a1ab4 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -203,10 +203,6 @@ impl<A: Access> LayeredAccess for TracingAccessor<A> {
type Writer = TracingWrapper<A::Writer>;
type Lister = TracingWrapper<A::Lister>;
type Deleter = TracingWrapper<A::Deleter>;
- type BlockingReader = TracingWrapper<A::BlockingReader>;
- type BlockingWriter = TracingWrapper<A::BlockingWriter>;
- type BlockingLister = TracingWrapper<A::BlockingLister>;
- type BlockingDeleter = TracingWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -280,70 +276,6 @@ impl<A: Access> LayeredAccess for TracingAccessor<A> {
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
self.inner.presign(path, args).await
}
-
- #[tracing::instrument(level = "debug", skip(self))]
- fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- self.inner.blocking_create_dir(path, args)
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let span = span!(Level::DEBUG, "read", path, ?args);
-
- let (rp, r) = {
- let _enter = span.enter();
- self.inner.blocking_read(path, args)?
- };
-
- Ok((rp, TracingWrapper::new(span, r)))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let span = span!(Level::DEBUG, "write", path, ?args);
-
- let (rp, r) = {
- let _enter = span.enter();
- self.inner.blocking_write(path, args)?
- };
-
- Ok((rp, TracingWrapper::new(span, r)))
- }
-
- #[tracing::instrument(level = "debug", skip(self))]
- fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- self.inner().blocking_copy(from, to, args)
- }
-
- #[tracing::instrument(level = "debug", skip(self))]
- fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- self.inner().blocking_rename(from, to, args)
- }
-
- #[tracing::instrument(level = "debug", skip(self))]
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.inner.blocking_stat(path, args)
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- let span = span!(Level::DEBUG, "delete");
-
- let (rp, r) = {
- let _enter = span.enter();
- self.inner.blocking_delete()?
- };
-
- Ok((rp, TracingWrapper::new(span, r)))
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let span = span!(Level::DEBUG, "list", path, ?args);
-
- let (rp, r) = {
- let _enter = span.enter();
- self.inner.blocking_list(path, args)?
- };
-
- Ok((rp, TracingWrapper::new(span, r)))
- }
}
pub struct TracingWrapper<R> {
@@ -365,14 +297,6 @@ impl<R: oio::Read> oio::Read for TracingWrapper<R> {
}
}
-impl<R: oio::BlockingRead> oio::BlockingRead for TracingWrapper<R> {
- fn read(&mut self) -> Result<Buffer> {
- let _enter = self.span.enter();
-
- self.inner.read()
- }
-}
-
impl<R: oio::Write> oio::Write for TracingWrapper<R> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
let _enter = self.span.enter();
@@ -393,20 +317,6 @@ impl<R: oio::Write> oio::Write for TracingWrapper<R> {
}
}
-impl<R: oio::BlockingWrite> oio::BlockingWrite for TracingWrapper<R> {
- fn write(&mut self, bs: Buffer) -> Result<()> {
- let _enter = self.span.enter();
-
- self.inner.write(bs)
- }
-
- fn close(&mut self) -> Result<Metadata> {
- let _enter = self.span.enter();
-
- self.inner.close()
- }
-}
-
impl<R: oio::List> oio::List for TracingWrapper<R> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
let _enter = self.span.enter();
@@ -415,14 +325,6 @@ impl<R: oio::List> oio::List for TracingWrapper<R> {
}
}
-impl<R: oio::BlockingList> oio::BlockingList for TracingWrapper<R> {
- fn next(&mut self) -> Result<Option<oio::Entry>> {
- let _enter = self.span.enter();
-
- self.inner.next()
- }
-}
-
impl<R: oio::Delete> oio::Delete for TracingWrapper<R> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
let _enter = self.span.enter();
@@ -436,17 +338,3 @@ impl<R: oio::Delete> oio::Delete for TracingWrapper<R> {
self.inner.flush().await
}
}
-
-impl<R: oio::BlockingDelete> oio::BlockingDelete for TracingWrapper<R> {
- fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- let _enter = self.span.enter();
-
- self.inner.delete(path, args)
- }
-
- fn flush(&mut self) -> Result<usize> {
- let _enter = self.span.enter();
-
- self.inner.flush()
- }
-}
diff --git a/core/src/layers/type_eraser.rs b/core/src/layers/type_eraser.rs
index cfe161aae..4fe7547ff 100644
--- a/core/src/layers/type_eraser.rs
+++ b/core/src/layers/type_eraser.rs
@@ -55,10 +55,6 @@ impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
type Writer = oio::Writer;
type Lister = oio::Lister;
type Deleter = oio::Deleter;
- type BlockingReader = oio::BlockingReader;
- type BlockingWriter = oio::BlockingWriter;
- type BlockingLister = oio::BlockingLister;
- type BlockingDeleter = oio::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -91,28 +87,4 @@ impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
.await
.map(|(rp, p)| (rp, Box::new(p) as oio::Lister))
}
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner
- .blocking_read(path, args)
- .map(|(rp, r)| (rp, Box::new(r) as oio::BlockingReader))
- }
-
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner
- .blocking_write(path, args)
- .map(|(rp, w)| (rp, Box::new(w) as oio::BlockingWriter))
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- self.inner
- .blocking_delete()
- .map(|(rp, p)| (rp, Box::new(p) as oio::BlockingDeleter))
- }
-
- fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- self.inner
- .blocking_list(path, args)
- .map(|(rp, p)| (rp, Box::new(p) as oio::BlockingLister))
- }
}
diff --git a/core/src/raw/oio/list/mod.rs b/core/src/raw/oio/list/mod.rs
index 60b8de0d9..c9bb97977 100644
--- a/core/src/raw/oio/list/mod.rs
+++ b/core/src/raw/oio/list/mod.rs
@@ -16,8 +16,6 @@
// under the License.
mod api;
-pub use api::BlockingList;
-pub use api::BlockingLister;
pub use api::List;
pub use api::Lister;
diff --git a/core/src/raw/oio/write/mod.rs b/core/src/raw/oio/write/mod.rs
index 9ce7d9427..d05c5ef65 100644
--- a/core/src/raw/oio/write/mod.rs
+++ b/core/src/raw/oio/write/mod.rs
@@ -16,8 +16,6 @@
// under the License.
mod api;
-pub use api::BlockingWrite;
-pub use api::BlockingWriter;
pub use api::Write;
pub use api::Writer;
diff --git a/core/src/raw/tests/utils.rs b/core/src/raw/tests/utils.rs
index 76eeae356..8f9fdb742 100644
--- a/core/src/raw/tests/utils.rs
+++ b/core/src/raw/tests/utils.rs
@@ -77,14 +77,5 @@ pub fn init_test_service() -> Result<Option<Operator>> {
.layer(layers::TimeoutLayer::new())
.layer(layers::RetryLayer::new().with_max_times(4));
- // Enable blocking layer if needed.
- if !op.info().full_capability().blocking {
- // Don't enable blocking layer for compfs
- if op.info().scheme() != Scheme::Compfs {
- let _guard = TEST_RUNTIME.enter();
- op = op.layer(layers::BlockingLayer::create().expect("blocking layer must be created"));
- }
- }
-
Ok(Some(op))
}
diff --git a/core/src/services/aliyun_drive/backend.rs b/core/src/services/aliyun_drive/backend.rs
index bcea12043..09159749b 100644
--- a/core/src/services/aliyun_drive/backend.rs
+++ b/core/src/services/aliyun_drive/backend.rs
@@ -238,10 +238,6 @@ impl Access for AliyunDriveBackend {
type Writer = AliyunDriveWriter;
type Lister = oio::PageLister<AliyunDriveLister>;
type Deleter = oio::OneShotDeleter<AliyunDriveDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/alluxio/backend.rs b/core/src/services/alluxio/backend.rs
index e43d204bf..70970fd8f 100644
--- a/core/src/services/alluxio/backend.rs
+++ b/core/src/services/alluxio/backend.rs
@@ -179,10 +179,6 @@ impl Access for AlluxioBackend {
type Writer = AlluxioWriters;
type Lister = oio::PageLister<AlluxioLister>;
type Deleter = oio::OneShotDeleter<AlluxioDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index 331d9b954..3ea769ab0 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -550,10 +550,6 @@ impl Access for AzblobBackend {
type Writer = AzblobWriters;
type Lister = oio::PageLister<AzblobLister>;
type Deleter = oio::BatchDeleter<AzblobDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/azdls/backend.rs b/core/src/services/azdls/backend.rs
index ed3b8ef38..beaf19684 100644
--- a/core/src/services/azdls/backend.rs
+++ b/core/src/services/azdls/backend.rs
@@ -262,10 +262,6 @@ impl Access for AzdlsBackend {
type Writer = AzdlsWriters;
type Lister = oio::PageLister<AzdlsLister>;
type Deleter = oio::OneShotDeleter<AzdlsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/azfile/backend.rs b/core/src/services/azfile/backend.rs
index 84820fad5..3ba22da17 100644
--- a/core/src/services/azfile/backend.rs
+++ b/core/src/services/azfile/backend.rs
@@ -272,10 +272,6 @@ impl Access for AzfileBackend {
type Writer = AzfileWriters;
type Lister = oio::PageLister<AzfileLister>;
type Deleter = oio::OneShotDeleter<AzfileDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/b2/backend.rs b/core/src/services/b2/backend.rs
index 7c6297031..653847837 100644
--- a/core/src/services/b2/backend.rs
+++ b/core/src/services/b2/backend.rs
@@ -269,10 +269,6 @@ impl Access for B2Backend {
type Writer = B2Writers;
type Lister = oio::PageLister<B2Lister>;
type Deleter = oio::OneShotDeleter<B2Deleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/compfs/backend.rs b/core/src/services/compfs/backend.rs
index 75fcf0a4b..07eec49e9 100644
--- a/core/src/services/compfs/backend.rs
+++ b/core/src/services/compfs/backend.rs
@@ -137,10 +137,6 @@ impl Access for CompfsBackend {
type Writer = CompfsWriter;
type Lister = Option<CompfsLister>;
type Deleter = OneShotDeleter<CompfsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/cos/backend.rs b/core/src/services/cos/backend.rs
index 854a8a33a..b38f8f1b6 100644
--- a/core/src/services/cos/backend.rs
+++ b/core/src/services/cos/backend.rs
@@ -318,10 +318,6 @@ impl Access for CosBackend {
type Writer = CosWriters;
type Lister = CosListers;
type Deleter = oio::OneShotDeleter<CosDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/dbfs/backend.rs b/core/src/services/dbfs/backend.rs
index 9ec6257ac..c43c1a036 100644
--- a/core/src/services/dbfs/backend.rs
+++ b/core/src/services/dbfs/backend.rs
@@ -147,10 +147,6 @@ impl Access for DbfsBackend {
type Writer = oio::OneShotWriter<DbfsWriter>;
type Lister = oio::PageLister<DbfsLister>;
type Deleter = oio::OneShotDeleter<DbfsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/src/services/dropbox/backend.rs b/core/src/services/dropbox/backend.rs
index 645157346..ccd42de88 100644
--- a/core/src/services/dropbox/backend.rs
+++ b/core/src/services/dropbox/backend.rs
@@ -40,10 +40,6 @@ impl Access for DropboxBackend {
type Writer = oio::OneShotWriter<DropboxWriter>;
type Lister = oio::PageLister<DropboxLister>;
type Deleter = oio::OneShotDeleter<DropboxDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/fs/backend.rs b/core/src/services/fs/backend.rs
index ad7f9e6bc..e6eac202b 100644
--- a/core/src/services/fs/backend.rs
+++ b/core/src/services/fs/backend.rs
@@ -172,7 +172,6 @@ impl Builder for FsBuilder {
copy: true,
rename: true,
- blocking: true,
shared: true,
@@ -200,10 +199,6 @@ impl Access for FsBackend {
type Writer = FsWriters;
type Lister = Option<FsLister<tokio::fs::ReadDir>>;
type Deleter = oio::OneShotDeleter<FsDeleter>;
- type BlockingReader = FsReader<std::fs::File>;
- type BlockingWriter = FsWriter<std::fs::File>;
- type BlockingLister = Option<FsLister<std::fs::ReadDir>>;
- type BlockingDeleter = oio::OneShotDeleter<FsDeleter>;
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -418,184 +413,6 @@ impl Access for FsBackend {
Ok(RpRename::default())
}
-
- fn blocking_create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- std::fs::create_dir_all(p).map_err(new_std_io_error)?;
-
- Ok(RpCreateDir::default())
- }
-
- fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let meta = std::fs::metadata(p).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))
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let mut f = std::fs::OpenOptions::new()
- .read(true)
- .open(p)
- .map_err(new_std_io_error)?;
-
- if args.range().offset() != 0 {
- use std::io::Seek;
-
- f.seek(SeekFrom::Start(args.range().offset()))
- .map_err(new_std_io_error)?;
- }
-
- let r = FsReader::new(
- self.core.clone(),
- f,
- args.range().size().unwrap_or(u64::MAX) as _,
- );
- Ok((RpRead::new(), r))
- }
-
- fn blocking_write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let (target_path, tmp_path) = if let Some(atomic_write_dir) = &self.core.atomic_write_dir {
- let target_path = self
- .core
- .blocking_ensure_write_abs_path(&self.core.root, path)?;
- let tmp_path = self
- .core
- .blocking_ensure_write_abs_path(atomic_write_dir, &tmp_file_of(path))?;
-
- // If the target file exists, we should append to the end of it directly.
- if op.append()
- && Path::new(&target_path)
- .try_exists()
- .map_err(new_std_io_error)?
- {
- (target_path, None)
- } else {
- (target_path, Some(tmp_path))
- }
- } else {
- let p = self
- .core
- .blocking_ensure_write_abs_path(&self.core.root, path)?;
-
- (p, None)
- };
-
- let mut f = std::fs::OpenOptions::new();
-
- if op.if_not_exists() {
- f.create_new(true);
- } else {
- f.create(true);
- }
-
- f.write(true);
-
- if op.append() {
- f.append(true);
- } else {
- f.truncate(true);
- }
-
- let f = f
- .open(tmp_path.as_ref().unwrap_or(&target_path))
- .map_err(|e| {
- match e.kind() {
- std::io::ErrorKind::AlreadyExists => {
- // Map io AlreadyExists to opendal ConditionNotMatch
- Error::new(
- ErrorKind::ConditionNotMatch,
- "The file already exists in the filesystem",
- )
- .set_source(e)
- }
- _ => new_std_io_error(e),
- }
- })?;
-
- Ok((RpWrite::new(), FsWriter::new(target_path, tmp_path, f)))
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- Ok((
- RpDelete::default(),
- oio::OneShotDeleter::new(FsDeleter::new(self.core.clone())),
- ))
- }
-
- fn blocking_list(&self, path: &str, _: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let f = match std::fs::read_dir(p) {
- Ok(rd) => rd,
- Err(e) => {
- return match e.kind() {
- // Return empty list if the directory not found
- std::io::ErrorKind::NotFound => Ok((RpList::default(), None)),
- // If the path is not a directory, return an empty list
- //
- // The path could be a file or a symbolic link in this case.
- // Returning a NotADirectory error to the user isn't helpful; instead,
- // providing an empty directory is a more user-friendly. In fact, the dir
- // `path/` does not exist.
- std::io::ErrorKind::NotADirectory => Ok((RpList::default(), None)),
- _ => Err(new_std_io_error(e)),
- };
- }
- };
-
- let rd = FsLister::new(&self.core.root, path, f);
- Ok((RpList::default(), Some(rd)))
- }
-
- fn blocking_copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let from = self.core.root.join(from.trim_end_matches('/'));
-
- // try to get the metadata of the source file to ensure it exists
- std::fs::metadata(&from).map_err(new_std_io_error)?;
-
- let to = self
- .core
- .blocking_ensure_write_abs_path(&self.core.root, to.trim_end_matches('/'))?;
-
- std::fs::copy(from, to).map_err(new_std_io_error)?;
-
- Ok(RpCopy::default())
- }
-
- fn blocking_rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- let from = self.core.root.join(from.trim_end_matches('/'));
-
- // try to get the metadata of the source file to ensure it exists
- std::fs::metadata(&from).map_err(new_std_io_error)?;
-
- let to = self
- .core
- .blocking_ensure_write_abs_path(&self.core.root, to.trim_end_matches('/'))?;
-
- std::fs::rename(from, to).map_err(new_std_io_error)?;
-
- Ok(RpRename::default())
- }
}
#[cfg(test)]
diff --git a/core/src/services/fs/delete.rs b/core/src/services/fs/delete.rs
index 81f39d9bf..2a541c9f8 100644
--- a/core/src/services/fs/delete.rs
+++ b/core/src/services/fs/delete.rs
@@ -51,25 +51,3 @@ impl oio::OneShotDelete for FsDeleter {
}
}
}
-
-impl oio::BlockingOneShotDelete for FsDeleter {
- fn blocking_delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let meta = std::fs::metadata(&p);
-
- match meta {
- Ok(meta) => {
- if meta.is_dir() {
- std::fs::remove_dir(&p).map_err(new_std_io_error)?;
- } else {
- std::fs::remove_file(&p).map_err(new_std_io_error)?;
- }
-
- Ok(())
- }
- Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
- Err(err) => Err(new_std_io_error(err)),
- }
- }
-}
diff --git a/core/src/services/fs/reader.rs b/core/src/services/fs/reader.rs
index 1629e670d..7ed69e2ae 100644
--- a/core/src/services/fs/reader.rs
+++ b/core/src/services/fs/reader.rs
@@ -82,40 +82,3 @@ impl oio::Read for FsReader<tokio::fs::File> {
Ok(Buffer::from(frozen))
}
}
-
-impl oio::BlockingRead for FsReader<std::fs::File> {
- fn read(&mut self) -> Result<Buffer> {
- if self.read >= self.size {
- return Ok(Buffer::new());
- }
-
- let mut bs = self.core.buf_pool.get();
- bs.reserve(self.buf_size);
-
- let size = (self.size - self.read).min(self.buf_size);
- let buf = &mut bs.spare_capacity_mut()[..size];
- let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
-
- // SAFETY: Read at most `limit` bytes into `read_buf`.
- unsafe {
- read_buf.assume_init(size);
- }
-
- let n = self
- .f
- .read(read_buf.initialize_unfilled())
- .map_err(new_std_io_error)?;
- read_buf.advance(n);
- self.read += n;
-
- // Safety: We make sure that bs contains `n` more bytes.
- let filled = read_buf.filled().len();
- unsafe { bs.set_len(filled) }
-
- let frozen = bs.split().freeze();
- // Return the buffer to the pool.
- self.core.buf_pool.put(bs);
-
- Ok(Buffer::from(frozen))
- }
-}
diff --git a/core/src/services/ftp/backend.rs b/core/src/services/ftp/backend.rs
index e98ddfff3..558eab27a 100644
--- a/core/src/services/ftp/backend.rs
+++ b/core/src/services/ftp/backend.rs
@@ -221,10 +221,6 @@ impl Access for FtpBackend {
type Writer = FtpWriter;
type Lister = FtpLister;
type Deleter = oio::OneShotDeleter<FtpDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/gcs/backend.rs b/core/src/services/gcs/backend.rs
index c48f2d227..18fd9ff58 100644
--- a/core/src/services/gcs/backend.rs
+++ b/core/src/services/gcs/backend.rs
@@ -412,10 +412,6 @@ impl Access for GcsBackend {
type Writer = GcsWriters;
type Lister = oio::PageLister<GcsLister>;
type Deleter = oio::BatchDeleter<GcsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/gdrive/backend.rs b/core/src/services/gdrive/backend.rs
index 88f164bdd..777e06887 100644
--- a/core/src/services/gdrive/backend.rs
+++ b/core/src/services/gdrive/backend.rs
@@ -42,10 +42,6 @@ impl Access for GdriveBackend {
type Writer = oio::OneShotWriter<GdriveWriter>;
type Lister = oio::PageLister<GdriveLister>;
type Deleter = oio::OneShotDeleter<GdriveDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/ghac/backend.rs b/core/src/services/ghac/backend.rs
index 564b42c0d..fca6fbe2c 100644
--- a/core/src/services/ghac/backend.rs
+++ b/core/src/services/ghac/backend.rs
@@ -239,10 +239,6 @@ impl Access for GhacBackend {
type Writer = GhacWriter;
type Lister = ();
type Deleter = ();
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/github/backend.rs b/core/src/services/github/backend.rs
index e361bed40..e3e01b138 100644
--- a/core/src/services/github/backend.rs
+++ b/core/src/services/github/backend.rs
@@ -205,10 +205,6 @@ impl Access for GithubBackend {
type Writer = GithubWriters;
type Lister = oio::PageLister<GithubLister>;
type Deleter = oio::OneShotDeleter<GithubDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/hdfs/backend.rs b/core/src/services/hdfs/backend.rs
index 26a8760b6..d4adb46d4 100644
--- a/core/src/services/hdfs/backend.rs
+++ b/core/src/services/hdfs/backend.rs
@@ -196,7 +196,6 @@ impl Builder for HdfsBuilder {
list_has_last_modified: true,
rename: true,
- blocking: true,
shared: true,
@@ -238,10 +237,6 @@ impl Access for HdfsBackend {
type Writer = HdfsWriter<hdrs::AsyncFile>;
type Lister = Option<HdfsLister>;
type Deleter = oio::OneShotDeleter<HdfsDeleter>;
- type BlockingReader = HdfsReader<hdrs::File>;
- type BlockingWriter = HdfsWriter<hdrs::File>;
- type BlockingLister = Option<HdfsLister>;
- type BlockingDeleter = oio::OneShotDeleter<HdfsDeleter>;
fn info(&self) -> Arc<AccessorInfo> {
self.info.clone()
@@ -431,186 +426,4 @@ impl Access for HdfsBackend {
Ok(RpRename::new())
}
-
- fn blocking_create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
- let p = build_rooted_abs_path(&self.root, path);
-
- self.client.create_dir(&p).map_err(new_std_io_error)?;
-
- Ok(RpCreateDir::default())
- }
-
- fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let meta = self.client.metadata(&p).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 mut m = Metadata::new(mode);
- m.set_content_length(meta.len());
- m.set_last_modified(meta.modified().into());
-
- Ok(RpStat::new(m))
- }
-
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let mut f = self
- .client
- .open_file()
- .read(true)
- .open(&p)
- .map_err(new_std_io_error)?;
-
- if args.range().offset() != 0 {
- use std::io::Seek;
-
- f.seek(SeekFrom::Start(args.range().offset()))
- .map_err(new_std_io_error)?;
- }
-
- Ok((
- RpRead::new(),
- HdfsReader::new(f, args.range().size().unwrap_or(u64::MAX) as _),
- ))
- }
-
- fn blocking_write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let target_path = build_rooted_abs_path(&self.root, path);
- let mut initial_size = 0;
- let target_exists = match self.client.metadata(&target_path) {
- Ok(meta) => {
- initial_size = meta.len();
- true
- }
- Err(err) => {
- if err.kind() != io::ErrorKind::NotFound {
- return Err(new_std_io_error(err));
- }
- false
- }
- };
-
- let should_append = op.append() && target_exists;
- let tmp_path = self.atomic_write_dir.as_ref().and_then(|atomic_write_dir| {
- // If the target file exists, we should append to the end of it directly.
- if should_append {
- None
- } else {
- Some(build_rooted_abs_path(atomic_write_dir, &tmp_file_of(path)))
- }
- });
-
- if !target_exists {
- let parent = get_parent(&target_path);
- self.client.create_dir(parent).map_err(new_std_io_error)?;
- }
- if !should_append {
- initial_size = 0;
- }
-
- let mut open_options = self.client.open_file();
- open_options.create(true);
- if should_append {
- open_options.append(true);
- } else {
- open_options.write(true);
- }
-
- let f = open_options
- .open(tmp_path.as_ref().unwrap_or(&target_path))
- .map_err(new_std_io_error)?;
-
- Ok((
- RpWrite::new(),
- HdfsWriter::new(
- target_path,
- tmp_path,
- f,
- Arc::clone(&self.client),
- target_exists,
- initial_size,
- ),
- ))
- }
-
- fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- Ok((
- RpDelete::default(),
- oio::OneShotDeleter::new(HdfsDeleter::new(Arc::new(self.clone()))),
- ))
- }
-
- fn blocking_list(&self, path: &str, _: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let p = build_rooted_abs_path(&self.root, path);
-
- let f = match self.client.read_dir(&p) {
- Ok(f) => f,
- Err(e) => {
- return if e.kind() == io::ErrorKind::NotFound {
- Ok((RpList::default(), None))
- } else {
- Err(new_std_io_error(e))
- }
- }
- };
-
- let rd = HdfsLister::new(&self.root, f, path);
-
- Ok((RpList::default(), Some(rd)))
- }
-
- fn blocking_rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- let from_path = build_rooted_abs_path(&self.root, from);
- self.client.metadata(&from_path).map_err(new_std_io_error)?;
-
- let to_path = build_rooted_abs_path(&self.root, to);
- let result = self.client.metadata(&to_path);
- match result {
- Err(err) => {
- // Early return if other error happened.
- if err.kind() != io::ErrorKind::NotFound {
- return Err(new_std_io_error(err));
- }
-
- let parent = PathBuf::from(&to_path)
- .parent()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "path should have parent but not, it must be malformed",
- )
- .with_context("input", &to_path)
- })?
- .to_path_buf();
-
- self.client
- .create_dir(&parent.to_string_lossy())
- .map_err(new_std_io_error)?;
- }
- Ok(metadata) => {
- if metadata.is_file() {
- self.client
- .remove_file(&to_path)
- .map_err(new_std_io_error)?;
- } else {
- return Err(Error::new(ErrorKind::IsADirectory, "path should be a file")
- .with_context("input", &to_path));
- }
- }
- }
-
- self.client
- .rename_file(&from_path, &to_path)
- .map_err(new_std_io_error)?;
-
- Ok(RpRename::new())
- }
}
diff --git a/core/src/services/hdfs/delete.rs b/core/src/services/hdfs/delete.rs
index 007c5ad69..ae129d268 100644
--- a/core/src/services/hdfs/delete.rs
+++ b/core/src/services/hdfs/delete.rs
@@ -59,32 +59,3 @@ impl oio::OneShotDelete for HdfsDeleter {
Ok(())
}
}
-
-impl oio::BlockingOneShotDelete for HdfsDeleter {
- fn blocking_delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let p = build_rooted_abs_path(&self.core.root, &path);
-
- let meta = self.core.client.metadata(&p);
-
- if let Err(err) = meta {
- return if err.kind() == io::ErrorKind::NotFound {
- Ok(())
- } else {
- Err(new_std_io_error(err))
- };
- }
-
- // Safety: Err branch has been checked, it's OK to unwrap.
- let meta = meta.ok().unwrap();
-
- let result = if meta.is_dir() {
- self.core.client.remove_dir(&p)
- } else {
- self.core.client.remove_file(&p)
- };
-
- result.map_err(new_std_io_error)?;
-
- Ok(())
- }
-}
diff --git a/core/src/services/hdfs/reader.rs b/core/src/services/hdfs/reader.rs
index 817e13163..e8c8593f5 100644
--- a/core/src/services/hdfs/reader.rs
+++ b/core/src/services/hdfs/reader.rs
@@ -80,36 +80,3 @@ impl oio::Read for HdfsReader<AsyncFile> {
Ok(Buffer::from(frozen))
}
}
-
-impl oio::BlockingRead for HdfsReader<File> {
- fn read(&mut self) -> Result<Buffer> {
- if self.read >= self.size {
- return Ok(Buffer::new());
- }
-
- let size = (self.size - self.read).min(self.buf_size);
- self.buf.reserve(size);
-
- let buf = &mut self.buf.spare_capacity_mut()[..size];
- let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
-
- // SAFETY: Read at most `limit` bytes into `read_buf`.
- unsafe {
- read_buf.assume_init(size);
- }
-
- let n = self
- .f
- .read(read_buf.initialize_unfilled())
- .map_err(new_std_io_error)?;
- read_buf.advance(n);
- self.read += n;
-
- // Safety: We make sure that bs contains `n` more bytes.
- let filled = read_buf.filled().len();
- unsafe { self.buf.set_len(filled) }
-
- let frozen = self.buf.split().freeze();
- Ok(Buffer::from(frozen))
- }
-}
diff --git a/core/src/services/hdfs_native/backend.rs b/core/src/services/hdfs_native/backend.rs
index d0ac3ff9b..b585373e7 100644
--- a/core/src/services/hdfs_native/backend.rs
+++ b/core/src/services/hdfs_native/backend.rs
@@ -145,10 +145,6 @@ impl Access for HdfsNativeBackend {
type Writer = HdfsNativeWriter;
type Lister = Option<HdfsNativeLister>;
type Deleter = oio::OneShotDeleter<HdfsNativeDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let am = AccessorInfo::default();
diff --git a/core/src/services/http/backend.rs b/core/src/services/http/backend.rs
index 02c955f17..3be5f1b3e 100644
--- a/core/src/services/http/backend.rs
+++ b/core/src/services/http/backend.rs
@@ -226,10 +226,6 @@ impl Access for HttpBackend {
type Writer = ();
type Lister = ();
type Deleter = ();
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/huggingface/backend.rs b/core/src/services/huggingface/backend.rs
index c26cae67c..fc868e3e3 100644
--- a/core/src/services/huggingface/backend.rs
+++ b/core/src/services/huggingface/backend.rs
@@ -213,10 +213,6 @@ impl Access for HuggingfaceBackend {
type Writer = ();
type Lister = oio::PageLister<HuggingfaceLister>;
type Deleter = ();
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/icloud/backend.rs b/core/src/services/icloud/backend.rs
index 99fe8dcae..6377c3dde 100644
--- a/core/src/services/icloud/backend.rs
+++ b/core/src/services/icloud/backend.rs
@@ -248,10 +248,6 @@ impl Access for IcloudBackend {
type Writer = ();
type Lister = ();
type Deleter = ();
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/ipfs/backend.rs b/core/src/services/ipfs/backend.rs
index 2db2bb1f1..f44cfe0c2 100644
--- a/core/src/services/ipfs/backend.rs
+++ b/core/src/services/ipfs/backend.rs
@@ -180,10 +180,6 @@ impl Access for IpfsBackend {
type Writer = ();
type Lister = oio::PageLister<DirStream>;
type Deleter = ();
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/ipmfs/backend.rs b/core/src/services/ipmfs/backend.rs
index a14fc2535..841967ff9 100644
--- a/core/src/services/ipmfs/backend.rs
+++ b/core/src/services/ipmfs/backend.rs
@@ -52,10 +52,6 @@ impl Access for IpmfsBackend {
type Writer = oio::OneShotWriter<IpmfsWriter>;
type Lister = oio::PageLister<IpmfsLister>;
type Deleter = oio::OneShotDeleter<IpmfsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/koofr/backend.rs b/core/src/services/koofr/backend.rs
index 3e13e2600..c3f790bbf 100644
--- a/core/src/services/koofr/backend.rs
+++ b/core/src/services/koofr/backend.rs
@@ -236,10 +236,6 @@ impl Access for KoofrBackend {
type Writer = KoofrWriters;
type Lister = oio::PageLister<KoofrLister>;
type Deleter = oio::OneShotDeleter<KoofrDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/lakefs/backend.rs b/core/src/services/lakefs/backend.rs
index 7df8f9762..c1fe4af11 100644
--- a/core/src/services/lakefs/backend.rs
+++ b/core/src/services/lakefs/backend.rs
@@ -218,10 +218,6 @@ impl Access for LakefsBackend {
type Writer = oio::OneShotWriter<LakefsWriter>;
type Lister = oio::PageLister<LakefsLister>;
type Deleter = oio::OneShotDeleter<LakefsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/memory/backend.rs b/core/src/services/memory/backend.rs
index d82186ab4..70ecf727b 100644
--- a/core/src/services/memory/backend.rs
+++ b/core/src/services/memory/backend.rs
@@ -90,10 +90,6 @@ impl typed_kv::Adapter for Adapter {
}
async fn get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
- self.blocking_get(path)
- }
-
- fn blocking_get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
match self.inner.lock().unwrap().get(path) {
None => Ok(None),
Some(bs) => Ok(Some(bs.to_owned())),
@@ -101,30 +97,18 @@ impl typed_kv::Adapter for Adapter {
}
async fn set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
- self.blocking_set(path, value)
- }
-
- fn blocking_set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
self.inner.lock().unwrap().insert(path.to_string(), value);
Ok(())
}
async fn delete(&self, path: &str) -> Result<()> {
- self.blocking_delete(path)
- }
-
- fn blocking_delete(&self, path: &str) -> Result<()> {
self.inner.lock().unwrap().remove(path);
Ok(())
}
async fn scan(&self, path: &str) -> Result<Vec<String>> {
- self.blocking_scan(path)
- }
-
- fn blocking_scan(&self, path: &str) -> Result<Vec<String>> {
let inner = self.inner.lock().unwrap();
if path.is_empty() {
@@ -155,25 +139,4 @@ mod tests {
let b2 = MemoryBuilder::default().build().unwrap();
assert_ne!(b1.info().name(), b2.info().name())
}
-
- #[test]
- fn test_blocking_scan() {
- let adapter = super::Adapter {
- inner: Arc::new(Mutex::new(BTreeMap::default())),
- };
-
- adapter.blocking_set("aaa/bbb/", Value::new_dir()).unwrap();
- adapter.blocking_set("aab/bbb/", Value::new_dir()).unwrap();
- adapter.blocking_set("aab/ccc/", Value::new_dir()).unwrap();
- adapter
- .blocking_set(&format!("aab{}aaa/", std::char::MAX), Value::new_dir())
- .unwrap();
- adapter.blocking_set("aac/bbb/", Value::new_dir()).unwrap();
-
- let data = adapter.blocking_scan("aab").unwrap();
- assert_eq!(data.len(), 3);
- for path in data {
- assert!(path.starts_with("aab"));
- }
- }
}
diff --git a/core/src/services/mini_moka/backend.rs b/core/src/services/mini_moka/backend.rs
index 23587e1e3..b883eee52 100644
--- a/core/src/services/mini_moka/backend.rs
+++ b/core/src/services/mini_moka/backend.rs
@@ -149,10 +149,6 @@ impl typed_kv::Adapter for Adapter {
}
async fn get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
- self.blocking_get(path)
- }
-
- fn blocking_get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
match self.inner.get(&path.to_string()) {
None => Ok(None),
Some(bs) => Ok(Some(bs)),
@@ -160,30 +156,18 @@ impl typed_kv::Adapter for Adapter {
}
async fn set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
- self.blocking_set(path, value)
- }
-
- fn blocking_set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
self.inner.insert(path.to_string(), value);
Ok(())
}
async fn delete(&self, path: &str) -> Result<()> {
- self.blocking_delete(path)
- }
-
- fn blocking_delete(&self, path: &str) -> Result<()> {
self.inner.invalidate(&path.to_string());
Ok(())
}
async fn scan(&self, path: &str) -> Result<Vec<String>> {
- self.blocking_scan(path)
- }
-
- fn blocking_scan(&self, path: &str) -> Result<Vec<String>> {
let keys = self.inner.iter().map(|kv| kv.key().to_string());
if path.is_empty() {
Ok(keys.collect())
diff --git a/core/src/services/moka/backend.rs b/core/src/services/moka/backend.rs
index 608bc902b..ed2da79cd 100644
--- a/core/src/services/moka/backend.rs
+++ b/core/src/services/moka/backend.rs
@@ -172,10 +172,6 @@ impl typed_kv::Adapter for Adapter {
}
async fn get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
- self.blocking_get(path)
- }
-
- fn blocking_get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
match self.inner.get(path) {
None => Ok(None),
Some(bs) => Ok(Some(bs)),
@@ -183,30 +179,18 @@ impl typed_kv::Adapter for Adapter {
}
async fn set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
- self.blocking_set(path, value)
- }
-
- fn blocking_set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
self.inner.insert(path.to_string(), value);
Ok(())
}
async fn delete(&self, path: &str) -> Result<()> {
- self.blocking_delete(path)
- }
-
- fn blocking_delete(&self, path: &str) -> Result<()> {
self.inner.invalidate(path);
Ok(())
}
async fn scan(&self, path: &str) -> Result<Vec<String>> {
- self.blocking_scan(path)
- }
-
- fn blocking_scan(&self, path: &str) -> Result<Vec<String>> {
let keys = self.inner.iter().map(|kv| kv.0.to_string());
if path.is_empty() {
Ok(keys.collect())
diff --git a/core/src/services/monoiofs/backend.rs b/core/src/services/monoiofs/backend.rs
index e8ee9fde4..0894f133f 100644
--- a/core/src/services/monoiofs/backend.rs
+++ b/core/src/services/monoiofs/backend.rs
@@ -106,10 +106,6 @@ impl Access for MonoiofsBackend {
type Writer = MonoiofsWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<MonoiofsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/obs/backend.rs b/core/src/services/obs/backend.rs
index 748e37d41..71dd360c4 100644
--- a/core/src/services/obs/backend.rs
+++ b/core/src/services/obs/backend.rs
@@ -321,10 +321,6 @@ impl Access for ObsBackend {
type Writer = ObsWriters;
type Lister = oio::PageLister<ObsLister>;
type Deleter = oio::OneShotDeleter<ObsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/onedrive/backend.rs b/core/src/services/onedrive/backend.rs
index 3413d68b2..bed50f7d3 100644
--- a/core/src/services/onedrive/backend.rs
+++ b/core/src/services/onedrive/backend.rs
@@ -48,10 +48,6 @@ impl Access for OnedriveBackend {
type Writer = oio::OneShotWriter<OneDriveWriter>;
type Lister = oio::PageLister<OneDriveLister>;
type Deleter = oio::OneShotDeleter<OneDriveDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index 96aa3eb75..151b59e12 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -522,10 +522,6 @@ impl Access for OssBackend {
type Writer = OssWriters;
type Lister = OssListers;
type Deleter = oio::BatchDeleter<OssDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/pcloud/backend.rs b/core/src/services/pcloud/backend.rs
index dd6f74922..23638a331 100644
--- a/core/src/services/pcloud/backend.rs
+++ b/core/src/services/pcloud/backend.rs
@@ -223,10 +223,6 @@ impl Access for PcloudBackend {
type Writer = PcloudWriters;
type Lister = oio::PageLister<PcloudLister>;
type Deleter = oio::OneShotDeleter<PcloudDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/persy/backend.rs b/core/src/services/persy/backend.rs
index 8ff8bad67..4228a5b22 100644
--- a/core/src/services/persy/backend.rs
+++ b/core/src/services/persy/backend.rs
@@ -162,7 +162,6 @@ impl kv::Adapter for Adapter {
read: true,
write: true,
delete: true,
- blocking: true,
shared: false,
..Default::default()
},
@@ -170,15 +169,6 @@ impl kv::Adapter for Adapter {
}
async fn get(&self, path: &str) -> Result<Option<Buffer>> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
- task::spawn_blocking(move || cloned_self.blocking_get(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)
- .and_then(|inner_result| inner_result)
- }
-
- fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
let mut read_id = self
.persy
.get::<String, persy::PersyId>(&self.index, &path.to_string())
@@ -192,16 +182,6 @@ impl kv::Adapter for Adapter {
}
async fn set(&self, path: &str, value: Buffer) -> Result<()> {
- let cloned_path = path.to_string();
- let cloned_self = self.clone();
-
- task::spawn_blocking(move || cloned_self.blocking_set(cloned_path.as_str(), value))
- .await
- .map_err(new_task_join_error)
- .and_then(|inner_result| inner_result)
- }
-
- fn blocking_set(&self, path: &str, value: Buffer) -> Result<()> {
let mut tx = self.persy.begin().map_err(parse_error)?;
let id = tx
.insert(&self.segment, &value.to_vec())
@@ -216,16 +196,6 @@ impl kv::Adapter for Adapter {
}
async fn delete(&self, path: &str) -> Result<()> {
- let cloned_path = path.to_string();
- let cloned_self = self.clone();
-
- task::spawn_blocking(move || cloned_self.blocking_delete(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)
- .and_then(|inner_result| inner_result)
- }
-
- fn blocking_delete(&self, path: &str) -> Result<()> {
let mut delete_id = self
.persy
.get::<String, persy::PersyId>(&self.index, &path.to_string())
diff --git a/core/src/services/redb/backend.rs b/core/src/services/redb/backend.rs
index b8951ac18..96890116d 100644
--- a/core/src/services/redb/backend.rs
+++ b/core/src/services/redb/backend.rs
@@ -120,7 +120,6 @@ impl kv::Adapter for Adapter {
Capability {
read: true,
write: true,
- blocking: true,
shared: false,
..Default::default()
},
@@ -128,16 +127,6 @@ impl kv::Adapter for Adapter {
}
async fn get(&self, path: &str) -> Result<Option<Buffer>> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_get(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)
- .and_then(|inner_result| inner_result)
- }
-
- fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
let read_txn = self.db.begin_read().map_err(parse_transaction_error)?;
let table_define: redb::TableDefinition<&str, &[u8]> =
@@ -156,16 +145,6 @@ impl kv::Adapter for Adapter {
}
async fn set(&self, path: &str, value: Buffer) -> Result<()> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_set(cloned_path.as_str(), value))
- .await
- .map_err(new_task_join_error)
- .and_then(|inner_result| inner_result)
- }
-
- fn blocking_set(&self, path: &str, value: Buffer) -> Result<()> {
let write_txn = self.db.begin_write().map_err(parse_transaction_error)?;
let table_define: redb::TableDefinition<&str, &[u8]> =
@@ -186,16 +165,6 @@ impl kv::Adapter for Adapter {
}
async fn delete(&self, path: &str) -> Result<()> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_delete(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)
- .and_then(|inner_result| inner_result)
- }
-
- fn blocking_delete(&self, path: &str) -> Result<()> {
let write_txn = self.db.begin_write().map_err(parse_transaction_error)?;
let table_define: redb::TableDefinition<&str, &[u8]> =
diff --git a/core/src/services/rocksdb/backend.rs b/core/src/services/rocksdb/backend.rs
index 3092bddef..a7fbc62be 100644
--- a/core/src/services/rocksdb/backend.rs
+++ b/core/src/services/rocksdb/backend.rs
@@ -118,7 +118,6 @@ impl kv::Adapter for Adapter {
read: true,
write: true,
list: true,
- blocking: true,
shared: false,
..Default::default()
},
@@ -126,60 +125,21 @@ impl kv::Adapter for Adapter {
}
async fn get(&self, path: &str) -> Result<Option<Buffer>> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_get(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)?
- }
-
- fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
let result = self.db.get(path).map_err(parse_rocksdb_error)?;
Ok(result.map(Buffer::from))
}
async fn set(&self, path: &str, value: Buffer) -> Result<()> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_set(cloned_path.as_str(), value))
- .await
- .map_err(new_task_join_error)?
- }
-
- fn blocking_set(&self, path: &str, value: Buffer) -> Result<()> {
self.db
.put(path, value.to_vec())
.map_err(parse_rocksdb_error)
}
async fn delete(&self, path: &str) -> Result<()> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_delete(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)?
- }
-
- fn blocking_delete(&self, path: &str) -> Result<()> {
self.db.delete(path).map_err(parse_rocksdb_error)
}
async fn scan(&self, path: &str) -> Result<Self::Scanner> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- let res = task::spawn_blocking(move || cloned_self.blocking_scan(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)??;
-
- Ok(Box::new(kv::ScanStdIter::new(res.into_iter().map(Ok))))
- }
-
- /// TODO: we only need key here.
- fn blocking_scan(&self, path: &str) -> Result<Vec<String>> {
let it = self.db.prefix_iterator(path).map(|r| r.map(|(k, _)| k));
let mut res = Vec::default();
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 1686e6981..264198adc 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -1034,10 +1034,6 @@ impl Access for S3Backend {
type Writer = S3Writers;
type Lister = S3Listers;
type Deleter = oio::BatchDeleter<S3Deleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/seafile/backend.rs b/core/src/services/seafile/backend.rs
index 6dd1d9667..69c924d35 100644
--- a/core/src/services/seafile/backend.rs
+++ b/core/src/services/seafile/backend.rs
@@ -242,10 +242,6 @@ impl Access for SeafileBackend {
type Writer = SeafileWriters;
type Lister = oio::PageLister<SeafileLister>;
type Deleter = oio::OneShotDeleter<SeafileDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/sftp/backend.rs b/core/src/services/sftp/backend.rs
index 917736255..4e3591995 100644
--- a/core/src/services/sftp/backend.rs
+++ b/core/src/services/sftp/backend.rs
@@ -244,10 +244,6 @@ impl Access for SftpBackend {
type Writer = SftpWriter;
type Lister = Option<SftpLister>;
type Deleter = oio::OneShotDeleter<SftpDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/sled/backend.rs b/core/src/services/sled/backend.rs
index 308859945..cc0e44e50 100644
--- a/core/src/services/sled/backend.rs
+++ b/core/src/services/sled/backend.rs
@@ -147,7 +147,6 @@ impl kv::Adapter for Adapter {
read: true,
write: true,
list: true,
- blocking: true,
shared: false,
..Default::default()
},
@@ -155,15 +154,6 @@ impl kv::Adapter for Adapter {
}
async fn get(&self, path: &str) -> Result<Option<Buffer>> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_get(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)?
- }
-
- fn blocking_get(&self, path: &str) -> Result<Option<Buffer>> {
Ok(self
.tree
.get(path)
@@ -172,15 +162,6 @@ impl kv::Adapter for Adapter {
}
async fn set(&self, path: &str, value: Buffer) -> Result<()> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_set(cloned_path.as_str(), value))
- .await
- .map_err(new_task_join_error)?
- }
-
- fn blocking_set(&self, path: &str, value: Buffer) -> Result<()> {
self.tree
.insert(path, value.to_vec())
.map_err(parse_error)?;
@@ -188,32 +169,12 @@ impl kv::Adapter for Adapter {
}
async fn delete(&self, path: &str) -> Result<()> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- task::spawn_blocking(move || cloned_self.blocking_delete(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)?
- }
-
- fn blocking_delete(&self, path: &str) -> Result<()> {
self.tree.remove(path).map_err(parse_error)?;
Ok(())
}
async fn scan(&self, path: &str) -> Result<Self::Scanner> {
- let cloned_self = self.clone();
- let cloned_path = path.to_string();
-
- let res = task::spawn_blocking(move || cloned_self.blocking_scan(cloned_path.as_str()))
- .await
- .map_err(new_task_join_error)??;
-
- Ok(Box::new(kv::ScanStdIter::new(res.into_iter().map(Ok))))
- }
-
- fn blocking_scan(&self, path: &str) -> Result<Vec<String>> {
let it = self.tree.scan_prefix(path).keys();
let mut res = Vec::default();
diff --git a/core/src/services/swift/backend.rs b/core/src/services/swift/backend.rs
index ba8529dc3..859d814c9 100644
--- a/core/src/services/swift/backend.rs
+++ b/core/src/services/swift/backend.rs
@@ -211,10 +211,6 @@ impl Access for SwiftBackend {
type Writer = oio::OneShotWriter<SwiftWriter>;
type Lister = oio::PageLister<SwiftLister>;
type Deleter = oio::OneShotDeleter<SwfitDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/upyun/backend.rs b/core/src/services/upyun/backend.rs
index f67ba915d..c2d2ce9b9 100644
--- a/core/src/services/upyun/backend.rs
+++ b/core/src/services/upyun/backend.rs
@@ -237,10 +237,6 @@ impl Access for UpyunBackend {
type Writer = UpyunWriters;
type Lister = oio::PageLister<UpyunLister>;
type Deleter = oio::OneShotDeleter<UpyunDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/vercel_artifacts/backend.rs b/core/src/services/vercel_artifacts/backend.rs
index e6137349e..c4da64618 100644
--- a/core/src/services/vercel_artifacts/backend.rs
+++ b/core/src/services/vercel_artifacts/backend.rs
@@ -46,10 +46,6 @@ impl Access for VercelArtifactsBackend {
type Writer = oio::OneShotWriter<VercelArtifactsWriter>;
type Lister = ();
type Deleter = ();
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/vercel_blob/backend.rs b/core/src/services/vercel_blob/backend.rs
index f6724cf77..c644be3b2 100644
--- a/core/src/services/vercel_blob/backend.rs
+++ b/core/src/services/vercel_blob/backend.rs
@@ -183,10 +183,6 @@ impl Access for VercelBlobBackend {
type Writer = VercelBlobWriters;
type Lister = oio::PageLister<VercelBlobLister>;
type Deleter = ();
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/webdav/backend.rs b/core/src/services/webdav/backend.rs
index 7b5e27cf0..5e6f4b9c1 100644
--- a/core/src/services/webdav/backend.rs
+++ b/core/src/services/webdav/backend.rs
@@ -246,10 +246,6 @@ impl Access for WebdavBackend {
type Writer = oio::OneShotWriter<WebdavWriter>;
type Lister = oio::PageLister<WebdavLister>;
type Deleter = oio::OneShotDeleter<WebdavDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/webhdfs/backend.rs b/core/src/services/webhdfs/backend.rs
index fd2de26e2..7dcc83979 100644
--- a/core/src/services/webhdfs/backend.rs
+++ b/core/src/services/webhdfs/backend.rs
@@ -261,10 +261,6 @@ impl Access for WebhdfsBackend {
type Writer = WebhdfsWriters;
type Lister = oio::PageLister<WebhdfsLister>;
type Deleter = oio::OneShotDeleter<WebhdfsDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
diff --git a/core/src/services/yandex_disk/backend.rs b/core/src/services/yandex_disk/backend.rs
index 39d5f36c5..510ee2a2a 100644
--- a/core/src/services/yandex_disk/backend.rs
+++ b/core/src/services/yandex_disk/backend.rs
@@ -185,10 +185,6 @@ impl Access for YandexDiskBackend {
type Writer = YandexDiskWriters;
type Lister = oio::PageLister<YandexDiskLister>;
type Deleter = oio::OneShotDeleter<YandexDiskDeleter>;
- type BlockingReader = ();
- type BlockingWriter = ();
- type BlockingLister = ();
- type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()