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 2024/05/31 15:03:11 UTC
(opendal) 02/06: Build pass
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch streaming-read
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit cdee6e7ab73bfd95049a26fd02d62740ed0b822f
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri May 31 22:44:21 2024 +0800
Build pass
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
bin/ofs/Cargo.lock | 4 +-
core/src/layers/retry.rs | 43 ++--
core/src/layers/timeout.rs | 2 +-
core/src/layers/type_eraser.rs | 2 +-
core/src/raw/futures_util.rs | 11 +
core/src/raw/http_util/client.rs | 89 +++++++-
core/src/raw/oio/read/api.rs | 29 +--
.../s3/mod.rs => raw/oio/read/from_stream.rs} | 33 ++-
core/src/raw/oio/read/mod.rs | 4 +
core/src/raw/ops.rs | 28 ++-
core/src/services/s3/backend.rs | 23 ++-
core/src/services/s3/core.rs | 4 +-
core/src/services/s3/mod.rs | 2 +-
core/src/types/buffer.rs | 6 +-
core/src/types/operator/operator.rs | 2 +-
core/src/types/operator/operator_futures.rs | 2 +-
core/src/types/read/buffer_stream.rs | 223 ++++++++++++++++-----
core/src/types/read/futures_async_reader.rs | 31 ++-
core/src/types/read/futures_bytes_stream.rs | 11 +-
core/src/types/read/reader.rs | 33 ++-
20 files changed, 436 insertions(+), 146 deletions(-)
diff --git a/bin/ofs/Cargo.lock b/bin/ofs/Cargo.lock
index 2ba0e09aeb..3cbe072632 100644
--- a/bin/ofs/Cargo.lock
+++ b/bin/ofs/Cargo.lock
@@ -331,9 +331,9 @@ dependencies = [
[[package]]
name = "crc32c"
-version = "0.6.5"
+version = "0.6.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89254598aa9b9fa608de44b3ae54c810f0f06d755e24c50177f1f8f31ff50ce2"
+checksum = "0227b9f93e535d49bc7ce914c066243424ce85ed90864cebd0874b184e9b6947"
dependencies = [
"rustc_version",
]
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 95ec1da93f..06682ef093 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -656,29 +656,28 @@ impl<R, I> RetryWrapper<R, I> {
}
impl<R: oio::Read, I: RetryInterceptor> oio::Read for RetryWrapper<R, I> {
+ /// FIXME: we should bring retry back.
async fn read(&mut self) -> Result<Buffer> {
- {
- || {
- self.inner
- .as_ref()
- .expect("inner must be valid")
- .read_at(offset, size)
- }
- }
- .retry(&self.builder)
- .when(|e| e.is_temporary())
- .notify(|err, dur| {
- self.notify.intercept(
- err,
- dur,
- &[
- ("operation", ReadOperation::Read.into_static()),
- ("path", &self.path),
- ],
- )
- })
- .await
- .map_err(|e| e.set_persistent())
+ self.inner
+ .as_mut()
+ .expect("inner must be valid")
+ .read()
+ .await
+ // { || self.inner.as_ref().expect("inner must be valid").read() }
+ // .retry(&self.builder)
+ // .when(|e| e.is_temporary())
+ // .notify(|err, dur| {
+ // self.notify.intercept(
+ // err,
+ // dur,
+ // &[
+ // ("operation", ReadOperation::Read.into_static()),
+ // ("path", &self.path),
+ // ],
+ // )
+ // })
+ // .await
+ // .map_err(|e| e.set_persistent())
}
}
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index d2f0361931..afc93bff18 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -288,7 +288,7 @@ impl<R> TimeoutWrapper<R> {
impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
async fn read(&mut self) -> Result<Buffer> {
- let fut = self.inner.read_at(offset, size);
+ let fut = self.inner.read();
Self::io_timeout(self.timeout, ReadOperation::Read.into_static(), fut).await
}
}
diff --git a/core/src/layers/type_eraser.rs b/core/src/layers/type_eraser.rs
index 72eee9d0e5..eab68f67ad 100644
--- a/core/src/layers/type_eraser.rs
+++ b/core/src/layers/type_eraser.rs
@@ -68,7 +68,7 @@ impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
self.inner
.read(path, args)
.await
- .map(|(rp, r)| (rp, Arc::new(r) as oio::Reader))
+ .map(|(rp, r)| (rp, Box::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 0010c2946c..4d5907cd5e 100644
--- a/core/src/raw/futures_util.rs
+++ b/core/src/raw/futures_util.rs
@@ -122,6 +122,17 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
self.results.clear();
}
+ /// Check if there are remaining space to push new tasks.
+ #[inline]
+ pub fn has_remaining(&self) -> bool {
+ self.tasks.len() < self.tasks.capacity()
+ }
+
+ #[inline]
+ pub fn has_result(&self) -> bool {
+ !self.results.is_empty()
+ }
+
/// Execute the task with given input.
///
/// - Execute the task in the current thread if is not concurrent.
diff --git a/core/src/raw/http_util/client.rs b/core/src/raw/http_util/client.rs
index 4ecd743ead..718b8e4e1e 100644
--- a/core/src/raw/http_util/client.rs
+++ b/core/src/raw/http_util/client.rs
@@ -24,12 +24,14 @@ use std::str::FromStr;
use bytes::Buf;
use bytes::Bytes;
-use futures::TryStreamExt;
+use futures::{StreamExt, TryStreamExt};
use http::Request;
use http::Response;
use super::parse_content_encoding;
use super::parse_content_length;
+use crate::raw::oio;
+use crate::raw::oio::from_stream;
use crate::*;
/// HttpClient that used across opendal.
@@ -161,6 +163,91 @@ impl HttpClient {
Ok(resp)
}
+
+ /// Fetch a request in async way.
+ pub async fn fetch(&self, req: Request<Buffer>) -> Result<Response<oio::Reader>> {
+ // Uri stores all string alike data in `Bytes` which means
+ // the clone here is cheap.
+ let uri = req.uri().clone();
+ let is_head = req.method() == http::Method::HEAD;
+
+ let (parts, body) = req.into_parts();
+
+ let mut req_builder = self
+ .client
+ .request(
+ parts.method,
+ reqwest::Url::from_str(&uri.to_string()).expect("input request url must be valid"),
+ )
+ .headers(parts.headers);
+
+ // Client under wasm doesn't support set version.
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ req_builder = req_builder.version(parts.version);
+ }
+
+ // Don't set body if body is empty.
+ if !body.is_empty() {
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ req_builder = req_builder.body(reqwest::Body::wrap_stream(body))
+ }
+ #[cfg(target_arch = "wasm32")]
+ {
+ req_builder = req_builder.body(reqwest::Body::from(body.to_bytes()))
+ }
+ }
+
+ let mut resp = req_builder.send().await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "send http request")
+ .with_operation("http_util::Client::send")
+ .with_context("url", uri.to_string())
+ .with_temporary(is_temporary_error(&err))
+ .set_source(err)
+ })?;
+
+ // Get content length from header so that we can check it.
+ //
+ // - If the request method is HEAD, we will ignore content length.
+ // - If response contains content_encoding, we should omit it's content length.
+ let content_length = if is_head || parse_content_encoding(resp.headers())?.is_some() {
+ None
+ } else {
+ parse_content_length(resp.headers())?
+ };
+
+ let mut hr = Response::builder()
+ .status(resp.status())
+ // Insert uri into response extension so that we can fetch
+ // it later.
+ .extension(uri.clone());
+
+ // Response builder under wasm doesn't support set version.
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ hr = hr.version(resp.version());
+ }
+
+ // Swap headers directly instead of copy the entire map.
+ mem::swap(hr.headers_mut().unwrap(), resp.headers_mut());
+
+ let bs: oio::Reader = Box::new(from_stream(
+ resp.bytes_stream()
+ .try_filter(|v| future::ready(!v.is_empty()))
+ .map_ok(|v| Buffer::from(v))
+ .map_err(move |err| {
+ Error::new(ErrorKind::Unexpected, "read data from http response")
+ .with_operation("http_util::Client::send")
+ .with_context("url", uri.to_string())
+ .with_temporary(is_temporary_error(&err))
+ .set_source(err)
+ }),
+ ));
+
+ let resp = hr.body(bs).expect("response must build succeed");
+ Ok(resp)
+ }
}
#[inline]
diff --git a/core/src/raw/oio/read/api.rs b/core/src/raw/oio/read/api.rs
index b6d13a346f..7241263605 100644
--- a/core/src/raw/oio/read/api.rs
+++ b/core/src/raw/oio/read/api.rs
@@ -18,11 +18,11 @@
use std::fmt::Display;
use std::fmt::Formatter;
use std::mem;
-use std::ops::Deref;
+use std::ops::{Deref, DerefMut};
use std::sync::Arc;
use bytes::Bytes;
-use futures::Future;
+use futures::{Future, Stream, StreamExt};
use crate::raw::*;
use crate::*;
@@ -80,12 +80,23 @@ pub type Reader = Box<dyn ReadDyn>;
/// once, at the outermost level of our API.
pub trait Read: Unpin + Send + Sync {
/// Read at the given offset with the given size.
- fn read(&mut self)
- -> impl Future<Output = Result<Buffer>> + MaybeSend;
+ fn read(&mut self) -> impl Future<Output = Result<Buffer>> + MaybeSend;
+
+ async fn read_all(&mut self) -> Result<Buffer> {
+ let mut bufs = vec![];
+ loop {
+ match self.read().await {
+ Ok(buf) if buf.is_empty() => break,
+ Ok(buf) => bufs.push(buf),
+ Err(err) => return Err(err),
+ }
+ }
+ Ok(bufs.into_iter().flatten().collect())
+ }
}
impl Read for () {
- async fn read(&self) -> Result<Buffer> {
+ async fn read(&mut self) -> Result<Buffer> {
Err(Error::new(
ErrorKind::Unsupported,
"output reader doesn't support read",
@@ -126,13 +137,7 @@ impl<T: Read + ?Sized> ReadDyn for T {
/// upon `&mut T` instead of `&mut Box<T>`. The later could result in infinite recursion.
impl<T: ReadDyn + ?Sized> Read for Box<T> {
async fn read(&mut self) -> Result<Buffer> {
- self.deref().read_dyn().await
- }
-}
-
-impl<T: ReadDyn + ?Sized> Read for Arc<T> {
- async fn read(&mut self) -> Result<Buffer> {
- self.deref().read_dyn().await
+ self.deref_mut().read_dyn().await
}
}
diff --git a/core/src/services/s3/mod.rs b/core/src/raw/oio/read/from_stream.rs
similarity index 57%
copy from core/src/services/s3/mod.rs
copy to core/src/raw/oio/read/from_stream.rs
index 01a986b6bb..4f687aa17f 100644
--- a/core/src/services/s3/mod.rs
+++ b/core/src/raw/oio/read/from_stream.rs
@@ -15,12 +15,29 @@
// specific language governing permissions and limitations
// under the License.
-mod backend;
-pub use backend::S3Builder as S3;
-pub use backend::S3Config;
+use crate::raw::oio::Read;
+use crate::*;
+use futures::Stream;
+use futures::StreamExt;
-mod core;
-mod error;
-mod lister;
-mod reader;
-mod writer;
+pub fn from_stream<S>(stream: S) -> FromStream<S>
+where
+ S: Stream<Item = Result<Buffer>> + Send + Sync + Unpin + 'static,
+{
+ FromStream(stream)
+}
+
+pub struct FromStream<S>(S);
+
+impl<S> Read for FromStream<S>
+where
+ S: Stream<Item = Result<Buffer>> + Send + Sync + Unpin + 'static,
+{
+ async fn read(&mut self) -> Result<Buffer> {
+ match self.0.next().await {
+ Some(Ok(buf)) => Ok(buf),
+ Some(Err(err)) => Err(err),
+ None => Ok(Buffer::new()),
+ }
+ }
+}
diff --git a/core/src/raw/oio/read/mod.rs b/core/src/raw/oio/read/mod.rs
index 5ec4114c51..39a4e2fb46 100644
--- a/core/src/raw/oio/read/mod.rs
+++ b/core/src/raw/oio/read/mod.rs
@@ -22,3 +22,7 @@ pub use api::Read;
pub use api::ReadDyn;
pub use api::ReadOperation;
pub use api::Reader;
+
+mod from_stream;
+pub use from_stream::from_stream;
+pub use from_stream::FromStream;
diff --git a/core/src/raw/ops.rs b/core/src/raw/ops.rs
index e22b25f31e..00ed8fe845 100644
--- a/core/src/raw/ops.rs
+++ b/core/src/raw/ops.rs
@@ -300,6 +300,7 @@ impl BatchOperation {
/// Args for `read` operation.
#[derive(Debug, Clone, Default)]
pub struct OpRead {
+ range: BytesRange,
if_match: Option<String>,
if_none_match: Option<String>,
override_content_type: Option<String>,
@@ -315,6 +316,17 @@ impl OpRead {
Self::default()
}
+ /// Set the range of the option
+ pub fn with_range(mut self, range: BytesRange) -> Self {
+ self.range = range;
+ self
+ }
+
+ /// Get range from option
+ pub fn range(&self) -> BytesRange {
+ self.range
+ }
+
/// Sets the content-disposition header that should be send back by the remote read operation.
pub fn with_override_content_disposition(mut self, content_disposition: &str) -> Self {
self.override_content_disposition = Some(content_disposition.into());
@@ -411,10 +423,6 @@ impl OpRead {
/// Args for reader operation.
#[derive(Debug, Clone)]
pub struct OpReader {
- /// The range of the read request.
- ///
- /// Not available for `reader`.
- range: BytesRange,
/// The concurrent requests that reader can send.
concurrent: usize,
/// The chunk size of each request.
@@ -426,7 +434,6 @@ pub struct OpReader {
impl Default for OpReader {
fn default() -> Self {
Self {
- range: BytesRange::default(),
concurrent: 1,
chunk: None,
gap: None,
@@ -440,17 +447,6 @@ impl OpReader {
Self::default()
}
- /// Set the range of the option
- pub fn with_range(mut self, range: BytesRange) -> Self {
- self.range = range;
- self
- }
-
- /// Get range from option
- pub fn range(&self) -> BytesRange {
- self.range
- }
-
/// Set the concurrent of the option
pub fn with_concurrent(mut self, concurrent: usize) -> Self {
self.concurrent = concurrent.max(1);
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 2f3a49b1f2..20500daa57 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -26,7 +26,7 @@ use std::sync::Arc;
use base64::prelude::BASE64_STANDARD;
use base64::Engine;
use bytes::Buf;
-use http::StatusCode;
+use http::{Response, StatusCode};
use log::debug;
use log::warn;
use md5::Digest;
@@ -44,9 +44,9 @@ use super::core::*;
use super::error::parse_error;
use super::error::parse_s3_error_code;
use super::lister::S3Lister;
-use super::reader::S3Reader;
use super::writer::S3Writer;
use super::writer::S3Writers;
+use crate::raw::oio::Read;
use crate::raw::*;
use crate::*;
@@ -1019,7 +1019,7 @@ pub struct S3Backend {
}
impl Access for S3Backend {
- type Reader = S3Reader;
+ type Reader = oio::Reader;
type Writer = S3Writers;
type Lister = oio::PageLister<S3Lister>;
type BlockingReader = ();
@@ -1099,10 +1099,19 @@ impl Access for S3Backend {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- Ok((
- RpRead::default(),
- S3Reader::new(self.core.clone(), path, args),
- ))
+ let resp = self.core.s3_get_object(path, args.range(), &args).await?;
+
+ let status = resp.status();
+ match status {
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ Ok((RpRead::default(), resp.into_body()))
+ }
+ _ => {
+ let (part, mut body) = resp.into_parts();
+ let buf = body.read_all().await?;
+ Err(parse_error(Response::from_parts(part, buf)).await?)
+ }
+ }
}
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
diff --git a/core/src/services/s3/core.rs b/core/src/services/s3/core.rs
index be4f7c34f0..b82918d4cc 100644
--- a/core/src/services/s3/core.rs
+++ b/core/src/services/s3/core.rs
@@ -398,12 +398,12 @@ impl S3Core {
path: &str,
range: BytesRange,
args: &OpRead,
- ) -> Result<Response<Buffer>> {
+ ) -> Result<Response<oio::Reader>> {
let mut req = self.s3_get_object_request(path, range, args)?;
self.sign(&mut req).await?;
- self.send(req).await
+ self.client.fetch(req).await
}
pub fn s3_put_object_request(
diff --git a/core/src/services/s3/mod.rs b/core/src/services/s3/mod.rs
index 01a986b6bb..12205efc42 100644
--- a/core/src/services/s3/mod.rs
+++ b/core/src/services/s3/mod.rs
@@ -22,5 +22,5 @@ pub use backend::S3Config;
mod core;
mod error;
mod lister;
-mod reader;
+// mod reader;
mod writer;
diff --git a/core/src/types/buffer.rs b/core/src/types/buffer.rs
index f72e9f1af1..f23e4eb391 100644
--- a/core/src/types/buffer.rs
+++ b/core/src/types/buffer.rs
@@ -19,6 +19,7 @@ use std::collections::VecDeque;
use std::convert::Infallible;
use std::fmt::Debug;
use std::fmt::Formatter;
+use std::future::Future;
use std::io::IoSlice;
use std::mem;
use std::ops::Bound;
@@ -32,7 +33,10 @@ use bytes::Buf;
use bytes::BufMut;
use bytes::Bytes;
use bytes::BytesMut;
-use futures::Stream;
+use futures::{Stream, TryStreamExt};
+
+use crate::raw::MaybeSend;
+use crate::*;
/// Buffer is a wrapper of contiguous `Bytes` and non-contiguous `[Bytes]`.
///
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 4685bb64e2..e564127f0b 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -570,7 +570,7 @@ impl Operator {
}
let path = Arc::new(path);
- let range = options.range();
+ let range = args.range();
let r = Reader::create(inner, path, args, options).await?;
let buf = r.read(range.to_range()).await?;
Ok(buf)
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index 832cae0194..4d16680087 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -224,7 +224,7 @@ impl<F: Future<Output = Result<Buffer>>> FutureRead<F> {
/// Set the range header for this operation.
pub fn range(self, range: impl RangeBounds<u64>) -> Self {
- self.map(|(args, op_reader)| (args, op_reader.with_range(range.into())))
+ self.map(|(args, op_reader)| (args.with_range(range.into()), op_reader))
}
/// Set the concurrent read task amount.
diff --git a/core/src/types/read/buffer_stream.rs b/core/src/types/read/buffer_stream.rs
index 21297367aa..488453f18c 100644
--- a/core/src/types/read/buffer_stream.rs
+++ b/core/src/types/read/buffer_stream.rs
@@ -15,11 +15,14 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::ops::Range;
use std::pin::Pin;
+use std::sync::Arc;
use std::task::Context;
use std::task::Poll;
+use crate::raw::oio::Read;
use futures::stream::Buffered;
use futures::stream::FusedStream;
use futures::stream::Iter;
@@ -30,87 +33,201 @@ use futures::StreamExt;
use crate::raw::*;
use crate::*;
-/// FutureIterator is an iterator that returns future generated by `Reader`'s
-/// read.
-struct FutureIterator {
- r: oio::Reader,
- chunk: Option<usize>,
+struct ReaderGenerator {
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
+ options: OpReader,
offset: u64,
end: u64,
}
-impl FutureIterator {
+impl ReaderGenerator {
#[inline]
- fn new(r: oio::Reader, chunk: Option<usize>, range: Range<u64>) -> Self {
- FutureIterator {
- r,
- chunk,
+ pub(crate) fn new(
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
+ options: OpReader,
+ range: Range<u64>,
+ ) -> Self {
+ Self {
+ acc,
+ path,
+ args,
+ options,
offset: range.start,
end: range.end,
}
}
-}
-
-impl Iterator for FutureIterator {
- type Item = BoxedFuture<'static, Result<Buffer>>;
- fn next(&mut self) -> Option<Self::Item> {
+ async fn next(&mut self) -> Result<Option<oio::Reader>> {
if self.offset >= self.end {
- return None;
+ return Ok(None);
}
let offset = self.offset;
- let mut limit = (self.end - self.offset) as usize;
- if let Some(chunk) = self.chunk {
- limit = limit.min(chunk)
+ let mut size = (self.end - self.offset) as usize;
+ if let Some(chunk) = self.options.chunk() {
+ size = size.min(chunk)
}
// Update self.offset before building future.
- self.offset += limit as u64;
- let r = self.r.clone();
- let fut = async move {
- let buf = r.read_at_dyn(offset, limit).await?;
- Ok(buf)
- };
+ self.offset += size as u64;
+ let args = self
+ .args
+ .clone()
+ .with_range(BytesRange::new(offset, Some(size as u64)));
+ let (_, r) = self.acc.read(&self.path, args).await?;
+ Ok(Some(r))
+ }
+}
- Some(Box::pin(fut))
+pub struct WholeReader {
+ generator: ReaderGenerator,
+ reader: Option<oio::Reader>,
+}
+
+impl WholeReader {
+ pub fn new(
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
+ options: OpReader,
+ range: Range<u64>,
+ ) -> Self {
+ let generator = ReaderGenerator::new(acc, path, args, options, range);
+
+ Self {
+ generator,
+ reader: None,
+ }
}
}
-/// BufferStream is a stream that returns [`Buffer`] from [`Reader`].
-///
-/// This stream will use concurrent read to fetch data from underlying storage.
-///
-/// # Notes
-///
-/// BufferStream uses `Buffered<Iter<ReadFutureIterator>>` internally,
-/// but we want to hide those details from users.
-pub struct BufferStream(Buffered<Iter<FutureIterator>>);
+impl oio::Read for WholeReader {
+ async fn read(&mut self) -> Result<Buffer> {
+ loop {
+ if self.reader.is_none() {
+ self.reader = self.generator.next().await?;
+ }
+ let Some(r) = self.reader.as_mut() else {
+ return Ok(Buffer::new());
+ };
-impl BufferStream {
- /// Create a new buffer stream from given reader.
- #[inline]
- pub(crate) fn new(r: oio::Reader, options: OpReader, range: Range<u64>) -> Self {
- let iter = FutureIterator::new(r, options.chunk(), range);
- let stream = stream::iter(iter).buffered(options.concurrent());
+ let buf = r.read().await?;
+ // Reset reader to None if this reader returns empty buffer.
+ if buf.is_empty() {
+ self.reader = None;
+ continue;
+ } else {
+ return Ok(buf);
+ }
+ }
+ }
+}
+
+pub struct ChunkedReader {
+ generator: ReaderGenerator,
+ tasks: ConcurrentTasks<oio::Reader, Buffer>,
+}
+
+impl ChunkedReader {
+ pub fn new(
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
+ options: OpReader,
+ range: Range<u64>,
+ ) -> Self {
+ let tasks = ConcurrentTasks::new(
+ args.executor().cloned(),
+ options.concurrent(),
+ |mut r: oio::Reader| {
+ Box::pin(async {
+ match r.read_all().await {
+ Ok(buf) => (r, Ok(buf)),
+ Err(err) => (r, Err(err)),
+ }
+ })
+ },
+ );
+ let generator = ReaderGenerator::new(acc, path, args, options, range);
- BufferStream(stream)
+ Self { generator, tasks }
}
}
-impl Stream for BufferStream {
- type Item = Result<Buffer>;
+impl oio::Read for ChunkedReader {
+ async fn read(&mut self) -> Result<Buffer> {
+ while self.tasks.has_remaining() {
+ if let Some(r) = self.generator.next().await? {
+ self.tasks.execute(r).await?;
+ }
+ if self.tasks.has_result() {
+ break;
+ }
+ }
+ Ok(self.tasks.next().await.transpose()?.unwrap_or_default())
+ }
+}
- #[inline]
- fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
- self.0.poll_next_unpin(cx)
+pub struct BufferStream {
+ state: State,
+}
+
+enum State {
+ Idle(Option<TwoWays<WholeReader, ChunkedReader>>),
+ Reading(BoxedStaticFuture<(TwoWays<WholeReader, ChunkedReader>, Result<Buffer>)>),
+}
+
+impl BufferStream {
+ pub fn new(
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
+ options: OpReader,
+ range: Range<u64>,
+ ) -> Self {
+ let reader = if options.chunk().is_some() {
+ TwoWays::Two(ChunkedReader::new(acc, path, args, options, range))
+ } else {
+ TwoWays::One(WholeReader::new(acc, path, args, options, range))
+ };
+ Self {
+ state: State::Idle(Some(reader)),
+ }
}
}
-impl FusedStream for BufferStream {
- fn is_terminated(&self) -> bool {
- self.0.is_terminated()
+impl Stream for BufferStream {
+ type Item = Result<Buffer>;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let mut this = self.get_mut();
+ loop {
+ match &mut this.state {
+ State::Idle(reader) => {
+ let mut reader = reader.take().unwrap();
+ let fut = async move {
+ let ret = reader.read().await;
+ (reader, ret)
+ };
+ this.state = State::Reading(Box::pin(fut));
+ }
+ State::Reading(fut) => {
+ let fut = fut.as_mut();
+ match fut.poll(cx) {
+ Poll::Ready((reader, buf)) => {
+ this.state = State::Idle(Some(reader));
+ return Poll::Ready(Some(buf));
+ }
+ Poll::Pending => return Poll::Pending,
+ }
+ }
+ }
+ }
}
}
@@ -126,14 +243,14 @@ mod tests {
#[test]
fn test_trait() {
- let v = BufferStream::new(Arc::new(Buffer::new()), OpReader::new(), 4..8);
+ let v = BufferStream::new(Box::new(Buffer::new()), OpReader::new(), 4..8);
let _: Box<dyn Unpin + MaybeSend + 'static> = Box::new(v);
}
#[test]
fn test_future_iterator() {
- let r: oio::Reader = Arc::new(Buffer::new());
+ let r: oio::Reader = Box::new(Buffer::new());
let it = FutureIterator::new(r.clone(), Some(1), 1..3);
let futs: Vec<_> = it.collect();
@@ -142,7 +259,7 @@ mod tests {
#[tokio::test]
async fn test_buffer_stream() {
- let r: oio::Reader = Arc::new(Buffer::from(vec![
+ let r: oio::Reader = Box::new(Buffer::from(vec![
Bytes::from("Hello"),
Bytes::from("World"),
]));
diff --git a/core/src/types/read/futures_async_reader.rs b/core/src/types/read/futures_async_reader.rs
index 2156fc36c0..26fd190f9f 100644
--- a/core/src/types/read/futures_async_reader.rs
+++ b/core/src/types/read/futures_async_reader.rs
@@ -19,6 +19,7 @@ use std::io;
use std::io::SeekFrom;
use std::ops::Range;
use std::pin::Pin;
+use std::sync::Arc;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
@@ -41,7 +42,9 @@ use crate::*;
///
/// FuturesAsyncReader also implements [`Unpin`], [`Send`] and [`Sync`]
pub struct FuturesAsyncReader {
- r: oio::Reader,
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
options: OpReader,
stream: BufferStream,
@@ -61,12 +64,26 @@ impl FuturesAsyncReader {
///
/// Extend this API to accept `impl RangeBounds`.
#[inline]
- pub(super) fn new(r: oio::Reader, options: OpReader, range: Range<u64>) -> Self {
+ pub(super) fn new(
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
+ options: OpReader,
+ range: Range<u64>,
+ ) -> Self {
let (start, end) = (range.start, range.end);
- let stream = BufferStream::new(r.clone(), options.clone(), range);
+ let stream = BufferStream::new(
+ acc.clone(),
+ path.clone(),
+ args.clone(),
+ options.clone(),
+ range,
+ );
FuturesAsyncReader {
- r,
+ acc,
+ path,
+ args,
options,
stream,
buf: Buffer::new(),
@@ -159,7 +176,9 @@ impl AsyncSeek for FuturesAsyncReader {
} else {
self.buf = Buffer::new();
self.stream = BufferStream::new(
- self.r.clone(),
+ self.acc.clone(),
+ self.path.clone(),
+ self.args.clone(),
self.options.clone(),
new_pos + self.start..self.end,
);
@@ -191,7 +210,7 @@ mod tests {
#[tokio::test]
async fn test_futures_async_read() {
- let r: oio::Reader = Arc::new(Buffer::from(vec![
+ let r: oio::Reader = Box::new(Buffer::from(vec![
Bytes::from("Hello"),
Bytes::from("World"),
]));
diff --git a/core/src/types/read/futures_bytes_stream.rs b/core/src/types/read/futures_bytes_stream.rs
index 7fc98007ba..dcbd7b8cdd 100644
--- a/core/src/types/read/futures_bytes_stream.rs
+++ b/core/src/types/read/futures_bytes_stream.rs
@@ -18,6 +18,7 @@
use std::io;
use std::ops::Range;
use std::pin::Pin;
+use std::sync::Arc;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
@@ -46,8 +47,14 @@ unsafe impl Sync for FuturesBytesStream {}
impl FuturesBytesStream {
/// NOTE: don't allow users to create FuturesStream directly.
#[inline]
- pub(crate) fn new(r: oio::Reader, options: OpReader, range: Range<u64>) -> Self {
- let stream = BufferStream::new(r, options, range);
+ pub(crate) fn new(
+ acc: Accessor,
+ path: Arc<String>,
+ args: OpRead,
+ options: OpReader,
+ range: Range<u64>,
+ ) -> Self {
+ let stream = BufferStream::new(acc, path, args, options, range);
FuturesBytesStream {
stream,
diff --git a/core/src/types/read/reader.rs b/core/src/types/read/reader.rs
index 867d09e6b7..5d85905b99 100644
--- a/core/src/types/read/reader.rs
+++ b/core/src/types/read/reader.rs
@@ -42,8 +42,7 @@ use crate::*;
pub struct Reader {
acc: Accessor,
path: Arc<String>,
-
- inner: oio::Reader,
+ args: OpRead,
options: OpReader,
/// Total size of the reader.
@@ -64,12 +63,10 @@ impl Reader {
args: OpRead,
options: OpReader,
) -> Result<Self> {
- let (_, r) = acc.read(&path, args).await?;
-
Ok(Reader {
acc,
path,
- inner: r,
+ args,
options,
size: Arc::new(AtomicContentLength::new()),
})
@@ -201,7 +198,13 @@ impl Reader {
/// Let's keep it inside for now.
async fn into_stream(self, range: impl RangeBounds<u64>) -> Result<BufferStream> {
let range = self.parse_range(range).await?;
- Ok(BufferStream::new(self.inner, self.options, range))
+ Ok(BufferStream::new(
+ self.acc,
+ self.path,
+ self.args,
+ self.options,
+ range,
+ ))
}
/// Convert reader into [`FuturesAsyncReader`] which implements [`futures::AsyncRead`],
@@ -267,7 +270,13 @@ impl Reader {
range: impl RangeBounds<u64>,
) -> Result<FuturesAsyncReader> {
let range = self.parse_range(range).await?;
- Ok(FuturesAsyncReader::new(self.inner, self.options, range))
+ Ok(FuturesAsyncReader::new(
+ self.acc,
+ self.path,
+ self.args,
+ self.options,
+ range,
+ ))
}
/// Convert reader into [`FuturesBytesStream`] which implements [`futures::Stream`].
@@ -322,7 +331,13 @@ impl Reader {
range: impl RangeBounds<u64>,
) -> Result<FuturesBytesStream> {
let range = self.parse_range(range).await?;
- Ok(FuturesBytesStream::new(self.inner, self.options, range))
+ Ok(FuturesBytesStream::new(
+ self.acc,
+ self.path,
+ self.args,
+ self.options,
+ range,
+ ))
}
}
@@ -344,7 +359,7 @@ mod tests {
let v = Reader {
acc: Arc::new(TypeEraseLayer.layer(())),
path: Arc::new("hello".to_string()),
- inner: Arc::new(Buffer::new()),
+ args: OpRead::new(),
options: OpReader::new(),
size: Arc::new(AtomicContentLength::new()),
};