From efebcb66be412fa2d11f5e9dcfd7b027aa005b3e Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 24 Jun 2026 19:04:04 +0800 Subject: [PATCH] feat(perf): add large PUT tuning and encode optimization (#3816) --- .../grafana/dashboards/rustfs.json | 943 +++++++++++++++++- crates/ecstore/src/erasure_coding/encode.rs | 181 +++- crates/ecstore/src/set_disk.rs | 54 +- crates/io-metrics/src/lib.rs | 21 + rustfs/src/app/object_usecase.rs | 38 +- rustfs/src/storage/concurrency/io_schedule.rs | 12 + rustfs/src/storage/concurrency/manager.rs | 6 +- rustfs/src/storage/concurrency/mod.rs | 8 +- .../src/storage/concurrency/request_guard.rs | 77 +- ...ect_put_large_stage_breakdown_artifacts.sh | 468 +++++++++ scripts/run_object_batch_bench_enhanced.sh | 38 + scripts/run_put_large_stage_breakdown.sh | 460 +++++++++ ..._put_large_stage_breakdown_with_capture.sh | 406 ++++++++ scripts/run_put_large_tuning_matrix.sh | 430 ++++++++ 14 files changed, 3120 insertions(+), 22 deletions(-) create mode 100755 scripts/collect_put_large_stage_breakdown_artifacts.sh create mode 100755 scripts/run_put_large_stage_breakdown.sh create mode 100755 scripts/run_put_large_stage_breakdown_with_capture.sh create mode 100755 scripts/run_put_large_tuning_matrix.sh diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index 4f958f1a1..60cb55174 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -10037,6 +10037,895 @@ ], "title": "Erasure Write Quorum Failures", "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 279 + }, + "id": 520, + "panels": [], + "title": "Large PUT Stage Breakdown", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ops" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 280 + }, + "id": 521, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (instance) (rate(rustfs_s3_put_object_total{job=~\"$job\"}[$__rate_interval]))", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "title": "PUT Request Rate by Instance", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 100 + }, + { + "color": "red", + "value": 500 + } + ] + }, + "unit": "ms" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 280 + }, + "id": 522, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.95, sum by (le) (rate(rustfs_s3_put_object_duration_ms_bucket{job=~\"$job\"}[$__rate_interval])))", + "legendFormat": "PutObject P95", + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.99, sum by (le) (rate(rustfs_s3_put_object_duration_ms_bucket{job=~\"$job\"}[$__rate_interval])))", + "legendFormat": "PutObject P99", + "refId": "B" + } + ], + "title": "PUT Latency (P95 / P99)", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ops" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 288 + }, + "id": 523, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (path) (rate(rustfs_s3_put_object_path_total{job=~\"$job\"}[$__rate_interval]))", + "legendFormat": "{{path}}", + "refId": "A" + } + ], + "title": "PUT Path Distribution", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 100 + }, + { + "color": "red", + "value": 500 + } + ] + }, + "unit": "ms" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 288 + }, + "id": 524, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.95, sum by (stage, le) (rate(rustfs_s3_put_object_stage_duration_ms_bucket{job=~\"$job\"}[$__rate_interval])))", + "legendFormat": "{{stage}}", + "refId": "A" + } + ], + "title": "PUT Stage Latency P95", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 100 + }, + { + "color": "red", + "value": 500 + } + ] + }, + "unit": "ms" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 296 + }, + "id": 525, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.99, sum by (stage, le) (rate(rustfs_s3_put_object_stage_duration_ms_bucket{job=~\"$job\"}[$__rate_interval])))", + "legendFormat": "{{stage}}", + "refId": "A" + } + ], + "title": "PUT Stage Latency P99", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "bytes" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 296 + }, + "id": 526, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (instance) (rustfs_ec_encode_inflight_bytes_current{job=~\"$job\"})", + "legendFormat": "EC inflight - {{instance}}", + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (instance) (rustfs_system_process_resident_memory_bytes{job=~\"$job\"}) or sum by (instance) (rustfs_memory_process_resident_bytes{job=~\"$job\"})", + "legendFormat": "RSS - {{instance}}", + "refId": "B" + } + ], + "title": "PUT Supporting Signals by Instance", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 100 + }, + { + "color": "red", + "value": 500 + } + ] + }, + "unit": "ms" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 304 + }, + "id": 527, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.95, sum by (instance, le) (rate(rustfs_s3_put_object_duration_ms_bucket{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval])))", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "title": "PUT Latency P95 by Instance", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 100 + }, + { + "color": "red", + "value": 500 + } + ] + }, + "unit": "ms" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 304 + }, + "id": 528, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "histogram_quantile(0.95, sum by (instance, le) (rate(rustfs_s3_put_object_stage_duration_ms_bucket{job=~\"$job\",instance=~\"$instance\",stage=~\"$stage\"}[$__rate_interval])))", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "title": "PUT Stage Latency P95 by Instance ($stage)", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "Bps" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 312 + }, + "id": 529, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum by (instance) (rate(rustfs_disk_write_bytes_total{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval]))", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "title": "Disk Write Throughput by Instance", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 70 + }, + { + "color": "red", + "value": 90 + } + ] + }, + "unit": "percent" + } + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 312 + }, + "id": 530, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom" + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\",instance=~\"$instance\"}[$__rate_interval])) * 100)", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "title": "Host CPU Usage by Instance", + "type": "timeseries" } ], "preload": false, @@ -10155,6 +11044,58 @@ "refresh": 2, "regex": "", "type": "query" + }, + { + "allValue": ".*", + "current": { + "text": "All", + "value": "$__all" + }, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "definition": "label_values(rustfs_s3_put_object_total,instance)", + "includeAll": true, + "label": "Instance", + "multi": true, + "name": "instance", + "options": [], + "query": { + "qryType": 1, + "query": "label_values(rustfs_s3_put_object_total,instance)", + "refId": "PrometheusVariableQueryEditor-VariableQuery" + }, + "refresh": 2, + "regex": "", + "sort": 1, + "type": "query" + }, + { + "allValue": ".*", + "current": { + "text": "All", + "value": "$__all" + }, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "definition": "label_values(rustfs_s3_put_object_stage_duration_ms_bucket,stage)", + "includeAll": true, + "label": "Stage", + "multi": true, + "name": "stage", + "options": [], + "query": { + "qryType": 1, + "query": "label_values(rustfs_s3_put_object_stage_duration_ms_bucket,stage)", + "refId": "PrometheusVariableQueryEditor-VariableQuery" + }, + "refresh": 2, + "regex": "", + "sort": 1, + "type": "query" } ] }, @@ -10180,5 +11121,5 @@ "timezone": "browser", "title": "RustFS", "uid": "rustfs-s3", - "version": 14 + "version": 16 } diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index 91508698a..2310702fc 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/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 = std::sync::OnceLock::new(); +static CACHED_BATCH_BLOCKS: std::sync::OnceLock = 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>) { } } +fn queued_batch_bytes(batch: &[Vec]) -> usize { + batch.iter().map(|block| queued_block_bytes(block)).sum() +} + +async fn drain_queued_batched_inflight_bytes(rx: &mut mpsc::Receiver>>) { + 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, encode_buf: Vec, len: usize) -> std::io::Result<(Vec, Vec)> { + 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( self: Arc, 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( + self: Arc, + mut reader: R, + writers: &mut [Option], + 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::>>(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( @@ -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() { diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 359e97778..4f66f6e8f 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/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 = 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)); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 4ba7b43a7..0768f162f 100644 --- a/crates/io-metrics/src/lib.rs +++ b/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) { @@ -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()); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 0252248bf..d72f4606c 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -53,7 +53,8 @@ use crate::error::ApiError; use crate::server::convert_ecstore_object_info; use crate::storage::access::{PostObjectRequestMarker, authorize_request, has_bypass_governance_header, req_info_mut}; use crate::storage::concurrency::{ - ConcurrencyManager, GetObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, + ConcurrencyManager, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, + get_put_concurrency_aware_buffer_size, }; use crate::storage::ecfs::*; use crate::storage::head_prefix::{head_prefix_not_found_message, probe_prefix_has_children}; @@ -117,6 +118,8 @@ use rustfs_zip::{ArchiveLimits, CompressionFormat}; use s3s::dto::*; use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH}; use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; + +const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024; use std::collections::HashMap; use std::ops::Add; use std::path::Path; @@ -1347,6 +1350,10 @@ pub struct DefaultObjectUsecase { } impl DefaultObjectUsecase { + fn should_use_large_put_concurrency_tuning(size: i64) -> bool { + size >= DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES + } + #[cfg(test)] pub fn without_context() -> Self { Self { context: None } @@ -2024,16 +2031,19 @@ impl DefaultObjectUsecase { let server_side_encryption_requested = server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some(); + let mut put_request_guard = PutObjectGuard::new(); + let concurrent_put_requests = PutObjectGuard::concurrent_requests(); + // Apply adaptive buffer sizing based on file size for optimal streaming performance. // Uses workload profile configuration (enabled by default) to select appropriate buffer size. // Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile. - // Concurrency-aware adjustment reduces buffer size under high concurrency to lower memory pressure. - // TODO: get_concurrency_aware_buffer_size reads ACTIVE_GET_REQUESTS (GET concurrency tracker), - // not PUT concurrency. Under pure PUT load the counter stays zero so buffers never shrink; - // unrelated GET load can shrink PUT buffers instead. Fix by adding ACTIVE_PUT_REQUESTS + - // PutObjectGuard and using PUT concurrency here. See PR #3514 review comment. + // Concurrency-aware adjustment reduces buffer size under high PUT concurrency to lower memory pressure. let base_buffer_size = get_buffer_size_opt_in(size); - let buffer_size = get_concurrency_aware_buffer_size(size, base_buffer_size); + let buffer_size = if Self::should_use_large_put_concurrency_tuning(size) { + get_put_concurrency_aware_buffer_size(size, base_buffer_size) + } else { + base_buffer_size + }; // Detect zero-copy opportunity before encryption/compression decisions // Zero-copy is beneficial for large unencrypted, uncompressed objects @@ -2377,6 +2387,7 @@ impl DefaultObjectUsecase { "PutObject store write returned" ); let result: S3Result> = Err(err.into()); + put_request_guard.finish_err(); let _ = helper.complete(&result); return result; } @@ -2462,6 +2473,19 @@ impl DefaultObjectUsecase { ); } + debug!( + target: "rustfs::app::object_usecase", + component = "app", + subsystem = "object", + bucket = %bucket, + key = %key, + concurrent_put_requests, + buffer_size, + "PutObject request completed" + ); + + put_request_guard.finish_ok(); + result } diff --git a/rustfs/src/storage/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index ff8eb3268..a62eb5843 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -39,6 +39,7 @@ use std::time::Duration; /// Global concurrent request counter for adaptive buffer sizing. pub(crate) static ACTIVE_GET_REQUESTS: AtomicUsize = AtomicUsize::new(0); +pub(crate) static ACTIVE_PUT_REQUESTS: AtomicUsize = AtomicUsize::new(0); #[derive(Debug, Clone, PartialEq)] pub enum IoLoadLevel { @@ -1377,6 +1378,17 @@ pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize compute_concurrency_aware_buffer_size(file_size, base_buffer_size, concurrent_requests, load_concurrency_thresholds()) } +pub fn get_put_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize) -> usize { + let concurrent_requests = ACTIVE_PUT_REQUESTS.load(Ordering::Relaxed); + + { + use metrics::gauge; + gauge!("rustfs_concurrent_put_requests").set(concurrent_requests as f64); + } + + compute_concurrency_aware_buffer_size(file_size, base_buffer_size, concurrent_requests, load_concurrency_thresholds()) +} + /// Advanced concurrency-aware buffer sizing with file size optimization /// /// This enhanced version considers both concurrency level and file size patterns diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index b81de5e9e..a294a1c63 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -18,7 +18,7 @@ use super::io_schedule::{ IoLoadLevel, IoLoadMetrics, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoSchedulerConfig, IoStrategy, get_advanced_buffer_size, }; -use super::request_guard::GetObjectGuard; +use super::request_guard::{GetObjectGuard, PutObjectGuard}; use crate::app::context::resolve_performance_metrics; use rustfs_concurrency::{ AdmissionState, GetObjectQueueSnapshot, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, @@ -145,6 +145,10 @@ impl ConcurrencyManager { GetObjectGuard::new() } + pub fn track_put_request() -> PutObjectGuard { + PutObjectGuard::new() + } + /// Get the bytes pool for buffer allocation /// /// Returns a reference to the BytesPool which can be used to acquire diff --git a/rustfs/src/storage/concurrency/mod.rs b/rustfs/src/storage/concurrency/mod.rs index cae7245e1..ff8e7d57c 100644 --- a/rustfs/src/storage/concurrency/mod.rs +++ b/rustfs/src/storage/concurrency/mod.rs @@ -47,10 +47,11 @@ pub mod request_guard; pub use io_schedule::{ IO_PRIORITY_METRICS, IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoSchedulerConfig, IoStrategy, get_advanced_buffer_size, get_buffer_size_opt_in, get_concurrency_aware_buffer_size, + get_put_concurrency_aware_buffer_size, }; // Request tracking -pub use request_guard::GetObjectGuard; +pub use request_guard::{GetObjectGuard, PutObjectGuard}; // Concurrency manager pub use manager::ConcurrencyManager; @@ -88,6 +89,11 @@ pub fn reset_active_get_requests() { io_schedule::ACTIVE_GET_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed); } +#[allow(dead_code)] +pub fn reset_active_put_requests() { + io_schedule::ACTIVE_PUT_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed); +} + /// Create a new I/O scheduler with default configuration. #[allow(dead_code)] pub fn create_io_scheduler() -> IoScheduler { diff --git a/rustfs/src/storage/concurrency/request_guard.rs b/rustfs/src/storage/concurrency/request_guard.rs index fbec62502..72bff1519 100644 --- a/rustfs/src/storage/concurrency/request_guard.rs +++ b/rustfs/src/storage/concurrency/request_guard.rs @@ -17,8 +17,11 @@ use std::sync::atomic::Ordering; use std::time::Instant; -use super::io_schedule::ACTIVE_GET_REQUESTS; -use rustfs_io_metrics::{record_get_object_request_result, record_get_object_request_start}; +use super::io_schedule::{ACTIVE_GET_REQUESTS, ACTIVE_PUT_REQUESTS}; +use rustfs_io_metrics::{ + record_get_object_request_result, record_get_object_request_start, record_put_object_request_result, + record_put_object_request_start, +}; /// RAII guard for tracking active GetObject requests. #[derive(Debug)] @@ -108,6 +111,66 @@ impl Drop for GetObjectGuard { } } +/// RAII guard for tracking active PutObject requests. +#[derive(Debug)] +pub struct PutObjectGuard { + start_time: Instant, + result: Option<&'static str>, +} + +impl PutObjectGuard { + pub fn new() -> Self { + ACTIVE_PUT_REQUESTS.fetch_add(1, Ordering::Relaxed); + let concurrent = ACTIVE_PUT_REQUESTS.load(Ordering::Relaxed); + record_put_object_request_start(concurrent); + + Self { + start_time: Instant::now(), + result: None, + } + } + + pub fn finish_ok(&mut self) { + self.result = Some("ok"); + } + + pub fn finish_err(&mut self) { + self.result = Some("error"); + } + + pub fn concurrent_count() -> usize { + ACTIVE_PUT_REQUESTS.load(Ordering::Relaxed) + } + + pub fn concurrent_requests() -> usize { + Self::concurrent_count() + } +} + +impl Default for PutObjectGuard { + fn default() -> Self { + Self::new() + } +} + +impl Drop for PutObjectGuard { + fn drop(&mut self) { + let duration_secs = self.start_time.elapsed().as_secs_f64(); + let status = self.result.unwrap_or("unknown"); + record_put_object_request_result(status, duration_secs); + + if let Err(previous) = + ACTIVE_PUT_REQUESTS.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)) + { + debug_assert_eq!( + previous, 0, + "ACTIVE_PUT_REQUESTS underflow attempt in PutObjectGuard::drop; previous value = {}", + previous + ); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -128,4 +191,14 @@ mod tests { std::thread::sleep(std::time::Duration::from_millis(10)); assert!(guard.elapsed().as_millis() >= 10); } + + #[test] + fn test_put_guard_increments_counter() { + let initial = PutObjectGuard::concurrent_count(); + { + let _guard = PutObjectGuard::new(); + assert_eq!(PutObjectGuard::concurrent_count(), initial + 1); + } + assert_eq!(PutObjectGuard::concurrent_count(), initial); + } } diff --git a/scripts/collect_put_large_stage_breakdown_artifacts.sh b/scripts/collect_put_large_stage_breakdown_artifacts.sh new file mode 100755 index 000000000..67b0b1073 --- /dev/null +++ b/scripts/collect_put_large_stage_breakdown_artifacts.sh @@ -0,0 +1,468 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Helper artifact collector for rustfs/backlog#706 large-object PUT stage-breakdown runs. +# Designed to pair with scripts/run_put_large_stage_breakdown.sh and store +# supporting evidence under: +# /captures/