This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new b5654468a6 feat(core): Implement into_stream for Reader (#4473)
b5654468a6 is described below
commit b5654468a6cf6b8c1d0046a62f2d69824497b167
Author: Xuanwo <[email protected]>
AuthorDate: Fri Apr 12 20:36:40 2024 +0800
feat(core): Implement into_stream for Reader (#4473)
* feat(core): Implement into_stream for Reader
Signed-off-by: Xuanwo <[email protected]>
* Fix build
Signed-off-by: Xuanwo <[email protected]>
---------
Signed-off-by: Xuanwo <[email protected]>
---
core/src/docs/internals/accessor.rs | 2 +-
core/src/layers/complete.rs | 2 +-
core/src/layers/minitrace.rs | 8 +-
core/src/layers/oteltrace.rs | 6 +-
core/src/layers/tracing.rs | 8 +-
core/src/layers/type_eraser.rs | 3 +-
core/src/raw/futures_util.rs | 17 +++
core/src/raw/http_util/client.rs | 11 --
core/src/raw/mod.rs | 1 +
core/src/raw/oio/list/api.rs | 7 +-
core/src/raw/oio/list/page_list.rs | 2 +-
core/src/raw/oio/read/api.rs | 39 ++++++-
core/src/raw/oio/read/mod.rs | 1 +
core/src/raw/oio/write/api.rs | 6 +-
core/src/raw/oio/write/append_write.rs | 10 +-
core/src/raw/oio/write/block_write.rs | 25 +----
core/src/raw/oio/write/multipart_write.rs | 35 +-----
core/src/raw/oio/write/one_shot_write.rs | 5 +-
core/src/raw/oio/write/range_write.rs | 10 +-
core/src/types/reader.rs | 170 +++++++++++++++++-------------
core/tests/behavior/utils.rs | 5 +-
21 files changed, 195 insertions(+), 178 deletions(-)
diff --git a/core/src/docs/internals/accessor.rs
b/core/src/docs/internals/accessor.rs
index 174bbb265b..287a82cb85 100644
--- a/core/src/docs/internals/accessor.rs
+++ b/core/src/docs/internals/accessor.rs
@@ -64,7 +64,7 @@
//! pub trait Accessor {
//! fn create_dir<'async>(
//! &'async self,
-//! ) -> Pin<Box<dyn core::future::Future<Output = Result()> + Send +
'async>>
+//! ) -> Pin<Box<dyn core::future::Future<Output = Result()> + MaybeSend +
'async>>
//! where Self: Sync + 'async;
//! }
//! ```
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index 228d68e2ce..b51dfc2224 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -734,7 +734,7 @@ mod tests {
}
async fn read(&self, _: &str, _: OpRead) -> Result<(RpRead,
Self::Reader)> {
- Ok((RpRead::new(), Box::new(bytes::Bytes::new())))
+ Ok((RpRead::new(), Arc::new(bytes::Bytes::new())))
}
async fn write(&self, _: &str, _: OpWrite) -> Result<(RpWrite,
Self::Writer)> {
diff --git a/core/src/layers/minitrace.rs b/core/src/layers/minitrace.rs
index f4abba7387..879a614161 100644
--- a/core/src/layers/minitrace.rs
+++ b/core/src/layers/minitrace.rs
@@ -59,7 +59,7 @@ use crate::*;
/// use opendal::services;
/// use opendal::Operator;
///
-/// fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+/// fn main() -> Result<(), Box<dyn Error + MaybeSend + Sync + 'static>> {
/// let reporter =
///
minitrace_jaeger::JaegerReporter::new("127.0.0.1:6831".parse().unwrap(),
"opendal")
/// .unwrap();
@@ -309,19 +309,19 @@ impl<R: oio::BlockingRead> oio::BlockingRead for
MinitraceWrapper<R> {
}
impl<R: oio::Write> oio::Write for MinitraceWrapper<R> {
- fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
Send {
+ fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
MaybeSend {
let _g = self.span.set_local_parent();
let _span =
LocalSpan::enter_with_local_parent(WriteOperation::Write.into_static());
self.inner.write(bs)
}
- fn abort(&mut self) -> impl Future<Output = Result<()>> + Send {
+ fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
let _g = self.span.set_local_parent();
let _span =
LocalSpan::enter_with_local_parent(WriteOperation::Abort.into_static());
self.inner.abort()
}
- fn close(&mut self) -> impl Future<Output = Result<()>> + Send {
+ fn close(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
let _g = self.span.set_local_parent();
let _span =
LocalSpan::enter_with_local_parent(WriteOperation::Close.into_static());
self.inner.close()
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index 1aeaf9fc20..18842c1c0d 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -287,15 +287,15 @@ impl<R: oio::BlockingRead> oio::BlockingRead for
OtelTraceWrapper<R> {
}
impl<R: oio::Write> oio::Write for OtelTraceWrapper<R> {
- fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
Send {
+ fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
MaybeSend {
self.inner.write(bs)
}
- fn abort(&mut self) -> impl Future<Output = Result<()>> + Send {
+ fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner.abort()
}
- fn close(&mut self) -> impl Future<Output = Result<()>> + Send {
+ fn close(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner.close()
}
}
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index b53beca0cf..dcd6820463 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -57,7 +57,7 @@ use crate::*;
/// use tracing_subscriber::prelude::*;
/// use tracing_subscriber::EnvFilter;
///
-/// fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+/// fn main() -> Result<(), Box<dyn Error + MaybeSend + Sync + 'static>> {
/// let tracer = opentelemetry_jaeger::new_pipeline()
/// .with_service_name("opendal_example")
/// .install_simple()?;
@@ -289,7 +289,7 @@ impl<R: oio::Write> oio::Write for TracingWrapper<R> {
parent = &self.span,
level = "trace",
skip_all)]
- fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
Send {
+ fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
MaybeSend {
self.inner.write(bs)
}
@@ -297,7 +297,7 @@ impl<R: oio::Write> oio::Write for TracingWrapper<R> {
parent = &self.span,
level = "trace",
skip_all)]
- fn abort(&mut self) -> impl Future<Output = Result<()>> + Send {
+ fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner.abort()
}
@@ -305,7 +305,7 @@ impl<R: oio::Write> oio::Write for TracingWrapper<R> {
parent = &self.span,
level = "trace",
skip_all)]
- fn close(&mut self) -> impl Future<Output = Result<()>> + Send {
+ fn close(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner.close()
}
}
diff --git a/core/src/layers/type_eraser.rs b/core/src/layers/type_eraser.rs
index 37a0444eca..51bb9ba258 100644
--- a/core/src/layers/type_eraser.rs
+++ b/core/src/layers/type_eraser.rs
@@ -17,6 +17,7 @@
use std::fmt::Debug;
use std::fmt::Formatter;
+use std::sync::Arc;
use async_trait::async_trait;
@@ -71,7 +72,7 @@ impl<A: Accessor> LayeredAccessor for TypeEraseAccessor<A> {
self.inner
.read(path, args)
.await
- .map(|(rp, r)| (rp, Box::new(r) as oio::Reader))
+ .map(|(rp, r)| (rp, Arc::new(r) as oio::Reader))
}
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite,
Self::Writer)> {
diff --git a/core/src/raw/futures_util.rs b/core/src/raw/futures_util.rs
index 2e467b8a2c..850b137d7e 100644
--- a/core/src/raw/futures_util.rs
+++ b/core/src/raw/futures_util.rs
@@ -41,6 +41,23 @@ pub type BoxedStaticFuture<T> =
futures::future::BoxFuture<'static, T>;
#[cfg(target_arch = "wasm32")]
pub type BoxedStaticFuture<T> = futures::future::LocalBoxFuture<'static, T>;
+/// MaybeSend is a marker to determine whether a type is `Send` or not.
+/// We use this trait to wrap the `Send` requirement for wasm32 target.
+///
+/// # Safety
+///
+/// MaybeSend equivalent to `Send` on non-wasm32 target. And it's empty
+/// on wasm32 target.
+#[cfg(not(target_arch = "wasm32"))]
+pub unsafe trait MaybeSend: Send {}
+#[cfg(target_arch = "wasm32")]
+pub unsafe trait MaybeSend {}
+
+#[cfg(not(target_arch = "wasm32"))]
+unsafe impl<T: Send> MaybeSend for T {}
+#[cfg(target_arch = "wasm32")]
+unsafe impl<T> MaybeSend for T {}
+
/// CONCURRENT_LARGE_THRESHOLD is the threshold to determine whether to use
/// [`FuturesOrdered`] or not.
///
diff --git a/core/src/raw/http_util/client.rs b/core/src/raw/http_util/client.rs
index 1073996417..fea7ac0365 100644
--- a/core/src/raw/http_util/client.rs
+++ b/core/src/raw/http_util/client.rs
@@ -57,7 +57,6 @@ impl HttpClient {
}
/// Build a new http client in async context.
- #[cfg(not(target_arch = "wasm32"))]
pub fn build(builder: reqwest::ClientBuilder) -> Result<Self> {
Ok(Self {
client: builder.build().map_err(|err| {
@@ -66,16 +65,6 @@ impl HttpClient {
})
}
- /// Build a new http client in async context.
- #[cfg(target_arch = "wasm32")]
- pub fn build(mut builder: reqwest::ClientBuilder) -> Result<Self> {
- Ok(Self {
- client: builder.build().map_err(|err| {
- Error::new(ErrorKind::Unexpected, "http client build
failed").set_source(err)
- })?,
- })
- }
-
/// Get the async client from http client.
pub fn client(&self) -> reqwest::Client {
self.client.clone()
diff --git a/core/src/raw/mod.rs b/core/src/raw/mod.rs
index e58d6e9ced..8065c86d27 100644
--- a/core/src/raw/mod.rs
+++ b/core/src/raw/mod.rs
@@ -73,6 +73,7 @@ mod futures_util;
pub use futures_util::BoxedFuture;
pub use futures_util::BoxedStaticFuture;
pub use futures_util::ConcurrentFutures;
+pub use futures_util::MaybeSend;
mod enum_utils;
pub use enum_utils::*;
diff --git a/core/src/raw/oio/list/api.rs b/core/src/raw/oio/list/api.rs
index 48368c7a93..c071b7ed45 100644
--- a/core/src/raw/oio/list/api.rs
+++ b/core/src/raw/oio/list/api.rs
@@ -21,7 +21,7 @@ use std::future::Future;
use std::ops::DerefMut;
use crate::raw::oio::Entry;
-use crate::raw::BoxedFuture;
+use crate::raw::*;
use crate::*;
/// PageOperation is the name for APIs of lister.
@@ -67,10 +67,7 @@ pub trait List: Unpin + Send + Sync {
///
/// `Ok(None)` means all pages have been returned. Any following call
/// to `next` will always get the same result.
- #[cfg(not(target_arch = "wasm32"))]
- fn next(&mut self) -> impl Future<Output = Result<Option<Entry>>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn next(&mut self) -> impl Future<Output = Result<Option<Entry>>>;
+ fn next(&mut self) -> impl Future<Output = Result<Option<Entry>>> +
MaybeSend;
}
impl List for () {
diff --git a/core/src/raw/oio/list/page_list.rs
b/core/src/raw/oio/list/page_list.rs
index 2ed43bdd42..9ba141be80 100644
--- a/core/src/raw/oio/list/page_list.rs
+++ b/core/src/raw/oio/list/page_list.rs
@@ -34,7 +34,7 @@ use crate::*;
pub trait PageList: Send + Sync + Unpin + 'static {
/// next_page is used to fetch next page of entries from underlying
storage.
#[cfg(not(target_arch = "wasm32"))]
- fn next_page(&self, ctx: &mut PageContext) -> impl Future<Output =
Result<()>> + Send;
+ fn next_page(&self, ctx: &mut PageContext) -> impl Future<Output =
Result<()>> + MaybeSend;
#[cfg(target_arch = "wasm32")]
fn next_page(&self, ctx: &mut PageContext) -> impl Future<Output =
Result<()>>;
}
diff --git a/core/src/raw/oio/read/api.rs b/core/src/raw/oio/read/api.rs
index 0cfe3bfd59..46332eb20e 100644
--- a/core/src/raw/oio/read/api.rs
+++ b/core/src/raw/oio/read/api.rs
@@ -18,6 +18,7 @@
use std::fmt::Display;
use std::fmt::Formatter;
use std::ops::Deref;
+use std::sync::Arc;
use bytes::Bytes;
use futures::Future;
@@ -60,7 +61,7 @@ impl From<ReadOperation> for &'static str {
}
/// Reader is a type erased [`Read`].
-pub type Reader = Box<dyn ReadDyn>;
+pub type Reader = Arc<dyn ReadDyn>;
/// Read is the internal trait used by OpenDAL to read data from storage.
///
@@ -84,7 +85,11 @@ pub trait Read: Unpin + Send + Sync {
/// Storage services should try to read as much as possible, only return
bytes less than the
/// limit while reaching the end of the file.
#[cfg(not(target_arch = "wasm32"))]
- fn read_at(&self, offset: u64, limit: usize) -> impl Future<Output =
Result<Buffer>> + Send;
+ fn read_at(
+ &self,
+ offset: u64,
+ limit: usize,
+ ) -> impl Future<Output = Result<Buffer>> + MaybeSend;
#[cfg(target_arch = "wasm32")]
fn read_at(&self, offset: u64, limit: usize) -> impl Future<Output =
Result<Buffer>>;
}
@@ -112,14 +117,38 @@ impl Read for Bytes {
}
}
+/// ReadDyn is the dyn version of [`Read`] make it possible to use as
+/// `Box<dyn ReadDyn>`.
pub trait ReadDyn: Unpin + Send + Sync {
+ /// The dyn version of [`Read::read_at`].
+ ///
+ /// This function returns a boxed future to make it object safe.
fn read_at_dyn(&self, offset: u64, limit: usize) ->
BoxedFuture<Result<Buffer>>;
+
+ /// The static version of [`Read::read_at`].
+ ///
+ /// This function returns a `'static` future by moving `self` into the
+ /// future. Caller can call `Box::pin` to build a static boxed future.
+ fn read_at_static(
+ self,
+ offset: u64,
+ limit: usize,
+ ) -> impl Future<Output = Result<Buffer>> + MaybeSend + 'static
+ where
+ Self: Sized + 'static;
}
impl<T: Read + ?Sized> ReadDyn for T {
fn read_at_dyn(&self, offset: u64, limit: usize) ->
BoxedFuture<Result<Buffer>> {
Box::pin(self.read_at(offset, limit))
}
+
+ async fn read_at_static(self, offset: u64, limit: usize) -> Result<Buffer>
+ where
+ Self: Sized + 'static,
+ {
+ self.read_at(offset, limit).await
+ }
}
/// # NOTE
@@ -132,6 +161,12 @@ impl<T: ReadDyn + ?Sized> Read for Box<T> {
}
}
+impl<T: ReadDyn + ?Sized> Read for Arc<T> {
+ async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
+ self.deref().read_at_dyn(offset, limit).await
+ }
+}
+
/// BlockingReader is a boxed dyn `BlockingRead`.
pub type BlockingReader = Box<dyn BlockingRead>;
diff --git a/core/src/raw/oio/read/mod.rs b/core/src/raw/oio/read/mod.rs
index 3144ce874b..5ec4114c51 100644
--- a/core/src/raw/oio/read/mod.rs
+++ b/core/src/raw/oio/read/mod.rs
@@ -19,5 +19,6 @@ mod api;
pub use api::BlockingRead;
pub use api::BlockingReader;
pub use api::Read;
+pub use api::ReadDyn;
pub use api::ReadOperation;
pub use api::Reader;
diff --git a/core/src/raw/oio/write/api.rs b/core/src/raw/oio/write/api.rs
index 473ef642ef..de74c76ef0 100644
--- a/core/src/raw/oio/write/api.rs
+++ b/core/src/raw/oio/write/api.rs
@@ -83,19 +83,19 @@ pub trait Write: Unpin + Send + Sync {
/// It's possible that `n < bs.len()`, caller should pass the remaining
bytes
/// repeatedly until all bytes has been written.
#[cfg(not(target_arch = "wasm32"))]
- fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
Send;
+ fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>> +
MaybeSend;
#[cfg(target_arch = "wasm32")]
fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<usize>>;
/// Close the writer and make sure all data has been flushed.
#[cfg(not(target_arch = "wasm32"))]
- fn close(&mut self) -> impl Future<Output = Result<()>> + Send;
+ fn close(&mut self) -> impl Future<Output = Result<()>> + MaybeSend;
#[cfg(target_arch = "wasm32")]
fn close(&mut self) -> impl Future<Output = Result<()>>;
/// Abort the pending writer.
#[cfg(not(target_arch = "wasm32"))]
- fn abort(&mut self) -> impl Future<Output = Result<()>> + Send;
+ fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend;
#[cfg(target_arch = "wasm32")]
fn abort(&mut self) -> impl Future<Output = Result<()>>;
}
diff --git a/core/src/raw/oio/write/append_write.rs
b/core/src/raw/oio/write/append_write.rs
index 58811bb046..a7f0fbae6d 100644
--- a/core/src/raw/oio/write/append_write.rs
+++ b/core/src/raw/oio/write/append_write.rs
@@ -40,21 +40,15 @@ pub trait AppendWrite: Send + Sync + Unpin + 'static {
/// Get the current offset of the append object.
///
/// Returns `0` if the object is not exist.
- #[cfg(not(target_arch = "wasm32"))]
- fn offset(&self) -> impl Future<Output = Result<u64>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn offset(&self) -> impl Future<Output = Result<u64>>;
+ fn offset(&self) -> impl Future<Output = Result<u64>> + MaybeSend;
/// Append the data to the end of this object.
- #[cfg(not(target_arch = "wasm32"))]
fn append(
&self,
offset: u64,
size: u64,
body: Buffer,
- ) -> impl Future<Output = Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn append(&self, offset: u64, size: u64, body: Buffer) -> impl
Future<Output = Result<()>>;
+ ) -> impl Future<Output = Result<()>> + MaybeSend;
}
/// AppendWriter will implements [`Write`] based on append object.
diff --git a/core/src/raw/oio/write/block_write.rs
b/core/src/raw/oio/write/block_write.rs
index d3625c41d0..0ff7cb7dab 100644
--- a/core/src/raw/oio/write/block_write.rs
+++ b/core/src/raw/oio/write/block_write.rs
@@ -65,10 +65,7 @@ pub trait BlockWrite: Send + Sync + Unpin + 'static {
/// BlockWriter will call this API when:
///
/// - All the data has been written to the buffer and we can perform the
upload at once.
- #[cfg(not(target_arch = "wasm32"))]
- fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>>;
+ fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>> + MaybeSend;
/// write_block will write a block of the data and returns the result
/// [`Block`].
@@ -77,33 +74,19 @@ pub trait BlockWrite: Send + Sync + Unpin + 'static {
/// order.
///
/// - block_id is the id of the block.
- #[cfg(not(target_arch = "wasm32"))]
fn write_block(
&self,
block_id: Uuid,
size: u64,
body: Buffer,
- ) -> impl Future<Output = Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn write_block(
- &self,
- block_id: Uuid,
- size: u64,
- body: Buffer,
- ) -> impl Future<Output = Result<()>>;
+ ) -> impl Future<Output = Result<()>> + MaybeSend;
/// complete_block will complete the block upload to build the final
/// file.
- #[cfg(not(target_arch = "wasm32"))]
- fn complete_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output =
Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn complete_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output =
Result<()>>;
+ fn complete_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output =
Result<()>> + MaybeSend;
/// abort_block will cancel the block upload and purge all data.
- #[cfg(not(target_arch = "wasm32"))]
- fn abort_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output =
Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn abort_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output =
Result<()>>;
+ fn abort_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output =
Result<()>> + MaybeSend;
}
/// WriteBlockResult is the result returned by [`WriteBlockFuture`].
diff --git a/core/src/raw/oio/write/multipart_write.rs
b/core/src/raw/oio/write/multipart_write.rs
index 091e418d4d..7c8ab07601 100644
--- a/core/src/raw/oio/write/multipart_write.rs
+++ b/core/src/raw/oio/write/multipart_write.rs
@@ -65,10 +65,7 @@ pub trait MultipartWrite: Send + Sync + Unpin + 'static {
/// MultipartWriter will call this API when:
///
/// - All the data has been written to the buffer and we can perform the
upload at once.
- #[cfg(not(target_arch = "wasm32"))]
- fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>>;
+ fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>> + MaybeSend;
/// initiate_part will call start a multipart upload and return the upload
id.
///
@@ -77,10 +74,7 @@ pub trait MultipartWrite: Send + Sync + Unpin + 'static {
/// - the total size of data is unknown.
/// - the total size of data is known, but the size of current write
/// is less then the total size.
- #[cfg(not(target_arch = "wasm32"))]
- fn initiate_part(&self) -> impl Future<Output = Result<String>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn initiate_part(&self) -> impl Future<Output = Result<String>>;
+ fn initiate_part(&self) -> impl Future<Output = Result<String>> +
MaybeSend;
/// write_part will write a part of the data and returns the result
/// [`MultipartPart`].
@@ -89,43 +83,24 @@ pub trait MultipartWrite: Send + Sync + Unpin + 'static {
/// order.
///
/// - part_number is the index of the part, starting from 0.
- #[cfg(not(target_arch = "wasm32"))]
fn write_part(
&self,
upload_id: &str,
part_number: usize,
size: u64,
body: Buffer,
- ) -> impl Future<Output = Result<MultipartPart>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn write_part(
- &self,
- upload_id: &str,
- part_number: usize,
- size: u64,
- body: Buffer,
- ) -> impl Future<Output = Result<MultipartPart>>;
+ ) -> impl Future<Output = Result<MultipartPart>> + MaybeSend;
/// complete_part will complete the multipart upload to build the final
/// file.
- #[cfg(not(target_arch = "wasm32"))]
- fn complete_part(
- &self,
- upload_id: &str,
- parts: &[MultipartPart],
- ) -> impl Future<Output = Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
fn complete_part(
&self,
upload_id: &str,
parts: &[MultipartPart],
- ) -> impl Future<Output = Result<()>>;
+ ) -> impl Future<Output = Result<()>> + MaybeSend;
/// abort_part will cancel the multipart upload and purge all data.
- #[cfg(not(target_arch = "wasm32"))]
- fn abort_part(&self, upload_id: &str) -> impl Future<Output = Result<()>>
+ Send;
- #[cfg(target_arch = "wasm32")]
- fn abort_part(&self, upload_id: &str) -> impl Future<Output = Result<()>>;
+ fn abort_part(&self, upload_id: &str) -> impl Future<Output = Result<()>>
+ MaybeSend;
}
/// The result of [`MultipartWrite::write_part`].
diff --git a/core/src/raw/oio/write/one_shot_write.rs
b/core/src/raw/oio/write/one_shot_write.rs
index 3cdc11e932..0cd6b2a97a 100644
--- a/core/src/raw/oio/write/one_shot_write.rs
+++ b/core/src/raw/oio/write/one_shot_write.rs
@@ -32,10 +32,7 @@ pub trait OneShotWrite: Send + Sync + Unpin + 'static {
/// write_once write all data at once.
///
/// Implementations should make sure that the data is written correctly at
once.
- #[cfg(not(target_arch = "wasm32"))]
- fn write_once(&self, bs: Bytes) -> impl Future<Output = Result<()>> + Send;
- #[cfg(target_arch = "wasm32")]
- fn write_once(&self, bs: Bytes) -> impl Future<Output = Result<()>>;
+ fn write_once(&self, bs: Bytes) -> impl Future<Output = Result<()>> +
MaybeSend;
}
/// OneShotWrite is used to implement [`Write`] based on one shot.
diff --git a/core/src/raw/oio/write/range_write.rs
b/core/src/raw/oio/write/range_write.rs
index 34e0e7a108..7b72805ffc 100644
--- a/core/src/raw/oio/write/range_write.rs
+++ b/core/src/raw/oio/write/range_write.rs
@@ -63,10 +63,10 @@ pub trait RangeWrite: Send + Sync + Unpin + 'static {
/// RangeWriter will call this API when:
///
/// - All the data has been written to the buffer and we can perform the
upload at once.
- fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>> + Send;
+ fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output =
Result<()>> + MaybeSend;
/// Initiate range the range write, the returning value is the location.
- fn initiate_range(&self) -> impl Future<Output = Result<String>> + Send;
+ fn initiate_range(&self) -> impl Future<Output = Result<String>> +
MaybeSend;
/// write_range will write a range of data.
fn write_range(
@@ -75,7 +75,7 @@ pub trait RangeWrite: Send + Sync + Unpin + 'static {
offset: u64,
size: u64,
body: Buffer,
- ) -> impl Future<Output = Result<()>> + Send;
+ ) -> impl Future<Output = Result<()>> + MaybeSend;
/// complete_range will complete the range write by uploading the last
chunk.
fn complete_range(
@@ -84,10 +84,10 @@ pub trait RangeWrite: Send + Sync + Unpin + 'static {
offset: u64,
size: u64,
body: Buffer,
- ) -> impl Future<Output = Result<()>> + Send;
+ ) -> impl Future<Output = Result<()>> + MaybeSend;
/// abort_range will abort the range write by abort all already uploaded
data.
- fn abort_range(&self, location: &str) -> impl Future<Output = Result<()>>
+ Send;
+ fn abort_range(&self, location: &str) -> impl Future<Output = Result<()>>
+ MaybeSend;
}
/// WritePartResult is the result returned by [`WriteRangeFuture`].
diff --git a/core/src/types/reader.rs b/core/src/types/reader.rs
index 3ff7cb7588..7dc8944ef7 100644
--- a/core/src/types/reader.rs
+++ b/core/src/types/reader.rs
@@ -15,12 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-use std::ops::Bound;
use std::ops::Range;
use std::ops::RangeBounds;
-use bytes::Buf;
use bytes::BufMut;
+use futures::stream;
+use futures::Stream;
+use futures::StreamExt;
+use futures::TryStreamExt;
use crate::raw::*;
use crate::*;
@@ -36,6 +38,7 @@ use crate::*;
/// ## Direct
///
/// [`Reader`] provides public API including [`Reader::read`],
[`Reader:read_range`], and [`Reader::read_to_end`]. You can use those APIs
directly without extra copy.
+#[derive(Clone)]
pub struct Reader {
inner: oio::Reader,
}
@@ -63,43 +66,8 @@ impl Reader {
///
/// - Buffer length smaller than range means we have reached the end of
file.
pub async fn read(&self, range: impl RangeBounds<u64>) -> Result<Buffer> {
- let start = match range.start_bound().cloned() {
- Bound::Included(start) => start,
- Bound::Excluded(start) => start + 1,
- Bound::Unbounded => 0,
- };
-
- let end = match range.end_bound().cloned() {
- Bound::Included(end) => Some(end + 1),
- Bound::Excluded(end) => Some(end),
- Bound::Unbounded => None,
- };
-
- // If range is empty, return Ok(0) directly.
- if let Some(end) = end {
- if end <= start {
- return Ok(Buffer::new());
- }
- }
-
- let mut bufs = Vec::new();
- let mut offset = start;
-
- loop {
- // TODO: use service preferred io size instead.
- let limit = end.map(|end| end - offset).unwrap_or(4 * 1024 * 1024)
as usize;
- let bs = self.inner.read_at_dyn(offset, limit).await?;
- let n = bs.remaining();
- bufs.push(bs);
- if n < limit {
- return Ok(bufs.into_iter().flatten().collect());
- }
-
- offset += n as u64;
- if Some(offset) == end {
- return Ok(bufs.into_iter().flatten().collect());
- }
- }
+ let bufs: Vec<_> = self.into_stream(range).try_collect().await?;
+ Ok(bufs.into_iter().flatten().collect())
}
/// Read all data from reader into given [`BufMut`].
@@ -115,46 +83,28 @@ impl Reader {
buf: &mut impl BufMut,
range: impl RangeBounds<u64>,
) -> Result<usize> {
- let start = match range.start_bound().cloned() {
- Bound::Included(start) => start,
- Bound::Excluded(start) => start + 1,
- Bound::Unbounded => 0,
- };
-
- let end = match range.end_bound().cloned() {
- Bound::Included(end) => Some(end + 1),
- Bound::Excluded(end) => Some(end),
- Bound::Unbounded => None,
- };
-
- // If range is empty, return Ok(0) directly.
- if let Some(end) = end {
- if end <= start {
- return Ok(0);
- }
- }
+ let mut stream = self.into_stream(range);
- let mut offset = start;
let mut read = 0;
-
loop {
- // TODO: use service preferred io size instead.
- let limit = end.map(|end| end - offset).unwrap_or(4 * 1024 * 1024)
as usize;
- let bs = self.inner.read_at_dyn(offset, limit).await?;
- let n = bs.remaining();
+ let Some(bs) = stream.try_next().await? else {
+ return Ok(read);
+ };
+ read += bs.len();
buf.put(bs);
- read += n as u64;
- if n < limit {
- return Ok(read as _);
- }
-
- offset += n as u64;
- if Some(offset) == end {
- return Ok(read as _);
- }
}
}
+ /// Create a buffer stream to read specific range from given reader.
+ pub fn into_stream(
+ &self,
+ range: impl RangeBounds<u64>,
+ ) -> impl Stream<Item = Result<Buffer>> + Unpin + MaybeSend + 'static {
+ let futs = into_stream::ReadFutureIterator::new(self.inner.clone(),
range);
+
+ stream::iter(futs).then(|f| f)
+ }
+
/// Convert reader into [`FuturesIoAsyncReader`] which implements
[`futures::AsyncRead`],
/// [`futures::AsyncSeek`] and [`futures::AsyncBufRead`].
#[inline]
@@ -170,6 +120,82 @@ impl Reader {
}
}
+pub mod into_stream {
+ use std::sync::atomic::Ordering;
+ use std::{
+ ops::{Bound, RangeBounds},
+ sync::{atomic::AtomicBool, Arc},
+ };
+
+ use crate::raw::oio::ReadDyn;
+ use crate::raw::*;
+ use crate::*;
+
+ pub struct ReadFutureIterator {
+ r: oio::Reader,
+
+ offset: u64,
+ end: Option<u64>,
+ finished: Arc<AtomicBool>,
+ }
+
+ impl ReadFutureIterator {
+ pub fn new(r: oio::Reader, range: impl RangeBounds<u64>) -> Self {
+ let start = match range.start_bound().cloned() {
+ Bound::Included(start) => start,
+ Bound::Excluded(start) => start + 1,
+ Bound::Unbounded => 0,
+ };
+ let end = match range.end_bound().cloned() {
+ Bound::Included(end) => Some(end + 1),
+ Bound::Excluded(end) => Some(end),
+ Bound::Unbounded => None,
+ };
+
+ ReadFutureIterator {
+ r,
+ offset: start,
+ end,
+ finished: Arc::default(),
+ }
+ }
+ }
+
+ impl Iterator for ReadFutureIterator {
+ type Item = BoxedFuture<'static, Result<Buffer>>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ if self.offset >= self.end.unwrap_or(u64::MAX) {
+ return None;
+ }
+ if self.finished.load(Ordering::Relaxed) {
+ return None;
+ }
+
+ let offset = self.offset;
+ let limit = self
+ .end
+ .map(|end| end - self.offset)
+ .unwrap_or(4 * 1024 * 1024) as usize;
+ let finished = self.finished.clone();
+ let r = self.r.clone();
+
+ // Update self.offset before building future.
+ self.offset += limit as u64;
+ let fut = async move {
+ let buf = r.read_at_static(offset, limit).await?;
+ if buf.len() < limit || limit == 0 {
+ // Update finished marked if buf is less than limit.
+ finished.store(true, Ordering::Relaxed);
+ }
+ Ok(buf)
+ };
+
+ Some(Box::pin(fut))
+ }
+ }
+}
+
pub mod into_futures_async_read {
use std::io;
use std::io::SeekFrom;
diff --git a/core/tests/behavior/utils.rs b/core/tests/behavior/utils.rs
index 63f8bce841..a35b4e86b4 100644
--- a/core/tests/behavior/utils.rs
+++ b/core/tests/behavior/utils.rs
@@ -23,6 +23,7 @@ use futures::Future;
use libtest_mimic::Failed;
use libtest_mimic::Trial;
use opendal::raw::tests::TEST_RUNTIME;
+use opendal::raw::*;
use opendal::*;
use rand::distributions::uniform::SampleRange;
use rand::prelude::*;
@@ -61,7 +62,7 @@ pub fn gen_offset_length(size: usize) -> (u64, u64) {
/// Build a new async trail as a test case.
pub fn build_async_trial<F, Fut>(name: &str, op: &Operator, f: F) -> Trial
where
- F: FnOnce(Operator) -> Fut + Send + 'static,
+ F: FnOnce(Operator) -> Fut + MaybeSend + 'static,
Fut: Future<Output = anyhow::Result<()>>,
{
let handle = TEST_RUNTIME.handle().clone();
@@ -86,7 +87,7 @@ macro_rules! async_trials {
/// Build a new async trail as a test case.
pub fn build_blocking_trial<F>(name: &str, op: &Operator, f: F) -> Trial
where
- F: FnOnce(BlockingOperator) -> anyhow::Result<()> + Send + 'static,
+ F: FnOnce(BlockingOperator) -> anyhow::Result<()> + MaybeSend + 'static,
{
let op = op.blocking();
