feat(ecstore): MultiWriter concurrent write

ecstore/src/erasure_coding/encode.rs

@@ -4,6 +4,8 @@ use crate::disk::error::Error;
 use crate::disk::error_reduce::count_errs;
 use crate::disk::error_reduce::{OBJECT_OP_IGNORED_ERRS, reduce_write_quorum_errs};
 use bytes::Bytes;
+use futures::StreamExt;
+use futures::stream::FuturesUnordered;
 use std::sync::Arc;
 use std::vec;
 use tokio::io::AsyncRead;
@@ -26,33 +28,41 @@ impl<'a> MultiWriter<'a> {
         }
     }
 
-    #[allow(clippy::needless_range_loop)]
-    pub async fn write(&mut self, data: Vec<Bytes>) -> std::io::Result<()> {
-        for i in 0..self.writers.len() {
-            if self.errs[i].is_some() {
-                continue; // Skip if we already have an error for this writer
-            }
-
-            let writer_opt = &mut self.writers[i];
-            let shard = &data[i];
-
-            if let Some(writer) = writer_opt {
+    async fn write_shard(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>, shard: &Bytes) {
+        match writer_opt {
+            Some(writer) => {
                 match writer.write(shard).await {
                     Ok(n) => {
                         if n < shard.len() {
-                            self.errs[i] = Some(Error::ShortWrite);
-                            self.writers[i] = None; // Mark as failed
+                            *err = Some(Error::ShortWrite);
+                            *writer_opt = None; // Mark as failed
                         } else {
-                            self.errs[i] = None;
+                            *err = None;
                         }
                     }
                     Err(e) => {
-                        self.errs[i] = Some(Error::from(e));
+                        *err = Some(Error::from(e));
                     }
                 }
-            } else {
-                self.errs[i] = Some(Error::DiskNotFound);
             }
+            None => {
+                *err = Some(Error::DiskNotFound);
+            }
+        }
+    }
+
+    pub async fn write(&mut self, data: Vec<Bytes>) -> std::io::Result<()> {
+        assert_eq!(data.len(), self.writers.len());
+
+        {
+            let mut futures = FuturesUnordered::new();
+            for ((writer_opt, err), shard) in self.writers.iter_mut().zip(self.errs.iter_mut()).zip(data.iter()) {
+                if err.is_some() {
+                    continue; // Skip if we already have an error for this writer
+                }
+                futures.push(Self::write_shard(writer_opt, err, shard));
+            }
+            while let Some(()) = futures.next().await {}
         }
 
         let nil_count = self.errs.iter().filter(|&e| e.is_none()).count();