From 60b0fbf07564b9f19c1283d07750e897d3e257aa 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 11:56:51 +0800 Subject: [PATCH] refactor: bridge storage concurrency policies (#3683) --- ARCHITECTURE.md | 9 ++++-- docs/architecture/migration-progress.md | 42 +++++++++++++++++++------ rustfs/src/storage/backpressure.rs | 34 ++++++++++++++++++++ rustfs/src/storage/deadlock_detector.rs | 32 +++++++++++++++++-- 4 files changed, 102 insertions(+), 15 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 33af31a2b..55b8d2f71 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -244,11 +244,14 @@ Depth 8 — TOP: ### High -- **Dependency inversions.** `utils → config` and `common → filemeta/madmin` break - the layering model. These need to be untangled. +- **Dependency inversions.** Historical `utils → config` and + `common → filemeta/madmin` edges must stay removed so leaf/helper crates do + not regain upward dependencies. - **Three-layer BackpressureConfig/DeadlockConfig duplication** across io-core, - concurrency, and rustfs/storage. Should be defined once with builder/composition. + 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. ### Medium diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index db3a0013c..d096af63e 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,15 +5,16 @@ 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-cluster-control-plane-status-snapshots` -- Baseline: completed `C-001/C-002/C-003`. -- Stacked on: ECStore cluster control-plane read models. +- Branch: `overtrue/arch-storage-concurrency-policy-bridges` +- Baseline: completed `C-004/C-005/C-006`. +- Stacked on: ECStore cluster status snapshots. - PR type for this branch: `api-extraction` - Runtime behavior changes: none. -- Rust code changes: add pool-state, local-node storage, and peer-health status - read models to the ECStore ClusterControlPlane snapshot. +- Rust code changes: add storage-to-concurrency policy bridge projections for + object backpressure and request hang/deadlock detection. - CI/script changes: none. -- Docs changes: record the C-004/C-005/C-006 status snapshot slice. +- Docs changes: record the C-011 policy bridge precondition for later + controller status work. ## Phase 0 Tasks @@ -68,6 +69,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block budget, heal admission, and the Tokio runtime builder. - Must preserve: no Rust source changes, no scheduler/controller contract changes, and no runtime behavior changes. +- [x] `C-011-POLICY` Bridge storage concurrency policies. + - Completed slice: add explicit projections from storage object backpressure + and request hang/deadlock policies into the shared `rustfs-concurrency` + facade and reusable `rustfs-io-core` configs. + - Acceptance: storage keeps existing env/default ownership and runtime + behavior, while later controller/read-only status work can consume the + shared facade policy shape instead of duplicating field mapping. + - 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 policy 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 @@ -3101,14 +3115,24 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | C-004/C-005/C-006 extends ClusterControlPlane with pool-state, local-node storage, and peer-health status read models behind `api::cluster`. | -| Migration preservation | passed | New status surfaces remain static read-only projections; no placement, readiness, health loop, lock, or object I/O behavior changes. | -| Testing/verification | passed | ECStore cluster tests, ECStore/RustFS compile coverage, migration guard, formatting, diff hygiene, risk scan, full pre-commit, and three-expert review passed. | +| 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. | ## Verification Notes Passed before push: +- 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. + - `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-004/C-005/C-006 current slice: - `cargo test -p rustfs-ecstore cluster -- --nocapture`: passed, 7 tests. - `cargo check -p rustfs-ecstore --all-targets`: passed. diff --git a/rustfs/src/storage/backpressure.rs b/rustfs/src/storage/backpressure.rs index 942e31d1c..de54276ea 100644 --- a/rustfs/src/storage/backpressure.rs +++ b/rustfs/src/storage/backpressure.rs @@ -42,6 +42,8 @@ use tokio::io::{DuplexStream, duplex}; use tracing::{debug, warn}; use metrics::counter; +use rustfs_concurrency::PipeBackpressurePolicy; +use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig; /// Object-transfer duplex pipe backpressure policy. #[derive(Debug, Clone, Copy)] @@ -98,6 +100,20 @@ impl ObjectPipeBackpressurePolicy { pub fn low_watermark_bytes(&self) -> usize { (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize } + + /// Project this object-transfer policy into the shared concurrency facade policy. + pub fn to_concurrency_policy(&self) -> PipeBackpressurePolicy { + PipeBackpressurePolicy { + buffer_size: self.buffer_size, + high_watermark: self.high_watermark, + low_watermark: self.low_watermark, + } + } + + /// Project this object-transfer policy into the reusable io-core admission config. + pub fn to_core_config(&self) -> CoreBackpressureConfig { + self.to_concurrency_policy().to_core_config() + } } /// Backpressure state. @@ -487,6 +503,24 @@ mod tests { assert_eq!(config.low_watermark_bytes(), 500); } + #[test] + fn test_backpressure_policy_projects_to_concurrency_and_core_config() { + let config = ObjectPipeBackpressurePolicy { + buffer_size: 2000, + high_watermark: 75, + low_watermark: 40, + }; + let concurrency = config.to_concurrency_policy(); + let core = config.to_core_config(); + + assert_eq!(concurrency.buffer_size, config.buffer_size); + assert_eq!(concurrency.high_watermark, config.high_watermark); + assert_eq!(concurrency.low_watermark, config.low_watermark); + assert_eq!(core.high_water_mark, 0.75); + assert_eq!(core.low_water_mark, 0.40); + assert!(core.enabled); + } + #[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 0e35a332c..ba4411d1e 100644 --- a/rustfs/src/storage/deadlock_detector.rs +++ b/rustfs/src/storage/deadlock_detector.rs @@ -66,6 +66,7 @@ use tokio::sync::broadcast; use tracing::{debug, error, warn}; use metrics::counter; +use rustfs_concurrency::DeadlockMonitorPolicy; use rustfs_io_core::DeadlockDetectorConfig as CoreDeadlockConfig; /// Request identifier type. @@ -124,10 +125,15 @@ impl RequestHangDetectionPolicy { /// Convert the request-level policy into the shared io-core deadlock config. pub fn to_core_config(&self) -> CoreDeadlockConfig { - CoreDeadlockConfig { + self.to_concurrency_policy().to_core_config() + } + + /// Convert the request-level policy into the shared concurrency facade policy. + pub fn to_concurrency_policy(&self) -> DeadlockMonitorPolicy { + DeadlockMonitorPolicy { enabled: self.enabled, - detection_interval: self.check_interval, - max_hold_time: self.hang_threshold, + check_interval: self.check_interval, + hang_threshold: self.hang_threshold, } } } @@ -612,6 +618,26 @@ mod tests { assert_eq!(config.hang_threshold, Duration::from_secs(10)); } + #[test] + fn test_request_hang_policy_projects_to_concurrency_and_core_config() { + 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 core = config.to_core_config(); + + assert!(concurrency.enabled); + assert_eq!(concurrency.check_interval, config.check_interval); + assert_eq!(concurrency.hang_threshold, config.hang_threshold); + assert!(core.enabled); + assert_eq!(core.detection_interval, config.check_interval); + assert_eq!(core.max_hold_time, config.hang_threshold); + assert!(config.capture_backtrace); + } + #[test] fn test_request_resource_tracker() { let tracker = RequestResourceTracker::new("req-1".to_string(), "GetObject bucket/key");