refactor(core/raw): Migrate oio::Read to async in trait

.github/workflows/ci_core.yml

@@ -80,8 +80,8 @@ jobs:
   check_msrv:
     runs-on: ubuntu-latest
     env:
-      # OpenDAL's MSRV is 1.67.
-      OPENDAL_MSRV: "1.67"
+      # OpenDAL's MSRV is 1.75.
+      OPENDAL_MSRV: "1.75"
     steps:
       - uses: actions/checkout@v4
       - name: Setup msrv of rust

bin/oay/Cargo.toml

@@ -26,7 +26,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.41.1"
 
 [features]

bin/ofs/Cargo.toml

@@ -27,7 +27,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 
 [dependencies]
 anyhow = "1"

bin/oli/Cargo.toml

@@ -26,7 +26,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.41.1"
 
 [features]

bindings/c/Cargo.toml

@@ -24,7 +24,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.44.3"
 
 [lib]

bindings/cpp/Cargo.toml

@@ -24,7 +24,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.44.3"
 
 [lib]

bindings/dotnet/Cargo.toml

@@ -25,7 +25,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 
 [lib]
 crate-type = ["cdylib"]

bindings/haskell/Cargo.toml

@@ -24,7 +24,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.44.3"
 
 [lib]

bindings/java/Cargo.toml

@@ -24,7 +24,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.45.1"
 
 [lib]

bindings/lua/Cargo.toml

@@ -25,7 +25,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 
 [features]
 default = ["mlua/lua52"]

bindings/nodejs/Cargo.toml

@@ -24,7 +24,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.45.1"
 
 [features]

bindings/nodejs/generated.d.ts

@@ -563,6 +563,8 @@ export class Reader {
    * > &mut self in async napi methods should be marked as unsafe
    *
    * Read bytes from this reader into given buffer.
+   *
+   * TODO: change api into stream based.
    */
   read(buf: Buffer): Promise<bigint>
 }

bindings/nodejs/src/lib.rs

@@ -27,7 +27,6 @@ use std::time::Duration;
 use futures::TryStreamExt;
 use napi::bindgen_prelude::*;
 use opendal::raw::oio::BlockingRead;
-use opendal::raw::oio::ReadExt;
 
 #[napi]
 pub struct Operator(opendal::Operator);
@@ -665,9 +664,14 @@ impl Reader {
     /// > &mut self in async napi methods should be marked as unsafe
     ///
     /// Read bytes from this reader into given buffer.
+    ///
+    /// TODO: change api into stream based.
     #[napi]
     pub async unsafe fn read(&mut self, mut buf: Buffer) -> Result<usize> {
-        self.0.read(buf.as_mut()).await.map_err(format_napi_error)
+        let buf = buf.as_mut();
+        let bs = self.0.read(buf.len()).await.map_err(format_napi_error)?;
+        buf[..bs.len()].copy_from_slice(&bs);
+        Ok(bs.len())
     }
 }
 

bindings/ocaml/Cargo.toml

@@ -25,7 +25,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 
 [lib]
 crate-type = ["staticlib", "cdylib"]

bindings/php/Cargo.toml

@@ -25,7 +25,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 
 [lib]
 crate-type = ["cdylib"]

bindings/python/Cargo.toml

@@ -24,7 +24,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.45.1"
 
 [features]

bindings/python/src/file.rs

@@ -25,7 +25,6 @@ use std::io::Write;
 use std::ops::DerefMut;
 use std::sync::Arc;
 
-use futures::AsyncReadExt;
 use futures::AsyncSeekExt;
 use futures::AsyncWriteExt;
 use pyo3::exceptions::PyIOError;
@@ -243,12 +242,11 @@ impl AsyncFile {
 
             let buffer = match size {
                 Some(size) => {
-                    let mut buffer = vec![0; size];
-                    reader
-                        .read_exact(&mut buffer)
+                    let buffer = reader
+                        .read_exact(size)
                         .await
                         .map_err(|err| PyIOError::new_err(err.to_string()))?;
-                    buffer
+                    buffer.to_vec()
                 }
                 None => {
                     let mut buffer = Vec::new();

bindings/ruby/Cargo.toml

@@ -25,7 +25,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 
 [lib]
 crate-type = ["cdylib"]

core/Cargo.toml

@@ -27,7 +27,7 @@ edition = "2021"
 homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.45.1"
 
 [package.metadata.docs.rs]
@@ -235,7 +235,7 @@ async-tls = { version = "0.13.0", optional = true }
 # Required dependencies
 anyhow = { version = "1.0.30", features = ["std"] }
 async-trait = "0.1.68"
-backon = "0.4.1"
+backon = "0.4.3"
 base64 = "0.21"
 bytes = "1.4"
 chrono = { version = "0.4.28", default-features = false, features = [
@@ -256,7 +256,7 @@ reqwest = { version = "0.11.18", features = [
 ], default-features = false }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-tokio = { version = "1.27", features = ["sync"] }
+tokio = { version = "1.27", features = ["sync", "io-util"] }
 uuid = { version = "1", features = ["serde", "v4"] }
 
 # Test only dependencies

core/benches/ops/read.rs

@@ -17,7 +17,6 @@
 
 use criterion::Criterion;
 use futures::io;
-use futures::AsyncReadExt;
 use opendal::raw::tests::init_test_service;
 use opendal::raw::tests::TEST_RUNTIME;
 use opendal::Operator;
@@ -112,25 +111,24 @@ fn bench_read_parallel(c: &mut Criterion, name: &str, op: Operator) {
         let content = gen_bytes(&mut rng, (size.bytes() * 2) as usize);
         let path = uuid::Uuid::new_v4().to_string();
         let offset = (size.bytes() / 2) as u64;
-        let buf = vec![0; size.bytes() as usize];
+        let buf_size = size.bytes() as usize;
         let temp_data = TempData::generate(op.clone(), &path, content.clone());
 
         for parallel in [1, 2, 4, 8, 16] {
             group.throughput(criterion::Throughput::Bytes(parallel * size.bytes() as u64));
             group.bench_with_input(
                 format!("{}x{}", parallel, size.to_string()),
-                &(op.clone(), &path, buf.clone()),
-                |b, (op, path, buf)| {
+                &(op.clone(), &path, buf_size),
+                |b, (op, path, buf_size)| {
                     b.to_async(&*TEST_RUNTIME).iter(|| async {
                         let futures = (0..parallel)
                             .map(|_| async {
-                                let mut buf = buf.clone();
                                 let mut r = op
                                     .reader_with(path)
                                     .range(offset..=offset + size.bytes() as u64)
                                     .await
                                     .unwrap();
-                                r.read_exact(&mut buf).await.unwrap();
+                                r.read_exact(*buf_size).await.unwrap();
 
                                 let mut d = 0;
                                 // mock same little cpu work

core/benches/vs_fs/Cargo.toml

@@ -21,7 +21,7 @@ edition = "2021"
 license = "Apache-2.0"
 name = "opendal-benchmark-vs-fs"
 publish = false
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.0.0"
 
 [dependencies]

core/benches/vs_s3/Cargo.toml

@@ -21,7 +21,7 @@ edition = "2021"
 license = "Apache-2.0"
 name = "opendal-benchmark-vs-s3"
 publish = false
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.0.0"
 
 [dependencies]

core/edge/file_write_on_full_disk/Cargo.toml

@@ -20,7 +20,7 @@ edition = "2021"
 license = "Apache-2.0"
 name = "edge_test_file_write_on_full_disk"
 publish = false
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.0.0"
 
 [dependencies]

core/edge/s3_aws_assume_role_with_web_identity/Cargo.toml

@@ -20,7 +20,7 @@ edition = "2021"
 license = "Apache-2.0"
 name = "edge_test_aws_s3_assume_role_with_web_identity"
 publish = false
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.0.0"
 
 [dependencies]

core/edge/s3_read_on_wasm/Cargo.toml

@@ -20,7 +20,7 @@ edition = "2021"
 license = "Apache-2.0"
 name = "edge_test_s3_read_on_wasm"
 publish = false
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.0.0"
 
 [lib]

core/fuzz/Cargo.toml

@@ -20,7 +20,7 @@ edition = "2021"
 license = "Apache-2.0"
 name = "opendal-fuzz"
 publish = false
-rust-version = "1.67"
+rust-version = "1.75"
 version = "0.0.0"
 
 [package.metadata]

core/fuzz/fuzz_reader.rs

@@ -96,27 +96,25 @@ impl Arbitrary<'_> for FuzzInput {
         let mut actions = vec![];
 
         for _ in 0..count {
-            let action = match u.int_in_range(0..=4)? {
+            let action = match u.int_in_range(0..=3)? {
                 // Read
                 0 => {
                     let size = u.int_in_range(0..=total_size * 2)?;
                     ReadAction::Read(size)
                 }
-                // Next
-                1 => ReadAction::Next,
                 // Seek Start
-                2 => {
+                1 => {
                     // NOTE: seek out of the end of file is valid.
                     let offset = u.int_in_range(0..=total_size * 2)?;
                     ReadAction::Seek(SeekFrom::Start(offset as u64))
                 }
                 // Seek Current
-                3 => {
+                2 => {
                     let offset = u.int_in_range(-(total_size as i64)..=(total_size as i64))?;
                     ReadAction::Seek(SeekFrom::Current(offset))
                 }
                 // Seek End
-                4 => {
+                3 => {
                     let offset = u.int_in_range(-(total_size as i64)..=(total_size as i64))?;
                     ReadAction::Seek(SeekFrom::End(offset))
                 }