feat: implement concurrent MultipartUploadWriter

core/src/raw/ops.rs

@@ -516,6 +516,7 @@ impl OpStat {
 pub struct OpWrite {
     append: bool,
     buffer: Option<usize>,
+    concurrent: usize,
 
     content_type: Option<String>,
     content_disposition: Option<String>,
@@ -601,6 +602,17 @@ impl OpWrite {
         self.cache_control = Some(cache_control.to_string());
         self
     }
+
+    /// Get the concurrent.
+    pub fn concurrent(&self) -> usize {
+        self.concurrent
+    }
+
+    /// Set the maximum concurrent write task amount.
+    pub fn with_concurrent(mut self, concurrent: usize) -> Self {
+        self.concurrent = concurrent;
+        self
+    }
 }
 
 /// Args for `copy` operation.

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))
     }

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))

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))

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))

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))
     }

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))
     }

core/src/types/operator/operator_futures.rs

@@ -478,6 +478,12 @@ impl FutureWrite {
             .map_args(|(args, bs)| (args.with_cache_control(v), bs));
         self
     }
+
+    /// Set the maximum concurrent write task amount.
+    pub fn concurrent(mut self, v: usize) -> Self {
+        self.0 = self.0.map_args(|(args, bs)| (args.with_buffer(v), bs));
+        self
+    }
 }
 
 impl Future for FutureWrite {
@@ -543,6 +549,12 @@ impl FutureWriter {
         self.0 = self.0.map_args(|args| args.with_cache_control(v));
         self
     }
+
+    /// Set the maximum concurrent write task amount.
+    pub fn concurrent(mut self, v: usize) -> Self {
+        self.0 = self.0.map_args(|args| args.with_concurrent(v));
+        self
+    }
 }
 
 impl Future for FutureWriter {