From 2a8be5566d27ecc3a98a86922460382aa5ad38c8 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 28 Aug 2026 06:52:50 +0800 Subject: [PATCH] refactor(concurrency): consolidate ForegroundPressure into workload owner (#6757) ForegroundPressure had two definitions with byte-identical pressure computation: one in ecstore data-movement backpressure and one in the heal manager queue. That duplication is a violation of the ARCHITECTURE.md invariant that each type has exactly one definition, and it means any future change to the utilization math has to land twice. Add the canonical `ForegroundPressure` and a `foreground_pressure(snapshot, read_threshold_pct, write_threshold_pct)` function to `crates/concurrency/src/workload.rs`, which already owns `WorkloadClass`, `AdmissionState`, and the admission snapshot contract. Both existing consumers already depend on `rustfs-concurrency`, so no crate edge is added. The filter_map pipeline is transferred verbatim, preserving all five boundary behaviors (zero threshold, zero limit, missing entry, missing active count, and the `Saturated` full-utilization special case), the mul-before-div percentage normalization, the `>=` threshold comparison, and the read-then-write ordering that makes `max_by_key` break utilization ties toward the write class. The enable switch is deliberately left out: ecstore gates on `config.enabled` while heal gates on `mainline_throttle_enable` plus a both-thresholds-zero check, so each call site keeps its own condition. This is the expand step only. The ecstore and heal copies are untouched and are removed by the follow-up migrate task. Refs rustfs/backlog#2047 --- crates/concurrency/src/workload.rs | 271 +++++++++++++++++++++++++++++ 1 file changed, 271 insertions(+) diff --git a/crates/concurrency/src/workload.rs b/crates/concurrency/src/workload.rs index f7e2128ea..d728d6793 100644 --- a/crates/concurrency/src/workload.rs +++ b/crates/concurrency/src/workload.rs @@ -178,6 +178,76 @@ pub trait WorkloadAdmissionSnapshotProvider { fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot; } +/// Foreground workload pressure observed against a configured utilization threshold. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ForegroundPressure { + /// Foreground workload class whose utilization reached its threshold. + pub class: WorkloadClass, + /// Observed utilization percentage for the class. + pub usage_pct: usize, + /// Configured threshold percentage that the observed utilization reached. + pub threshold_pct: usize, +} + +impl ForegroundPressure { + /// Return a stable reason label for logs and metrics. + pub const fn reason(self) -> &'static str { + match self.class { + WorkloadClass::ForegroundRead => "foreground_read_pressure", + WorkloadClass::ForegroundWrite => "foreground_write_pressure", + _ => "foreground_pressure", + } + } +} + +/// Return the strongest foreground pressure in `snapshot`, if any. +/// +/// A zero threshold disables its class. `Saturated` counts as full utilization +/// regardless of the reported limit; otherwise a class contributes only when it +/// reports a non-zero limit, with a missing active count read as zero. When both +/// classes are above their threshold the higher utilization wins. +/// +/// Callers own the enable switch: this function evaluates thresholds only. +pub fn foreground_pressure( + snapshot: &WorkloadAdmissionRegistrySnapshot, + read_threshold_pct: usize, + write_threshold_pct: usize, +) -> Option { + [ + (WorkloadClass::ForegroundRead, read_threshold_pct), + (WorkloadClass::ForegroundWrite, write_threshold_pct), + ] + .into_iter() + .filter_map(|(class, threshold_pct)| { + if threshold_pct == 0 { + return None; + } + + let entry = snapshot.get(class)?; + let usage_pct = if matches!(entry.state, AdmissionState::Saturated) { + 100 + } else { + let limit = entry.limit?; + if limit == 0 { + return None; + } + entry + .active + .unwrap_or(0) + .saturating_mul(100) + .checked_div(limit) + .unwrap_or(100) + }; + + (usage_pct >= threshold_pct).then_some(ForegroundPressure { + class, + usage_pct, + threshold_pct, + }) + }) + .max_by_key(|pressure| pressure.usage_pct) +} + #[cfg(test)] mod tests { use super::*; @@ -314,4 +384,205 @@ mod tests { assert!(err.to_string().contains("unexpected")); } + + fn counted( + class: WorkloadClass, + state: AdmissionState, + active: Option, + limit: Option, + ) -> WorkloadAdmissionSnapshot { + WorkloadAdmissionSnapshot::new(class, state).with_counts(active, None, limit) + } + + fn registry(entries: Vec) -> WorkloadAdmissionRegistrySnapshot { + WorkloadAdmissionRegistrySnapshot::new(entries) + } + + #[test] + fn foreground_pressure_reason_labels_cover_non_foreground_classes() { + let read = ForegroundPressure { + class: WorkloadClass::ForegroundRead, + usage_pct: 90, + threshold_pct: 80, + }; + let write = ForegroundPressure { + class: WorkloadClass::ForegroundWrite, + usage_pct: 90, + threshold_pct: 80, + }; + let repair = ForegroundPressure { + class: WorkloadClass::Repair, + usage_pct: 90, + threshold_pct: 80, + }; + + assert_eq!(read.reason(), "foreground_read_pressure"); + assert_eq!(write.reason(), "foreground_write_pressure"); + assert_eq!(repair.reason(), "foreground_pressure"); + } + + #[test] + fn foreground_pressure_is_disabled_when_both_thresholds_are_zero() { + let snapshot = registry(vec![ + counted(WorkloadClass::ForegroundRead, AdmissionState::Saturated, Some(8), Some(8)), + counted(WorkloadClass::ForegroundWrite, AdmissionState::Saturated, Some(8), Some(8)), + ]); + + assert_eq!(foreground_pressure(&snapshot, 0, 0), None); + } + + #[test] + fn foreground_pressure_skips_only_the_class_whose_threshold_is_zero() { + let snapshot = registry(vec![ + counted(WorkloadClass::ForegroundRead, AdmissionState::Open, Some(10), Some(10)), + counted(WorkloadClass::ForegroundWrite, AdmissionState::Open, Some(9), Some(10)), + ]); + + assert_eq!( + foreground_pressure(&snapshot, 0, 80), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundWrite, + usage_pct: 90, + threshold_pct: 80, + }) + ); + assert_eq!( + foreground_pressure(&snapshot, 80, 0), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundRead, + usage_pct: 100, + threshold_pct: 80, + }) + ); + } + + #[test] + fn foreground_pressure_ignores_missing_entries() { + let snapshot = registry(vec![counted(WorkloadClass::Scanner, AdmissionState::Saturated, Some(8), Some(8))]); + + assert_eq!(foreground_pressure(&snapshot, 1, 1), None); + } + + #[test] + fn foreground_pressure_ignores_missing_and_zero_limits() { + let missing_limit = registry(vec![counted( + WorkloadClass::ForegroundRead, + AdmissionState::Throttled, + Some(8), + None, + )]); + let zero_limit = registry(vec![counted( + WorkloadClass::ForegroundWrite, + AdmissionState::Throttled, + Some(8), + Some(0), + )]); + + assert_eq!(foreground_pressure(&missing_limit, 1, 1), None); + assert_eq!(foreground_pressure(&zero_limit, 1, 1), None); + } + + #[test] + fn foreground_pressure_treats_saturated_as_full_without_reading_limit() { + let snapshot = registry(vec![ + counted(WorkloadClass::ForegroundRead, AdmissionState::Saturated, None, None), + counted(WorkloadClass::ForegroundWrite, AdmissionState::Saturated, Some(0), Some(0)), + ]); + + assert_eq!( + foreground_pressure(&snapshot, 100, 0), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundRead, + usage_pct: 100, + threshold_pct: 100, + }) + ); + assert_eq!( + foreground_pressure(&snapshot, 0, 100), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundWrite, + usage_pct: 100, + threshold_pct: 100, + }) + ); + } + + #[test] + fn foreground_pressure_reads_missing_active_as_zero() { + let snapshot = registry(vec![counted(WorkloadClass::ForegroundRead, AdmissionState::Open, None, Some(8))]); + + assert_eq!(foreground_pressure(&snapshot, 1, 1), None); + } + + #[test] + fn foreground_pressure_returns_the_higher_utilization_when_both_classes_exceed() { + let read_higher = registry(vec![ + counted(WorkloadClass::ForegroundRead, AdmissionState::Open, Some(19), Some(20)), + counted(WorkloadClass::ForegroundWrite, AdmissionState::Open, Some(17), Some(20)), + ]); + let write_higher = registry(vec![ + counted(WorkloadClass::ForegroundRead, AdmissionState::Open, Some(17), Some(20)), + counted(WorkloadClass::ForegroundWrite, AdmissionState::Open, Some(19), Some(20)), + ]); + + assert_eq!( + foreground_pressure(&read_higher, 80, 80), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundRead, + usage_pct: 95, + threshold_pct: 80, + }) + ); + assert_eq!( + foreground_pressure(&write_higher, 80, 80), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundWrite, + usage_pct: 95, + threshold_pct: 80, + }) + ); + } + + #[test] + fn foreground_pressure_breaks_utilization_ties_toward_the_write_class() { + let snapshot = registry(vec![ + counted(WorkloadClass::ForegroundRead, AdmissionState::Open, Some(18), Some(20)), + counted(WorkloadClass::ForegroundWrite, AdmissionState::Open, Some(18), Some(20)), + ]); + + assert_eq!( + foreground_pressure(&snapshot, 80, 80), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundWrite, + usage_pct: 90, + threshold_pct: 80, + }) + ); + } + + #[test] + fn foreground_pressure_triggers_exactly_at_the_threshold_and_not_below() { + let at_threshold = registry(vec![counted( + WorkloadClass::ForegroundRead, + AdmissionState::Open, + Some(8), + Some(10), + )]); + let below_threshold = registry(vec![counted( + WorkloadClass::ForegroundRead, + AdmissionState::Open, + Some(7), + Some(10), + )]); + + assert_eq!( + foreground_pressure(&at_threshold, 80, 80), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundRead, + usage_pct: 80, + threshold_pct: 80, + }) + ); + assert_eq!(foreground_pressure(&below_threshold, 80, 80), None); + } }