From ce7d3119b2d0474e103c2cf1c6c59517f849bb21 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 12 Jul 2026 00:19:12 +0800 Subject: [PATCH] perf(io-metrics): throttle collector percentile sort off the per-IO path (#4744) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MetricsCollector::record_io_operation` recomputed P50/P95/P99 on every disk IO by taking a read lock, collecting up to `max_latency_samples` (1000) into a `Vec` and fully sorting it — an O(n log n) alloc+sort per operation on the GET read path (gated by stage metrics). Both consumers sample only periodically: the autotuner reads `avg_io_latency_us` on its tuning tick, and the P95/P99 values are never read internally — they only feed OTEL export. Nothing needs per-IO freshness. Split the two costs: - The window mean (stored in `avg_io_latency_us`, read by the autotuner) is now maintained in O(1) via a running sum kept in step with the window's push/pop, and refreshed on every op — no sort, no warm-up regression. - The P95/P99 sort is throttled to once per `PERCENTILE_RECOMPUTE_INTERVAL` (128) operations. A bounded lag is invisible to the periodic autotuner tick and OTEL export while the per-op sort cost is amortised ~128x away. Semantics are otherwise unchanged: same sliding window, same percentile indices, same mean value. Tests updated — the percentile test forces a recompute to exercise the math directly, and a new test asserts the mean tracks every op while the percentiles only recompute at the interval boundary. Addresses rustfs/backlog#1185 (P1). Co-authored-by: heihutu --- crates/io-metrics/src/collector.rs | 140 +++++++++++++++++++++-------- 1 file changed, 101 insertions(+), 39 deletions(-) diff --git a/crates/io-metrics/src/collector.rs b/crates/io-metrics/src/collector.rs index 39ad035d9..3fe88c023 100644 --- a/crates/io-metrics/src/collector.rs +++ b/crates/io-metrics/src/collector.rs @@ -20,10 +20,28 @@ use super::performance::PerformanceMetrics; use std::collections::VecDeque; use std::sync::Arc; -use std::sync::atomic::Ordering; +use std::sync::atomic::{AtomicU32, Ordering}; use std::time::Duration; use tokio::sync::RwLock; +/// How many recorded operations elapse between P95/P99 recomputations. +/// +/// The sliding-window mean (`avg`) is refreshed on every operation — it is O(1) +/// and the autotuner reads it — but the percentiles need an O(n log n) sort of +/// the window, so they are recomputed at most once per this many operations. +/// Both consumers (the autotuner tick and OTEL export) sample on their own +/// periodic cadence, never per-IO, so a bounded lag here is invisible to them +/// while the per-op sort cost is amortised away. +const PERCENTILE_RECOMPUTE_INTERVAL: u32 = 128; + +/// Sliding window of I/O latency samples plus a running sum, so the window mean +/// is maintained in O(1) as samples enter and leave. +struct LatencyWindow { + samples: VecDeque, + /// Sum of `samples` in microseconds, kept in step with every push/pop. + sum_micros: u128, +} + /// Metrics collector for tracking I/O operations and computing latency percentiles. /// /// Maintains a sliding window of I/O latency samples and updates P95/P99 metrics. @@ -31,10 +49,12 @@ use tokio::sync::RwLock; pub struct MetricsCollector { /// The underlying metrics (shared reference) metrics: Arc, - /// I/O latency samples for percentile calculation - io_latency_samples: RwLock>, + /// Sliding window of I/O latency samples (+ running sum) for mean/percentile calculation + io_latency: RwLock, /// Maximum number of latency samples to keep max_latency_samples: usize, + /// Operations recorded since the last P95/P99 recompute (throttle counter). + ops_since_percentile: AtomicU32, } impl MetricsCollector { @@ -47,8 +67,12 @@ impl MetricsCollector { pub fn new(metrics: Arc, max_latency_samples: usize) -> Self { Self { metrics, - io_latency_samples: RwLock::new(VecDeque::new()), + io_latency: RwLock::new(LatencyWindow { + samples: VecDeque::new(), + sum_micros: 0, + }), max_latency_samples, + ops_since_percentile: AtomicU32::new(0), } } @@ -88,47 +112,61 @@ impl MetricsCollector { // Report to metrics crate for OTEL export crate::record_data_transfer(bytes, duration.as_millis() as f64); - // Record latency sample for percentile calculation - let mut samples = self.io_latency_samples.write().await; - samples.push_back(duration); + // Update the sliding window and the O(1) running mean under a short write + // lock, and decide — still under the lock, so the count is exact — whether + // this op crosses the percentile-recompute throttle. + let (mean_us, recompute_percentiles) = { + let mut window = self.io_latency.write().await; + window.samples.push_back(duration); + window.sum_micros += duration.as_micros(); - // Keep only the most recent samples (O(1) removal from front) - if samples.len() > self.max_latency_samples { - samples.pop_front(); + // Keep only the most recent samples (O(1) removal from front). + if window.samples.len() > self.max_latency_samples + && let Some(old) = window.samples.pop_front() + { + window.sum_micros -= old.as_micros(); + } + + let len = window.samples.len() as u128; + let mean_us = window.sum_micros.checked_div(len).unwrap_or(0) as u64; + + let n = self.ops_since_percentile.fetch_add(1, Ordering::Relaxed) + 1; + let recompute = n >= PERCENTILE_RECOMPUTE_INTERVAL; + if recompute { + self.ops_since_percentile.store(0, Ordering::Relaxed); + } + (mean_us, recompute) + }; + + // The mean is cheap and the autotuner reads it, so refresh it every op. + self.metrics.avg_io_latency_us.store(mean_us, Ordering::Relaxed); + crate::record_io_latency(mean_us as f64 / 1000.0); // Convert to ms + + // The P95/P99 sort is the expensive part and only feeds OTEL export, so it + // is throttled to once per PERCENTILE_RECOMPUTE_INTERVAL operations. + if recompute_percentiles { + self.update_latency_percentiles().await; } - - // Update latency percentiles - drop(samples); // Release write lock before calling update - self.update_latency_percentiles().await; } - /// Update the latency percentile metrics (P50, P95, P99). + /// Recompute the P95/P99 latency percentiles from the current window. /// - /// Calculates percentiles from the sliding window of latency samples - /// and updates both PerformanceMetrics and reports to metrics crate. + /// This sorts a snapshot of the window (O(n log n)), so it is driven on a + /// throttle from the record path rather than per operation. The mean is not + /// computed here — it is maintained in O(1) on every `record_io_operation`. async fn update_latency_percentiles(&self) { - let samples: tokio::sync::RwLockReadGuard<'_, VecDeque> = self.io_latency_samples.read().await; - if samples.is_empty() { - return; - } - - // Sort samples to calculate percentiles - let mut sorted: Vec = samples.iter().map(|d| d.as_micros()).collect(); - drop(samples); // Release read lock before sort - sorted.sort(); + // Snapshot the window micros under a read lock, then sort outside the lock. + let mut sorted: Vec = { + let window = self.io_latency.read().await; + if window.samples.is_empty() { + return; + } + window.samples.iter().map(|d| d.as_micros()).collect() + }; + sorted.sort_unstable(); let len = sorted.len(); - // Calculate average (P50) - let sum: u128 = sorted.iter().sum(); - let avg = (sum / len as u128) as u64; - - // Update PerformanceMetrics - self.metrics.avg_io_latency_us.store(avg, Ordering::Relaxed); - - // Report to metrics crate - crate::record_io_latency(avg as f64 / 1000.0); // Convert to ms - // Calculate P95 let p95_idx = ((len as f64) * 0.95) as usize; if let Some(&p95) = sorted.get(p95_idx.min(len - 1)) { @@ -146,7 +184,7 @@ impl MetricsCollector { /// Get the number of recorded latency samples. pub async fn sample_count(&self) -> usize { - self.io_latency_samples.read().await.len() + self.io_latency.read().await.samples.len() } /// Get the maximum number of samples this collector will retain. @@ -190,11 +228,13 @@ mod tests { collector.record_io_operation(0, Duration::from_micros(400), true).await; collector.record_io_operation(0, Duration::from_micros(500), true).await; - // Check average + // The mean is refreshed on every op. let avg = metrics.avg_io_latency_us.load(Ordering::Relaxed); assert_eq!(avg, 300); // (100+200+300+400+500) / 5 - // Check percentiles + // P95/P99 are throttled off the record path; force a recompute to exercise + // the percentile math directly. + collector.update_latency_percentiles().await; let p95 = metrics.p95_io_latency_us.load(Ordering::Relaxed); let p99 = metrics.p99_io_latency_us.load(Ordering::Relaxed); @@ -204,6 +244,28 @@ mod tests { assert_eq!(p99, 500); } + #[tokio::test] + async fn test_percentiles_throttled_but_mean_updates_per_op() { + let metrics = Arc::new(PerformanceMetrics::new()); + let collector = MetricsCollector::new(metrics.clone(), 1000); + + // Fewer ops than the recompute interval: the mean tracks every op, but the + // percentiles must not have been recomputed yet (they stay at their init 0). + for _ in 0..(PERCENTILE_RECOMPUTE_INTERVAL - 1) { + collector.record_io_operation(0, Duration::from_micros(200), true).await; + } + assert_eq!(metrics.avg_io_latency_us.load(Ordering::Relaxed), 200); + assert_eq!( + metrics.p99_io_latency_us.load(Ordering::Relaxed), + 0, + "percentiles must not be recomputed before the throttle interval" + ); + + // The op that crosses the interval triggers exactly one recompute. + collector.record_io_operation(0, Duration::from_micros(200), true).await; + assert_eq!(metrics.p99_io_latency_us.load(Ordering::Relaxed), 200); + } + #[tokio::test] async fn test_sample_limit() { let metrics = Arc::new(PerformanceMetrics::new());