From 3172f875aeb21723b2308104f5a5c56f5b6ff463 Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Wed, 30 Apr 2025 14:41:16 +0200 Subject: [PATCH] pass config --- src/model/garage.rs | 39 +++++++++++++++++++++++++++------ src/model/index_counter.rs | 3 +++ src/table/data.rs | 25 ++++++++++++++------- src/table/merkle.rs | 27 +++++++++++++---------- src/table/table.rs | 11 ++++++++-- src/util/config.rs | 45 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 122 insertions(+), 28 deletions(-) diff --git a/src/model/garage.rs b/src/model/garage.rs index 11c0d90f..10a36ac1 100644 --- a/src/model/garage.rs +++ b/src/model/garage.rs @@ -175,7 +175,13 @@ impl Garage { // ---- admin tables ---- info!("Initialize bucket_table..."); - let bucket_table = Table::new(BucketTable, control_rep_param.clone(), system.clone(), &db); + let bucket_table = Table::new( + BucketTable, + control_rep_param.clone(), + system.clone(), + &db, + &config.experimental.merkle_backpressure, + ); info!("Initialize bucket_alias_table..."); let bucket_alias_table = Table::new( @@ -183,9 +189,16 @@ impl Garage { control_rep_param.clone(), system.clone(), &db, + &config.experimental.merkle_backpressure, ); info!("Initialize key_table_table..."); - let key_table = Table::new(KeyTable, control_rep_param, system.clone(), &db); + let key_table = Table::new( + KeyTable, + control_rep_param, + system.clone(), + &db, + &config.experimental.merkle_backpressure, + ); // ---- S3 tables ---- info!("Initialize block_ref_table..."); @@ -196,6 +209,7 @@ impl Garage { meta_rep_param.clone(), system.clone(), &db, + &config.experimental.merkle_backpressure, ); info!("Initialize version_table..."); @@ -206,10 +220,12 @@ impl Garage { meta_rep_param.clone(), system.clone(), &db, + &config.experimental.merkle_backpressure, ); info!("Initialize multipart upload counter table..."); - let mpu_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db); + let mpu_counter_table = + IndexCounter::new(system.clone(), meta_rep_param.clone(), &db, &config); info!("Initialize multipart upload table..."); let mpu_table = Table::new( @@ -220,10 +236,12 @@ impl Garage { meta_rep_param.clone(), system.clone(), &db, + &config.experimental.merkle_backpressure, ); info!("Initialize object counter table..."); - let object_counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), &db); + let object_counter_table = + IndexCounter::new(system.clone(), meta_rep_param.clone(), &db, &config); info!("Initialize object_table..."); #[allow(clippy::redundant_clone)] @@ -236,6 +254,7 @@ impl Garage { meta_rep_param.clone(), system.clone(), &db, + &config.experimental.merkle_backpressure, ); info!("Load lifecycle worker state..."); @@ -245,7 +264,7 @@ impl Garage { // ---- K2V ---- #[cfg(feature = "k2v")] - let k2v = GarageK2V::new(system.clone(), &db, meta_rep_param); + let k2v = GarageK2V::new(system.clone(), &db, meta_rep_param, &config); // ---- setup block refcount recalculation ---- // this function can be used to fix inconsistencies in the RC table @@ -335,9 +354,14 @@ impl Garage { #[cfg(feature = "k2v")] impl GarageK2V { - fn new(system: Arc, db: &db::Db, meta_rep_param: TableShardedReplication) -> Self { + fn new( + system: Arc, + db: &db::Db, + meta_rep_param: TableShardedReplication, + config: &Config, + ) -> Self { info!("Initialize K2V counter table..."); - let counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), db); + let counter_table = IndexCounter::new(system.clone(), meta_rep_param.clone(), db, config); info!("Initialize K2V subscription manager..."); let subscriptions = Arc::new(SubscriptionManager::new()); @@ -351,6 +375,7 @@ impl GarageK2V { meta_rep_param, system.clone(), db, + &config.experimental.merkle_backpressure, ); info!("Initialize K2V RPC handler..."); diff --git a/src/model/index_counter.rs b/src/model/index_counter.rs index aa13ee7b..e4f79753 100644 --- a/src/model/index_counter.rs +++ b/src/model/index_counter.rs @@ -10,6 +10,7 @@ use garage_db as db; use garage_rpc::layout::LayoutHelper; use garage_rpc::system::System; use garage_util::background::BackgroundRunner; +use garage_util::config::Config; use garage_util::data::*; use garage_util::error::*; use garage_util::migrate::Migrate; @@ -173,6 +174,7 @@ impl IndexCounter { system: Arc, replication: TableShardedReplication, db: &db::Db, + config: &Config, ) -> Arc { Arc::new(Self { this_node: system.id, @@ -186,6 +188,7 @@ impl IndexCounter { replication, system, db, + &config.experimental.merkle_backpressure, ), }) } diff --git a/src/table/data.rs b/src/table/data.rs index 47f67e94..46c546f7 100644 --- a/src/table/data.rs +++ b/src/table/data.rs @@ -8,6 +8,7 @@ use tokio::sync::Notify; use garage_db as db; +use garage_util::config::MerkleBackpressureEnum; use garage_util::data::*; use garage_util::error::*; use garage_util::migrate::Migrate; @@ -21,11 +22,6 @@ use crate::replication::*; use crate::schema::*; use crate::util::*; -pub(crate) const MERKLE_SLEEP_INITIAL: Duration = Duration::from_micros(1); -pub(crate) const MERKLE_SLEEP_MAX: Duration = Duration::from_secs(30); -pub(crate) const MERKLE_SLEEP_ADD_DECREASE: Duration = Duration::from_micros(1); -pub(crate) const MERKLE_SLEEP_MULT_INCREASE: f32 = 1.1; - pub struct TableData { system: Arc, @@ -37,7 +33,6 @@ pub struct TableData { pub(crate) merkle_tree: db::Tree, pub(crate) merkle_todo: db::Tree, pub(crate) merkle_todo_notify: Notify, - // @FIXME: replace with a tokio::sync::watch ---V pub(crate) merkle_todo_sleep: Arc>, pub(crate) insert_queue: db::Tree, @@ -46,10 +41,18 @@ pub struct TableData { pub(crate) gc_todo: db::Tree, pub(crate) metrics: TableMetrics, + + pub(crate) config: MerkleBackpressureEnum, } impl TableData { - pub fn new(system: Arc, instance: F, replication: R, db: &db::Db) -> Arc { + pub fn new( + system: Arc, + instance: F, + replication: R, + db: &db::Db, + config: &MerkleBackpressureEnum, + ) -> Arc { let store = db .open_tree(format!("{}:table", F::TABLE_NAME)) .expect("Unable to open DB tree"); @@ -60,7 +63,12 @@ impl TableData { let merkle_todo = db .open_tree(format!("{}:merkle_todo", F::TABLE_NAME)) .expect("Unable to open DB Merkle TODO tree"); - let merkle_todo_sleep = Arc::new(Mutex::new(MERKLE_SLEEP_INITIAL)); + + let initial = match config { + MerkleBackpressureEnum::None => Duration::ZERO, + MerkleBackpressureEnum::Aimd(aimd) => Duration::from_micros(aimd.initial_us), + }; + let merkle_todo_sleep = Arc::new(Mutex::new(initial)); let insert_queue = db .open_tree(format!("{}:insert_queue", F::TABLE_NAME)) @@ -92,6 +100,7 @@ impl TableData { insert_queue_notify: Arc::new(Notify::new()), gc_todo, metrics, + config: config.clone(), }) } diff --git a/src/table/merkle.rs b/src/table/merkle.rs index a1c9a9f1..341dfd4f 100644 --- a/src/table/merkle.rs +++ b/src/table/merkle.rs @@ -12,6 +12,7 @@ use tokio::sync::watch; use garage_db as db; use garage_util::background::*; +use garage_util::config::{MerkleBackpressureAimd, MerkleBackpressureEnum}; use garage_util::data::*; use garage_util::encode::{nonversioned_decode, nonversioned_encode}; use garage_util::error::Error; @@ -74,6 +75,11 @@ impl MerkleUpdater { pub(crate) fn new(data: Arc>) -> Arc { let empty_node_hash = blake2sum(&nonversioned_encode(&MerkleNode::Empty).unwrap()[..]); + match &data.config { + MerkleBackpressureEnum::None => info!("Merkle Backpressure is not activated"), + MerkleBackpressureEnum::Aimd(v) => info!("Merkle backpressure is activated (initial={}us, max={}us, underload={}us, overload={}x)", v.initial_us, v.max_us, v.underload_us, v.overload_mult), + } + Arc::new(Self { data, empty_node_hash, @@ -88,14 +94,17 @@ impl MerkleUpdater { fn updater_loop_iter(&self) -> Result { if let Some((key, valhash)) = self.data.merkle_todo.first()? { self.update_item(&key, &valhash)?; - self.adapt_backpressure()?; + match &self.data.config { + MerkleBackpressureEnum::None => (), + MerkleBackpressureEnum::Aimd(a) => self.adapt_aimd_backpressure(a)?, + }; Ok(WorkerState::Busy) } else { Ok(WorkerState::Idle) } } - fn adapt_backpressure(&self) -> Result<(), Error> { + fn adapt_aimd_backpressure(&self, config: &MerkleBackpressureAimd) -> Result<(), Error> { // Capture evolution of the merkle todo length let current_merkle_len = self.data.merkle_todo.len()?; @@ -103,19 +112,15 @@ impl MerkleUpdater { { let a = self.data.merkle_todo_sleep.clone(); let mut v = a.lock().unwrap(); - if current_merkle_len == 0 { - // @FIXME not sure if it's correct - // If we have nothing in queue, we reset the backpressure - *v = MERKLE_SLEEP_INITIAL; - } else if current_merkle_len < self.previous_merkle_len.load(Ordering::Relaxed) { + if current_merkle_len < self.previous_merkle_len.load(Ordering::Relaxed) { // If we decrease the queue size, we can decrease the sleep time - *v = v.saturating_sub(MERKLE_SLEEP_ADD_DECREASE); + *v = v.saturating_sub(Duration::from_micros(config.underload_us)); } else { // If we are late, we increase the queue size - *v = v.mul_f32(MERKLE_SLEEP_MULT_INCREASE); + *v = v.mul_f64(config.overload_mult); } - *v = v.min(MERKLE_SLEEP_MAX); - *v = v.max(MERKLE_SLEEP_INITIAL); + *v = v.min(Duration::from_micros(config.initial_us)); + *v = v.max(Duration::from_micros(config.max_us)); } self.previous_merkle_len diff --git a/src/table/table.rs b/src/table/table.rs index 4b807dc7..3a17c191 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -15,6 +15,7 @@ use opentelemetry::{ use garage_db as db; use garage_util::background::BackgroundRunner; +use garage_util::config::MerkleBackpressureEnum; use garage_util::data::*; use garage_util::error::Error; use garage_util::metrics::RecordDuration; @@ -69,12 +70,18 @@ impl Rpc for TableRpc { impl Table { // =============== PUBLIC INTERFACE FUNCTIONS (new, insert, get, etc) =============== - pub fn new(instance: F, replication: R, system: Arc, db: &db::Db) -> Arc { + pub fn new( + instance: F, + replication: R, + system: Arc, + db: &db::Db, + config: &MerkleBackpressureEnum, + ) -> Arc { let endpoint = system .netapp .endpoint(format!("garage_table/table.rs/Rpc:{}", F::TABLE_NAME)); - let data = TableData::new(system.clone(), instance, replication, db); + let data = TableData::new(system.clone(), instance, replication, db, config); let merkle_updater = MerkleUpdater::new(data.clone()); diff --git a/src/util/config.rs b/src/util/config.rs index 73fc4ff4..4a9cf131 100644 --- a/src/util/config.rs +++ b/src/util/config.rs @@ -135,6 +135,10 @@ pub struct Config { /// Configuration for the admin API endpoint #[serde(default = "Default::default")] pub admin: AdminConfig, + + /// --- Experimental + #[serde(default = "Default::default")] + pub experimental: ExperimentalConfig, } /// Value for data_dir: either a single directory or a list of dirs with attributes @@ -255,6 +259,31 @@ pub struct KubernetesDiscoveryConfig { pub skip_crd: bool, } +#[derive(Deserialize, Debug, Clone, Default)] +pub struct ExperimentalConfig { + pub merkle_backpressure: MerkleBackpressureEnum, +} + +#[derive(Deserialize, Debug, Clone, Default)] +#[serde(rename_all = "lowercase", tag = "kind")] +pub enum MerkleBackpressureEnum { + #[default] + None, + Aimd(MerkleBackpressureAimd), +} + +#[derive(Deserialize, Debug, Clone, Default)] +pub struct MerkleBackpressureAimd { + #[serde(default = "default_initial_us")] + pub initial_us: u64, + #[serde(default = "default_max_us")] + pub max_us: u64, + #[serde(default = "default_underload_us")] + pub underload_us: u64, + #[serde(default = "default_overload_mult")] + pub overload_mult: f64, +} + /// Read and parse configuration pub fn read_config(config_file: PathBuf) -> Result { let config = std::fs::read_to_string(config_file)?; @@ -281,6 +310,22 @@ fn default_compression() -> Option { Some(1) } +fn default_initial_us() -> u64 { + 10 +} + +fn default_max_us() -> u64 { + 30 * 1000 * 1000 +} + +fn default_underload_us() -> u64 { + 10 +} + +fn default_overload_mult() -> f64 { + 1.01 +} + fn deserialize_compression<'de, D>(deserializer: D) -> Result, D::Error> where D: de::Deserializer<'de>,