From 29899f4731974ea322270b93404a91b94768ff27 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 23:28:45 +0800 Subject: [PATCH] refactor(replication): isolate queue backpressure decisions (#4202) --- .../bucket/replication/replication_pool.rs | 177 +++++++++--------- crates/replication/src/lib.rs | 9 +- crates/replication/src/queue.rs | 51 ++++- crates/replication/src/runtime.rs | 143 ++++++++++++++ 4 files changed, 282 insertions(+), 98 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 7fbef38b8..8c8f82d63 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -34,11 +34,12 @@ use super::runtime_boundary as runtime_sources; use super::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; use lazy_static::lazy_static; use rustfs_replication::{ - DeletedObjectReplicationInfo, LARGE_WORKER_COUNT, ReplicationHealQueueAction, ReplicationHealQueueResult, - ReplicationHealResyncDeletes, ReplicationOperation, ReplicationPoolOpts, ReplicationPriority, ReplicationQueueAdmission, - WORKER_MAX_LIMIT, initial_worker_counts, mrf_worker_size_to_count, next_large_worker_count, next_mrf_worker_count, - next_regular_worker_count, replication_heal_queue_action, resized_worker_counts, should_auto_resume_resync, - should_grow_large_workers, should_queue_large_object, + DeletedObjectReplicationInfo, LARGE_WORKER_COUNT, ReplicationBackpressureRecommendation, ReplicationBackpressureState, + ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, ReplicationOperation, + ReplicationPoolOpts, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, WORKER_MAX_LIMIT, + initial_worker_counts, large_worker_backpressure_resize, mrf_worker_size_to_count, replication_backpressure_recommendation, + replication_heal_queue_action, resized_worker_counts, should_auto_resume_resync, should_queue_large_object, + worker_queue_for_replication_type, }; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; use std::sync::Arc; @@ -396,6 +397,73 @@ impl ReplicationPool { workers.get(index).cloned() } + async fn worker_queue_channel( + &self, + op_type: &ReplicationType, + bucket: &str, + object: &str, + size: i64, + ) -> Option> { + match worker_queue_for_replication_type(op_type) { + ReplicationWorkerQueue::Mrf => Some(self.mrf_replica_tx.clone()), + ReplicationWorkerQueue::Regular => self.get_worker_ch(bucket, object, size).await, + } + } + + async fn apply_queue_backpressure(&self, queue_type: &'static str, include_mrf_workers: bool, message: &'static str) { + let priority = self.priority.read().await.clone(); + let max_workers = *self.max_workers.read().await; + let current_workers = self.workers.read().await.len(); + let current_mrf_workers = self.mrf_worker_size.load(Ordering::SeqCst); + let recommendation = replication_backpressure_recommendation( + &priority, + ReplicationBackpressureState { + current_workers, + active_workers: self.active_workers(), + current_mrf_workers, + active_mrf_workers: self.active_mrf_workers(), + max_workers, + include_mrf_workers, + }, + ); + + match recommendation { + ReplicationBackpressureRecommendation::KeepFast => { + debug!( + event = EVENT_REPLICATION_BACKPRESSURE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + queue_type, + priority = "fast", + recommendation = "none", + "{message}" + ); + } + ReplicationBackpressureRecommendation::SetPriorityAuto => { + debug!( + event = EVENT_REPLICATION_BACKPRESSURE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + queue_type, + priority = "slow", + recommendation = "set_priority_auto", + "{message}" + ); + } + ReplicationBackpressureRecommendation::Resize(resize) => { + if let Some(regular_workers) = resize.regular_workers { + self.resize_workers(regular_workers.new_count, regular_workers.existing_count) + .await; + } + + if let Some(mrf_workers) = resize.mrf_workers { + self.resize_failed_workers(mrf_workers).await; + } + } + ReplicationBackpressureRecommendation::Noop => {} + } + } + /// Queues a replica task pub async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission { // If object is large, queue it to a static set of large workers @@ -418,6 +486,7 @@ impl ReplicationPool { // Try to add more workers if possible let max_l_workers = *self.max_l_workers.read().await; let existing = lrg_workers.len(); + let resize = large_worker_backpressure_resize(existing, self.active_lrg_workers(), max_l_workers); drop(lrg_workers); // Queue to MRF if worker is busy. @@ -425,10 +494,8 @@ impl ReplicationPool { queue_mrf_save_admission(&self.mrf_save_tx, ri.to_mrf_entry(), &ri.bucket, &ri.name, "large_object") .await; - if should_grow_large_workers(self.active_lrg_workers(), max_l_workers) { - let workers = next_large_worker_count(existing, max_l_workers); - - self.resize_lrg_workers(workers, existing).await; + if let Some(resize) = resize { + self.resize_lrg_workers(resize.new_count, resize.existing_count).await; } return admission; } @@ -440,10 +507,7 @@ impl ReplicationPool { // Handle regular sized objects - let ch = match ri.op_type { - ReplicationType::Heal | ReplicationType::ExistingObject => Some(self.mrf_replica_tx.clone()), - _ => self.get_worker_ch(&ri.bucket, &ri.name, ri.size).await, - }; + let ch = self.worker_queue_channel(&ri.op_type, &ri.bucket, &ri.name, ri.size).await; let Some(channel) = ch else { return ReplicationQueueAdmission::Missed; @@ -457,57 +521,17 @@ impl ReplicationPool { let admission = queue_mrf_save_admission(&self.mrf_save_tx, ri.to_mrf_entry(), &ri.bucket, &ri.name, "object").await; // Try to scale up workers based on priority - let priority = self.priority.read().await.clone(); - let max_workers = *self.max_workers.read().await; - - match priority { - ReplicationPriority::Fast => { - debug!( - event = EVENT_REPLICATION_BACKPRESSURE, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - queue_type = "object", - priority = "fast", - recommendation = "none", - "Replication queue is backpressured" - ); - } - ReplicationPriority::Slow => { - debug!( - event = EVENT_REPLICATION_BACKPRESSURE, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - queue_type = "object", - priority = "slow", - recommendation = "set_priority_auto", - "Replication queue is backpressured" - ); - } - ReplicationPriority::Auto => { - let workers = self.workers.read().await; - if let Some(new_count) = next_regular_worker_count(workers.len(), self.active_workers(), max_workers) { - let existing = workers.len(); - - drop(workers); - self.resize_workers(new_count, existing).await; - } - - let current_mrf = self.mrf_worker_size.load(Ordering::SeqCst); - if let Some(new_mrf) = next_mrf_worker_count(current_mrf, self.active_mrf_workers(), max_workers) { - self.resize_failed_workers(new_mrf).await; - } - } - } + self.apply_queue_backpressure("object", true, "Replication queue is backpressured") + .await; admission } /// Queues a replica delete task pub async fn queue_replica_delete_task(&self, doi: DeletedObjectReplicationInfo) -> ReplicationQueueAdmission { - let ch = match doi.op_type { - ReplicationType::Heal | ReplicationType::ExistingObject => Some(self.mrf_replica_tx.clone()), - _ => self.get_worker_ch(&doi.bucket, &doi.delete_object.object_name, 0).await, - }; + let ch = self + .worker_queue_channel(&doi.op_type, &doi.bucket, &doi.delete_object.object_name, 0) + .await; let Some(channel) = ch else { return ReplicationQueueAdmission::Missed; @@ -526,41 +550,8 @@ impl ReplicationPool { ) .await; - let priority = self.priority.read().await.clone(); - let max_workers = *self.max_workers.read().await; - - match priority { - ReplicationPriority::Fast => { - debug!( - event = EVENT_REPLICATION_BACKPRESSURE, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - queue_type = "delete", - priority = "fast", - recommendation = "none", - "Replication delete queue is backpressured" - ); - } - ReplicationPriority::Slow => { - debug!( - event = EVENT_REPLICATION_BACKPRESSURE, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION, - queue_type = "delete", - priority = "slow", - recommendation = "set_priority_auto", - "Replication delete queue is backpressured" - ); - } - ReplicationPriority::Auto => { - let workers = self.workers.read().await; - if let Some(new_count) = next_regular_worker_count(workers.len(), self.active_workers(), max_workers) { - let existing = workers.len(); - drop(workers); - self.resize_workers(new_count, existing).await; - } - } - } + self.apply_queue_backpressure("delete", false, "Replication delete queue is backpressured") + .await; admission } diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 2cbb11b50..29b8bf4c3 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -46,7 +46,8 @@ pub use operation::{ }; pub use queue::{ ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, ReplicationOperation, - ReplicationPriority, ReplicationQueueAdmission, replication_heal_queue_action, + ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, mrf_save_admission, replication_heal_queue_action, + worker_queue_for_replication_type, }; pub use resync::{ BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, @@ -56,8 +57,10 @@ pub use resync::{ pub use rule::ReplicationRuleExt; pub use runtime::{ LARGE_WORKER_COUNT, MIN_LARGE_OBJ_SIZE, MRF_WORKER_AUTO_DEFAULT, MRF_WORKER_MAX_LIMIT, MRF_WORKER_MIN_LIMIT, - ReplicationPoolOpts, ReplicationWorkerCounts, WORKER_AUTO_DEFAULT, WORKER_MAX_LIMIT, WORKER_MIN_LIMIT, initial_worker_counts, - mrf_worker_size_to_count, next_large_worker_count, next_mrf_worker_count, next_regular_worker_count, resized_worker_counts, + ReplicationBackpressureRecommendation, ReplicationBackpressureResize, ReplicationBackpressureState, ReplicationPoolOpts, + ReplicationWorkerCounts, ReplicationWorkerResize, WORKER_AUTO_DEFAULT, WORKER_MAX_LIMIT, WORKER_MIN_LIMIT, + initial_worker_counts, large_worker_backpressure_resize, mrf_worker_size_to_count, next_large_worker_count, + next_mrf_worker_count, next_regular_worker_count, replication_backpressure_recommendation, resized_worker_counts, should_grow_large_workers, should_queue_large_object, worker_counts_for_priority, }; pub use rustfs_filemeta::{ diff --git a/crates/replication/src/queue.rs b/crates/replication/src/queue.rs index 0a06e0e3f..016db9d61 100644 --- a/crates/replication/src/queue.rs +++ b/crates/replication/src/queue.rs @@ -40,6 +40,27 @@ impl ReplicationQueueAdmission { } } +pub fn mrf_save_admission(saved: bool) -> ReplicationQueueAdmission { + if saved { + ReplicationQueueAdmission::Queued + } else { + ReplicationQueueAdmission::Missed + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicationWorkerQueue { + Regular, + Mrf, +} + +pub fn worker_queue_for_replication_type(op_type: &ReplicationType) -> ReplicationWorkerQueue { + match op_type { + ReplicationType::Heal | ReplicationType::ExistingObject => ReplicationWorkerQueue::Mrf, + _ => ReplicationWorkerQueue::Regular, + } +} + #[derive(Debug, Clone, Default)] pub struct ReplicationHealQueueResult { pub object_info: ReplicateObjectInfo, @@ -247,11 +268,14 @@ mod tests { use rustfs_storage_api::DeletedObject; use uuid::Uuid; - use super::{ReplicationHealQueueAction, ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission}; + use super::{ + ReplicationHealQueueAction, ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, + }; use crate::{ DeletedObjectReplicationInfo, REPLICATE_EXISTING, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, - ResyncTargetDecision, VersionPurgeStatusType, replication_heal_queue_action, + ResyncTargetDecision, VersionPurgeStatusType, mrf_save_admission, replication_heal_queue_action, + worker_queue_for_replication_type, }; #[test] @@ -268,6 +292,29 @@ mod tests { assert_eq!(admission, ReplicationQueueAdmission::Missed); } + #[test] + fn mrf_save_admission_maps_send_result() { + assert_eq!(mrf_save_admission(true), ReplicationQueueAdmission::Queued); + assert_eq!(mrf_save_admission(false), ReplicationQueueAdmission::Missed); + } + + #[test] + fn worker_queue_routes_heal_and_existing_objects_to_mrf() { + assert_eq!(worker_queue_for_replication_type(&ReplicationType::Heal), ReplicationWorkerQueue::Mrf); + assert_eq!( + worker_queue_for_replication_type(&ReplicationType::ExistingObject), + ReplicationWorkerQueue::Mrf + ); + assert_eq!( + worker_queue_for_replication_type(&ReplicationType::Object), + ReplicationWorkerQueue::Regular + ); + assert_eq!( + worker_queue_for_replication_type(&ReplicationType::Delete), + ReplicationWorkerQueue::Regular + ); + } + #[test] fn heal_queue_action_routes_failed_objects_to_heal_queue() { let mut roi = replicate_object_info(ReplicationStatusType::Failed); diff --git a/crates/replication/src/runtime.rs b/crates/replication/src/runtime.rs index cb4122368..3b47cf496 100644 --- a/crates/replication/src/runtime.rs +++ b/crates/replication/src/runtime.rs @@ -23,6 +23,36 @@ pub const MRF_WORKER_AUTO_DEFAULT: usize = 4; pub const LARGE_WORKER_COUNT: usize = 10; pub const MIN_LARGE_OBJ_SIZE: i64 = 128 * 1024 * 1024; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicationWorkerResize { + pub new_count: usize, + pub existing_count: usize, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicationBackpressureState { + pub current_workers: usize, + pub active_workers: i32, + pub current_mrf_workers: i32, + pub active_mrf_workers: i32, + pub max_workers: usize, + pub include_mrf_workers: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicationBackpressureResize { + pub regular_workers: Option, + pub mrf_workers: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicationBackpressureRecommendation { + KeepFast, + SetPriorityAuto, + Resize(ReplicationBackpressureResize), + Noop, +} + #[derive(Debug, Clone)] pub struct ReplicationPoolOpts { pub priority: ReplicationPriority, @@ -133,6 +163,49 @@ pub fn next_mrf_worker_count(current_mrf_workers: i32, active_mrf_workers: i32, } } +pub fn large_worker_backpressure_resize( + existing_workers: usize, + active_lrg_workers: i32, + max_l_workers: usize, +) -> Option { + should_grow_large_workers(active_lrg_workers, max_l_workers).then_some(ReplicationWorkerResize { + new_count: next_large_worker_count(existing_workers, max_l_workers), + existing_count: existing_workers, + }) +} + +pub fn replication_backpressure_recommendation( + priority: &ReplicationPriority, + state: ReplicationBackpressureState, +) -> ReplicationBackpressureRecommendation { + match priority { + ReplicationPriority::Fast => ReplicationBackpressureRecommendation::KeepFast, + ReplicationPriority::Slow => ReplicationBackpressureRecommendation::SetPriorityAuto, + ReplicationPriority::Auto => { + let regular_workers = + next_regular_worker_count(state.current_workers, state.active_workers, state.max_workers).map(|new_count| { + ReplicationWorkerResize { + new_count, + existing_count: state.current_workers, + } + }); + let mrf_workers = state + .include_mrf_workers + .then(|| next_mrf_worker_count(state.current_mrf_workers, state.active_mrf_workers, state.max_workers)) + .flatten(); + + if regular_workers.is_none() && mrf_workers.is_none() { + ReplicationBackpressureRecommendation::Noop + } else { + ReplicationBackpressureRecommendation::Resize(ReplicationBackpressureResize { + regular_workers, + mrf_workers, + }) + } + } + } +} + fn auto_worker_count(current_workers: usize, auto_default: usize) -> usize { if current_workers < auto_default { current_workers.saturating_add(1).min(auto_default) @@ -229,4 +302,74 @@ mod tests { assert!(should_grow_large_workers(1, LARGE_WORKER_COUNT)); assert_eq!(next_large_worker_count(LARGE_WORKER_COUNT, LARGE_WORKER_COUNT), LARGE_WORKER_COUNT); } + + #[test] + fn large_worker_resize_recommendation_respects_growth_limit() { + assert_eq!( + large_worker_backpressure_resize(3, 2, LARGE_WORKER_COUNT), + Some(ReplicationWorkerResize { + new_count: 4, + existing_count: 3 + }) + ); + assert_eq!( + large_worker_backpressure_resize(LARGE_WORKER_COUNT, usize_to_i32_saturating(LARGE_WORKER_COUNT), LARGE_WORKER_COUNT), + None + ); + } + + #[test] + fn queue_backpressure_recommendation_preserves_priority_semantics() { + let state = ReplicationBackpressureState { + current_workers: 4, + active_workers: 4, + current_mrf_workers: 2, + active_mrf_workers: 2, + max_workers: WORKER_MAX_LIMIT, + include_mrf_workers: true, + }; + + assert_eq!( + replication_backpressure_recommendation(&ReplicationPriority::Fast, state), + ReplicationBackpressureRecommendation::KeepFast + ); + assert_eq!( + replication_backpressure_recommendation(&ReplicationPriority::Slow, state), + ReplicationBackpressureRecommendation::SetPriorityAuto + ); + + assert_eq!( + replication_backpressure_recommendation(&ReplicationPriority::Auto, state), + ReplicationBackpressureRecommendation::Resize(ReplicationBackpressureResize { + regular_workers: Some(ReplicationWorkerResize { + new_count: 5, + existing_count: 4 + }), + mrf_workers: Some(3), + }) + ); + } + + #[test] + fn delete_queue_backpressure_skips_mrf_growth() { + let state = ReplicationBackpressureState { + current_workers: 4, + active_workers: 4, + current_mrf_workers: 2, + active_mrf_workers: 2, + max_workers: WORKER_MAX_LIMIT, + include_mrf_workers: false, + }; + + assert_eq!( + replication_backpressure_recommendation(&ReplicationPriority::Auto, state), + ReplicationBackpressureRecommendation::Resize(ReplicationBackpressureResize { + regular_workers: Some(ReplicationWorkerResize { + new_count: 5, + existing_count: 4 + }), + mrf_workers: None, + }) + ); + } }