From 7fddf0af9ce3c148ecb63949f711207b2174f1c8 Mon Sep 17 00:00:00 2001
From: Quentin Dufour <quentin@deuxfleurs.fr>
Date: Tue, 29 Apr 2025 10:50:47 +0200
Subject: [PATCH] first implementation

---
 src/table/data.rs   | 22 ++++++++++++++++------
 src/table/merkle.rs | 27 ++++++++++++++++++++++++++-
 2 files changed, 42 insertions(+), 7 deletions(-)

diff --git a/src/table/data.rs b/src/table/data.rs
index 1e0cc2ef..0cd5a05a 100644
--- a/src/table/data.rs
+++ b/src/table/data.rs
@@ -1,6 +1,6 @@
 use core::borrow::Borrow;
 use std::convert::TryInto;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use serde_bytes::ByteBuf;
@@ -21,6 +21,11 @@ use crate::replication::*;
 use crate::schema::*;
 use crate::util::*;
 
+pub(crate) const MERKLE_SLEEP_INITIAL: Duration = Duration::from_micros(100);
+pub(crate) const MERKLE_SLEEP_MAX: Duration = Duration::from_secs(30);
+pub(crate) const MERKLE_SLEEP_ADD_DECREASE: Duration = Duration::from_micros(100);
+pub(crate) const MERKLE_SLEEP_MULT_INCREASE: f32 = 1.2;
+
 pub struct TableData<F: TableSchema, R: TableReplication> {
 	system: Arc<System>,
 
@@ -32,7 +37,8 @@ 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,
-	pub(crate) merkle_todo_sleep: Duration,
+	// @FIXME: replace with a tokio::sync::watch ---V
+	pub(crate) merkle_todo_sleep: Arc<Mutex<Duration>>,
 
 	pub(crate) insert_queue: db::Tree,
 	pub(crate) insert_queue_notify: Arc<Notify>,
@@ -54,7 +60,7 @@ 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 = Duration::from_secs(1);
+		let merkle_todo_sleep = Arc::new(Mutex::new(MERKLE_SLEEP_INITIAL));
 
 		let insert_queue = db
 			.open_tree(format!("{}:insert_queue", F::TABLE_NAME))
@@ -281,8 +287,9 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 
 		// Synchronize with the Merkle Worker
 		self.merkle_todo_notify.notify_one(); // Wake-up it
+		let backpressure = self.merkle_todo_sleep.clone().lock().unwrap().clone();
 
-		Ok(Some((new_entry, self.merkle_todo_sleep)))
+		Ok(Some((new_entry, backpressure)))
 	}
 
 	pub(crate) fn delete_if_equal(
@@ -312,7 +319,8 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 
 		self.metrics.internal_delete_counter.add(1);
 		self.merkle_todo_notify.notify_one();
-		Ok((removed, self.merkle_todo_sleep))
+		let backpressure = self.merkle_todo_sleep.clone().lock().unwrap().clone();
+		Ok((removed, backpressure))
 	}
 
 	pub(crate) fn delete_if_equal_hash(
@@ -343,7 +351,9 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 		self.metrics.internal_delete_counter.add(1);
 		self.merkle_todo_notify.notify_one();
 
-		Ok((true, self.merkle_todo_sleep))
+		let bp = self.merkle_todo_sleep.clone().lock().unwrap().clone();
+
+		Ok((true, bp))
 	}
 
 	// ---- Insert queue functions ----
diff --git a/src/table/merkle.rs b/src/table/merkle.rs
index 596d5805..e03b733a 100644
--- a/src/table/merkle.rs
+++ b/src/table/merkle.rs
@@ -82,13 +82,38 @@ 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(|| self.update_item(&key, &valhash))?;
 			Ok(WorkerState::Busy)
 		} else {
 			Ok(WorkerState::Idle)
 		}
 	}
 
+	fn adapt_backpressure<FX>(&self, func: FX) -> Result<(), Error>
+	where
+		FX: FnOnce() -> Result<(), Error>,
+	{
+		// Capture evolution of the merkle todo length
+		let qlen_before = self.data.merkle_todo.len()?;
+		let ret = func()?;
+		let qlen_after = self.data.merkle_todo.len()?;
+
+		// Algorithm inspired by Additive Increase Multiplicative Decrease (AIMD)
+		{
+			let a = self.data.merkle_todo_sleep.clone();
+			let mut v = a.lock().unwrap();
+			if qlen_after > qlen_before {
+				*v = v.mul_f32(MERKLE_SLEEP_MULT_INCREASE);
+			} else {
+				*v = v.saturating_sub(MERKLE_SLEEP_ADD_DECREASE);
+			}
+			*v = v.min(MERKLE_SLEEP_MAX);
+			*v = v.max(MERKLE_SLEEP_INITIAL);
+		}
+
+		Ok(ret)
+	}
+
 	fn update_item(&self, k: &[u8], vhash_by: &[u8]) -> Result<(), Error> {
 		let khash = blake2sum(k);