feat(ecstore): add deeper zero-copy ingest experiment (#3847)

* feat(storage): add multipart put stage metrics

* feat(scripts): add multipart put focus runner

* docs(operations): add multipart put server-path guides

* chore(scripts): add local rustfs restart helper

* docs(observability): add local metrics backend guide

* docs(observability): add localized multipart guides

* fix(ecstore): validate multipart batching path

* feat(obs): add erasure encode overlap metrics

* docs(ops): update overlap retest summary

* docs(ops): add batchblocks retest matrix

* docs(ops): extend overlap candidate summary

* docs(ops): capture 8-run overlap summary

* feat(storage): switch rename_data to msgpack map

* test(storage): add rename_data payload checks

* feat(object): add zero_copy_eager put path

* docs(ops): add zero_copy_eager put guide

* docs(ops): add deeper zero-copy next steps

* feat(ecstore): add bytesmut erasure ingest gate

* docs(ops): add bytesmut ingest summary

* docs(ops): extend bytesmut ingest matrix summary

* docs(ops): extend bytesmut larger-object summary

* docs(ops): capture bytesmut variability summary

* chore(scripts): add deeper zero-copy capture flow

* chore(scripts): add deeper zero-copy capture support

* docs(ops): add capture-backed bytesmut retest

* docs(ops): update deeper zero-copy retests

* chore(docs): keep issue-712 notes local only

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/ecstore/src/erasure_coding/encode.rs

@@ -19,7 +19,7 @@ use crate::disk::error_reduce::{
 use crate::erasure_coding::BitrotWriterWrapper;
 use crate::erasure_coding::Erasure;
 use crate::runtime_sources;
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use std::sync::Arc;
@@ -32,14 +32,17 @@ 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 ENV_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST: &str = "RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST";
 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;
+const DEFAULT_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST: bool = false;
 
 /// 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();
+static CACHED_BYTESMUT_INGEST: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
 
 #[inline(always)]
 fn stage_timer_if_enabled() -> Option<Instant> {
@@ -79,6 +82,11 @@ fn erasure_encode_max_inflight_bytes() -> usize {
     })
 }
 
+fn use_bytesmut_ingest() -> bool {
+    *CACHED_BYTESMUT_INGEST.get_or_init(|| {
+        rustfs_utils::get_env_bool(ENV_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST, DEFAULT_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST)
+    })
+}
 fn queued_block_bytes(block: &[Bytes]) -> usize {
     block.iter().map(Bytes::len).sum()
 }
@@ -279,6 +287,24 @@ impl Erasure {
         Ok((res?, returned_buf))
     }
 
+    async fn encode_block_bytes_mut(self: Arc<Self>, encode_buf: BytesMut, len: usize) -> std::io::Result<Vec<Bytes>> {
+        let encode_stage_start = stage_timer_if_enabled();
+        let encode_once = move || self.encode_data_bytes_mut(encode_buf, len);
+
+        let res = 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}")))?,
+        };
+
+        record_internal_stage_if_enabled("erasure_encode_cpu", encode_stage_start);
+        res
+    }
+
     async fn encode_small_direct<R>(
         self: Arc<Self>,
         mut reader: R,
@@ -346,39 +372,75 @@ impl Erasure {
 
         let task = tokio::spawn(async move {
             let block_size = self.block_size;
+            let use_bytesmut_ingest = use_bytesmut_ingest();
             let mut total = 0;
-            let mut buf = vec![0u8; block_size];
-            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);
-                        rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
-                        let send_wait_stage_start = stage_timer_if_enabled();
-                        if let Err(err) = tx.send(res).await {
-                            rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
-                            return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
+            if use_bytesmut_ingest {
+                let mut buf = BytesMut::with_capacity(block_size);
+                buf.resize(block_size, 0);
+                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 = buf;
+                            let res = self.clone().encode_block_bytes_mut(encode_buf, n).await?;
+                            buf = BytesMut::with_capacity(block_size);
+                            buf.resize(block_size, 0);
+                            let queued_bytes = queued_block_bytes(&res);
+                            rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
+                            let send_wait_stage_start = stage_timer_if_enabled();
+                            if let Err(err) = tx.send(res).await {
+                                rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
+                                return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
+                            }
+                            record_internal_stage_if_enabled("erasure_encode_send_wait", send_wait_stage_start);
                         }
-                        record_internal_stage_if_enabled("erasure_encode_send_wait", send_wait_stage_start);
-                    }
-                    Ok(None) => {
-                        break;
-                    }
-                    Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
-                        // Check if the inner error is a checksum mismatch - if so, propagate it
-                        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()));
+                        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);
                         }
-                        return Err(e);
+                        Err(e) => return Err(e),
                     }
-                    Err(e) => {
-                        return Err(e);
+                }
+            } else {
+                let mut buf = vec![0u8; block_size];
+                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);
+                            rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
+                            let send_wait_stage_start = stage_timer_if_enabled();
+                            if let Err(err) = tx.send(res).await {
+                                rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
+                                return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
+                            }
+                            record_internal_stage_if_enabled("erasure_encode_send_wait", send_wait_stage_start);
+                        }
+                        Ok(None) => {
+                            break;
+                        }
+                        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                            // Check if the inner error is a checksum mismatch - if so, propagate it
+                            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);
+                        }
                     }
                 }
             }

crates/ecstore/src/erasure_coding/erasure.rs

@@ -565,6 +565,53 @@ impl Erasure {
         Ok(shards)
     }
 
+    /// Encode data from an owned `BytesMut` buffer, avoiding the initial copy
+    /// from a borrowed slice into a fresh `BytesMut`.
+    pub fn encode_data_bytes_mut(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
+        let shard_size_fn = if self.uses_legacy {
+            calc_shard_size_legacy
+        } else {
+            calc_shard_size
+        };
+        let per_shard_size = shard_size_fn(data_len, self.data_shards);
+        if per_shard_size == 0 {
+            return Ok(vec![Bytes::new(); self.total_shard_count()]);
+        }
+        let need_total_size = per_shard_size * self.total_shard_count();
+
+        if data_buffer.len() > data_len {
+            data_buffer.truncate(data_len);
+        }
+        data_buffer.resize(need_total_size, 0u8);
+
+        {
+            let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();
+
+            if self.parity_shards > 0 {
+                if self.uses_legacy {
+                    if let Some(encoder) = self.legacy_encoder.as_ref() {
+                        encoder.encode(data_slices)?;
+                    } else {
+                        warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
+                    }
+                } else if let Some(encoder) = self.encoder.as_ref() {
+                    encoder.encode(data_slices)?;
+                } else {
+                    warn!("parity_shards > 0, but encoder is None");
+                }
+            }
+        }
+
+        let mut data_buffer = data_buffer.freeze();
+        let mut shards = Vec::with_capacity(self.total_shard_count());
+        for _ in 0..self.total_shard_count() {
+            let shard = data_buffer.split_to(per_shard_size);
+            shards.push(shard);
+        }
+
+        Ok(shards)
+    }
+
     /// Decode and reconstruct missing data shards in-place.
     ///
     /// # Arguments
@@ -799,6 +846,21 @@ mod tests {
         }
     }
 
+    #[test]
+    fn encode_data_bytes_mut_matches_borrowed_path() {
+        for uses_legacy in [false, true] {
+            let erasure = Erasure::new_with_options(4, 2, 64, uses_legacy);
+            for data in [Vec::new(), b"small payload".to_vec(), (0_u8..37).collect::<Vec<_>>()] {
+                let borrowed = erasure.encode_data(&data).expect("borrowed encode should succeed");
+                let bytes_mut = BytesMut::from(&data[..]);
+                let owned = erasure
+                    .encode_data_bytes_mut(bytes_mut, data.len())
+                    .expect("bytesmut encode should succeed");
+                assert_eq!(owned, borrowed);
+            }
+        }
+    }
+
     #[test]
     fn decode_data_keeps_missing_parity_shard_unreconstructed() {
         let erasure = Erasure::new(2, 2, 64);