gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-17 18:27:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(storage): extend PUT path tuning and observability (#3829)

Browse Source 
* 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
This commit is contained in:
houseme
2026-06-25 19:24:35 +08:00
committed by GitHub
parent 3942186f3d
commit a30357d21e
24 changed files with 3037 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/erasure_coding/encode.rs
+56 -14
View File
@@ -23,6 +23,7 @@ use bytes::Bytes;
use futures::StreamExt;
use futures::stream::FuturesUnordered;
use std::sync::Arc;
use std::time::Instant;
use std::vec;
use tokio::io::AsyncRead;
use tokio::runtime::RuntimeFlavor;
@@ -40,6 +41,18 @@ const DEFAULT_RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS: usize = 4;
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();
#[inline(always)]
fn stage_timer_if_enabled() -> Option<Instant> {
rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now)
}
#[inline(always)]
fn record_internal_stage_if_enabled(stage: &'static str, started_at: Option<Instant>) {
if let Some(started_at) = started_at {
rustfs_io_metrics::record_stage_duration(stage, started_at.elapsed().as_secs_f64() * 1000.0);
}
}
fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usize) -> usize {
if expanded_block_bytes == 0 {
return 1;
@@ -57,6 +70,15 @@ fn encode_batch_block_count() -> usize {
})
}
fn erasure_encode_max_inflight_bytes() -> usize {
*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,
)
})
}
fn queued_block_bytes(block: &[Bytes]) -> usize {
block.iter().map(Bytes::len).sum()
}
@@ -237,6 +259,7 @@ 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_stage_start = stage_timer_if_enabled();
let encode_once = move || {
let res = self.encode_data(&encode_buf[..len]);
(res, encode_buf)
@@ -252,6 +275,7 @@ impl Erasure {
.map_err(|err| std::io::Error::other(format!("EC encode task failed: {err}")))?,
};
record_internal_stage_if_enabled("erasure_encode_cpu", encode_stage_start);
Ok((res?, returned_buf))
}
@@ -316,12 +340,7 @@ impl Erasure {
// Bound queued encoded blocks by memory budget to avoid per-request spikes.
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 max_inflight_bytes = erasure_encode_max_inflight_bytes();
let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes);
let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(inflight_blocks);
@@ -339,10 +358,12 @@ impl Erasure {
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;
@@ -369,30 +390,41 @@ impl Erasure {
let mut write_err = None;
while let Some(block) = rx.recv().await {
loop {
let recv_wait_stage_start = stage_timer_if_enabled();
let Some(block) = rx.recv().await else {
break;
};
record_internal_stage_if_enabled("erasure_encode_recv_wait", recv_wait_stage_start);
if block.is_empty() {
break;
}
let queued_bytes = queued_block_bytes(&block);
rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
let write_stage_start = stage_timer_if_enabled();
if let Err(err) = writers.write(block).await {
write_err = Some(err);
break;
}
record_internal_stage_if_enabled("erasure_encode_write", write_stage_start);
}
if let Some(err) = write_err {
task.abort();
let _ = task.await;
drain_queued_inflight_bytes(&mut rx).await;
let shutdown_stage_start = stage_timer_if_enabled();
if let Err(shutdown_err) = writers.shutdown().await {
error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err);
}
record_internal_stage_if_enabled("erasure_encode_shutdown", shutdown_stage_start);
return Err(err);
}
let (reader, total) = task.await??;
let shutdown_stage_start = stage_timer_if_enabled();
writers.shutdown().await?;
record_internal_stage_if_enabled("erasure_encode_shutdown", shutdown_stage_start);
Ok((reader, total))
}
@@ -413,12 +445,7 @@ impl Erasure {
}
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 max_inflight_bytes = 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);
@@ -444,10 +471,12 @@ impl Erasure {
if pending_batch.len() >= batch_blocks {
rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes);
let send_wait_stage_start = stage_timer_if_enabled();
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}")));
}
record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start);
pending_batch = Vec::with_capacity(batch_blocks);
pending_batch_bytes = 0;
}
@@ -471,10 +500,12 @@ impl Erasure {
if !pending_batch.is_empty() {
rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes);
let send_wait_stage_start = stage_timer_if_enabled();
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}")));
}
record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start);
}
Ok((reader, total))
@@ -483,14 +514,21 @@ impl Erasure {
let mut writers = MultiWriter::new(writers, quorum);
let mut write_err = None;
while let Some(batch) = rx.recv().await {
loop {
let recv_wait_stage_start = stage_timer_if_enabled();
let Some(batch) = rx.recv().await else {
break;
};
record_internal_stage_if_enabled("erasure_encode_batched_recv_wait", recv_wait_stage_start);
rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_batch_bytes(&batch));
let write_stage_start = stage_timer_if_enabled();
for block in batch {
if let Err(err) = writers.write(block).await {
write_err = Some(err);
break;
}
}
record_internal_stage_if_enabled("erasure_encode_batched_write", write_stage_start);
if write_err.is_some() {
break;
}
@@ -500,14 +538,18 @@ impl Erasure {
task.abort();
let _ = task.await;
drain_queued_batched_inflight_bytes(&mut rx).await;
let shutdown_stage_start = stage_timer_if_enabled();
if let Err(shutdown_err) = writers.shutdown().await {
error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err);
}
record_internal_stage_if_enabled("erasure_encode_batched_shutdown", shutdown_stage_start);
return Err(err);
}
let (reader, total) = task.await??;
let shutdown_stage_start = stage_timer_if_enabled();
writers.shutdown().await?;
record_internal_stage_if_enabled("erasure_encode_batched_shutdown", shutdown_stage_start);
Ok((reader, total))
}