pass config

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<System>, db: &db::Db, meta_rep_param: TableShardedReplication) -> Self {
+	fn new(
+		system: Arc<System>,
+		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...");

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<T: CountedItem> IndexCounter<T> {
 		system: Arc<System>,
 		replication: TableShardedReplication,
 		db: &db::Db,
+		config: &Config,
 	) -> Arc<Self> {
 		Arc::new(Self {
 			this_node: system.id,
@@ -186,6 +188,7 @@ impl<T: CountedItem> IndexCounter<T> {
 				replication,
 				system,
 				db,
+				&config.experimental.merkle_backpressure,
 			),
 		})
 	}

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<F: TableSchema, R: TableReplication> {
 	system: Arc<System>,
 
@@ -37,7 +33,6 @@ pub struct TableData<F: TableSchema, R: TableReplication> {
 	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<Mutex<Duration>>,
 
 	pub(crate) insert_queue: db::Tree,
@@ -46,10 +41,18 @@ pub struct TableData<F: TableSchema, R: TableReplication> {
 	pub(crate) gc_todo: db::Tree,
 
 	pub(crate) metrics: TableMetrics,
+
+	pub(crate) config: MerkleBackpressureEnum,
 }
 
 impl<F: TableSchema, R: TableReplication> TableData<F, R> {
-	pub fn new(system: Arc<System>, instance: F, replication: R, db: &db::Db) -> Arc<Self> {
+	pub fn new(
+		system: Arc<System>,
+		instance: F,
+		replication: R,
+		db: &db::Db,
+		config: &MerkleBackpressureEnum,
+	) -> Arc<Self> {
 		let store = db
 			.open_tree(format!("{}:table", F::TABLE_NAME))
 			.expect("Unable to open DB tree");
@@ -60,7 +63,12 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 		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<F: TableSchema, R: TableReplication> TableData<F, R> {
 			insert_queue_notify: Arc::new(Notify::new()),
 			gc_todo,
 			metrics,
+			config: config.clone(),
 		})
 	}
 

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<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
 	pub(crate) fn new(data: Arc<TableData<F, R>>) -> Arc<Self> {
 		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<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
 	fn updater_loop_iter(&self) -> Result<WorkerState, Error> {
 		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<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
 		{
 			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

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<F: TableSchema> Rpc for TableRpc<F> {
 impl<F: TableSchema, R: TableReplication> Table<F, R> {
 	// =============== PUBLIC INTERFACE FUNCTIONS (new, insert, get, etc) ===============
 
-	pub fn new(instance: F, replication: R, system: Arc<System>, db: &db::Db) -> Arc<Self> {
+	pub fn new(
+		instance: F,
+		replication: R,
+		system: Arc<System>,
+		db: &db::Db,
+		config: &MerkleBackpressureEnum,
+	) -> Arc<Self> {
 		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());
 

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<Config, Error> {
 	let config = std::fs::read_to_string(config_file)?;
@@ -281,6 +310,22 @@ fn default_compression() -> Option<i32> {
 	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<Option<i32>, D::Error>
 where
 	D: de::Deserializer<'de>,