(opendal) branch fix-gcs-concurrent-write created (now 5bd3806171)

[xu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xuanwo pushed a change to branch fix-gcs-concurrent-write
in repository https://gitbox.apache.org/repos/asf/opendal.git


      at 5bd3806171 fix(core): Gcs's RangeWrite doesn't support concurrent write

This branch includes the following new commits:

     new 5bd3806171 fix(core): Gcs's RangeWrite doesn't support concurrent write

The 1 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) 01/01: fix(core): Gcs's RangeWrite doesn't support concurrent write

[xu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xuanwo pushed a commit to branch fix-gcs-concurrent-write
in repository https://gitbox.apache.org/repos/asf/opendal.git

commit 5bd3806171cb73f32cfc3a1259445c7b094d5e3c
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Jun 26 01:05:56 2024 +0800

    fix(core): Gcs's RangeWrite doesn't support concurrent write
    
    Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
 core/src/services/gcs/backend.rs   |  6 +++---
 core/tests/behavior/async_write.rs | 26 ++++++++++++++++++--------
 2 files changed, 21 insertions(+), 11 deletions(-)

diff --git a/core/src/services/gcs/backend.rs b/core/src/services/gcs/backend.rs
index 3f1cb93724..248de35916 100644
--- a/core/src/services/gcs/backend.rs
+++ b/core/src/services/gcs/backend.rs
@@ -362,7 +362,7 @@ impl Access for GcsBackend {
                 // It's recommended that you use at least 8 MiB for the chunk size.
                 //
                 // Reference: [Perform resumable uploads](https://cloud.google.com/storage/docs/performing-resumable-uploads)
-                write_multi_align_size: Some(256 * 1024 * 1024),
+                write_multi_align_size: Some(8 * 1024 * 1024),
 
                 delete: true,
                 copy: true,
@@ -433,10 +433,10 @@ impl Access for GcsBackend {
     }
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
-        let concurrent = args.concurrent();
         let executor = args.executor().cloned();
         let w = GcsWriter::new(self.core.clone(), path, args);
-        let w = oio::RangeWriter::new(w, executor, concurrent);
+        // Gcs can't support concurrent write, always use concurrent=1 for now.
+        let w = oio::RangeWriter::new(w, executor, 1);
 
         Ok((RpWrite::default(), w))
     }
diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs
index c95fdd5f82..969ea31398 100644
--- a/core/tests/behavior/async_write.rs
+++ b/core/tests/behavior/async_write.rs
@@ -300,30 +300,40 @@ pub async fn test_writer_write_with_concurrent(op: Operator) -> Result<()> {
     }
 
     let path = TEST_FIXTURE.new_file_path();
-    let size = 5 * 1024 * 1024; // write file with 5 MiB
-    let content_a = gen_fixed_bytes(size);
-    let content_b = gen_fixed_bytes(size);
+    // We need at least 3 part to make sure concurrent happened.
+    let (content_a, size_a) = gen_bytes_with_range(5 * 1024 * 1024..6 * 1024 * 1024);
+    let (content_b, size_b) = gen_bytes_with_range(5 * 1024 * 1024..6 * 1024 * 1024);
+    let (content_c, size_c) = gen_bytes_with_range(5 * 1024 * 1024..6 * 1024 * 1024);
 
-    let mut w = op.writer_with(&path).concurrent(2).await?;
+    let mut w = op.writer_with(&path).concurrent(3).await?;
     w.write(content_a.clone()).await?;
     w.write(content_b.clone()).await?;
+    w.write(content_c.clone()).await?;
     w.close().await?;
 
     let meta = op.stat(&path).await.expect("stat must succeed");
-    assert_eq!(meta.content_length(), (size * 2) as u64);
+    assert_eq!(meta.content_length(), (size_a + size_b + size_c) as u64);
 
     let bs = op.read(&path).await?.to_bytes();
-    assert_eq!(bs.len(), size * 2, "read size");
+    assert_eq!(bs.len(), size_a + size_b + size_c, "read size");
     assert_eq!(
-        format!("{:x}", Sha256::digest(&bs[..size])),
+        format!("{:x}", Sha256::digest(&bs[..size_a])),
         format!("{:x}", Sha256::digest(content_a)),
         "read content a"
     );
     assert_eq!(
-        format!("{:x}", Sha256::digest(&bs[size..])),
+        format!("{:x}", Sha256::digest(&bs[size_a..size_a + size_b])),
         format!("{:x}", Sha256::digest(content_b)),
         "read content b"
     );
+    assert_eq!(
+        format!(
+            "{:x}",
+            Sha256::digest(&bs[size_a + size_b..size_a + size_b + size_c])
+        ),
+        format!("{:x}", Sha256::digest(content_c)),
+        "read content b"
+    );
 
     Ok(())
 }