upload multipart in parallel

Author: parmesant

src/storage/object_storage.rs

@@ -844,7 +844,11 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
                 let stream_relative_path = format!("{stream_name}/{file_suffix}");
 
                 // Try uploading the file, handle potential errors without breaking the loop
-                if let Err(e) = self.upload_file(&stream_relative_path, &path).await {
+                // if let Err(e) = self.upload_multipart(key, path)
+                if let Err(e) = self
+                    .upload_multipart(&RelativePathBuf::from(&stream_relative_path), &path)
+                    .await
+                {
                     error!("Failed to upload file {filename:?}: {e}");
                     continue; // Skip to the next file
                 }

src/storage/s3.rs

@@ -535,38 +535,30 @@ impl S3 {
             let mut data = Vec::new();
             file.read_to_end(&mut data).await?;
             self.client.put(location, data.into()).await?;
-            // async_writer.put_part(data.into()).await?;
-            // async_writer.complete().await?;
+
             return Ok(());
         } else {
-            let mut data = Vec::new();
-            file.read_to_end(&mut data).await?;
+            let mut buf = [0; MIN_MULTIPART_UPLOAD_SIZE];
 
-            // let mut upload_parts = Vec::new();
+            let mut tasks = FuturesUnordered::new();
+            // let mut tasks = Vec::new();
 
-            let has_final_partial_part = total_size % MIN_MULTIPART_UPLOAD_SIZE > 0;
-            let num_full_parts = total_size / MIN_MULTIPART_UPLOAD_SIZE;
-            let total_parts = num_full_parts + if has_final_partial_part { 1 } else { 0 };
-
-            // Upload each part
-            for part_number in 0..(total_parts) {
-                let start_pos = part_number * MIN_MULTIPART_UPLOAD_SIZE;
-                let end_pos = if part_number == num_full_parts && has_final_partial_part {
-                    // Last part might be smaller than 5MB (which is allowed)
-                    total_size
-                } else {
-                    // All other parts must be at least 5MB
-                    start_pos + MIN_MULTIPART_UPLOAD_SIZE
-                };
+            while let Ok(b) = file.read(&mut buf).await {
+                if b == 0 {
+                    break;
+                }
 
                 // Extract this part's data
-                let part_data = data[start_pos..end_pos].to_vec();
+                let part_data = buf.to_vec();
 
                 // Upload the part
-                async_writer.put_part(part_data.into()).await?;
+                tasks.push(tokio::spawn(async_writer.put_part(part_data.into())));
+            }
 
-                // upload_parts.push(part_number as u64 + 1);
+            while let Some(res) = tasks.next().await {
+                res??;
             }
+
             async_writer.complete().await?;
         }
         Ok(())