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/03/11 07:54:48 UTC
(opendal) 02/24: Save work
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch read-async-trait
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit c53addaafb6705a985de091f2aa724329170a8f8
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 7 19:13:04 2024 +0800
Save work
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/concurrent_limit.rs | 12 +-
core/src/layers/error_context.rs | 20 +-
core/src/layers/logging.rs | 176 ++++----
core/src/layers/retry.rs | 207 +++-------
core/src/layers/timeout.rs | 64 +--
core/src/raw/enum_utils.rs | 72 ++--
core/src/raw/http_util/body.rs | 36 +-
core/src/raw/oio/read/api.rs | 50 ++-
core/src/raw/oio/read/buffer_reader.rs | 2 +-
core/src/raw/oio/read/file_read.rs | 537 ++++++++++---------------
core/src/raw/oio/read/futures_read.rs | 18 +-
core/src/raw/oio/read/into_read_from_stream.rs | 28 +-
core/src/raw/oio/read/into_streamable_read.rs | 21 +-
core/src/raw/oio/read/lazy_read.rs | 37 +-
core/src/raw/oio/read/range_read.rs | 537 ++++++++++---------------
core/src/raw/oio/read/tokio_read.rs | 38 +-
core/src/raw/oio/stream/api.rs | 10 +-
core/src/types/reader.rs | 103 ++---
18 files changed, 829 insertions(+), 1139 deletions(-)
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index e37b74ce04..1ec0eee9b8 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -257,16 +257,16 @@ impl<R> ConcurrentLimitWrapper<R> {
}
impl<R: oio::Read> oio::Read for ConcurrentLimitWrapper<R> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- self.inner.poll_read(cx, buf)
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.inner.read(buf).await
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- self.inner.poll_seek(cx, pos)
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ self.inner.seek(pos).await
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- self.inner.poll_next(cx)
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ self.inner.next().await
}
}
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 9a3a0130ef..8c6aabc4ac 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -349,8 +349,8 @@ pub struct ErrorContextWrapper<T> {
}
impl<T: oio::Read> oio::Read for ErrorContextWrapper<T> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- self.inner.poll_read(cx, buf).map_err(|err| {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.inner.read(buf).await.map_err(|err| {
err.with_operation(ReadOperation::Read)
.with_context("service", self.scheme)
.with_context("path", &self.path)
@@ -358,19 +358,21 @@ impl<T: oio::Read> oio::Read for ErrorContextWrapper<T> {
})
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- self.inner.poll_seek(cx, pos).map_err(|err| {
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ self.inner.seek(pos).await.map_err(|err| {
err.with_operation(ReadOperation::Seek)
.with_context("service", self.scheme)
.with_context("path", &self.path)
})
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- self.inner.poll_next(cx).map_err(|err| {
- err.with_operation(ReadOperation::Next)
- .with_context("service", self.scheme)
- .with_context("path", &self.path)
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ self.inner.next().await.map(|v| {
+ v.map_err(|err| {
+ err.with_operation(ReadOperation::Next)
+ .with_context("service", self.scheme)
+ .with_context("path", &self.path)
+ })
})
}
}
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 5dab01287e..36dfed943e 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -991,153 +991,113 @@ impl<R> Drop for LoggingReader<R> {
}
impl<R: oio::Read> oio::Read for LoggingReader<R> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
let buf_size = buf.len();
- match self.inner.poll_read(cx, buf) {
- Poll::Ready(res) => match res {
- Ok(n) => {
- self.read += n as u64;
- trace!(
- target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> buf size: {}B, read {}B ",
- self.ctx.scheme,
- ReadOperation::Read,
- self.path,
- self.read,
- buf_size,
- n
- );
- Poll::Ready(Ok(n))
- }
- Err(err) => {
- if let Some(lvl) = self.ctx.error_level(&err) {
- log!(
- target: LOGGING_TARGET,
- lvl,
- "service={} operation={} path={} read={} -> read failed: {}",
- self.ctx.scheme,
- ReadOperation::Read,
- self.path,
- self.read,
- self.ctx.error_print(&err)
- )
- }
- Poll::Ready(Err(err))
- }
- },
- Poll::Pending => {
+ match self.inner.read(buf).await {
+ Ok(n) => {
+ self.read += n as u64;
trace!(
target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> buf size: {}B, read pending",
+ "service={} operation={} path={} read={} -> buf size: {}B, read {}B ",
self.ctx.scheme,
ReadOperation::Read,
self.path,
self.read,
- buf_size
+ buf_size,
+ n
);
- Poll::Pending
+ Ok(n)
}
- }
- }
-
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
- match self.inner.poll_seek(cx, pos) {
- Poll::Ready(res) => match res {
- Ok(n) => {
- trace!(
+ Err(err) => {
+ if let Some(lvl) = self.ctx.error_level(&err) {
+ log!(
target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> seek to {pos:?}, current offset {n}",
+ lvl,
+ "service={} operation={} path={} read={} -> read failed: {}",
self.ctx.scheme,
- ReadOperation::Seek,
+ ReadOperation::Read,
self.path,
self.read,
- );
- Poll::Ready(Ok(n))
- }
- Err(err) => {
- if let Some(lvl) = self.ctx.error_level(&err) {
- log!(
- target: LOGGING_TARGET,
- lvl,
- "service={} operation={} path={} read={} -> seek to {pos:?} failed: {}",
- self.ctx.scheme,
- ReadOperation::Seek,
- self.path,
- self.read,
- self.ctx.error_print(&err),
- )
- }
- Poll::Ready(Err(err))
+ self.ctx.error_print(&err)
+ )
}
- },
- Poll::Pending => {
+ Err(err)
+ }
+ }
+ }
+
+ async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
+ match self.inner.seek(pos).await {
+ Ok(n) => {
trace!(
target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> seek to {pos:?} pending",
+ "service={} operation={} path={} read={} -> seek to {pos:?}, current offset {n}",
self.ctx.scheme,
ReadOperation::Seek,
self.path,
- self.read
+ self.read,
);
- Poll::Pending
+ Ok(n)
}
- }
- }
-
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- match self.inner.poll_next(cx) {
- Poll::Ready(res) => match res {
- Some(Ok(bs)) => {
- self.read += bs.len() as u64;
- trace!(
+ Err(err) => {
+ if let Some(lvl) = self.ctx.error_level(&err) {
+ log!(
target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> next returns {}B",
+ lvl,
+ "service={} operation={} path={} read={} -> seek to {pos:?} failed: {}",
self.ctx.scheme,
- ReadOperation::Next,
+ ReadOperation::Seek,
self.path,
self.read,
- bs.len()
- );
- Poll::Ready(Some(Ok(bs)))
- }
- Some(Err(err)) => {
- if let Some(lvl) = self.ctx.error_level(&err) {
- log!(
- target: LOGGING_TARGET,
- lvl,
- "service={} operation={} path={} read={} -> next failed: {}",
- self.ctx.scheme,
- ReadOperation::Next,
- self.path,
- self.read,
- self.ctx.error_print(&err),
- )
- }
- Poll::Ready(Some(Err(err)))
+ self.ctx.error_print(&err),
+ )
}
- None => {
- trace!(
+ Err(err)
+ }
+ }
+ }
+
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ match self.inner.next().await {
+ Some(Ok(bs)) => {
+ self.read += bs.len() as u64;
+ trace!(
+ target: LOGGING_TARGET,
+ "service={} operation={} path={} read={} -> next returns {}B",
+ self.ctx.scheme,
+ ReadOperation::Next,
+ self.path,
+ self.read,
+ bs.len()
+ );
+ Some(Ok(bs))
+ }
+ Some(Err(err)) => {
+ if let Some(lvl) = self.ctx.error_level(&err) {
+ log!(
target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> next returns None",
+ lvl,
+ "service={} operation={} path={} read={} -> next failed: {}",
self.ctx.scheme,
ReadOperation::Next,
self.path,
self.read,
- );
- Poll::Ready(None)
+ self.ctx.error_print(&err),
+ )
}
- },
- Poll::Pending => {
+ Some(Err(err))
+ }
+ None => {
trace!(
target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> next returns pending",
+ "service={} operation={} path={} read={} -> next returns None",
self.ctx.scheme,
ReadOperation::Next,
self.path,
- self.read
+ self.read,
);
- Poll::Pending
+ None
}
}
}
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index abef993128..945f2a515e 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -671,146 +671,61 @@ impl<R, I> RetryWrapper<R, I> {
}
impl<R: oio::Read, I: RetryInterceptor> oio::Read for RetryWrapper<R, I> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- if let Some(sleep) = self.sleep.as_mut() {
- ready!(sleep.poll_unpin(cx));
- self.sleep = None;
- }
-
- match ready!(self.inner.poll_read(cx, buf)) {
- Ok(v) => {
- self.current_backoff = None;
- Poll::Ready(Ok(v))
- }
- Err(err) if !err.is_temporary() => {
- self.current_backoff = None;
- Poll::Ready(Err(err))
- }
- Err(err) => {
- let backoff = match self.current_backoff.as_mut() {
- Some(backoff) => backoff,
- None => {
- self.current_backoff = Some(self.builder.build());
- self.current_backoff.as_mut().unwrap()
- }
- };
-
- match backoff.next() {
- None => {
- self.current_backoff = None;
- Poll::Ready(Err(err))
- }
- Some(dur) => {
- self.notify.intercept(
- &err,
- dur,
- &[
- ("operation", ReadOperation::Read.into_static()),
- ("path", &self.path),
- ],
- );
- self.sleep = Some(Box::pin(tokio::time::sleep(dur)));
- self.poll_read(cx, buf)
- }
- }
- }
- }
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ // { || self.inner.read(buf) }
+ // .retry(&self.builder)
+ // .when(|e| e.is_temporary())
+ // .notify(|err, dur| {
+ // self.notify.intercept(
+ // err,
+ // dur,
+ // &[
+ // ("operation", ReadOperation::Read.into_static()),
+ // ("path", &self.path),
+ // ],
+ // )
+ // })
+ // .map(|v| v.map_err(|e| e.set_persistent()))
+ // .await
+ todo!()
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
- if let Some(sleep) = self.sleep.as_mut() {
- ready!(sleep.poll_unpin(cx));
- self.sleep = None;
- }
-
- match ready!(self.inner.poll_seek(cx, pos)) {
- Ok(v) => {
- self.current_backoff = None;
- Poll::Ready(Ok(v))
- }
- Err(err) if !err.is_temporary() => {
- self.current_backoff = None;
- Poll::Ready(Err(err))
- }
- Err(err) => {
- let backoff = match self.current_backoff.as_mut() {
- Some(backoff) => backoff,
- None => {
- self.current_backoff = Some(self.builder.build());
- self.current_backoff.as_mut().unwrap()
- }
- };
-
- match backoff.next() {
- None => {
- self.current_backoff = None;
- Poll::Ready(Err(err))
- }
- Some(dur) => {
- self.notify.intercept(
- &err,
- dur,
- &[
- ("operation", ReadOperation::Seek.into_static()),
- ("path", &self.path),
- ],
- );
- self.sleep = Some(Box::pin(tokio::time::sleep(dur)));
- self.poll_seek(cx, pos)
- }
- }
- }
- }
+ async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
+ todo!()
+ // { || self.inner.seek(pos) }
+ // .retry(&self.builder)
+ // .when(|e| e.is_temporary())
+ // .notify(|err, dur| {
+ // self.notify.intercept(
+ // err,
+ // dur,
+ // &[
+ // ("operation", ReadOperation::Seek.into_static()),
+ // ("path", &self.path),
+ // ],
+ // )
+ // })
+ // .map(|v| v.map_err(|e| e.set_persistent()))
+ // .await
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- if let Some(sleep) = self.sleep.as_mut() {
- ready!(sleep.poll_unpin(cx));
- self.sleep = None;
- }
-
- match ready!(self.inner.poll_next(cx)) {
- None => {
- self.current_backoff = None;
- Poll::Ready(None)
- }
- Some(Ok(v)) => {
- self.current_backoff = None;
- Poll::Ready(Some(Ok(v)))
- }
- Some(Err(err)) if !err.is_temporary() => {
- self.current_backoff = None;
- Poll::Ready(Some(Err(err)))
- }
- Some(Err(err)) => {
- let backoff = match self.current_backoff.as_mut() {
- Some(backoff) => backoff,
- None => {
- self.current_backoff = Some(self.builder.build());
- self.current_backoff.as_mut().unwrap()
- }
- };
-
- match backoff.next() {
- None => {
- self.current_backoff = None;
- Poll::Ready(Some(Err(err)))
- }
- Some(dur) => {
- self.notify.intercept(
- &err,
- dur,
- &[
- ("operation", ReadOperation::Next.into_static()),
- ("path", &self.path),
- ],
- );
- self.sleep = Some(Box::pin(tokio::time::sleep(dur)));
- self.poll_next(cx)
- }
- }
- }
- }
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ todo!()
+ // { || self.inner.next() }
+ // .retry(&self.builder)
+ // .when(|e| e.is_temporary())
+ // .notify(|err, dur| {
+ // self.notify.intercept(
+ // err,
+ // dur,
+ // &[
+ // ("operation", ReadOperation::Next.into_static()),
+ // ("path", &self.path),
+ // ],
+ // )
+ // })
+ // .map(|v| v.map_err(|e| e.set_persistent()))
+ // .await
}
}
@@ -1263,11 +1178,11 @@ mod tests {
}
impl oio::Read for MockReader {
- fn poll_read(&mut self, _: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
let mut attempt = self.attempt.lock().unwrap();
*attempt += 1;
- Poll::Ready(match *attempt {
+ match *attempt {
1 => Err(
Error::new(ErrorKind::Unexpected, "retryable_error from reader")
.set_temporary(),
@@ -1288,25 +1203,25 @@ mod tests {
}
5 => Ok(0),
_ => unreachable!(),
- })
+ }
}
- fn poll_seek(&mut self, _: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
+ async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
self.pos = match pos {
io::SeekFrom::Current(n) => (self.pos as i64 + n) as u64,
io::SeekFrom::Start(n) => n,
io::SeekFrom::End(n) => (13 + n) as u64,
};
- Poll::Ready(Ok(self.pos))
+ Ok(self.pos)
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
let mut bs = vec![0; 1];
- match ready!(self.poll_read(cx, &mut bs)) {
- Ok(0) => Poll::Ready(None),
- Ok(v) => Poll::Ready(Some(Ok(Bytes::from(bs[..v].to_vec())))),
- Err(err) => Poll::Ready(Some(Err(err))),
+ match self.read(&mut bs).await {
+ Ok(0) => None,
+ Ok(v) => Some(Ok(Bytes::from(bs[..v].to_vec()))),
+ Err(err) => Some(Err(err)),
}
}
}
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index 6f322251ef..29fbbd26cb 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -286,6 +286,20 @@ impl<R> TimeoutWrapper<R> {
}
}
+ #[inline]
+ async fn io_timeout<F: Future<Output = Result<T>>, T>(
+ timeout: Duration,
+ op: &'static str,
+ fut: F,
+ ) -> Result<T> {
+ tokio::time::timeout(timeout, fut).await.map_err(|_| {
+ Error::new(ErrorKind::Unexpected, "io operation timeout reached")
+ .with_operation(op)
+ .with_context("timeout", timeout.as_secs_f64().to_string())
+ .set_temporary()
+ })?
+ }
+
#[inline]
fn poll_timeout(&mut self, cx: &mut Context<'_>, op: &'static str) -> Result<()> {
let sleep = self
@@ -308,28 +322,29 @@ impl<R> TimeoutWrapper<R> {
}
impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- self.poll_timeout(cx, ReadOperation::Read.into_static())?;
-
- let v = ready!(self.inner.poll_read(cx, buf));
- self.sleep = None;
- Poll::Ready(v)
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ let fut = self.inner.read(buf);
+ Self::io_timeout(self.timeout, ReadOperation::Read.into_static(), fut).await
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- self.poll_timeout(cx, ReadOperation::Seek.into_static())?;
-
- let v = ready!(self.inner.poll_seek(cx, pos));
- self.sleep = None;
- Poll::Ready(v)
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ let fut = self.inner.seek(pos);
+ Self::io_timeout(self.timeout, ReadOperation::Seek.into_static(), fut).await
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- self.poll_timeout(cx, ReadOperation::Next.into_static())?;
-
- let v = ready!(self.inner.poll_next(cx));
- self.sleep = None;
- Poll::Ready(v)
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ match tokio::time::timeout(self.timeout, self.inner.next()).await {
+ Ok(v) => v,
+ Err(err) => {
+ return Some(Err(Error::new(
+ ErrorKind::Unexpected,
+ "io operation timeout reached",
+ )
+ .with_operation(ReadOperation::Next.into_static())
+ .with_context("timeout", self.timeout.as_secs_f64().to_string())
+ .set_temporary()))
+ }
+ }
}
}
@@ -371,6 +386,7 @@ impl<R: oio::List> oio::List for TimeoutWrapper<R> {
#[cfg(test)]
mod tests {
+ use std::future::{pending, Future};
use std::io::SeekFrom;
use std::sync::Arc;
use std::task::Context;
@@ -434,16 +450,16 @@ mod tests {
struct MockReader;
impl oio::Read for MockReader {
- fn poll_read(&mut self, _: &mut Context<'_>, _: &mut [u8]) -> Poll<Result<usize>> {
- Poll::Pending
+ fn read(&mut self, _: &mut [u8]) -> impl Future<Output = Result<usize>> {
+ pending()
}
- fn poll_seek(&mut self, _: &mut Context<'_>, _: SeekFrom) -> Poll<Result<u64>> {
- Poll::Pending
+ fn seek(&mut self, _: SeekFrom) -> impl Future<Output = Result<u64>> {
+ pending()
}
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- Poll::Pending
+ fn next(&mut self) -> impl Future<Output = Option<Result<Bytes>>> {
+ pending()
}
}
diff --git a/core/src/raw/enum_utils.rs b/core/src/raw/enum_utils.rs
index 518ccc6740..fe871046f3 100644
--- a/core/src/raw/enum_utils.rs
+++ b/core/src/raw/enum_utils.rs
@@ -58,24 +58,24 @@ pub enum TwoWays<ONE, TWO> {
}
impl<ONE: oio::Read, TWO: oio::Read> oio::Read for TwoWays<ONE, TWO> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
match self {
- Self::One(v) => v.poll_read(cx, buf),
- Self::Two(v) => v.poll_read(cx, buf),
+ Self::One(v) => v.read(buf).await,
+ Self::Two(v) => v.read(buf).await,
}
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
match self {
- Self::One(v) => v.poll_seek(cx, pos),
- Self::Two(v) => v.poll_seek(cx, pos),
+ Self::One(v) => v.seek(pos).await,
+ Self::Two(v) => v.seek(pos).await,
}
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
match self {
- Self::One(v) => v.poll_next(cx),
- Self::Two(v) => v.poll_next(cx),
+ Self::One(v) => v.next().await,
+ Self::Two(v) => v.next().await,
}
}
}
@@ -139,27 +139,27 @@ pub enum ThreeWays<ONE, TWO, THREE> {
}
impl<ONE: oio::Read, TWO: oio::Read, THREE: oio::Read> oio::Read for ThreeWays<ONE, TWO, THREE> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
match self {
- Self::One(v) => v.poll_read(cx, buf),
- Self::Two(v) => v.poll_read(cx, buf),
- Self::Three(v) => v.poll_read(cx, buf),
+ Self::One(v) => v.read(buf).await,
+ Self::Two(v) => v.read(buf).await,
+ Self::Three(v) => v.read(buf).await,
}
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
match self {
- Self::One(v) => v.poll_seek(cx, pos),
- Self::Two(v) => v.poll_seek(cx, pos),
- Self::Three(v) => v.poll_seek(cx, pos),
+ Self::One(v) => v.seek(pos).await,
+ Self::Two(v) => v.seek(pos).await,
+ Self::Three(v) => v.seek(pos).await,
}
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
match self {
- Self::One(v) => v.poll_next(cx),
- Self::Two(v) => v.poll_next(cx),
- Self::Three(v) => v.poll_next(cx),
+ Self::One(v) => v.next().await,
+ Self::Two(v) => v.next().await,
+ Self::Three(v) => v.next().await,
}
}
}
@@ -241,30 +241,30 @@ where
THREE: oio::Read,
FOUR: oio::Read,
{
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
match self {
- Self::One(v) => v.poll_read(cx, buf),
- Self::Two(v) => v.poll_read(cx, buf),
- Self::Three(v) => v.poll_read(cx, buf),
- Self::Four(v) => v.poll_read(cx, buf),
+ Self::One(v) => v.read(buf).await,
+ Self::Two(v) => v.read(buf).await,
+ Self::Three(v) => v.read(buf).await,
+ Self::Four(v) => v.read(buf).await,
}
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
match self {
- Self::One(v) => v.poll_seek(cx, pos),
- Self::Two(v) => v.poll_seek(cx, pos),
- Self::Three(v) => v.poll_seek(cx, pos),
- Self::Four(v) => v.poll_seek(cx, pos),
+ Self::One(v) => v.seek(pos).await,
+ Self::Two(v) => v.seek(pos).await,
+ Self::Three(v) => v.seek(pos).await,
+ Self::Four(v) => v.seek(pos).await,
}
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
match self {
- Self::One(v) => v.poll_next(cx),
- Self::Two(v) => v.poll_next(cx),
- Self::Three(v) => v.poll_next(cx),
- Self::Four(v) => v.poll_next(cx),
+ Self::One(v) => v.next().await,
+ Self::Two(v) => v.next().await,
+ Self::Three(v) => v.next().await,
+ Self::Four(v) => v.next().await,
}
}
}
diff --git a/core/src/raw/http_util/body.rs b/core/src/raw/http_util/body.rs
index c73aaf3741..42cbb2e5df 100644
--- a/core/src/raw/http_util/body.rs
+++ b/core/src/raw/http_util/body.rs
@@ -25,7 +25,9 @@ use std::task::Poll;
use bytes::Buf;
use bytes::BufMut;
use bytes::Bytes;
+use futures::StreamExt;
+use crate::raw::oio::Read;
use crate::raw::*;
use crate::*;
@@ -160,9 +162,9 @@ impl IncomingAsyncBody {
}
impl oio::Read for IncomingAsyncBody {
- fn poll_read(&mut self, cx: &mut Context<'_>, mut buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, mut buf: &mut [u8]) -> Result<usize> {
if buf.is_empty() || self.size == Some(0) {
- return Poll::Ready(Ok(0));
+ return Ok(0);
}
// Avoid extra poll of next if we already have chunks.
@@ -170,7 +172,7 @@ impl oio::Read for IncomingAsyncBody {
chunk
} else {
loop {
- match ready!(self.inner.poll_next(cx)) {
+ match self.inner.next().await {
// It's possible for underlying stream to return empty bytes, we should continue
// to fetch next one.
Some(Ok(bs)) if bs.is_empty() => continue,
@@ -178,12 +180,12 @@ impl oio::Read for IncomingAsyncBody {
self.consumed += bs.len() as u64;
break bs;
}
- Some(Err(err)) => return Poll::Ready(Err(err)),
+ Some(Err(err)) => return Err(err),
None => {
if let Some(size) = self.size {
Self::check(size, self.consumed)?;
}
- return Poll::Ready(Ok(0));
+ return Ok(0);
}
}
}
@@ -196,28 +198,28 @@ impl oio::Read for IncomingAsyncBody {
self.chunk = Some(bs);
}
- Poll::Ready(Ok(amt))
+ Ok(amt)
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
- let (_, _) = (cx, pos);
+ async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
+ let (_) = (pos);
- Poll::Ready(Err(Error::new(
+ Err(Error::new(
ErrorKind::Unsupported,
"output reader doesn't support seeking",
- )))
+ ))
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
if self.size == Some(0) {
- return Poll::Ready(None);
+ return None;
}
if let Some(bs) = self.chunk.take() {
- return Poll::Ready(Some(Ok(bs)));
+ return Some(Ok(bs));
}
- let res = match ready!(self.inner.poll_next(cx)) {
+ let res = match self.inner.next().await {
Some(Ok(bs)) => {
self.consumed += bs.len() as u64;
Some(Ok(bs))
@@ -225,13 +227,15 @@ impl oio::Read for IncomingAsyncBody {
Some(Err(err)) => Some(Err(err)),
None => {
if let Some(size) = self.size {
- Self::check(size, self.consumed)?;
+ if let Err(err) = Self::check(size, self.consumed) {
+ return Some(Err(err));
+ }
}
None
}
};
- Poll::Ready(res)
+ res
}
}
diff --git a/core/src/raw/oio/read/api.rs b/core/src/raw/oio/read/api.rs
index 0d5c3e18f1..5d5d36b54d 100644
--- a/core/src/raw/oio/read/api.rs
+++ b/core/src/raw/oio/read/api.rs
@@ -92,12 +92,12 @@ pub type Reader = Box<dyn DynRead>;
/// is optional. We use `Read` to make users life easier.
pub trait Read: Unpin + Send + Sync {
/// Read bytes asynchronously.
- async fn read(&mut self, buf: &mut [u8]) -> Result<usize>;
+ fn read(&mut self, buf: &mut [u8]) -> impl Future<Output = Result<usize>> + Send;
/// Seek asynchronously.
///
/// Returns `Unsupported` error if underlying reader doesn't support seek.
- async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64>;
+ fn seek(&mut self, pos: io::SeekFrom) -> impl Future<Output = Result<u64>> + Send;
/// Stream [`Bytes`] from underlying reader.
///
@@ -106,7 +106,7 @@ pub trait Read: Unpin + Send + Sync {
/// This API exists for avoiding bytes copying inside async runtime.
/// Users can poll bytes from underlying reader and decide when to
/// read/consume them.
- async fn next(&mut self) -> Option<Result<Bytes>>;
+ fn next(&mut self) -> impl Future<Output = Option<Result<Bytes>>> + Send;
}
impl Read for () {
@@ -135,19 +135,19 @@ impl Read for () {
/// `Box<dyn Read>` won't implement `Read` automatically. To make Reader
/// work as expected, we must add this impl.
-impl<T: Read + ?Sized> Read for Box<T> {
- async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
- (**self).read(buf).await
- }
-
- async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
- (**self).seek(pos).await
- }
-
- async fn next(&mut self) -> Option<Result<Bytes>> {
- (**self).next().await
- }
-}
+// impl<T: Read + ?Sized> Read for Box<T> {
+// async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+// (**self).read(buf).await
+// }
+//
+// async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
+// (**self).seek(pos).await
+// }
+//
+// async fn next(&mut self) -> Option<Result<Bytes>> {
+// (**self).next().await
+// }
+// }
/// Impl ReadExt for all T: Read
impl<T: Read> ReadExt for T {}
@@ -206,7 +206,7 @@ where
}
pub trait DynRead: Unpin + Send + Sync {
- fn dyn_read(&mut self, buf: &mut [u8]) -> BoxedFuture<Result<usize>>;
+ fn dyn_read<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxedFuture<'a, Result<usize>>;
fn dyn_seek(&mut self, pos: io::SeekFrom) -> BoxedFuture<Result<u64>>;
@@ -214,7 +214,7 @@ pub trait DynRead: Unpin + Send + Sync {
}
impl<T: Read + ?Sized> DynRead for T {
- fn dyn_read(&mut self, buf: &mut [u8]) -> BoxedFuture<Result<usize>> {
+ fn dyn_read<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxedFuture<'a, Result<usize>> {
Box::pin(self.read(buf))
}
@@ -227,6 +227,20 @@ impl<T: Read + ?Sized> DynRead for T {
}
}
+impl<T: DynRead + ?Sized> Read for Box<T> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.dyn_read(buf).await
+ }
+
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ self.dyn_seek(pos).await
+ }
+
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ self.dyn_next().await
+ }
+}
+
/// BlockingReader is a boxed dyn `BlockingRead`.
pub type BlockingReader = Box<dyn BlockingRead>;
diff --git a/core/src/raw/oio/read/buffer_reader.rs b/core/src/raw/oio/read/buffer_reader.rs
index 6c3ac32505..83da9970dd 100644
--- a/core/src/raw/oio/read/buffer_reader.rs
+++ b/core/src/raw/oio/read/buffer_reader.rs
@@ -297,7 +297,7 @@ where
}
fn next(&mut self) -> Option<Result<Bytes>> {
- match self.fill_buf() {
+ match self.blocking_fill_buf() {
Ok(bytes) => {
if bytes.is_empty() {
return None;
diff --git a/core/src/raw/oio/read/file_read.rs b/core/src/raw/oio/read/file_read.rs
index 69bdbff788..1136c28312 100644
--- a/core/src/raw/oio/read/file_read.rs
+++ b/core/src/raw/oio/read/file_read.rs
@@ -44,25 +44,12 @@ pub struct FileReader<A: Accessor, R> {
size: Option<u64>,
cur: u64,
+ reader: Option<R>,
buf: oio::AdaptiveBuf,
- state: State<R>,
/// Do we need to reset our cursor?
seek_dirty: bool,
}
-enum State<R> {
- Idle,
- Send(BoxedStaticFuture<Result<(RpRead, R)>>),
- Read(R),
-}
-
-/// # Safety
-///
-/// wasm32 is a special target that we only have one event-loop for this state.
-unsafe impl<R> Send for State<R> {}
-/// Safety: State will only be accessed under &mut.
-unsafe impl<R> Sync for State<R> {}
-
impl<A, R> FileReader<A, R>
where
A: Accessor,
@@ -82,7 +69,7 @@ where
size: None,
cur: 0,
buf: oio::AdaptiveBuf::default(),
- state: State::<R>::Idle,
+ reader: None,
seek_dirty: false,
}
}
@@ -93,80 +80,76 @@ where
A: Accessor<Reader = R>,
R: oio::Read,
{
- fn read_future(&self) -> BoxedStaticFuture<Result<(RpRead, R)>> {
- let acc = self.acc.clone();
- let path = self.path.clone();
-
- // FileReader doesn't support range, we will always use full range to open a file.
- let op = self.op.clone().with_range(BytesRange::from(..));
+ async fn reader(&mut self) -> Result<&mut R> {
+ if self.reader.is_none() {
+ // FileReader doesn't support range, we will always use full range to open a file.
+ let op = self.op.clone().with_range(BytesRange::from(..));
+ let (_, r) = self.acc.read(&self.path, op).await?;
+ self.reader = Some(r);
+ }
- Box::pin(async move { acc.read(&path, op).await })
+ Ok(self.reader.as_mut().expect("reader must be valid"))
}
/// calculate_offset will make sure that the offset has been set.
- fn poll_offset(
- cx: &mut Context<'_>,
- r: &mut R,
- range: BytesRange,
- ) -> Poll<Result<(Option<u64>, Option<u64>)>> {
+ async fn offset(r: &mut R, range: BytesRange) -> Result<(Option<u64>, Option<u64>)> {
let (offset, size) = match (range.offset(), range.size()) {
(None, None) => (0, None),
(None, Some(size)) => {
- let start = ready!(r.poll_seek(cx, SeekFrom::End(-(size as i64))))?;
+ let start = r.seek(SeekFrom::End(-(size as i64))).await?;
(start, Some(size))
}
(Some(offset), None) => {
- let start = ready!(r.poll_seek(cx, SeekFrom::Start(offset)))?;
+ let start = r.seek(SeekFrom::Start(offset)).await?;
(start, None)
}
(Some(offset), Some(size)) => {
- let start = ready!(r.poll_seek(cx, SeekFrom::Start(offset)))?;
+ let start = r.seek(SeekFrom::Start(offset)).await?;
(start, Some(size))
}
};
- Poll::Ready(Ok((Some(offset), size)))
+ Ok((Some(offset), size))
}
- fn poll_seek_inner(
- cx: &mut Context<'_>,
+ async fn seek_inner(
r: &mut R,
offset: Option<u64>,
size: Option<u64>,
cur: u64,
pos: SeekFrom,
- ) -> Poll<Result<u64>> {
+ ) -> Result<u64> {
let offset = offset.expect("offset should be set for calculate_position");
match pos {
SeekFrom::Start(n) => {
// It's valid for user to seek outsides end of the file.
- r.poll_seek(cx, SeekFrom::Start(offset + n))
+ r.seek(SeekFrom::Start(offset + n)).await
}
SeekFrom::End(n) => {
let size =
size.expect("size should be set for calculate_position when seek with end");
if size as i64 + n < 0 {
- return Poll::Ready(Err(Error::new(
+ return Err(Error::new(
ErrorKind::InvalidInput,
"seek to a negative position is invalid",
)
- .with_context("position", format!("{pos:?}"))));
+ .with_context("position", format!("{pos:?}")));
}
// size is known, we can convert SeekFrom::End into SeekFrom::Start.
let pos = SeekFrom::Start(offset + (size as i64 + n) as u64);
- r.poll_seek(cx, pos)
+ r.seek(pos).await
}
SeekFrom::Current(n) => {
if cur as i64 + n < 0 {
- return Poll::Ready(Err(Error::new(
+ return Err(Error::new(
ErrorKind::InvalidInput,
"seek to a negative position is invalid",
)
- .with_context("position", format!("{pos:?}"))));
+ .with_context("position", format!("{pos:?}")));
}
let pos = SeekFrom::Start(offset + (cur as i64 + n) as u64);
- r.poll_seek(cx, pos)
+ r.seek(pos).await
}
}
}
@@ -177,6 +160,17 @@ where
A: Accessor<BlockingReader = R>,
R: oio::BlockingRead,
{
+ fn blocking_reader(&mut self) -> Result<&mut R> {
+ if self.reader.is_none() {
+ // FileReader doesn't support range, we will always use full range to open a file.
+ let op = self.op.clone().with_range(BytesRange::from(..));
+ let (_, r) = self.acc.blocking_read(&self.path, op)?;
+ self.reader = Some(r);
+ }
+
+ Ok(self.reader.as_mut().expect("reader must be valid"))
+ }
+
/// calculate_offset will make sure that the offset has been set.
fn calculate_offset(r: &mut R, range: BytesRange) -> Result<(Option<u64>, Option<u64>)> {
let (offset, size) = match (range.offset(), range.size()) {
@@ -198,7 +192,7 @@ where
Ok((Some(offset), size))
}
- fn seek_inner(
+ fn blocking_seek_inner(
r: &mut R,
offset: Option<u64>,
size: Option<u64>,
@@ -246,149 +240,106 @@ where
A: Accessor<Reader = R>,
R: oio::Read,
{
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- match &mut self.state {
- State::Idle => {
- self.state = State::Send(self.read_future());
- self.poll_read(cx, buf)
- }
- State::Send(fut) => {
- let (_, r) = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If send future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
- self.state = State::Read(r);
- self.poll_read(cx, buf)
- }
- State::Read(r) => {
- // We should know where to start read the data.
- if self.offset.is_none() {
- (self.offset, self.size) = ready!(Self::poll_offset(cx, r, self.op.range()))?;
- }
-
- let size = if let Some(size) = self.size {
- // Sanity check.
- if self.cur >= size {
- return Poll::Ready(Ok(0));
- }
- cmp::min(buf.len(), (size - self.cur) as usize)
- } else {
- buf.len()
- };
-
- match ready!(r.poll_read(cx, &mut buf[..size])) {
- Ok(0) => Poll::Ready(Ok(0)),
- Ok(n) => {
- self.cur += n as u64;
- Poll::Ready(Ok(n))
- }
- // We don't need to reset state here since it's ok to poll the same reader.
- Err(err) => Poll::Ready(Err(err)),
- }
- }
- }
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ todo!()
+ // let r = self.reader().await?;
+ //
+ // // We should know where to start read the data.
+ // if self.offset.is_none() {
+ // (self.offset, self.size) = Self::offset(r, self.op.range()).await?;
+ // }
+ //
+ // let size = if let Some(size) = self.size {
+ // // Sanity check.
+ // if self.cur >= size {
+ // return Ok(0);
+ // }
+ // cmp::min(buf.len(), (size - self.cur) as usize)
+ // } else {
+ // buf.len()
+ // };
+ //
+ // match r.read(&mut buf[..size]).await {
+ // Ok(0) => Ok(0),
+ // Ok(n) => {
+ // self.cur += n as u64;
+ // Ok(n)
+ // }
+ // // We don't need to reset state here since it's ok to poll the same reader.
+ // Err(err) => Err(err),
+ // }
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- match &mut self.state {
- State::Idle => {
- self.state = State::Send(self.read_future());
- self.poll_seek(cx, pos)
- }
- State::Send(fut) => {
- let (_, r) = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If send future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
- self.state = State::Read(r);
- self.poll_seek(cx, pos)
- }
- State::Read(r) => {
- // We should know where to start read the data.
- if self.offset.is_none() {
- (self.offset, self.size) = ready!(Self::poll_offset(cx, r, self.op.range()))?;
- }
-
- // Fetch size when seek end.
- let current_offset = self.offset.unwrap() + self.cur;
- if matches!(pos, SeekFrom::End(_)) && self.size.is_none() {
- let size = ready!(r.poll_seek(cx, SeekFrom::End(0)))?;
- self.size = Some(size - self.offset.unwrap());
- self.seek_dirty = true;
- }
- if self.seek_dirty {
- // Reset cursor.
- ready!(r.poll_seek(cx, SeekFrom::Start(current_offset)))?;
- self.seek_dirty = false;
- }
-
- let pos = ready!(Self::poll_seek_inner(
- cx,
- r,
- self.offset,
- self.size,
- self.cur,
- pos
- ))?;
- self.cur = pos - self.offset.unwrap();
- Poll::Ready(Ok(self.cur))
- }
- }
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ todo!()
+ // let r = self.reader().await?;
+ //
+ // // We should know where to start read the data.
+ // if self.offset.is_none() {
+ // (self.offset, self.size) = Self::offset(r, self.op.range()).await?;
+ // }
+ //
+ // // Fetch size when seek end.
+ // let current_offset = self.offset.unwrap() + self.cur;
+ // if matches!(pos, SeekFrom::End(_)) && self.size.is_none() {
+ // let size = r.seek(SeekFrom::End(0)).await?;
+ // self.size = Some(size - self.offset.unwrap());
+ // self.seek_dirty = true;
+ // }
+ // if self.seek_dirty {
+ // // Reset cursor.
+ // r.seek(SeekFrom::Start(current_offset)).await?;
+ // self.seek_dirty = false;
+ // }
+ //
+ // let pos = Self::seek_inner(r, self.offset, self.size, self.cur, pos).await?;
+ // self.cur = pos - self.offset.unwrap();
+ // Ok(self.cur)
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- match &mut self.state {
- State::Idle => {
- self.state = State::Send(self.read_future());
- self.poll_next(cx)
- }
- State::Send(fut) => {
- let (_, r) = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If send future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
- self.state = State::Read(r);
- self.poll_next(cx)
- }
- State::Read(r) => {
- // We should know where to start read the data.
- if self.offset.is_none() {
- (self.offset, self.size) = ready!(Self::poll_offset(cx, r, self.op.range()))?;
- }
-
- self.buf.reserve();
-
- let mut buf = self.buf.initialized_mut();
- let buf = buf.initialized_mut();
-
- let size = if let Some(size) = self.size {
- // Sanity check.
- if self.cur >= size {
- return Poll::Ready(None);
- }
- cmp::min(buf.len(), (size - self.cur) as usize)
- } else {
- buf.len()
- };
-
- match ready!(r.poll_read(cx, &mut buf[..size])) {
- Ok(0) => Poll::Ready(None),
- Ok(n) => {
- self.cur += n as u64;
- self.buf.record(n);
- Poll::Ready(Some(Ok(self.buf.split(n))))
- }
- // We don't need to reset state here since it's ok to poll the same reader.
- Err(err) => Poll::Ready(Some(Err(err))),
- }
- }
- }
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ todo!()
+ // let cur = self.cur;
+ // let size = self.size;
+ //
+ // let r = match self.reader().await {
+ // Ok(r) => r,
+ // Err(err) => return Some(Err(err)),
+ // };
+ //
+ // // We should know where to start read the data.
+ // if self.offset.is_none() {
+ // (self.offset, self.size) = match Self::offset(r, self.op.range()).await {
+ // Ok((offset, size)) => (offset, size),
+ // Err(err) => return Some(Err(err)),
+ // }
+ // }
+ //
+ // self.buf.reserve();
+ //
+ // let mut buf = self.buf.initialized_mut();
+ // let buf = buf.initialized_mut();
+ //
+ // let size = if let Some(size) = size {
+ // // Sanity check.
+ // if cur >= size {
+ // return None;
+ // }
+ // cmp::min(buf.len(), (size - cur) as usize)
+ // } else {
+ // buf.len()
+ // };
+ //
+ // match r.read(&mut buf[..size]).await {
+ // Ok(0) => None,
+ // Ok(n) => {
+ // self.cur += n as u64;
+ // self.buf.record(n);
+ // Some(Ok(self.buf.split(n)))
+ // }
+ // // We don't need to reset state here since it's ok to poll the same reader.
+ // Err(err) => Some(Err(err)),
+ // }
}
}
@@ -398,144 +349,102 @@ where
R: oio::BlockingRead,
{
fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
- match &mut self.state {
- State::Idle => {
- // FileReader doesn't support range, we will always use full range to open a file.
- let op = self.op.clone().with_range(BytesRange::from(..));
-
- let (_, r) = self.acc.blocking_read(&self.path, op)?;
- self.state = State::Read(r);
- self.read(buf)
- }
-
- State::Read(r) => {
- // We should know where to start read the data.
- if self.offset.is_none() {
- (self.offset, self.size) = Self::calculate_offset(r, self.op.range())?;
- }
-
- let size = if let Some(size) = self.size {
- // Sanity check.
- if self.cur >= size {
- return Ok(0);
- }
- cmp::min(buf.len(), (size - self.cur) as usize)
- } else {
- buf.len()
- };
-
- match r.read(&mut buf[..size]) {
- Ok(0) => Ok(0),
- Ok(n) => {
- self.cur += n as u64;
- Ok(n)
- }
- // We don't need to reset state here since it's ok to poll the same reader.
- Err(err) => Err(err),
- }
- }
- State::Send(_) => {
- unreachable!(
- "It's invalid to go into State::Send for BlockingRead, please report this bug"
- )
- }
- }
+ todo!()
+ // let r = self.blocking_reader()?;
+ //
+ // // We should know where to start read the data.
+ // if self.offset.is_none() {
+ // (self.offset, self.size) = Self::calculate_offset(r, self.op.range())?;
+ // }
+ //
+ // let size = if let Some(size) = self.size {
+ // // Sanity check.
+ // if self.cur >= size {
+ // return Ok(0);
+ // }
+ // cmp::min(buf.len(), (size - self.cur) as usize)
+ // } else {
+ // buf.len()
+ // };
+ //
+ // match r.read(&mut buf[..size]) {
+ // Ok(0) => Ok(0),
+ // Ok(n) => {
+ // self.cur += n as u64;
+ // Ok(n)
+ // }
+ // // We don't need to reset state here since it's ok to poll the same reader.
+ // Err(err) => Err(err),
+ // }
}
fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
- match &mut self.state {
- State::Idle => {
- // FileReader doesn't support range, we will always use full range to open a file.
- let op = self.op.clone().with_range(BytesRange::from(..));
-
- let (_, r) = self.acc.blocking_read(&self.path, op)?;
- self.state = State::Read(r);
- self.seek(pos)
- }
- State::Read(r) => {
- // We should know where to start read the data.
- if self.offset.is_none() {
- (self.offset, self.size) = Self::calculate_offset(r, self.op.range())?;
- }
- // Fetch size when seek end.
- let current_offset = self.offset.unwrap() + self.cur;
- if matches!(pos, SeekFrom::End(_)) && self.size.is_none() {
- let size = r.seek(SeekFrom::End(0))?;
- self.size = Some(size - self.offset.unwrap());
- self.seek_dirty = true;
- }
- if self.seek_dirty {
- // Reset cursor.
- r.seek(SeekFrom::Start(current_offset))?;
- self.seek_dirty = false;
- }
-
- let pos = Self::seek_inner(r, self.offset, self.size, self.cur, pos)?;
- self.cur = pos - self.offset.unwrap();
- Ok(self.cur)
- }
- State::Send(_) => {
- unreachable!(
- "It's invalid to go into State::Send for BlockingRead, please report this bug"
- )
- }
- }
+ todo!()
+
+ // let r = self.blocking_reader()?;
+ //
+ // // We should know where to start read the data.
+ // if self.offset.is_none() {
+ // (self.offset, self.size) = Self::calculate_offset(r, self.op.range())?;
+ // }
+ // // Fetch size when seek end.
+ // let current_offset = self.offset.unwrap() + self.cur;
+ // if matches!(pos, SeekFrom::End(_)) && self.size.is_none() {
+ // let size = r.seek(SeekFrom::End(0))?;
+ // self.size = Some(size - self.offset.unwrap());
+ // self.seek_dirty = true;
+ // }
+ // if self.seek_dirty {
+ // // Reset cursor.
+ // r.seek(SeekFrom::Start(current_offset))?;
+ // self.seek_dirty = false;
+ // }
+ //
+ // let pos = Self::blocking_seek_inner(r, self.offset, self.size, self.cur, pos)?;
+ // self.cur = pos - self.offset.unwrap();
+ // Ok(self.cur)
}
fn next(&mut self) -> Option<Result<Bytes>> {
- match &mut self.state {
- State::Idle => {
- // FileReader doesn't support range, we will always use full range to open a file.
- let op = self.op.clone().with_range(BytesRange::from(..));
-
- let r = match self.acc.blocking_read(&self.path, op) {
- Ok((_, r)) => r,
- Err(err) => return Some(Err(err)),
- };
- self.state = State::Read(r);
- self.next()
- }
-
- State::Read(r) => {
- // We should know where to start read the data.
- if self.offset.is_none() {
- (self.offset, self.size) = match Self::calculate_offset(r, self.op.range()) {
- Ok(v) => v,
- Err(err) => return Some(Err(err)),
- }
- }
-
- self.buf.reserve();
-
- let mut buf = self.buf.initialized_mut();
- let buf = buf.initialized_mut();
-
- let size = if let Some(size) = self.size {
- // Sanity check.
- if self.cur >= size {
- return None;
- }
- cmp::min(buf.len(), (size - self.cur) as usize)
- } else {
- buf.len()
- };
-
- match r.read(&mut buf[..size]) {
- Ok(0) => None,
- Ok(n) => {
- self.cur += n as u64;
- self.buf.record(n);
- Some(Ok(self.buf.split(n)))
- }
- // We don't need to reset state here since it's ok to poll the same reader.
- Err(err) => Some(Err(err)),
- }
- }
- State::Send(_) => {
- unreachable!(
- "It's invalid to go into State::Send for BlockingRead, please report this bug"
- )
- }
- }
+ todo!()
+
+ // let r = match self.blocking_reader() {
+ // Ok(r) => r,
+ // Err(err) => return Some(Err(err)),
+ // };
+ //
+ // // We should know where to start read the data.
+ // if self.offset.is_none() {
+ // (self.offset, self.size) = match Self::calculate_offset(r, self.op.range()) {
+ // Ok(v) => v,
+ // Err(err) => return Some(Err(err)),
+ // }
+ // }
+ //
+ // self.buf.reserve();
+ //
+ // let mut buf = self.buf.initialized_mut();
+ // let buf = buf.initialized_mut();
+ //
+ // let size = if let Some(size) = self.size {
+ // // Sanity check.
+ // if self.cur >= size {
+ // return None;
+ // }
+ // cmp::min(buf.len(), (size - self.cur) as usize)
+ // } else {
+ // buf.len()
+ // };
+ //
+ // match r.read(&mut buf[..size]) {
+ // Ok(0) => None,
+ // Ok(n) => {
+ // self.cur += n as u64;
+ // self.buf.record(n);
+ // Some(Ok(self.buf.split(n)))
+ // }
+ // // We don't need to reset state here since it's ok to poll the same reader.
+ // Err(err) => Some(Err(err)),
+ // }
}
}
diff --git a/core/src/raw/oio/read/futures_read.rs b/core/src/raw/oio/read/futures_read.rs
index 8b3bd3999d..2aacfb1438 100644
--- a/core/src/raw/oio/read/futures_read.rs
+++ b/core/src/raw/oio/read/futures_read.rs
@@ -22,7 +22,9 @@ use std::task::Poll;
use bytes::Bytes;
use futures::AsyncRead;
+use futures::AsyncReadExt;
use futures::AsyncSeek;
+use futures::AsyncSeekExt;
use crate::raw::*;
use crate::*;
@@ -43,28 +45,26 @@ impl<R> oio::Read for FuturesReader<R>
where
R: AsyncRead + AsyncSeek + Unpin + Send + Sync,
{
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- Pin::new(&mut self.inner).poll_read(cx, buf).map_err(|err| {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.inner.read(buf).await.map_err(|err| {
new_std_io_error(err)
.with_operation(oio::ReadOperation::Read)
.with_context("source", "FuturesReader")
})
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- Pin::new(&mut self.inner).poll_seek(cx, pos).map_err(|err| {
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ self.inner.seek(pos).await.map_err(|err| {
new_std_io_error(err)
.with_operation(oio::ReadOperation::Seek)
.with_context("source", "FuturesReader")
})
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- let _ = cx;
-
- Poll::Ready(Some(Err(Error::new(
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ Some(Err(Error::new(
ErrorKind::Unsupported,
"FuturesReader doesn't support poll_next",
- ))))
+ )))
}
}
diff --git a/core/src/raw/oio/read/into_read_from_stream.rs b/core/src/raw/oio/read/into_read_from_stream.rs
index 9bb734eda2..139cd6ee39 100644
--- a/core/src/raw/oio/read/into_read_from_stream.rs
+++ b/core/src/raw/oio/read/into_read_from_stream.rs
@@ -45,43 +45,39 @@ where
S: futures::Stream<Item = Result<T>> + Send + Sync + Unpin + 'static,
T: Into<Bytes>,
{
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
if !self.buf.is_empty() {
let len = std::cmp::min(buf.len(), self.buf.len());
buf[..len].copy_from_slice(&self.buf[..len]);
self.buf.advance(len);
- return Poll::Ready(Ok(len));
+ return Ok(len);
}
- match futures::ready!(self.inner.poll_next_unpin(cx)) {
+ match self.inner.next().await {
Some(Ok(bytes)) => {
let bytes = bytes.into();
let len = std::cmp::min(buf.len(), bytes.len());
buf[..len].copy_from_slice(&bytes[..len]);
self.buf = bytes.slice(len..);
- Poll::Ready(Ok(len))
+ Ok(len)
}
- Some(Err(err)) => Poll::Ready(Err(err)),
- None => Poll::Ready(Ok(0)),
+ Some(Err(err)) => Err(err),
+ None => Ok(0),
}
}
- fn poll_seek(&mut self, _: &mut Context<'_>, _: SeekFrom) -> Poll<Result<u64>> {
- Poll::Ready(Err(Error::new(
+ async fn seek(&mut self, _: SeekFrom) -> Result<u64> {
+ Err(Error::new(
ErrorKind::Unsupported,
"FromStreamReader can't support operation",
- )))
+ ))
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
if !self.buf.is_empty() {
- return Poll::Ready(Some(Ok(std::mem::take(&mut self.buf))));
+ return Some(Ok(std::mem::take(&mut self.buf)));
}
- match futures::ready!(self.inner.poll_next_unpin(cx)) {
- Some(Ok(bytes)) => Poll::Ready(Some(Ok(bytes.into()))),
- Some(Err(err)) => Poll::Ready(Some(Err(err))),
- None => Poll::Ready(None),
- }
+ self.inner.next().await.map(|v| v.map(|v| v.into()))
}
}
diff --git a/core/src/raw/oio/read/into_streamable_read.rs b/core/src/raw/oio/read/into_streamable_read.rs
index c3f90e24d5..c5cd5f6141 100644
--- a/core/src/raw/oio/read/into_streamable_read.rs
+++ b/core/src/raw/oio/read/into_streamable_read.rs
@@ -43,25 +43,25 @@ pub struct StreamableReader<R> {
}
impl<R: oio::Read> oio::Read for StreamableReader<R> {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- self.r.poll_read(cx, buf)
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.r.read(buf).await
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- self.r.poll_seek(cx, pos)
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ self.r.seek(pos).await
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
let dst = self.buf.spare_capacity_mut();
let mut buf = ReadBuf::uninit(dst);
unsafe { buf.assume_init(self.cap) };
- match ready!(self.r.poll_read(cx, buf.initialized_mut())) {
- Err(err) => Poll::Ready(Some(Err(err))),
- Ok(0) => Poll::Ready(None),
+ match self.r.read(buf.initialized_mut()).await {
+ Err(err) => Some(Err(err)),
+ Ok(0) => None,
Ok(n) => {
buf.set_filled(n);
- Poll::Ready(Some(Ok(Bytes::from(buf.filled().to_vec()))))
+ Some(Ok(Bytes::from(buf.filled().to_vec())))
}
}
}
@@ -94,6 +94,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for StreamableReader<R> {
#[cfg(test)]
mod tests {
+ use crate::raw::oio::Read;
use bytes::BufMut;
use bytes::BytesMut;
use rand::prelude::*;
@@ -102,8 +103,6 @@ mod tests {
#[tokio::test]
async fn test_into_stream() {
- use oio::ReadExt;
-
let mut rng = ThreadRng::default();
// Generate size between 1B..16MB.
let size = rng.gen_range(1..16 * 1024 * 1024);
diff --git a/core/src/raw/oio/read/lazy_read.rs b/core/src/raw/oio/read/lazy_read.rs
index 78e089d0c9..b193262671 100644
--- a/core/src/raw/oio/read/lazy_read.rs
+++ b/core/src/raw/oio/read/lazy_read.rs
@@ -67,15 +67,6 @@ where
Ok(self.reader.as_mut().expect("reader must be valid"))
}
-
- fn blocking_reader(&mut self) -> Result<&mut R> {
- if self.reader.is_none() {
- let (_, r) = self.acc.blocking_read(&self.path, self.op.clone())?;
- self.reader = Some(r);
- }
-
- Ok(self.reader.as_mut().expect("reader must be valid"))
- }
}
impl<A, R> oio::Read for LazyReader<A, R>
@@ -92,7 +83,26 @@ where
}
async fn next(&mut self) -> Option<Result<Bytes>> {
- self.reader().await?.next().await
+ let r = match self.reader().await {
+ Ok(r) => r,
+ Err(err) => return Some(Err(err)),
+ };
+ r.next().await
+ }
+}
+
+impl<A, R> LazyReader<A, R>
+where
+ A: Accessor<BlockingReader = R>,
+ R: oio::BlockingRead,
+{
+ fn blocking_reader(&mut self) -> Result<&mut R> {
+ if self.reader.is_none() {
+ let (_, r) = self.acc.blocking_read(&self.path, self.op.clone())?;
+ self.reader = Some(r);
+ }
+
+ Ok(self.reader.as_mut().expect("reader must be valid"))
}
}
@@ -110,6 +120,11 @@ where
}
fn next(&mut self) -> Option<Result<Bytes>> {
- self.blocking_reader()?.next()
+ let r = match self.blocking_reader() {
+ Ok(r) => r,
+ Err(err) => return Some(Err(err)),
+ };
+
+ r.next()
}
}
diff --git a/core/src/raw/oio/read/range_read.rs b/core/src/raw/oio/read/range_read.rs
index 925c9ad700..f15241897f 100644
--- a/core/src/raw/oio/read/range_read.rs
+++ b/core/src/raw/oio/read/range_read.rs
@@ -17,13 +17,10 @@
use std::future::Future;
use std::io::SeekFrom;
-use std::pin::Pin;
use std::sync::Arc;
-use std::task::ready;
-use std::task::Context;
-use std::task::Poll;
use bytes::Bytes;
+use futures::AsyncSeek;
use crate::raw::*;
use crate::*;
@@ -45,23 +42,9 @@ pub struct RangeReader<A: Accessor, R> {
offset: Option<u64>,
size: Option<u64>,
cur: u64,
- state: State<R>,
+ reader: Option<R>,
}
-enum State<R> {
- Idle,
- SendStat(BoxedStaticFuture<Result<RpStat>>),
- SendRead(BoxedStaticFuture<Result<(RpRead, R)>>),
- Read(R),
-}
-
-/// # Safety
-///
-/// wasm32 is a special target that we only have one event-loop for this state.
-unsafe impl<R> Send for State<R> {}
-/// Safety: State will only be accessed under &mut.
-unsafe impl<R> Sync for State<R> {}
-
impl<A, R> RangeReader<A, R>
where
A: Accessor,
@@ -95,7 +78,7 @@ where
offset,
size,
cur: 0,
- state: State::<R>::Idle,
+ reader: None,
}
}
@@ -106,7 +89,7 @@ where
if size > total_size {
// If returns an error, we should reset
// state to Idle so that we can retry it.
- self.state = State::Idle;
+ self.reader = None;
return Err(Error::new(
ErrorKind::InvalidInput,
"read to a negative or overflowing position is invalid",
@@ -195,10 +178,7 @@ where
A: Accessor<Reader = R>,
R: oio::Read,
{
- fn read_future(&self) -> BoxedStaticFuture<Result<(RpRead, R)>> {
- let acc = self.acc.clone();
- let path = self.path.clone();
-
+ async fn read_future(&self) -> Result<(RpRead, R)> {
let mut op = self.op.clone();
// cur != 0 means we have read some data out, we should convert
// the op into deterministic to avoid ETag changes.
@@ -208,13 +188,10 @@ where
// Alter OpRead with correct calculated range.
op = op.with_range(self.calculate_range());
- Box::pin(async move { acc.read(&path, op).await })
+ self.acc.read(&self.path, op).await
}
- fn stat_future(&self) -> BoxedStaticFuture<Result<RpStat>> {
- let acc = self.acc.clone();
- let path = self.path.clone();
-
+ async fn stat_future(&self) -> Result<RpStat> {
// Handle if-match and if-none-match correctly.
let mut args = OpStat::default();
// TODO: stat should support range to check if ETag matches.
@@ -227,7 +204,7 @@ where
}
}
- Box::pin(async move { acc.stat(&path, args).await })
+ self.acc.stat(&self.path, args).await
}
}
@@ -277,192 +254,123 @@ where
A: Accessor<Reader = R>,
R: oio::Read,
{
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
// Sanity check for normal cases.
if buf.is_empty() || self.cur >= self.size.unwrap_or(u64::MAX) {
- return Poll::Ready(Ok(0));
+ return Ok(0);
}
- match &mut self.state {
- State::Idle => {
- self.state = if self.offset.is_none() {
- // Offset is none means we are doing tailing reading.
- // we should stat first to get the correct offset.
- State::SendStat(self.stat_future())
- } else {
- State::SendRead(self.read_future())
- };
-
- self.poll_read(cx, buf)
- }
- State::SendStat(fut) => {
- let rp = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If stat future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
+ if self.offset.is_none() {
+ let rp = self.stat_future().await?;
+ let length = rp.into_metadata().content_length();
+ self.ensure_offset(length)?;
+ }
+ if self.reader.is_none() {
+ let (rp, r) = self.read_future().await?;
- let length = rp.into_metadata().content_length();
- self.ensure_offset(length)?;
+ self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
+ self.reader = Some(r);
+ }
- self.state = State::Idle;
- self.poll_read(cx, buf)
+ let r = self.reader.as_mut().expect("reader must be valid");
+ match r.read(buf).await {
+ Ok(0) => {
+ // Reset state to Idle after all data has been consumed.
+ self.reader = None;
+ Ok(0)
}
- State::SendRead(fut) => {
- let (rp, r) = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If read future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
-
- self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
-
- self.state = State::Read(r);
- self.poll_read(cx, buf)
+ Ok(n) => {
+ self.cur += n as u64;
+ Ok(n)
+ }
+ Err(e) => {
+ self.reader = None;
+ Err(e)
}
- State::Read(r) => match ready!(Pin::new(r).poll_read(cx, buf)) {
- Ok(0) => {
- // Reset state to Idle after all data has been consumed.
- self.state = State::Idle;
- Poll::Ready(Ok(0))
- }
- Ok(n) => {
- self.cur += n as u64;
- Poll::Ready(Ok(n))
- }
- Err(e) => {
- self.state = State::Idle;
- Poll::Ready(Err(e))
- }
- },
}
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- match &mut self.state {
- State::Idle => {
- let (base, amt) = match pos {
- SeekFrom::Start(n) => (0, n as i64),
- SeekFrom::Current(n) => (self.cur as i64, n),
- SeekFrom::End(n) => {
- if let Some(size) = self.size {
- (size as i64, n)
- } else {
- self.state = State::SendStat(self.stat_future());
- return self.poll_seek(cx, pos);
- }
- }
- };
-
- let seek_pos = match base.checked_add(amt) {
- Some(n) if n >= 0 => n as u64,
- _ => {
- return Poll::Ready(Err(Error::new(
- ErrorKind::InvalidInput,
- "invalid seek to a negative or overflowing position",
- )))
- }
- };
-
- self.cur = seek_pos;
- Poll::Ready(Ok(self.cur))
- }
- State::SendStat(fut) => {
- let rp = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If stat future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ // There is an optimization here that we can calculate if users trying to seek
+ // the same position, for example, `reader.seek(SeekFrom::Current(0))`.
+ // In this case, we can just return current position without dropping reader.
+ if pos == SeekFrom::Current(0) || pos == SeekFrom::Start(self.cur) {
+ return Ok(self.cur);
+ }
- let length = rp.into_metadata().content_length();
- self.ensure_offset(length)?;
+ // We are seeking to other places, let's drop existing reader.
+ self.reader = None;
- self.state = State::Idle;
- self.poll_seek(cx, pos)
- }
- State::SendRead(_) => {
- // It's impossible for us to go into this state while
- // poll_seek. We can just drop this future and check state.
- self.state = State::Idle;
- self.poll_seek(cx, pos)
- }
- State::Read(_) => {
- // There is an optimization here that we can calculate if users trying to seek
- // the same position, for example, `reader.seek(SeekFrom::Current(0))`.
- // In this case, we can just return current position without dropping reader.
- if pos == SeekFrom::Current(0) || pos == SeekFrom::Start(self.cur) {
- return Poll::Ready(Ok(self.cur));
+ let (base, amt) = match pos {
+ SeekFrom::Start(n) => (0, n as i64),
+ SeekFrom::Current(n) => (self.cur as i64, n),
+ SeekFrom::End(n) => {
+ if let Some(size) = self.size {
+ (size as i64, n)
+ } else {
+ let rp = self.stat_future().await?;
+ let length = rp.into_metadata().content_length();
+ self.ensure_offset(length)?;
+
+ (length as i64, n)
}
+ }
+ };
- self.state = State::Idle;
- self.poll_seek(cx, pos)
+ let seek_pos = match base.checked_add(amt) {
+ Some(n) if n >= 0 => n as u64,
+ _ => {
+ return Err(Error::new(
+ ErrorKind::InvalidInput,
+ "invalid seek to a negative or overflowing position",
+ ))
}
- }
+ };
+
+ self.cur = seek_pos;
+ Ok(self.cur)
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
// Sanity check for normal cases.
if self.cur >= self.size.unwrap_or(u64::MAX) {
- return Poll::Ready(None);
+ return None;
}
- match &mut self.state {
- State::Idle => {
- self.state = if self.offset.is_none() {
- // Offset is none means we are doing tailing reading.
- // we should stat first to get the correct offset.
- State::SendStat(self.stat_future())
- } else {
- State::SendRead(self.read_future())
- };
-
- self.poll_next(cx)
+ if self.offset.is_none() {
+ let rp = match self.stat_future().await {
+ Ok(v) => v,
+ Err(err) => return Some(Err(err)),
+ };
+ let length = rp.into_metadata().content_length();
+ if let Err(err) = self.ensure_offset(length) {
+ return Some(Err(err));
}
- State::SendStat(fut) => {
- let rp = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If stat future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
-
- let length = rp.into_metadata().content_length();
- self.ensure_offset(length)?;
+ }
+ if self.reader.is_none() {
+ let (rp, r) = match self.read_future().await {
+ Ok((rp, r)) => (rp, r),
+ Err(err) => return Some(Err(err)),
+ };
+
+ self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
+ self.reader = Some(r);
+ }
- self.state = State::Idle;
- self.poll_next(cx)
+ let r = self.reader.as_mut().expect("reader must be valid");
+ match r.next().await {
+ Some(Ok(bs)) => {
+ self.cur += bs.len() as u64;
+ Some(Ok(bs))
}
- State::SendRead(fut) => {
- let (rp, r) = ready!(Pin::new(fut).poll(cx)).map_err(|err| {
- // If read future returns an error, we should reset
- // state to Idle so that we can retry it.
- self.state = State::Idle;
- err
- })?;
-
- // Set size if read returns size hint.
- self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
-
- self.state = State::Read(r);
- self.poll_next(cx)
+ Some(Err(err)) => {
+ self.reader = None;
+ Some(Err(err))
+ }
+ None => {
+ self.reader = None;
+ None
}
- State::Read(r) => match ready!(Pin::new(r).poll_next(cx)) {
- Some(Ok(bs)) => {
- self.cur += bs.len() as u64;
- Poll::Ready(Some(Ok(bs)))
- }
- Some(Err(err)) => {
- self.state = State::Idle;
- Poll::Ready(Some(Err(err)))
- }
- None => {
- self.state = State::Idle;
- Poll::Ready(None)
- }
- },
}
}
}
@@ -478,155 +386,116 @@ where
return Ok(0);
}
- match &mut self.state {
- State::Idle => {
- // Offset is none means we are doing tailing reading.
- // we should stat first to get the correct offset.
- if self.offset.is_none() {
- let rp = self.stat_action()?;
-
- let length = rp.into_metadata().content_length();
- self.ensure_offset(length)?;
- }
-
- let (rp, r) = self.read_action()?;
+ if self.offset.is_none() {
+ let rp = self.stat_action()?;
+ let length = rp.into_metadata().content_length();
+ self.ensure_offset(length)?;
+ }
+ if self.reader.is_none() {
+ let (rp, r) = self.read_action()?;
- // Set size if read returns size hint.
- self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
+ self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
+ self.reader = Some(r);
+ }
- self.state = State::Read(r);
- self.read(buf)
- }
- State::Read(r) => {
- match r.read(buf) {
- Ok(0) => {
- // Reset state to Idle after all data has been consumed.
- self.state = State::Idle;
- Ok(0)
- }
- Ok(n) => {
- self.cur += n as u64;
- Ok(n)
- }
- Err(e) => {
- self.state = State::Idle;
- Err(e)
- }
- }
+ let r = self.reader.as_mut().expect("reader must be valid");
+ match r.read(buf) {
+ Ok(0) => {
+ // Reset state to Idle after all data has been consumed.
+ self.reader = None;
+ Ok(0)
}
- State::SendStat(_) => {
- unreachable!("It's invalid to go into State::SendStat for BlockingRead, please report this bug")
+ Ok(n) => {
+ self.cur += n as u64;
+ Ok(n)
}
- State::SendRead(_) => {
- unreachable!("It's invalid to go into State::SendRead for BlockingRead, please report this bug")
+ Err(e) => {
+ self.reader = None;
+ Err(e)
}
}
}
fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
- match &mut self.state {
- State::Idle => {
- let (base, amt) = match pos {
- SeekFrom::Start(n) => (0, n as i64),
- SeekFrom::End(n) => {
- if let Some(size) = self.size {
- (size as i64, n)
- } else {
- let rp = self.stat_action()?;
- let length = rp.into_metadata().content_length();
- self.ensure_offset(length)?;
-
- let size = self.size.expect("size must be valid after fill_range");
- (size as i64, n)
- }
- }
- SeekFrom::Current(n) => (self.cur as i64, n),
- };
-
- let seek_pos = match base.checked_add(amt) {
- Some(n) if n >= 0 => n as u64,
- _ => {
- return Err(Error::new(
- ErrorKind::InvalidInput,
- "invalid seek to a negative or overflowing position",
- ));
- }
- };
-
- self.cur = seek_pos;
- Ok(self.cur)
- }
- State::Read(_) => {
- // There is an optimization here that we can calculate if users trying to seek
- // the same position, for example, `reader.seek(SeekFrom::Current(0))`.
- // In this case, we can just return current position without dropping reader.
- if pos == SeekFrom::Current(0) || pos == SeekFrom::Start(self.cur) {
- return Ok(self.cur);
- }
+ // There is an optimization here that we can calculate if users trying to seek
+ // the same position, for example, `reader.seek(SeekFrom::Current(0))`.
+ // In this case, we can just return current position without dropping reader.
+ if pos == SeekFrom::Current(0) || pos == SeekFrom::Start(self.cur) {
+ return Ok(self.cur);
+ }
- self.state = State::Idle;
- self.seek(pos)
- }
- State::SendStat(_) => {
- unreachable!("It's invalid to go into State::SendStat for BlockingRead, please report this bug")
+ // We are seeking to other places, let's drop existing reader.
+ self.reader = None;
+
+ let (base, amt) = match pos {
+ SeekFrom::Start(n) => (0, n as i64),
+ SeekFrom::Current(n) => (self.cur as i64, n),
+ SeekFrom::End(n) => {
+ if let Some(size) = self.size {
+ (size as i64, n)
+ } else {
+ let rp = self.stat_action()?;
+ let length = rp.into_metadata().content_length();
+ self.ensure_offset(length)?;
+
+ (length as i64, n)
+ }
}
- State::SendRead(_) => {
- unreachable!("It's invalid to go into State::SendRead for BlockingRead, please report this bug")
+ };
+
+ let seek_pos = match base.checked_add(amt) {
+ Some(n) if n >= 0 => n as u64,
+ _ => {
+ return Err(Error::new(
+ ErrorKind::InvalidInput,
+ "invalid seek to a negative or overflowing position",
+ ))
}
- }
+ };
+
+ self.cur = seek_pos;
+ Ok(self.cur)
}
fn next(&mut self) -> Option<Result<Bytes>> {
- match &mut self.state {
- State::Idle => {
- // Sanity check for normal cases.
- if self.cur >= self.size.unwrap_or(u64::MAX) {
- return None;
- }
-
- // Offset is none means we are doing tailing reading.
- // we should stat first to get the correct offset.
- if self.offset.is_none() {
- let rp = match self.stat_action() {
- Ok(rp) => rp,
- Err(err) => return Some(Err(err)),
- };
+ // Sanity check for normal cases.
+ if self.cur >= self.size.unwrap_or(u64::MAX) {
+ return None;
+ }
- let length = rp.into_metadata().content_length();
- if let Err(err) = self.ensure_offset(length) {
- return Some(Err(err));
- }
- }
+ if self.offset.is_none() {
+ let rp = match self.stat_action() {
+ Ok(rp) => rp,
+ Err(err) => return Some(Err(err)),
+ };
+ let length = rp.into_metadata().content_length();
+ if let Err(err) = self.ensure_offset(length) {
+ return Some(Err(err));
+ }
+ }
+ if self.reader.is_none() {
+ let (rp, r) = match self.read_action() {
+ Ok((rp, r)) => (rp, r),
+ Err(err) => return Some(Err(err)),
+ };
+
+ self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
+ self.reader = Some(r);
+ }
- let r = match self.read_action() {
- Ok((rp, r)) => {
- self.ensure_size(rp.range().unwrap_or_default().size(), rp.size());
- r
- }
- Err(err) => return Some(Err(err)),
- };
- self.state = State::Read(r);
- self.next()
+ let r = self.reader.as_mut().expect("reader must be valid");
+ match r.next() {
+ Some(Ok(bs)) => {
+ self.cur += bs.len() as u64;
+ Some(Ok(bs))
}
- State::Read(r) => match r.next() {
- Some(Ok(bs)) => {
- self.cur += bs.len() as u64;
- Some(Ok(bs))
- }
- Some(Err(err)) => {
- self.state = State::Idle;
- Some(Err(err))
- }
- None => {
- self.state = State::Idle;
- None
- }
- },
- State::SendStat(_) => {
- unreachable!("It's invalid to go into State::SendStat for BlockingRead, please report this bug")
+ Some(Err(err)) => {
+ self.reader = None;
+ Some(Err(err))
}
- State::SendRead(_) => {
- unreachable!("It's invalid to go into State::SendRead for BlockingRead, please report this bug")
+ None => {
+ self.reader = None;
+ None
}
}
}
@@ -706,32 +575,30 @@ mod tests {
}
impl oio::Read for MockReader {
- fn poll_read(&mut self, cx: &mut Context, buf: &mut [u8]) -> Poll<Result<usize>> {
- Pin::new(&mut self.inner).poll_read(cx, buf).map_err(|err| {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.inner.read(buf).await.map_err(|err| {
Error::new(ErrorKind::Unexpected, "read data from mock").set_source(err)
})
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- let (_, _) = (cx, pos);
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ let _ = pos;
- Poll::Ready(Err(Error::new(
+ Err(Error::new(
ErrorKind::Unsupported,
"output reader doesn't support seeking",
- )))
+ ))
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+ async fn next(&mut self) -> Option<Result<Bytes>> {
let mut bs = vec![0; 4 * 1024];
- let n = ready!(Pin::new(&mut self.inner)
- .poll_read(cx, &mut bs)
- .map_err(
- |err| Error::new(ErrorKind::Unexpected, "read data from mock").set_source(err)
- )?);
+ let n = self.inner.read(&mut bs).await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "read data from mock").set_source(err)
+ })?;
if n == 0 {
- Poll::Ready(None)
+ None
} else {
- Poll::Ready(Some(Ok(Bytes::from(bs[..n].to_vec()))))
+ Some(Ok(Bytes::from(bs[..n].to_vec())))
}
}
}
diff --git a/core/src/raw/oio/read/tokio_read.rs b/core/src/raw/oio/read/tokio_read.rs
index 37aabc1473..a38c2b1427 100644
--- a/core/src/raw/oio/read/tokio_read.rs
+++ b/core/src/raw/oio/read/tokio_read.rs
@@ -23,7 +23,9 @@ use std::task::Poll;
use bytes::Bytes;
use tokio::io::AsyncRead;
+use tokio::io::AsyncReadExt;
use tokio::io::AsyncSeek;
+use tokio::io::AsyncSeekExt;
use tokio::io::ReadBuf;
use crate::raw::*;
@@ -50,44 +52,26 @@ impl<R> oio::Read for TokioReader<R>
where
R: AsyncRead + AsyncSeek + Unpin + Send + Sync,
{
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- let mut buf = ReadBuf::new(buf);
-
- ready!(Pin::new(&mut self.inner).poll_read(cx, &mut buf)).map_err(|err| {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.inner.read(buf).await.map_err(|err| {
new_std_io_error(err)
.with_operation(oio::ReadOperation::Read)
.with_context("source", "TokioReader")
- })?;
-
- Poll::Ready(Ok(buf.filled().len()))
+ })
}
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: SeekFrom) -> Poll<Result<u64>> {
- if self.seek_pos != Some(pos) {
- Pin::new(&mut self.inner).start_seek(pos).map_err(|err| {
- new_std_io_error(err)
- .with_operation(oio::ReadOperation::Seek)
- .with_context("source", "TokioReader")
- })?;
- self.seek_pos = Some(pos)
- }
-
- // NOTE: don't return error by `?` here, we need to reset seek_pos.
- let pos = ready!(Pin::new(&mut self.inner).poll_complete(cx).map_err(|err| {
+ async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ self.inner.seek(pos).await.map_err(|err| {
new_std_io_error(err)
.with_operation(oio::ReadOperation::Seek)
.with_context("source", "TokioReader")
- }));
- self.seek_pos = None;
- Poll::Ready(pos)
+ })
}
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- let _ = cx;
-
- Poll::Ready(Some(Err(Error::new(
+ async fn next(&mut self) -> Option<Result<Bytes>> {
+ Some(Err(Error::new(
ErrorKind::Unsupported,
"TokioReader doesn't support poll_next",
- ))))
+ )))
}
}
diff --git a/core/src/raw/oio/stream/api.rs b/core/src/raw/oio/stream/api.rs
index fe95bb656a..7a672d2ff2 100644
--- a/core/src/raw/oio/stream/api.rs
+++ b/core/src/raw/oio/stream/api.rs
@@ -54,11 +54,11 @@ impl<T: Stream + ?Sized> Stream for Box<T> {
}
}
-impl Stream for dyn raw::oio::Read {
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- raw::oio::Read::poll_next(self, cx)
- }
-}
+// impl<T: raw::oio::Read> Stream for T {
+// fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+// raw::oio::Read::poll_next(self, cx)
+// }
+// }
impl futures::Stream for dyn Stream {
type Item = Result<Bytes>;
diff --git a/core/src/types/reader.rs b/core/src/types/reader.rs
index 5c6d97b30d..7e2e90416a 100644
--- a/core/src/types/reader.rs
+++ b/core/src/types/reader.rs
@@ -76,19 +76,19 @@ impl Reader {
}
}
-impl oio::Read for Reader {
- fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
- self.inner.poll_read(cx, buf)
- }
-
- fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
- self.inner.poll_seek(cx, pos)
- }
-
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- self.inner.poll_next(cx)
- }
-}
+// impl oio::Read for Reader {
+// fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+// self.inner.poll_read(cx, buf)
+// }
+//
+// fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
+// self.inner.poll_seek(cx, pos)
+// }
+//
+// fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+// self.inner.poll_next(cx)
+// }
+// }
impl AsyncRead for Reader {
fn poll_read(
@@ -96,7 +96,8 @@ impl AsyncRead for Reader {
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
- Pin::new(&mut self.inner).poll_read(cx, buf)
+ todo!()
+ // Pin::new(&mut self.inner).poll_read(cx, buf)
}
}
@@ -106,7 +107,8 @@ impl AsyncSeek for Reader {
cx: &mut Context<'_>,
pos: io::SeekFrom,
) -> Poll<io::Result<u64>> {
- Pin::new(&mut self.inner).poll_seek(cx, pos)
+ todo!()
+ // Pin::new(&mut self.inner).poll_seek(cx, pos)
}
}
@@ -116,43 +118,49 @@ impl tokio::io::AsyncRead for Reader {
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
- // Safety: We make sure that we will set filled correctly.
- unsafe { buf.assume_init(buf.remaining()) }
- let n = ready!(self.inner.poll_read(cx, buf.initialize_unfilled()))?;
- buf.advance(n);
- Poll::Ready(Ok(()))
+ todo!()
+
+ // // Safety: We make sure that we will set filled correctly.
+ // unsafe { buf.assume_init(buf.remaining()) }
+ // let n = ready!(self.inner.poll_read(cx, buf.initialize_unfilled()))?;
+ // buf.advance(n);
+ // Poll::Ready(Ok(()))
}
}
impl tokio::io::AsyncSeek for Reader {
fn start_seek(self: Pin<&mut Self>, pos: io::SeekFrom) -> io::Result<()> {
- let this = self.get_mut();
- if let SeekState::Start(_) = this.seek_state {
- return Err(io::Error::new(
- io::ErrorKind::Other,
- "another search is in progress.",
- ));
- }
- this.seek_state = SeekState::Start(pos);
- Ok(())
+ todo!()
+
+ // let this = self.get_mut();
+ // if let SeekState::Start(_) = this.seek_state {
+ // return Err(io::Error::new(
+ // io::ErrorKind::Other,
+ // "another search is in progress.",
+ // ));
+ // }
+ // this.seek_state = SeekState::Start(pos);
+ // Ok(())
}
fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
- let state = self.seek_state;
- match state {
- SeekState::Init => {
- // AsyncSeek recommends calling poll_complete before start_seek.
- // We don't have to guarantee that the value returned by
- // poll_complete called without start_seek is correct,
- // so we'll return 0.
- Poll::Ready(Ok(0))
- }
- SeekState::Start(pos) => {
- let n = ready!(self.inner.poll_seek(cx, pos))?;
- self.get_mut().seek_state = SeekState::Init;
- Poll::Ready(Ok(n))
- }
- }
+ todo!()
+
+ // let state = self.seek_state;
+ // match state {
+ // SeekState::Init => {
+ // // AsyncSeek recommends calling poll_complete before start_seek.
+ // // We don't have to guarantee that the value returned by
+ // // poll_complete called without start_seek is correct,
+ // // so we'll return 0.
+ // Poll::Ready(Ok(0))
+ // }
+ // SeekState::Start(pos) => {
+ // let n = ready!(self.inner.poll_seek(cx, pos))?;
+ // self.get_mut().seek_state = SeekState::Init;
+ // Poll::Ready(Ok(n))
+ // }
+ // }
}
}
@@ -169,9 +177,10 @@ impl Stream for Reader {
type Item = io::Result<Bytes>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
- Pin::new(&mut self.inner)
- .poll_next(cx)
- .map_err(|err| io::Error::new(io::ErrorKind::Interrupted, err))
+ todo!()
+ // Pin::new(&mut self.inner)
+ // .poll_next(cx)
+ // .map_err(|err| io::Error::new(io::ErrorKind::Interrupted, err))
}
}