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/12 06:16:12 UTC
(opendal) branch cleanup created (now 56f01dd8b4)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 56f01dd8b4 add comments
This branch includes the following new commits:
new 6f7fe0d050 Polish comments
new 4409085eff Remove not exist op
new 56f01dd8b4 add comments
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 03/03: add comments
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 56f01dd8b4cc6f624db7f649db73fb97b3c26a7a
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Mar 12 14:15:59 2024 +0800
add comments
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/types/reader.rs | 38 ++++++++++++++++++++++++--------------
1 file changed, 24 insertions(+), 14 deletions(-)
diff --git a/core/src/types/reader.rs b/core/src/types/reader.rs
index 10af69f955..2af5ca5144 100644
--- a/core/src/types/reader.rs
+++ b/core/src/types/reader.rs
@@ -34,25 +34,35 @@ use crate::*;
///
/// # Usage
///
-/// Reader implements the following APIs:
+/// [`Reader`] provides multiple ways to read data from given reader. Please note that it's
+/// undefined behavior to use `Reader` in different ways.
///
-/// - `AsyncRead`
-/// - `AsyncSeek`
-/// - `Stream<Item = <io::Result<Bytes>>>`
+/// ## Direct
///
-/// For reading data, we can use `AsyncRead` and `Stream`. The mainly
-/// different is where the `copy` happens.
+/// [`Reader`] provides public API including [`Reader::read`], [`Reader::seek`] and
+/// [`Reader::read_to_end`]. You can use those APIs directly without extra copy.
///
-/// `AsyncRead` requires user to prepare a buffer for `Reader` to fill.
-/// And `Stream` will stream out a `Bytes` for user to decide when to copy
-/// it's content.
+/// ## Bytes Stream
///
-/// For example, users may have their only CPU/IO bound workers and don't
-/// want to do copy inside IO workers. They can use `Stream` to get a `Bytes`
-/// and consume it in side CPU workers inside.
+/// [`Reader`] can be used as `Stream<Item = <io::Result<Bytes>>>`.
///
-/// Besides, `Stream` **COULD** reduce an extra copy if underlying reader is
-/// stream based (like services s3, azure which based on HTTP).
+/// It also implements [`Send`], [`Sync`] and [`Unpin`].
+///
+/// ## Futures AsyncRead
+///
+/// [`Reader`] can be used as [`futures::AsyncRead`] and [`futures::AsyncSeek`].
+///
+/// It also implements [`Send`], [`Sync`] and [`Unpin`].
+///
+/// [`Reader`] provides [`Reader::into_futures_read`] to remove extra APIs upon self.
+///
+/// ## Tokio AsyncRead
+///
+/// [`Reader`] can be used as [`tokio::io::AsyncRead`] and [`tokio::io::AsyncSeek`].
+///
+/// It also implements [`Send`], [`Sync`] and [`Unpin`].
+///
+/// [`Reader`] provides [`Reader::into_tokio_read`] to remove extra APIs upon self.
pub struct Reader {
state: State,
}
(opendal) 01/03: Polish comments
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 6f7fe0d050b15b88ad0dee8aabc89fb4e7809158
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Mar 12 13:43:55 2024 +0800
Polish comments
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/chaos.rs | 4 +--
core/src/layers/concurrent_limit.rs | 4 +--
core/src/layers/dtrace.rs | 4 +--
core/src/layers/error_context.rs | 4 +--
core/src/layers/logging.rs | 4 +--
core/src/layers/metrics.rs | 4 +--
core/src/layers/minitrace.rs | 4 +--
core/src/layers/oteltrace.rs | 4 +--
core/src/layers/prometheus.rs | 4 +--
core/src/layers/prometheus_client.rs | 4 +--
core/src/layers/retry.rs | 4 +--
core/src/layers/throttle.rs | 4 +--
core/src/layers/timeout.rs | 4 +--
core/src/layers/tracing.rs | 4 +--
core/src/raw/enum_utils.rs | 24 ++++++++--------
core/src/raw/http_util/body.rs | 4 +--
core/src/raw/oio/cursor.rs | 4 +--
core/src/raw/oio/read/api.rs | 40 ++++++++++++++------------
core/src/raw/oio/read/buffer_reader.rs | 14 ++++-----
core/src/raw/oio/read/file_read.rs | 4 +--
core/src/raw/oio/read/futures_read.rs | 8 +++---
core/src/raw/oio/read/into_read_from_stream.rs | 4 +--
core/src/raw/oio/read/into_streamable_read.rs | 4 +--
core/src/raw/oio/read/lazy_read.rs | 4 +--
core/src/raw/oio/read/range_read.rs | 8 +++---
core/src/raw/oio/read/tokio_read.rs | 8 +++---
core/src/services/dbfs/reader.rs | 4 +--
core/src/services/ftp/util.rs | 8 +++---
core/src/services/hdfs_native/reader.rs | 4 +--
core/src/types/reader.rs | 4 +--
30 files changed, 103 insertions(+), 99 deletions(-)
diff --git a/core/src/layers/chaos.rs b/core/src/layers/chaos.rs
index 223e1e256a..079500c444 100644
--- a/core/src/layers/chaos.rs
+++ b/core/src/layers/chaos.rs
@@ -174,9 +174,9 @@ impl<R> ChaosReader<R> {
}
impl<R: oio::Read> oio::Read for ChaosReader<R> {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
if self.i_feel_lucky() {
- self.inner.read(size).await
+ self.inner.read(limit).await
} else {
Err(Self::unexpected_eof())
}
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index c313eaafcc..36fbcb92e7 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -261,8 +261,8 @@ impl<R: oio::Read> oio::Read for ConcurrentLimitWrapper<R> {
self.inner.seek(pos).await
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- self.inner.read(size).await
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ self.inner.read(limit).await
}
}
diff --git a/core/src/layers/dtrace.rs b/core/src/layers/dtrace.rs
index e32fe1b07d..4e488dd499 100644
--- a/core/src/layers/dtrace.rs
+++ b/core/src/layers/dtrace.rs
@@ -344,10 +344,10 @@ impl<R> DtraceLayerWrapper<R> {
}
impl<R: oio::Read> oio::Read for DtraceLayerWrapper<R> {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
let c_path = CString::new(self.path.clone()).unwrap();
probe_lazy!(opendal, reader_read_start, c_path.as_ptr());
- match self.inner.read(size).await {
+ match self.inner.read(limit).await {
Ok(bs) => {
probe_lazy!(opendal, reader_read_ok, c_path.as_ptr(), bs.len());
Ok(bs)
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 3cf0036bd9..7e0e65a187 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -357,8 +357,8 @@ impl<T: oio::Read> oio::Read for ErrorContextWrapper<T> {
})
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- self.inner.read(size).await.map_err(|err| {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ self.inner.read(limit).await.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 45d6c266a6..9bfecf5b2c 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -1022,8 +1022,8 @@ impl<R: oio::Read> oio::Read for LoggingReader<R> {
}
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- match self.inner.read(size).await {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ match self.inner.read(limit).await {
Ok(bs) => {
self.read += bs.len() as u64;
trace!(
diff --git a/core/src/layers/metrics.rs b/core/src/layers/metrics.rs
index 7cb3dcf9f6..c6001b8678 100644
--- a/core/src/layers/metrics.rs
+++ b/core/src/layers/metrics.rs
@@ -773,8 +773,8 @@ impl<R> Drop for MetricWrapper<R> {
}
impl<R: oio::Read> oio::Read for MetricWrapper<R> {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- match self.inner.read(size).await {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ match self.inner.read(limit).await {
Ok(bytes) => {
self.bytes += bytes.len() as u64;
Ok(bytes)
diff --git a/core/src/layers/minitrace.rs b/core/src/layers/minitrace.rs
index 596a3b0a18..fbf4b2f064 100644
--- a/core/src/layers/minitrace.rs
+++ b/core/src/layers/minitrace.rs
@@ -298,8 +298,8 @@ impl<R> MinitraceWrapper<R> {
impl<R: oio::Read> oio::Read for MinitraceWrapper<R> {
#[trace(enter_on_poll = true)]
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- self.inner.read(size).await
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ self.inner.read(limit).await
}
#[trace(enter_on_poll = true)]
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index 1c3825e5a5..f8f85be255 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -278,8 +278,8 @@ impl<R> OtelTraceWrapper<R> {
}
impl<R: oio::Read> oio::Read for OtelTraceWrapper<R> {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- self.inner.read(size).await
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ self.inner.read(limit).await
}
async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
diff --git a/core/src/layers/prometheus.rs b/core/src/layers/prometheus.rs
index 68a74c0341..0037217d99 100644
--- a/core/src/layers/prometheus.rs
+++ b/core/src/layers/prometheus.rs
@@ -686,13 +686,13 @@ impl<R> PrometheusMetricWrapper<R> {
}
impl<R: oio::Read> oio::Read for PrometheusMetricWrapper<R> {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
let labels = self.stats.generate_metric_label(
self.scheme.into_static(),
Operation::Read.into_static(),
&self.path,
);
- match self.inner.read(size).await {
+ match self.inner.read(limit).await {
Ok(bytes) => {
self.stats
.bytes_total
diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs
index 7f298dc491..7c12f3ecfd 100644
--- a/core/src/layers/prometheus_client.rs
+++ b/core/src/layers/prometheus_client.rs
@@ -539,8 +539,8 @@ impl<R> PrometheusMetricWrapper<R> {
}
impl<R: oio::Read> oio::Read for PrometheusMetricWrapper<R> {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- match self.inner.read(size).await {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ match self.inner.read(limit).await {
Ok(bytes) => {
self.bytes_total += bytes.len();
Ok(bytes)
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index bb5256e7eb..a563ebd799 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -703,14 +703,14 @@ impl<R: oio::Read, I: RetryInterceptor> oio::Read for RetryWrapper<R, I> {
res
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
use backon::RetryableWithContext;
let inner = self.inner.take().expect("inner must be valid");
let (inner, res) = {
|mut r: R| async move {
- let res = r.read(size).await;
+ let res = r.read(limit).await;
(r, res)
}
diff --git a/core/src/layers/throttle.rs b/core/src/layers/throttle.rs
index 55b13959eb..3e42839f46 100644
--- a/core/src/layers/throttle.rs
+++ b/core/src/layers/throttle.rs
@@ -185,9 +185,9 @@ impl<R> ThrottleWrapper<R> {
}
impl<R: oio::Read> oio::Read for ThrottleWrapper<R> {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
// TODO: How can we handle buffer reads with a limiter?
- self.inner.read(size).await
+ self.inner.read(limit).await
}
async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index 8b1e5d3b21..40a3318f4d 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -327,8 +327,8 @@ impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
Self::io_timeout(self.timeout, ReadOperation::Seek.into_static(), fut).await
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- let fut = self.inner.read(size);
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ let fut = self.inner.read(limit);
Self::io_timeout(self.timeout, ReadOperation::Next.into_static(), fut).await
}
}
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index 4dc4359946..6dfd34138d 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -272,8 +272,8 @@ impl<R: oio::Read> oio::Read for TracingWrapper<R> {
parent = &self.span,
level = "trace",
skip_all)]
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- self.inner.read(size).await
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ self.inner.read(limit).await
}
#[tracing::instrument(
diff --git a/core/src/raw/enum_utils.rs b/core/src/raw/enum_utils.rs
index ca6ec68c2f..e2d2a10dff 100644
--- a/core/src/raw/enum_utils.rs
+++ b/core/src/raw/enum_utils.rs
@@ -65,10 +65,10 @@ impl<ONE: oio::Read, TWO: oio::Read> oio::Read for TwoWays<ONE, TWO> {
}
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
match self {
- Self::One(v) => v.read(size).await,
- Self::Two(v) => v.read(size).await,
+ Self::One(v) => v.read(limit).await,
+ Self::Two(v) => v.read(limit).await,
}
}
}
@@ -140,11 +140,11 @@ impl<ONE: oio::Read, TWO: oio::Read, THREE: oio::Read> oio::Read for ThreeWays<O
}
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
match self {
- Self::One(v) => v.read(size).await,
- Self::Two(v) => v.read(size).await,
- Self::Three(v) => v.read(size).await,
+ Self::One(v) => v.read(limit).await,
+ Self::Two(v) => v.read(limit).await,
+ Self::Three(v) => v.read(limit).await,
}
}
}
@@ -235,12 +235,12 @@ where
}
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
match self {
- Self::One(v) => v.read(size).await,
- Self::Two(v) => v.read(size).await,
- Self::Three(v) => v.read(size).await,
- Self::Four(v) => v.read(size).await,
+ Self::One(v) => v.read(limit).await,
+ Self::Two(v) => v.read(limit).await,
+ Self::Three(v) => v.read(limit).await,
+ Self::Four(v) => v.read(limit).await,
}
}
}
diff --git a/core/src/raw/http_util/body.rs b/core/src/raw/http_util/body.rs
index ac0174090d..4c39dfa74e 100644
--- a/core/src/raw/http_util/body.rs
+++ b/core/src/raw/http_util/body.rs
@@ -164,7 +164,7 @@ impl IncomingAsyncBody {
}
impl oio::Read for IncomingAsyncBody {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
if self.size == Some(0) {
return Ok(Bytes::new());
}
@@ -182,7 +182,7 @@ impl oio::Read for IncomingAsyncBody {
};
}
- let size = min(size, self.chunk.len());
+ let size = min(limit, self.chunk.len());
self.consumed += size as u64;
let bs = self.chunk.split_to(size);
Ok(bs)
diff --git a/core/src/raw/oio/cursor.rs b/core/src/raw/oio/cursor.rs
index a8a82a0a57..a920652451 100644
--- a/core/src/raw/oio/cursor.rs
+++ b/core/src/raw/oio/cursor.rs
@@ -92,13 +92,13 @@ impl oio::Read for Cursor {
Ok(n)
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
if self.is_empty() {
Ok(Bytes::new())
} else {
// The clone here is required as we don't want to change it.
let mut bs = self.inner.clone().split_off(self.pos as usize);
- let bs = bs.split_to(min(bs.len(), size));
+ let bs = bs.split_to(min(bs.len(), limit));
self.pos += bs.len() as u64;
Ok(bs)
}
diff --git a/core/src/raw/oio/read/api.rs b/core/src/raw/oio/read/api.rs
index d7f681fe96..233e86a918 100644
--- a/core/src/raw/oio/read/api.rs
+++ b/core/src/raw/oio/read/api.rs
@@ -76,29 +76,33 @@ impl From<ReadOperation> for &'static str {
/// Reader is a type erased [`Read`].
pub type Reader = Box<dyn ReadDyn>;
-/// Read is the trait that OpenDAL returns to callers.
+/// Read is the internal trait used by OpenDAL to read data from storage.
///
-/// Read is compose of the following trait
+/// Users should not use or import this trait unless they are implementing an `Accessor`.
///
-/// - `AsyncRead`
-/// - `AsyncSeek`
-/// - `Stream<Item = Result<Bytes>>`
+/// # Notes
///
-/// `AsyncRead` is required to be implemented, `AsyncSeek` and `Stream`
-/// is optional. We use `Read` to make users life easier.
+/// ## Object Safety
+///
+/// `Read` uses `async in trait`, making it not object safe, preventing the use of `Box<dyn Read>`.
+/// To address this, we've introduced [`ReadDyn`] and its compatible type `Box<dyn ReadDyn>`.
+///
+/// `ReadDyn` uses `Box::pin()` to transform the returned future into a [`BoxedFuture`], introducing
+/// an additional layer of indirection and an extra allocation. Ideally, `ReadDyn` should occur only
+/// once, at the outermost level of our API.
pub trait Read: Unpin + Send + Sync {
/// Fetch more bytes from underlying reader.
///
- /// `size` is used to hint the data that user want to read at most. Implementer
- /// MUST NOT return more than `size` bytes. However, implementer can decide
+ /// `limit` is used to hint the data that user want to read at most. Implementer
+ /// MUST NOT return more than `limit` bytes. However, implementer can decide
/// whether to split or merge the read requests underground.
///
/// Returning `bytes`'s `length == 0` means:
///
/// - This reader has reached its “end of file” and will likely no longer be able to produce bytes.
- /// - The `size` specified was `0`.
+ /// - The `limit` specified was `0`.
#[cfg(not(target_arch = "wasm32"))]
- fn read(&mut self, size: usize) -> impl Future<Output = Result<Bytes>> + Send;
+ fn read(&mut self, limit: usize) -> impl Future<Output = Result<Bytes>> + Send;
#[cfg(target_arch = "wasm32")]
fn read(&mut self, size: usize) -> impl Future<Output = Result<Bytes>>;
@@ -112,8 +116,8 @@ pub trait Read: Unpin + Send + Sync {
}
impl Read for () {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- let _ = size;
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ let _ = limit;
Err(Error::new(
ErrorKind::Unsupported,
@@ -132,14 +136,14 @@ impl Read for () {
}
pub trait ReadDyn: Unpin + Send + Sync {
- fn read_dyn(&mut self, size: usize) -> BoxedFuture<Result<Bytes>>;
+ fn read_dyn(&mut self, limit: usize) -> BoxedFuture<Result<Bytes>>;
fn seek_dyn(&mut self, pos: io::SeekFrom) -> BoxedFuture<Result<u64>>;
}
impl<T: Read + ?Sized> ReadDyn for T {
- fn read_dyn(&mut self, size: usize) -> BoxedFuture<Result<Bytes>> {
- Box::pin(self.read(size))
+ fn read_dyn(&mut self, limit: usize) -> BoxedFuture<Result<Bytes>> {
+ Box::pin(self.read(limit))
}
fn seek_dyn(&mut self, pos: io::SeekFrom) -> BoxedFuture<Result<u64>> {
@@ -152,8 +156,8 @@ impl<T: Read + ?Sized> ReadDyn for T {
/// Take care about the `deref_mut()` here. This makes sure that we are calling functions
/// 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, size: usize) -> Result<Bytes> {
- self.deref_mut().read_dyn(size).await
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ self.deref_mut().read_dyn(limit).await
}
async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
diff --git a/core/src/raw/oio/read/buffer_reader.rs b/core/src/raw/oio/read/buffer_reader.rs
index d6d723fa7b..33709b2355 100644
--- a/core/src/raw/oio/read/buffer_reader.rs
+++ b/core/src/raw/oio/read/buffer_reader.rs
@@ -160,16 +160,16 @@ where
}
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- if size == 0 {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ if limit == 0 {
return Ok(Bytes::new());
}
// If we don't have any buffered data and we're doing a massive read
// (larger than our internal buffer), bypass our internal buffer
// entirely.
- if self.pos == self.filled && size >= self.capacity() {
- let res = self.r.read(size).await;
+ if self.pos == self.filled && limit >= self.capacity() {
+ let res = self.r.read(limit).await;
self.discard_buffer();
return match res {
Ok(bs) => {
@@ -185,7 +185,7 @@ where
if bytes.is_empty() {
return Ok(Bytes::new());
}
- let size = min(bytes.len(), size);
+ let size = min(bytes.len(), limit);
let bytes = Bytes::copy_from_slice(&bytes[..size]);
self.consume(bytes.len());
Ok(bytes)
@@ -391,8 +391,8 @@ mod tests {
))
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- oio::Read::read(&mut self.inner, size).await
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ oio::Read::read(&mut self.inner, limit).await
}
}
diff --git a/core/src/raw/oio/read/file_read.rs b/core/src/raw/oio/read/file_read.rs
index bb43e7be70..6a27733634 100644
--- a/core/src/raw/oio/read/file_read.rs
+++ b/core/src/raw/oio/read/file_read.rs
@@ -246,7 +246,7 @@ where
Ok(self.cur)
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
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(..));
@@ -261,7 +261,7 @@ where
(self.offset, self.size) = Self::offset(r, self.op.range()).await?;
}
- r.read(size).await
+ r.read(limit).await
}
}
diff --git a/core/src/raw/oio/read/futures_read.rs b/core/src/raw/oio/read/futures_read.rs
index 81f53447a0..7b1bd2e6aa 100644
--- a/core/src/raw/oio/read/futures_read.rs
+++ b/core/src/raw/oio/read/futures_read.rs
@@ -55,17 +55,17 @@ where
})
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
// Make sure buf has enough space.
- if self.buf.capacity() < size {
- self.buf.reserve(size);
+ if self.buf.capacity() < limit {
+ self.buf.reserve(limit);
}
let buf = self.buf.spare_capacity_mut();
let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
// SAFETY: Read at most `size` bytes into `read_buf`.
unsafe {
- read_buf.assume_init(size);
+ read_buf.assume_init(limit);
}
let n = self
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 a130de2ee4..ca1c43e9b3 100644
--- a/core/src/raw/oio/read/into_read_from_stream.rs
+++ b/core/src/raw/oio/read/into_read_from_stream.rs
@@ -50,7 +50,7 @@ where
))
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
if self.buf.is_empty() {
self.buf = match self.inner.next().await.transpose()? {
Some(v) => v.into(),
@@ -58,7 +58,7 @@ where
};
}
- let bs = self.buf.split_to(min(size, self.buf.len()));
+ let bs = self.buf.split_to(min(limit, self.buf.len()));
Ok(bs)
}
}
diff --git a/core/src/raw/oio/read/into_streamable_read.rs b/core/src/raw/oio/read/into_streamable_read.rs
index 890b129d71..e234eb654c 100644
--- a/core/src/raw/oio/read/into_streamable_read.rs
+++ b/core/src/raw/oio/read/into_streamable_read.rs
@@ -45,8 +45,8 @@ impl<R: oio::Read> oio::Read for StreamableReader<R> {
self.r.seek(pos).await
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- let size = min(self.buf.capacity(), size);
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ let size = min(self.buf.capacity(), limit);
let dst = self.buf.spare_capacity_mut();
let mut buf = ReadBuf::uninit(dst);
diff --git a/core/src/raw/oio/read/lazy_read.rs b/core/src/raw/oio/read/lazy_read.rs
index b39929d3a5..89be1a3b6f 100644
--- a/core/src/raw/oio/read/lazy_read.rs
+++ b/core/src/raw/oio/read/lazy_read.rs
@@ -73,9 +73,9 @@ where
self.reader().await?.seek(pos).await
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
let r = self.reader().await?;
- r.read(size).await
+ r.read(limit).await
}
}
diff --git a/core/src/raw/oio/read/range_read.rs b/core/src/raw/oio/read/range_read.rs
index d557cf0e3d..d0551f86d2 100644
--- a/core/src/raw/oio/read/range_read.rs
+++ b/core/src/raw/oio/read/range_read.rs
@@ -293,7 +293,7 @@ where
Ok(self.cur)
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
// Sanity check for normal cases.
if self.cur >= self.size.unwrap_or(u64::MAX) {
return Ok(Bytes::new());
@@ -318,7 +318,7 @@ where
}
let r = self.reader.as_mut().expect("reader must be valid");
- match r.read(size).await {
+ match r.read(limit).await {
Ok(bs) => {
self.cur += bs.len() as u64;
Ok(bs)
@@ -538,8 +538,8 @@ mod tests {
))
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- let mut bs = vec![0; size];
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ let mut bs = vec![0; limit];
let n = self.inner.read(&mut bs).await.map_err(|err| {
Error::new(ErrorKind::Unexpected, "read data from mock").set_source(err)
})?;
diff --git a/core/src/raw/oio/read/tokio_read.rs b/core/src/raw/oio/read/tokio_read.rs
index 71d5340e20..6dea5cec39 100644
--- a/core/src/raw/oio/read/tokio_read.rs
+++ b/core/src/raw/oio/read/tokio_read.rs
@@ -55,17 +55,17 @@ where
})
}
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
// Make sure buf has enough space.
- if self.buf.capacity() < size {
- self.buf.reserve(size);
+ if self.buf.capacity() < limit {
+ self.buf.reserve(limit);
}
let buf = self.buf.spare_capacity_mut();
let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
// SAFETY: Read at most `size` bytes into `read_buf`.
unsafe {
- read_buf.assume_init(size);
+ read_buf.assume_init(limit);
}
let n = self
diff --git a/core/src/services/dbfs/reader.rs b/core/src/services/dbfs/reader.rs
index bac1c4be73..856d0eaed4 100644
--- a/core/src/services/dbfs/reader.rs
+++ b/core/src/services/dbfs/reader.rs
@@ -86,8 +86,8 @@ impl DbfsReader {
unsafe impl Sync for DbfsReader {}
impl oio::Read for DbfsReader {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- let _ = size;
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ let _ = limit;
Err(Error::new(
ErrorKind::Unsupported,
diff --git a/core/src/services/ftp/util.rs b/core/src/services/ftp/util.rs
index 3e46abe513..3a7e5abac9 100644
--- a/core/src/services/ftp/util.rs
+++ b/core/src/services/ftp/util.rs
@@ -53,7 +53,7 @@ impl FtpReader {
}
impl oio::Read for FtpReader {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
if self.conn.is_none() {
return Err(Error::new(
ErrorKind::Unexpected,
@@ -62,15 +62,15 @@ impl oio::Read for FtpReader {
}
// Make sure buf has enough space.
- if self.buf.capacity() < size {
- self.buf.reserve(size);
+ if self.buf.capacity() < limit {
+ self.buf.reserve(limit);
}
let buf = self.buf.spare_capacity_mut();
let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
// SAFETY: Read at most `size` bytes into `read_buf`.
unsafe {
- read_buf.assume_init(size);
+ read_buf.assume_init(limit);
}
let data = self.reader.read(read_buf.initialize_unfilled()).await;
diff --git a/core/src/services/hdfs_native/reader.rs b/core/src/services/hdfs_native/reader.rs
index c661d45229..9ae5ed020a 100644
--- a/core/src/services/hdfs_native/reader.rs
+++ b/core/src/services/hdfs_native/reader.rs
@@ -34,8 +34,8 @@ impl HdfsNativeReader {
}
impl Read for HdfsNativeReader {
- async fn read(&mut self, size: usize) -> Result<Bytes> {
- let _ = size;
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ let _ = limit;
todo!()
}
diff --git a/core/src/types/reader.rs b/core/src/types/reader.rs
index 20236bed47..10af69f955 100644
--- a/core/src/types/reader.rs
+++ b/core/src/types/reader.rs
@@ -121,11 +121,11 @@ impl Reader {
/// Read at most `size` bytes of data from reader.
#[inline]
- pub async fn read(&mut self, size: usize) -> Result<Bytes> {
+ pub async fn read(&mut self, limit: usize) -> Result<Bytes> {
let State::Idle(Some(r)) = &mut self.state else {
return Err(Error::new(ErrorKind::Unexpected, "reader must be valid"));
};
- r.read_dyn(size).await
+ r.read_dyn(limit).await
}
/// Read exact `size` bytes of data from reader.
(opendal) 02/03: Remove not exist op
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 4409085effff246facf3fbcdcffd90b7211bb42d
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Mar 12 13:47:11 2024 +0800
Remove not exist op
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/error_context.rs | 12 ++++++------
core/src/layers/logging.rs | 36 ++++++++++++++++++------------------
core/src/layers/retry.rs | 2 +-
core/src/layers/timeout.rs | 10 +++++-----
core/src/raw/oio/read/api.rs | 7 ++-----
5 files changed, 32 insertions(+), 35 deletions(-)
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 7e0e65a187..43b08ea540 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -349,17 +349,17 @@ pub struct ErrorContextWrapper<T> {
}
impl<T: oio::Read> oio::Read for ErrorContextWrapper<T> {
- async fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
- self.inner.seek(pos).await.map_err(|err| {
- err.with_operation(ReadOperation::Seek)
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ self.inner.read(limit).await.map_err(|err| {
+ err.with_operation(ReadOperation::Read)
.with_context("service", self.scheme)
.with_context("path", &self.path)
})
}
- async fn read(&mut self, limit: usize) -> Result<Bytes> {
- self.inner.read(limit).await.map_err(|err| {
- err.with_operation(ReadOperation::Next)
+ 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)
})
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 9bfecf5b2c..2f9fed02b1 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -991,27 +991,29 @@ impl<R> Drop for LoggingReader<R> {
}
impl<R: oio::Read> oio::Read for LoggingReader<R> {
- async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
- match self.inner.seek(pos).await {
- Ok(n) => {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ match self.inner.read(limit).await {
+ Ok(bs) => {
+ self.read += bs.len() as u64;
trace!(
target: LOGGING_TARGET,
- "service={} operation={} path={} read={} -> seek to {pos:?}, current offset {n}",
+ "service={} operation={} path={} read={} -> next returns {}B",
self.ctx.scheme,
- ReadOperation::Seek,
+ ReadOperation::Read,
self.path,
self.read,
+ bs.len()
);
- Ok(n)
+ Ok(bs)
}
Err(err) => {
if let Some(lvl) = self.ctx.error_level(&err) {
log!(
target: LOGGING_TARGET,
lvl,
- "service={} operation={} path={} read={} -> seek to {pos:?} failed: {}",
+ "service={} operation={} path={} read={} -> next failed: {}",
self.ctx.scheme,
- ReadOperation::Seek,
+ ReadOperation::Read,
self.path,
self.read,
self.ctx.error_print(&err),
@@ -1022,29 +1024,27 @@ impl<R: oio::Read> oio::Read for LoggingReader<R> {
}
}
- async fn read(&mut self, limit: usize) -> Result<Bytes> {
- match self.inner.read(limit).await {
- Ok(bs) => {
- self.read += bs.len() as u64;
+ 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={} -> next returns {}B",
+ "service={} operation={} path={} read={} -> seek to {pos:?}, current offset {n}",
self.ctx.scheme,
- ReadOperation::Next,
+ ReadOperation::Seek,
self.path,
self.read,
- bs.len()
);
- Ok(bs)
+ Ok(n)
}
Err(err) => {
if let Some(lvl) = self.ctx.error_level(&err) {
log!(
target: LOGGING_TARGET,
lvl,
- "service={} operation={} path={} read={} -> next failed: {}",
+ "service={} operation={} path={} read={} -> seek to {pos:?} failed: {}",
self.ctx.scheme,
- ReadOperation::Next,
+ ReadOperation::Seek,
self.path,
self.read,
self.ctx.error_print(&err),
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index a563ebd799..5c4fce8dcd 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -723,7 +723,7 @@ impl<R: oio::Read, I: RetryInterceptor> oio::Read for RetryWrapper<R, I> {
err,
dur,
&[
- ("operation", ReadOperation::Next.into_static()),
+ ("operation", ReadOperation::Read.into_static()),
("path", &self.path),
],
)
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index 40a3318f4d..f80d29cf1b 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -322,15 +322,15 @@ impl<R> TimeoutWrapper<R> {
}
impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
+ async fn read(&mut self, limit: usize) -> Result<Bytes> {
+ let fut = self.inner.read(limit);
+ Self::io_timeout(self.timeout, ReadOperation::Read.into_static(), fut).await
+ }
+
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
}
-
- async fn read(&mut self, limit: usize) -> Result<Bytes> {
- let fut = self.inner.read(limit);
- Self::io_timeout(self.timeout, ReadOperation::Next.into_static(), fut).await
- }
}
impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
diff --git a/core/src/raw/oio/read/api.rs b/core/src/raw/oio/read/api.rs
index 233e86a918..22865375ab 100644
--- a/core/src/raw/oio/read/api.rs
+++ b/core/src/raw/oio/read/api.rs
@@ -31,12 +31,10 @@ use crate::*;
#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq)]
#[non_exhaustive]
pub enum ReadOperation {
- /// Operation for [`Read::poll_read`]
+ /// Operation for [`Read::read`]
Read,
- /// Operation for [`Read::poll_seek`]
+ /// Operation for [`Read::seek`]
Seek,
- /// Operation for [`Read::poll_next`]
- Next,
/// Operation for [`BlockingRead::read`]
BlockingRead,
/// Operation for [`BlockingRead::seek`]
@@ -65,7 +63,6 @@ impl From<ReadOperation> for &'static str {
match v {
Read => "Reader::read",
Seek => "Reader::seek",
- Next => "Reader::next",
BlockingRead => "BlockingReader::read",
BlockingSeek => "BlockingReader::seek",
BlockingNext => "BlockingReader::next",