feat: implement concurrent MultipartUploadWriter

apache · Jan 4, 2024 · 7004275 · 7004275
1 parent 996ee16
commit 7004275
Show file tree

Hide file tree

Showing 7 changed files with 183 additions and 108 deletions.
diff --git a/core/src/raw/oio/write/multipart_upload_write.rs b/core/src/raw/oio/write/multipart_upload_write.rs
diff --git a/core/src/services/b2/backend.rs b/core/src/services/b2/backend.rs
@@ -376,9 +376,10 @@ impl Accessor for B2Backend {
     }
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let concurrent = args.concurrent();
         let writer = B2Writer::new(self.core.clone(), path, args);
 
-        let w = oio::MultipartUploadWriter::new(writer);
+        let w = oio::MultipartUploadWriter::new(writer, concurrent);
 
         Ok((RpWrite::default(), w))
     }

diff --git a/core/src/services/cos/backend.rs b/core/src/services/cos/backend.rs
@@ -340,7 +340,7 @@ impl Accessor for CosBackend {
         let w = if args.append() {
             CosWriters::Two(oio::AppendObjectWriter::new(writer))
         } else {
-            CosWriters::One(oio::MultipartUploadWriter::new(writer))
+            CosWriters::One(oio::MultipartUploadWriter::new(writer, args.concurrent()))
         };
 
         Ok((RpWrite::default(), w))

diff --git a/core/src/services/obs/backend.rs b/core/src/services/obs/backend.rs
@@ -350,7 +350,7 @@ impl Accessor for ObsBackend {
         let w = if args.append() {
             ObsWriters::Two(oio::AppendObjectWriter::new(writer))
         } else {
-            ObsWriters::One(oio::MultipartUploadWriter::new(writer))
+            ObsWriters::One(oio::MultipartUploadWriter::new(writer, args.concurrent()))
         };
 
         Ok((RpWrite::default(), w))

diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
@@ -492,7 +492,7 @@ impl Accessor for OssBackend {
         let w = if args.append() {
             OssWriters::Two(oio::AppendObjectWriter::new(writer))
         } else {
-            OssWriters::One(oio::MultipartUploadWriter::new(writer))
+            OssWriters::One(oio::MultipartUploadWriter::new(writer, args.concurrent()))
         };
 
         Ok((RpWrite::default(), w))

diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
@@ -1085,9 +1085,10 @@ impl Accessor for S3Backend {
     }
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let concurrent = args.concurrent();
         let writer = S3Writer::new(self.core.clone(), path, args);
 
-        let w = oio::MultipartUploadWriter::new(writer);
+        let w = oio::MultipartUploadWriter::new(writer, concurrent);
 
         Ok((RpWrite::default(), w))
     }

diff --git a/core/src/services/upyun/backend.rs b/core/src/services/upyun/backend.rs
@@ -316,9 +316,10 @@ impl Accessor for UpyunBackend {
     }
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let concurrent = args.concurrent();
         let writer = UpyunWriter::new(self.core.clone(), args, path.to_string());
 
-        let w = oio::MultipartUploadWriter::new(writer);
+        let w = oio::MultipartUploadWriter::new(writer, concurrent);
 
         Ok((RpWrite::default(), w))
     }