feat(perf): add large PUT tuning and encode optimization (#3816)

crates/ecstore/src/erasure_coding/encode.rs

@@ -25,16 +25,20 @@ use futures::stream::FuturesUnordered;
 use std::sync::Arc;
 use std::vec;
 use tokio::io::AsyncRead;
+use tokio::runtime::RuntimeFlavor;
 use tokio::sync::mpsc;
 use tracing::error;
 
 const ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: &str = "RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES";
+const ENV_RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS: &str = "RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS";
 const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: usize = 32 * 1024 * 1024;
 const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS: usize = 32;
+const DEFAULT_RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS: usize = 4;
 
 /// Cached value of `RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES` env var.
 /// Read once at first use via `OnceLock` to avoid per-encode syscall.
 static CACHED_MAX_INFLIGHT_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
+static CACHED_BATCH_BLOCKS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
 
 fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usize) -> usize {
     if expanded_block_bytes == 0 {
@@ -46,6 +50,13 @@ fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usiz
         .clamp(1, DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS)
 }
 
+fn encode_batch_block_count() -> usize {
+    *CACHED_BATCH_BLOCKS.get_or_init(|| {
+        rustfs_utils::get_env_usize(ENV_RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS, DEFAULT_RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS)
+            .clamp(1, DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS)
+    })
+}
+
 fn queued_block_bytes(block: &[Bytes]) -> usize {
     block.iter().map(Bytes::len).sum()
 }
@@ -56,6 +67,16 @@ async fn drain_queued_inflight_bytes(rx: &mut mpsc::Receiver<Vec<Bytes>>) {
     }
 }
 
+fn queued_batch_bytes(batch: &[Vec<Bytes>]) -> usize {
+    batch.iter().map(|block| queued_block_bytes(block)).sum()
+}
+
+async fn drain_queued_batched_inflight_bytes(rx: &mut mpsc::Receiver<Vec<Vec<Bytes>>>) {
+    while let Some(batch) = rx.recv().await {
+        rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_batch_bytes(&batch));
+    }
+}
+
 fn dominant_error_summary_label(summary: &WriteQuorumFailureSummary) -> &'static str {
     summary.dominant_error_label
 }
@@ -215,6 +236,25 @@ impl<'a> MultiWriter<'a> {
 }
 
 impl Erasure {
+    async fn encode_block(self: Arc<Self>, encode_buf: Vec<u8>, len: usize) -> std::io::Result<(Vec<Bytes>, Vec<u8>)> {
+        let encode_once = move || {
+            let res = self.encode_data(&encode_buf[..len]);
+            (res, encode_buf)
+        };
+
+        let (res, returned_buf) = match tokio::runtime::Handle::current().runtime_flavor() {
+            RuntimeFlavor::MultiThread => tokio::task::block_in_place(encode_once),
+            RuntimeFlavor::CurrentThread => tokio::task::spawn_blocking(encode_once)
+                .await
+                .map_err(|err| std::io::Error::other(format!("EC encode task failed: {err}")))?,
+            _ => tokio::task::spawn_blocking(encode_once)
+                .await
+                .map_err(|err| std::io::Error::other(format!("EC encode task failed: {err}")))?,
+        };
+
+        Ok((res?, returned_buf))
+    }
+
     async fn encode_small_direct<R>(
         self: Arc<Self>,
         mut reader: R,
@@ -294,16 +334,9 @@ impl Erasure {
                     Ok(Some(n)) => {
                         debug_assert!(n > 0, "non-zero block_size prevents zero-length reads");
                         total += n;
-                        let erasure = self.clone();
                         let encode_buf = std::mem::take(&mut buf);
-                        let (res, returned_buf) = tokio::task::spawn_blocking(move || {
-                            let res = erasure.encode_data(&encode_buf[..n]);
-                            (res, encode_buf)
-                        })
-                        .await
-                        .map_err(|err| std::io::Error::other(format!("EC encode task failed: {err}")))?;
+                        let (res, returned_buf) = self.clone().encode_block(encode_buf, n).await?;
                         buf = returned_buf;
-                        let res = res?;
                         let queued_bytes = queued_block_bytes(&res);
                         rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
                         if let Err(err) = tx.send(res).await {
@@ -363,6 +396,121 @@ impl Erasure {
         Ok((reader, total))
     }
 
+    pub async fn encode_batched<R>(
+        self: Arc<Self>,
+        mut reader: R,
+        writers: &mut [Option<BitrotWriterWrapper>],
+        quorum: usize,
+    ) -> std::io::Result<(R, usize)>
+    where
+        R: AsyncRead + Send + Sync + Unpin + 'static,
+    {
+        if self.block_size == 0 {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "erasure block_size must be non-zero",
+            ));
+        }
+
+        let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count());
+        let max_inflight_bytes = *CACHED_MAX_INFLIGHT_BYTES.get_or_init(|| {
+            rustfs_utils::get_env_usize(
+                ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES,
+                DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES,
+            )
+        });
+        let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes);
+        let batch_blocks = encode_batch_block_count().min(inflight_blocks);
+        let channel_capacity = inflight_blocks.div_ceil(batch_blocks).max(1);
+        let (tx, mut rx) = mpsc::channel::<Vec<Vec<Bytes>>>(channel_capacity);
+
+        let task = tokio::spawn(async move {
+            let block_size = self.block_size;
+            let mut total = 0;
+            let mut buf = vec![0u8; block_size];
+            let mut pending_batch = Vec::with_capacity(batch_blocks);
+            let mut pending_batch_bytes = 0usize;
+            loop {
+                match rustfs_utils::read_full_or_eof(&mut reader, &mut buf).await {
+                    Ok(Some(n)) => {
+                        debug_assert!(n > 0, "non-zero block_size prevents zero-length reads");
+                        total += n;
+                        let encode_buf = std::mem::take(&mut buf);
+                        let (res, returned_buf) = self.clone().encode_block(encode_buf, n).await?;
+                        buf = returned_buf;
+                        let queued_bytes = queued_block_bytes(&res);
+                        pending_batch_bytes = pending_batch_bytes.saturating_add(queued_bytes);
+                        pending_batch.push(res);
+
+                        if pending_batch.len() >= batch_blocks {
+                            rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes);
+                            if let Err(err) = tx.send(pending_batch).await {
+                                rustfs_io_metrics::remove_ec_encode_inflight_bytes(pending_batch_bytes);
+                                return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
+                            }
+                            pending_batch = Vec::with_capacity(batch_blocks);
+                            pending_batch_bytes = 0;
+                        }
+                    }
+                    Ok(None) => {
+                        break;
+                    }
+                    Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                        if let Some(inner) = e.get_ref()
+                            && rustfs_rio::is_checksum_mismatch(inner)
+                        {
+                            return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()));
+                        }
+                        return Err(e);
+                    }
+                    Err(e) => {
+                        return Err(e);
+                    }
+                }
+            }
+
+            if !pending_batch.is_empty() {
+                rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes);
+                if let Err(err) = tx.send(pending_batch).await {
+                    rustfs_io_metrics::remove_ec_encode_inflight_bytes(pending_batch_bytes);
+                    return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
+                }
+            }
+
+            Ok((reader, total))
+        });
+
+        let mut writers = MultiWriter::new(writers, quorum);
+        let mut write_err = None;
+
+        while let Some(batch) = rx.recv().await {
+            rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_batch_bytes(&batch));
+            for block in batch {
+                if let Err(err) = writers.write(block).await {
+                    write_err = Some(err);
+                    break;
+                }
+            }
+            if write_err.is_some() {
+                break;
+            }
+        }
+
+        if let Some(err) = write_err {
+            task.abort();
+            let _ = task.await;
+            drain_queued_batched_inflight_bytes(&mut rx).await;
+            if let Err(shutdown_err) = writers.shutdown().await {
+                error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err);
+            }
+            return Err(err);
+        }
+
+        let (reader, total) = task.await??;
+        writers.shutdown().await?;
+        Ok((reader, total))
+    }
+
     /// Fast path for small inline objects: skip tokio::spawn + mpsc channel.
     /// Reads all data, encodes directly, writes shards sequentially.
     pub async fn encode_inline_small<R>(
@@ -497,6 +645,23 @@ mod tests {
         assert!(err.to_string().contains("block_size"));
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn encode_works_on_current_thread_runtime() {
+        let committed = Arc::new(Mutex::new(Vec::new()));
+        let writer = DeferredCommitWriter::new(committed);
+        let mut writers = vec![Some(BitrotWriterWrapper::new(
+            CustomWriter::new_tokio_writer(writer),
+            16,
+            HashAlgorithm::HighwayHash256S,
+        ))];
+
+        let erasure = Arc::new(Erasure::new(1, 0, 16));
+        let reader = tokio::io::BufReader::new(Cursor::new(b"current-thread payload".to_vec()));
+        let (_reader, written) = erasure.encode(reader, &mut writers, 1).await.unwrap();
+
+        assert_eq!(written, b"current-thread payload".len());
+    }
+
     /// encode_inline_small: empty reader returns (reader, 0) without writing to any shard.
     #[tokio::test]
     async fn encode_inline_small_empty_stream_returns_zero() {

crates/ecstore/src/set_disk.rs

@@ -149,6 +149,9 @@ const EVENT_SET_DISK_HEAL: &str = "set_disk_heal";
 const EVENT_SET_DISK_COMMIT_TAIL_SLOW: &str = "set_disk_commit_tail_slow";
 const EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY: &str = "set_disk_put_object_stage_summary";
 const SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS: u128 = 5_000;
+const ENV_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: &str = "RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES";
+const DEFAULT_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: usize = 64 * 1024 * 1024;
+static CACHED_PUT_LARGE_BATCH_MIN_SIZE_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
 
 use crate::rio::{EtagResolvable, HashReader, HashReaderMut, TryGetIndex as _};
 
@@ -812,6 +815,7 @@ enum SmallWritePath {
     Inline,
     SingleBlockNonInline,
     Pipeline,
+    PipelineBatchedLarge,
 }
 
 impl SmallWritePath {
@@ -820,10 +824,17 @@ impl SmallWritePath {
             SmallWritePath::Inline => "write_inline",
             SmallWritePath::SingleBlockNonInline => "write_single_block_non_inline",
             SmallWritePath::Pipeline => "write_pipeline",
+            SmallWritePath::PipelineBatchedLarge => "write_pipeline_batched_large",
         }
     }
 }
 
+fn put_large_batch_min_size_bytes() -> usize {
+    *CACHED_PUT_LARGE_BATCH_MIN_SIZE_BYTES.get_or_init(|| {
+        rustfs_utils::get_env_usize(ENV_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES, DEFAULT_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES)
+    })
+}
+
 fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
     if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
         SmallWritePath::Inline
@@ -834,6 +845,20 @@ fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_siz
     }
 }
 
+fn classify_put_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
+    if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
+        return SmallWritePath::Inline;
+    }
+    if should_use_single_block_non_inline_fast_path(is_inline_buffer, object_size, block_size) {
+        return SmallWritePath::SingleBlockNonInline;
+    }
+
+    match usize::try_from(object_size) {
+        Ok(size) if !is_inline_buffer && size >= put_large_batch_min_size_bytes() => SmallWritePath::PipelineBatchedLarge,
+        _ => SmallWritePath::Pipeline,
+    }
+}
+
 #[async_trait::async_trait]
 impl rustfs_storage_api::ObjectIO for SetDisks {
     type Error = Error;
@@ -1154,7 +1179,7 @@ impl rustfs_storage_api::ObjectIO for SetDisks {
                 HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
             );
 
-            let write_path = classify_small_write_path(is_inline_buffer, data.size(), fi.erasure.block_size);
+            let write_path = classify_put_write_path(is_inline_buffer, data.size(), fi.erasure.block_size);
             rustfs_io_metrics::record_put_object_path(write_path.metric_label());
 
             let encode_stage_start = Instant::now();
@@ -1179,6 +1204,15 @@ impl rustfs_storage_api::ObjectIO for SetDisks {
                         return Err(e.into());
                     }
                 },
+                SmallWritePath::PipelineBatchedLarge => {
+                    match Arc::new(erasure).encode_batched(stream, &mut writers, write_quorum).await {
+                        Ok((r, w)) => (r, w),
+                        Err(e) => {
+                            error!("encode_batched err {:?}", e);
+                            return Err(e.into());
+                        }
+                    }
+                }
                 SmallWritePath::Pipeline => match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await {
                     Ok((r, w)) => (r, w),
                     Err(e) => {
@@ -3421,7 +3455,7 @@ impl rustfs_storage_api::MultipartOperations for SetDisks {
                     .encode_single_block_non_inline(stream, &mut writers, write_quorum)
                     .await?
             }
-            SmallWritePath::Inline | SmallWritePath::Pipeline => {
+            SmallWritePath::Inline | SmallWritePath::Pipeline | SmallWritePath::PipelineBatchedLarge => {
                 Arc::new(erasure).encode(stream, &mut writers, write_quorum).await?
             }
         }; // TODO: delete temporary directory on error
@@ -7430,6 +7464,22 @@ mod tests {
         assert!(matches!(classify_small_write_path(false, 8192, 4096), SmallWritePath::Pipeline));
     }
 
+    #[test]
+    fn put_object_large_batch_path_only_applies_to_large_ordinary_puts() {
+        assert!(matches!(
+            classify_put_write_path(false, 64 * 1024 * 1024, 1024 * 1024),
+            SmallWritePath::PipelineBatchedLarge
+        ));
+        assert!(matches!(
+            classify_put_write_path(false, 32 * 1024 * 1024, 1024 * 1024),
+            SmallWritePath::Pipeline
+        ));
+        assert!(matches!(
+            classify_put_write_path(true, 64 * 1024 * 1024, 1024 * 1024),
+            SmallWritePath::Pipeline
+        ));
+    }
+
     #[test]
     fn put_object_part_fast_path_selection_matches_single_block_non_inline_rules() {
         assert!(should_use_single_block_non_inline_fast_path(false, 4096, 4096));

crates/io-metrics/src/lib.rs

@@ -225,6 +225,20 @@ pub fn record_get_object_request_result(status: &str, duration_secs: f64) {
     histogram!("rustfs_io_get_object_request_duration_seconds", "status" => status.to_string()).record(duration_secs);
 }
 
+/// Record PutObject request start.
+#[inline(always)]
+pub fn record_put_object_request_start(concurrent_requests: usize) {
+    counter!("rustfs_io_put_object_requests_total").increment(1);
+    gauge!("rustfs_io_put_object_concurrent_requests").set(concurrent_requests as f64);
+}
+
+/// Record PutObject request result.
+#[inline(always)]
+pub fn record_put_object_request_result(status: &str, duration_secs: f64) {
+    counter!("rustfs_io_put_object_request_results_total", "status" => status.to_string()).increment(1);
+    histogram!("rustfs_io_put_object_request_duration_seconds", "status" => status.to_string()).record(duration_secs);
+}
+
 /// Record GetObject timeout for a specific stage.
 #[inline(always)]
 pub fn record_get_object_timeout(stage: Option<&str>, elapsed_secs: Option<f64>) {
@@ -877,6 +891,13 @@ mod tests {
         record_put_object(100.0, 512, false);
     }
 
+    #[test]
+    fn test_record_put_object_request_metrics() {
+        record_put_object_request_start(3);
+        record_put_object_request_result("ok", 0.25);
+        record_put_object_request_result("error", 0.5);
+    }
+
     #[test]
     fn test_record_put_object_path_and_stage() {
         let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());