diff --git a/ecstore/src/erasure_coding/encode.rs b/ecstore/src/erasure_coding/encode.rs index 6517c26e0..899b8f570 100644 --- a/ecstore/src/erasure_coding/encode.rs +++ b/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) -> 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, err: &mut Option, 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) -> 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();