diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index a1d1e6fba..27f933441 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -32,8 +32,10 @@ use super::runtime_boundary as runtime_sources; use super::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; use lazy_static::lazy_static; use rustfs_replication::{ - DeletedObjectReplicationInfo, ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority, - ReplicationQueueAdmission, + DeletedObjectReplicationInfo, LARGE_WORKER_COUNT, ReplicationHealQueueResult, ReplicationOperation, ReplicationPoolOpts, + ReplicationPriority, ReplicationQueueAdmission, WORKER_MAX_LIMIT, initial_worker_counts, next_large_worker_count, + next_mrf_worker_count, next_regular_worker_count, resized_worker_counts, should_grow_large_workers, + should_queue_large_object, }; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; use std::sync::Arc; @@ -65,33 +67,6 @@ fn should_auto_resume_resync(status: ResyncStatusType) -> bool { matches!(status, ResyncStatusType::ResyncPending | ResyncStatusType::ResyncStarted) } -// Worker limits -pub const WORKER_MAX_LIMIT: usize = 500; -pub const WORKER_MIN_LIMIT: usize = 50; -pub const WORKER_AUTO_DEFAULT: usize = 100; -pub const MRF_WORKER_MAX_LIMIT: usize = 8; -pub const MRF_WORKER_MIN_LIMIT: usize = 2; -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; // 128MiB - -/// Replication pool options -#[derive(Debug, Clone)] -pub struct ReplicationPoolOpts { - pub priority: ReplicationPriority, - pub max_workers: Option, - pub max_l_workers: Option, -} - -impl Default for ReplicationPoolOpts { - fn default() -> Self { - Self { - priority: ReplicationPriority::Auto, - max_workers: None, - max_l_workers: None, - } - } -} /// Main replication pool structure #[derive(Debug)] pub struct ReplicationPool { @@ -138,23 +113,14 @@ pub struct ReplicationPool { impl ReplicationPool { /// Creates a new replication pool with specified options pub async fn new(opts: ReplicationPoolOpts, stats: Arc, storage: Arc) -> Arc { + let worker_counts = initial_worker_counts(&opts); let max_workers = opts.max_workers.unwrap_or(WORKER_MAX_LIMIT); - - let (workers, failed_workers) = match opts.priority { - ReplicationPriority::Fast => (WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT), - ReplicationPriority::Slow => (WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT), - ReplicationPriority::Auto => (WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT), - }; - - let workers = std::cmp::min(workers, max_workers); - let failed_workers = std::cmp::min(failed_workers, max_workers); - let max_l_workers = opts.max_l_workers.unwrap_or(LARGE_WORKER_COUNT); // Create MRF channels let (mrf_replica_tx, mrf_replica_rx) = mpsc::channel(100000); let (mrf_save_tx, mrf_save_rx) = mpsc::channel(100000); - let (mrf_worker_kill_tx, _mrf_worker_kill_rx) = mpsc::channel(failed_workers); + let (mrf_worker_kill_tx, _mrf_worker_kill_rx) = mpsc::channel(worker_counts.mrf_workers); let (mrf_stop_tx, _mrf_stop_rx) = mpsc::channel(1); let pool = Arc::new(Self { @@ -181,8 +147,8 @@ impl ReplicationPool { // Initialize workers pool.resize_lrg_workers(max_l_workers, 0).await; - pool.resize_workers(workers, 0).await; - pool.resize_failed_workers(failed_workers as i32).await; + pool.resize_workers(worker_counts.workers, 0).await; + pool.resize_failed_workers(worker_counts.mrf_workers_i32()).await; // Start background tasks pool.start_mrf_processor().await; @@ -399,39 +365,20 @@ impl ReplicationPool { max_workers: Option, max_l_workers: Option, ) { - let (workers, mrf_workers) = match pri { - ReplicationPriority::Fast => (WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT), - ReplicationPriority::Slow => (WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT), - ReplicationPriority::Auto => { - let mut workers = WORKER_AUTO_DEFAULT; - let mut mrf_workers = MRF_WORKER_AUTO_DEFAULT; + let current_workers = self.workers.read().await.len(); + let current_mrf = usize::try_from(self.mrf_worker_size.load(Ordering::SeqCst)).unwrap_or_default(); + let worker_counts = resized_worker_counts(&pri, max_workers, current_workers, current_mrf); - let current_workers = self.workers.read().await.len(); - if current_workers < WORKER_AUTO_DEFAULT { - workers = std::cmp::min(current_workers + 1, WORKER_AUTO_DEFAULT); - } - - let current_mrf = self.mrf_worker_size.load(Ordering::SeqCst) as usize; - if current_mrf < MRF_WORKER_AUTO_DEFAULT { - mrf_workers = std::cmp::min(current_mrf + 1, MRF_WORKER_AUTO_DEFAULT); - } - (workers, mrf_workers) - } - }; - - let (final_workers, final_mrf_workers) = if let Some(max_w) = max_workers { + if let Some(max_w) = max_workers { *self.max_workers.write().await = max_w; - (std::cmp::min(workers, max_w), std::cmp::min(mrf_workers, max_w)) - } else { - (workers, mrf_workers) - }; + } let max_l_workers_val = max_l_workers.unwrap_or(LARGE_WORKER_COUNT); *self.max_l_workers.write().await = max_l_workers_val; *self.priority.write().await = pri; - self.resize_workers(final_workers, 0).await; - self.resize_failed_workers(final_mrf_workers as i32).await; + self.resize_workers(worker_counts.workers, 0).await; + self.resize_failed_workers(worker_counts.mrf_workers_i32()).await; self.resize_lrg_workers(max_l_workers_val, 0).await; } @@ -456,7 +403,7 @@ impl ReplicationPool { /// 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 - if ri.size >= MIN_LARGE_OBJ_SIZE { + if should_queue_large_object(ri.size) { use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; @@ -490,8 +437,8 @@ impl ReplicationPool { // Try to add more workers if possible let max_l_workers = *self.max_l_workers.read().await; let existing = lrg_workers.len(); - if self.active_lrg_workers() < std::cmp::min(max_l_workers, LARGE_WORKER_COUNT) as i32 { - let workers = std::cmp::min(existing + 1, max_l_workers); + if should_grow_large_workers(self.active_lrg_workers(), max_l_workers) { + let workers = next_large_worker_count(existing, max_l_workers); drop(lrg_workers); self.resize_lrg_workers(workers, existing).await; @@ -562,25 +509,16 @@ impl ReplicationPool { ); } ReplicationPriority::Auto => { - let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT); - let active_workers = self.active_workers(); - - if active_workers < max_w as i32 { - let workers = self.workers.read().await; - let new_count = std::cmp::min(workers.len() + 1, max_w); + 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 max_mrf_workers = std::cmp::min(max_workers, MRF_WORKER_MAX_LIMIT); - let active_mrf = self.active_mrf_workers(); - - if active_mrf < max_mrf_workers as i32 { - let current_mrf = self.mrf_worker_size.load(Ordering::SeqCst); - let new_mrf = std::cmp::min(current_mrf + 1, max_mrf_workers as i32); - + 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; } } @@ -645,10 +583,8 @@ impl ReplicationPool { ); } ReplicationPriority::Auto => { - let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT); - if self.active_workers() < max_w as i32 { - let workers = self.workers.read().await; - let new_count = std::cmp::min(workers.len() + 1, max_w); + 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; diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 8869ba3bb..dc77f4920 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -19,6 +19,7 @@ pub mod operation; pub mod queue; pub mod resync; pub mod rule; +pub mod runtime; pub mod stats; pub mod tagging; @@ -32,6 +33,12 @@ pub use resync::{ decode_resync_file, encode_resync_file, }; 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, + next_large_worker_count, next_mrf_worker_count, next_regular_worker_count, resized_worker_counts, should_grow_large_workers, + should_queue_large_object, worker_counts_for_priority, +}; pub use rustfs_filemeta::{ REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, diff --git a/crates/replication/src/runtime.rs b/crates/replication/src/runtime.rs new file mode 100644 index 000000000..898e362b1 --- /dev/null +++ b/crates/replication/src/runtime.rs @@ -0,0 +1,223 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::ReplicationPriority; + +pub const WORKER_MAX_LIMIT: usize = 500; +pub const WORKER_MIN_LIMIT: usize = 50; +pub const WORKER_AUTO_DEFAULT: usize = 100; +pub const MRF_WORKER_MAX_LIMIT: usize = 8; +pub const MRF_WORKER_MIN_LIMIT: usize = 2; +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)] +pub struct ReplicationPoolOpts { + pub priority: ReplicationPriority, + pub max_workers: Option, + pub max_l_workers: Option, +} + +impl Default for ReplicationPoolOpts { + fn default() -> Self { + Self { + priority: ReplicationPriority::Auto, + max_workers: None, + max_l_workers: None, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicationWorkerCounts { + pub workers: usize, + pub mrf_workers: usize, +} + +impl ReplicationWorkerCounts { + pub const fn new(workers: usize, mrf_workers: usize) -> Self { + Self { workers, mrf_workers } + } + + pub fn capped_by_worker_limit(self, max_workers: usize) -> Self { + Self { + workers: self.workers.min(max_workers), + mrf_workers: self.mrf_workers.min(max_workers), + } + } + + pub fn mrf_workers_i32(self) -> i32 { + usize_to_i32_saturating(self.mrf_workers) + } +} + +pub fn initial_worker_counts(opts: &ReplicationPoolOpts) -> ReplicationWorkerCounts { + worker_counts_for_priority(&opts.priority, WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT) + .capped_by_worker_limit(opts.max_workers.unwrap_or(WORKER_MAX_LIMIT)) +} + +pub fn resized_worker_counts( + priority: &ReplicationPriority, + max_workers: Option, + current_workers: usize, + current_mrf_workers: usize, +) -> ReplicationWorkerCounts { + let counts = worker_counts_for_priority(priority, current_workers, current_mrf_workers); + if let Some(max_workers) = max_workers { + counts.capped_by_worker_limit(max_workers) + } else { + counts + } +} + +pub fn worker_counts_for_priority( + priority: &ReplicationPriority, + current_workers: usize, + current_mrf_workers: usize, +) -> ReplicationWorkerCounts { + match priority { + ReplicationPriority::Fast => ReplicationWorkerCounts::new(WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT), + ReplicationPriority::Slow => ReplicationWorkerCounts::new(WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT), + ReplicationPriority::Auto => ReplicationWorkerCounts::new( + auto_worker_count(current_workers, WORKER_AUTO_DEFAULT), + auto_worker_count(current_mrf_workers, MRF_WORKER_AUTO_DEFAULT), + ), + } +} + +pub fn should_queue_large_object(size: i64) -> bool { + size >= MIN_LARGE_OBJ_SIZE +} + +pub fn should_grow_large_workers(active_lrg_workers: i32, max_l_workers: usize) -> bool { + active_lrg_workers < usize_to_i32_saturating(max_l_workers.min(LARGE_WORKER_COUNT)) +} + +pub fn next_large_worker_count(existing_workers: usize, max_l_workers: usize) -> usize { + existing_workers.saturating_add(1).min(max_l_workers) +} + +pub fn next_regular_worker_count(current_workers: usize, active_workers: i32, max_workers: usize) -> Option { + let max_workers = max_workers.min(WORKER_MAX_LIMIT); + if active_workers < usize_to_i32_saturating(max_workers) { + Some(current_workers.saturating_add(1).min(max_workers)) + } else { + None + } +} + +pub fn next_mrf_worker_count(current_mrf_workers: i32, active_mrf_workers: i32, max_workers: usize) -> Option { + let max_mrf_workers = max_workers.min(MRF_WORKER_MAX_LIMIT); + let max_mrf_workers = usize_to_i32_saturating(max_mrf_workers); + if active_mrf_workers < max_mrf_workers { + Some(current_mrf_workers.saturating_add(1).min(max_mrf_workers)) + } else { + None + } +} + +fn auto_worker_count(current_workers: usize, auto_default: usize) -> usize { + if current_workers < auto_default { + current_workers.saturating_add(1).min(auto_default) + } else { + auto_default + } +} + +fn usize_to_i32_saturating(value: usize) -> i32 { + match i32::try_from(value) { + Ok(value) => value, + Err(_) => i32::MAX, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn initial_worker_counts_preserve_priority_defaults() { + let fast = ReplicationPoolOpts { + priority: ReplicationPriority::Fast, + ..Default::default() + }; + assert_eq!( + initial_worker_counts(&fast), + ReplicationWorkerCounts::new(WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT) + ); + + let slow = ReplicationPoolOpts { + priority: ReplicationPriority::Slow, + ..Default::default() + }; + assert_eq!( + initial_worker_counts(&slow), + ReplicationWorkerCounts::new(WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT) + ); + + let auto = ReplicationPoolOpts::default(); + assert_eq!( + initial_worker_counts(&auto), + ReplicationWorkerCounts::new(WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT) + ); + } + + #[test] + fn worker_counts_respect_configured_max_workers() { + let opts = ReplicationPoolOpts { + priority: ReplicationPriority::Fast, + max_workers: Some(3), + max_l_workers: None, + }; + + assert_eq!(initial_worker_counts(&opts), ReplicationWorkerCounts::new(3, 3)); + assert_eq!( + resized_worker_counts(&ReplicationPriority::Fast, Some(5), WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT), + ReplicationWorkerCounts::new(5, 5) + ); + } + + #[test] + fn auto_priority_grows_toward_defaults() { + assert_eq!( + worker_counts_for_priority(&ReplicationPriority::Auto, 0, 0), + ReplicationWorkerCounts::new(1, 1) + ); + assert_eq!( + worker_counts_for_priority(&ReplicationPriority::Auto, WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT), + ReplicationWorkerCounts::new(WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT) + ); + } + + #[test] + fn backpressure_growth_caps_at_limits() { + assert_eq!(next_regular_worker_count(4, 4, WORKER_MAX_LIMIT), Some(5)); + assert_eq!( + next_regular_worker_count(WORKER_MAX_LIMIT, usize_to_i32_saturating(WORKER_MAX_LIMIT), WORKER_MAX_LIMIT), + None + ); + + assert_eq!(next_mrf_worker_count(2, 2, WORKER_MAX_LIMIT), Some(3)); + assert_eq!( + next_mrf_worker_count(8, usize_to_i32_saturating(MRF_WORKER_MAX_LIMIT), WORKER_MAX_LIMIT), + None + ); + + assert!(should_queue_large_object(MIN_LARGE_OBJ_SIZE)); + assert!(!should_queue_large_object(MIN_LARGE_OBJ_SIZE - 1)); + assert!(should_grow_large_workers(1, LARGE_WORKER_COUNT)); + assert_eq!(next_large_worker_count(LARGE_WORKER_COUNT, LARGE_WORKER_COUNT), LARGE_WORKER_COUNT); + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 71416db94..911501ccd 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -204,6 +204,7 @@ REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_d REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_operation_contract_backslide_hits.txt" REPLICATION_QUEUE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_queue_contract_backslide_hits.txt" REPLICATION_STATS_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_stats_contract_backslide_hits.txt" +REPLICATION_RUNTIME_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_runtime_contract_backslide_hits.txt" REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_resync_contract_backslide_hits.txt" REPLICATION_MRF_WIRE_FORMAT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_mrf_wire_format_backslide_hits.txt" STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/storage_replication_handle_boundary_bypass_hits.txt" @@ -2623,6 +2624,21 @@ if [[ -s "$REPLICATION_STATS_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then report_failure "replication stats contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_STATS_CONTRACT_BACKSLIDE_HITS_FILE")" fi +( + cd "$ROOT_DIR" + replication_runtime_status=0 + rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:const\s+(?:WORKER_MAX_LIMIT|WORKER_MIN_LIMIT|WORKER_AUTO_DEFAULT|MRF_WORKER_MAX_LIMIT|MRF_WORKER_MIN_LIMIT|MRF_WORKER_AUTO_DEFAULT|LARGE_WORKER_COUNT|MIN_LARGE_OBJ_SIZE)\b)|(?:struct\s+(?:ReplicationPoolOpts|ReplicationWorkerCounts)\b)|(?:fn\s+(?:initial_worker_counts|resized_worker_counts|worker_counts_for_priority|should_queue_large_object|should_grow_large_workers|next_large_worker_count|next_regular_worker_count|next_mrf_worker_count)\b))' \ + crates/ecstore/src/bucket/replication \ + --glob '*.rs' >"$REPLICATION_RUNTIME_CONTRACT_BACKSLIDE_HITS_FILE" || replication_runtime_status=$? + if [[ "$replication_runtime_status" -ne 0 && "$replication_runtime_status" -ne 1 ]]; then + exit "$replication_runtime_status" + fi +) + +if [[ -s "$REPLICATION_RUNTIME_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then + report_failure "replication runtime sizing contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_RUNTIME_CONTRACT_BACKSLIDE_HITS_FILE")" +fi + ( cd "$ROOT_DIR" rg -n --with-filename 'pub\s+(struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \