chore: Cleanup of oio::Read, docs, comments, naming (#4345)

* Polish comments Signed-off-by: Xuanwo <github@xuanwo.io> * Remove not exist op Signed-off-by: Xuanwo <github@xuanwo.io> * add comments Signed-off-by: Xuanwo <github@xuanwo.io> --------- Signed-off-by: Xuanwo <github@xuanwo.io>
apache · Mar 12, 2024 · e79daa5 · e79daa5
1 parent e704a89
commit e79daa5
Show file tree

Hide file tree

Showing 30 changed files with 153 additions and 142 deletions.
diff --git 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
@@ -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
@@ -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
@@ -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, size: usize) -> Result<Bytes> {
-        self.inner.read(size).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
@@ -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, size: usize) -> Result<Bytes> {
-        match self.inner.read(size).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/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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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)
             }
@@ -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/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
@@ -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, size: usize) -> Result<Bytes> {
-        let fut = self.inner.read(size);
-        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/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
@@ -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
@@ -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
@@ -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)
         }