From 7b81f2ed5307ae5e03145c76ee94947440d2de93 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sun, 21 Jun 2026 12:44:44 +0800 Subject: [PATCH] refactor: consume storage concurrency policies (#3684) --- ARCHITECTURE.md | 6 +-- docs/architecture/migration-progress.md | 41 ++++++++++++++----- rustfs/src/storage/backpressure.rs | 42 +++++++++++++++++--- rustfs/src/storage/deadlock_detector.rs | 53 +++++++++++++++++-------- 4 files changed, 109 insertions(+), 33 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 55b8d2f71..da869bf5b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -249,9 +249,9 @@ Depth 8 — TOP: not regain upward dependencies. - **Three-layer BackpressureConfig/DeadlockConfig duplication** across io-core, - concurrency, and rustfs/storage. Storage policies now expose explicit - projections into the concurrency/io-core policy shapes; later work should use - those bridges before deleting compatibility wrappers. + concurrency, and rustfs/storage. Storage policies now expose and consume + explicit projections into the concurrency/io-core policy shapes; later work + should use those bridges before deleting compatibility wrappers. ### Medium diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index d096af63e..b31834fb5 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,15 +5,15 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-storage-concurrency-policy-bridges` -- Baseline: completed `C-004/C-005/C-006`. -- Stacked on: ECStore cluster status snapshots. +- Branch: `overtrue/arch-storage-concurrency-policy-consumers` +- Baseline: completed `C-004/C-005/C-006/C-011`. +- Stacked on: storage concurrency policy bridges. - PR type for this branch: `api-extraction` - Runtime behavior changes: none. -- Rust code changes: add storage-to-concurrency policy bridge projections for - object backpressure and request hang/deadlock detection. +- Rust code changes: route storage object backpressure and request hang/deadlock + runtime config consumption through the shared concurrency facade policies. - CI/script changes: none. -- Docs changes: record the C-011 policy bridge precondition for later +- Docs changes: record the C-012 policy consumer precondition for later controller status work. ## Phase 0 Tasks @@ -82,6 +82,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block - Verification: storage backpressure/deadlock policy tests, compile coverage, formatting, diff hygiene, risk scan, architecture guard, pre-commit quality gate, and three-expert review. +- [x] `C-012-POLICY` Consume storage concurrency policy bridges. + - Completed slice: route object backpressure threshold derivation and request + hang/deadlock runtime policy reads through the shared `rustfs-concurrency` + facade policies. + - Acceptance: storage keeps env/default ownership and local state machines, + while threshold and hang-policy consumption is anchored on the shared + concurrency policy shapes. + - Must preserve: no worker start/stop, no object pipe state-machine change, no + deadlock detector lifecycle change, no metrics label change, and no S3 I/O + behavior change. + - Verification: storage backpressure/deadlock consumer tests, compile + coverage, formatting, diff hygiene, risk scan, architecture guard, + pre-commit quality gate, and three-expert review. - [x] `G-012` Inventory placement and repair invariants. - Acceptance: [`placement-repair-invariants.md`](placement-repair-invariants.md) records @@ -3115,14 +3128,24 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | C-011 exposes storage backpressure and request-hang policies through existing `rustfs-concurrency` and `rustfs-io-core` policy shapes without adding ownership cycles. | -| Migration preservation | passed | Bridge methods preserve storage env/default ownership, object pipe state, deadlock detector lifecycle, metrics labels, and S3 I/O behavior. | -| Testing/verification | passed | Focused storage policy tests, RustFS lib compile coverage, migration guard, formatting, diff hygiene, added-line risk scan, full pre-commit, and three-expert review passed. | +| Quality/architecture | passed | C-012 routes storage backpressure and request-hang runtime consumers through existing `rustfs-concurrency` policy shapes without adding ownership cycles. | +| Migration preservation | passed | Consumer changes preserve storage env/default ownership, object pipe state, deadlock detector lifecycle, metrics labels, and S3 I/O behavior. | +| Testing/verification | passed | Focused storage consumer tests, RustFS lib compile coverage, migration guard, formatting, diff hygiene, added-line risk scan, full pre-commit, and three-expert review passed. | ## Verification Notes Passed before push: +- Issue #660 C-012 current slice: + - `cargo test -p rustfs --lib storage::backpressure::tests:: -- --nocapture`: passed. + - `cargo test -p rustfs --lib storage::deadlock_detector::tests:: -- --nocapture`: passed. + - `cargo check -p rustfs --lib`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - Rust added-line risk scan on changed storage Rust files: passed. + - `make pre-commit`: passed. + - Issue #660 C-011 current slice: - `cargo test -p rustfs --lib storage::deadlock_detector::tests::test_request_hang_policy_projects_to_concurrency_and_core_config -- --nocapture`: passed. - `cargo test -p rustfs --lib storage::backpressure::tests::test_backpressure_policy_projects_to_concurrency_and_core_config -- --nocapture`: passed. diff --git a/rustfs/src/storage/backpressure.rs b/rustfs/src/storage/backpressure.rs index de54276ea..76eff2563 100644 --- a/rustfs/src/storage/backpressure.rs +++ b/rustfs/src/storage/backpressure.rs @@ -229,9 +229,10 @@ impl BackpressurePipe { /// Create a new backpressure-aware pipe with custom configuration. pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { - let (reader, writer) = duplex(config.buffer_size); - let high_watermark_bytes = config.high_watermark_bytes(); - let low_watermark_bytes = config.low_watermark_bytes(); + let policy = config.to_concurrency_policy(); + let (reader, writer) = duplex(policy.buffer_size); + let high_watermark_bytes = policy.high_watermark_bytes(); + let low_watermark_bytes = policy.low_watermark_bytes(); debug!( buffer_size = config.buffer_size, @@ -396,8 +397,9 @@ impl BackpressureMonitor { /// Create a new monitor with custom configuration. pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { - let high_watermark_bytes = config.high_watermark_bytes(); - let low_watermark_bytes = config.low_watermark_bytes(); + let policy = config.to_concurrency_policy(); + let high_watermark_bytes = policy.high_watermark_bytes(); + let low_watermark_bytes = policy.low_watermark_bytes(); Self { config, buffer_usage: AtomicUsize::new(0), @@ -521,6 +523,36 @@ mod tests { assert!(core.enabled); } + #[test] + fn test_backpressure_pipe_consumes_concurrency_policy_thresholds() { + let config = ObjectPipeBackpressurePolicy { + buffer_size: 2000, + high_watermark: 75, + low_watermark: 40, + }; + let concurrency = config.to_concurrency_policy(); + let pipe = BackpressurePipe::with_config(config); + + assert_eq!(pipe.capacity(), concurrency.buffer_size); + assert_eq!(pipe.high_watermark_bytes, concurrency.high_watermark_bytes()); + assert_eq!(pipe.low_watermark_bytes, concurrency.low_watermark_bytes()); + } + + #[test] + fn test_backpressure_monitor_consumes_concurrency_policy_thresholds() { + let config = ObjectPipeBackpressurePolicy { + buffer_size: 2000, + high_watermark: 75, + low_watermark: 40, + }; + let concurrency = config.to_concurrency_policy(); + let monitor = BackpressureMonitor::with_config(config); + + assert_eq!(monitor.meta().buffer_capacity, concurrency.buffer_size); + assert_eq!(monitor.high_watermark_bytes, concurrency.high_watermark_bytes()); + assert_eq!(monitor.low_watermark_bytes, concurrency.low_watermark_bytes()); + } + #[test] fn test_backpressure_state_display() { assert_eq!(format!("{}", BackpressureState::Normal), "normal"); diff --git a/rustfs/src/storage/deadlock_detector.rs b/rustfs/src/storage/deadlock_detector.rs index ba4411d1e..108b9718b 100644 --- a/rustfs/src/storage/deadlock_detector.rs +++ b/rustfs/src/storage/deadlock_detector.rs @@ -264,6 +264,8 @@ pub struct ResourceUsage { pub struct DeadlockDetector { /// Configuration. config: RequestHangDetectionPolicy, + /// Shared concurrency facade policy. + policy: DeadlockMonitorPolicy, /// Active request trackers. requests: Arc>>, /// Detection task handle. @@ -280,9 +282,11 @@ impl DeadlockDetector { /// Create a new deadlock detector. pub fn new(config: RequestHangDetectionPolicy) -> Self { let (shutdown_tx, _) = broadcast::channel(1); + let policy = config.to_concurrency_policy(); Self { config, + policy, requests: Arc::new(RwLock::new(HashMap::new())), detector_task: Arc::new(Mutex::new(None)), shutdown_tx, @@ -293,12 +297,12 @@ impl DeadlockDetector { /// Check if detection is enabled. pub fn is_enabled(&self) -> bool { - self.config.enabled + self.policy.enabled } /// Start the detection task. pub fn start(&self) { - if !self.config.enabled { + if !self.policy.enabled { debug!("Deadlock detection is disabled"); return; } @@ -309,22 +313,22 @@ impl DeadlockDetector { } let requests = self.requests.clone(); - let config = self.config.clone(); + let policy = self.policy; let deadlocks_detected = self.deadlocks_detected.clone(); let mut shutdown_rx = self.shutdown_tx.subscribe(); let running = self.running.clone(); let handle = tokio::spawn(async move { debug!( - check_interval_secs = config.check_interval.as_secs(), - hang_threshold_secs = config.hang_threshold.as_secs(), + check_interval_secs = policy.check_interval.as_secs(), + hang_threshold_secs = policy.hang_threshold.as_secs(), "Deadlock detector started" ); loop { tokio::select! { - _ = tokio::time::sleep(config.check_interval) => { - Self::detect_cycle(&requests, &config, &deadlocks_detected); + _ = tokio::time::sleep(policy.check_interval) => { + Self::detect_cycle(&requests, &policy, &deadlocks_detected); } _ = shutdown_rx.recv() => { debug!("Deadlock detector shutting down"); @@ -351,7 +355,7 @@ impl DeadlockDetector { /// Register a new request for tracking. pub fn register_request(&self, request_id: impl Into, description: impl Into) { - if !self.config.enabled { + if !self.policy.enabled { return; } @@ -365,7 +369,7 @@ impl DeadlockDetector { /// Unregister a request (it completed or was cancelled). pub fn unregister_request(&self, request_id: &str) { - if !self.config.enabled { + if !self.policy.enabled { return; } @@ -376,7 +380,7 @@ impl DeadlockDetector { /// Record a lock acquisition. pub fn record_lock_acquire(&self, request_id: &str, lock: LockInfo) { - if !self.config.enabled { + if !self.policy.enabled { return; } @@ -387,7 +391,7 @@ impl DeadlockDetector { /// Record a lock release. pub fn record_lock_release(&self, request_id: &str, lock_id: &LockId) { - if !self.config.enabled { + if !self.policy.enabled { return; } @@ -398,7 +402,7 @@ impl DeadlockDetector { /// Record that a request is waiting for a lock. pub fn record_lock_wait(&self, request_id: &str, lock: LockInfo) { - if !self.config.enabled { + if !self.policy.enabled { return; } @@ -409,7 +413,7 @@ impl DeadlockDetector { /// Clear the waiting lock (acquired or gave up). pub fn clear_lock_wait(&self, request_id: &str) { - if !self.config.enabled { + if !self.policy.enabled { return; } @@ -420,7 +424,7 @@ impl DeadlockDetector { /// Record resource usage. pub fn record_resource(&self, request_id: &str, resource_type: ResourceType, amount: usize) { - if !self.config.enabled { + if !self.policy.enabled { return; } @@ -442,13 +446,13 @@ impl DeadlockDetector { /// Detect deadlock cycles in the lock wait graph. fn detect_cycle( requests: &Arc>>, - config: &RequestHangDetectionPolicy, + policy: &DeadlockMonitorPolicy, deadlocks_detected: &Arc, ) { let requests_guard = requests.read().unwrap(); // Find hung requests - let hung_requests: Vec<_> = requests_guard.values().filter(|r| r.is_hung(config.hang_threshold)).collect(); + let hung_requests: Vec<_> = requests_guard.values().filter(|r| r.is_hung(policy.hang_threshold)).collect(); if hung_requests.is_empty() { return; @@ -638,6 +642,23 @@ mod tests { assert!(config.capture_backtrace); } + #[test] + fn test_deadlock_detector_consumes_concurrency_policy() { + let config = RequestHangDetectionPolicy { + enabled: true, + check_interval: Duration::from_secs(7), + hang_threshold: Duration::from_secs(11), + capture_backtrace: true, + }; + let concurrency = config.to_concurrency_policy(); + let detector = DeadlockDetector::new(config); + + assert!(detector.is_enabled()); + assert_eq!(detector.policy.check_interval, concurrency.check_interval); + assert_eq!(detector.policy.hang_threshold, concurrency.hang_threshold); + assert!(detector.config.capture_backtrace); + } + #[test] fn test_request_resource_tracker() { let tracker = RequestResourceTracker::new("req-1".to_string(), "GetObject bucket/key");