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:13 UTC
(opendal) 01/03: Polish comments
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.