diff --git a/src/table/data.rs b/src/table/data.rs index 2af72253..0b65c46b 100644 --- a/src/table/data.rs +++ b/src/table/data.rs @@ -266,7 +266,13 @@ impl TableData { // early return if nothing changed let (new_entry, new_bytes_hash) = match changed { Some((e, b)) => (e, b), - None => return Ok(None), + None => { + let maybe_bound = self.merkle_todo_bounded_queue.clone(); + if let Some(b) = &maybe_bound { + b.add_permits(1); + } + return Ok(None); + } }; // Handle GC in case of tombstone @@ -313,6 +319,10 @@ impl TableData { })?; if !removed { + let maybe_bound = self.merkle_todo_bounded_queue.clone(); + if let Some(b) = &maybe_bound { + b.add_permits(1); + } return Ok(false); } @@ -343,6 +353,10 @@ impl TableData { })?; if !removed { + let maybe_bound = self.merkle_todo_bounded_queue.clone(); + if let Some(b) = &maybe_bound { + b.add_permits(1); + } return Ok(false); } diff --git a/src/table/merkle.rs b/src/table/merkle.rs index 35233ed4..f1241ed0 100644 --- a/src/table/merkle.rs +++ b/src/table/merkle.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::time::{Duration, SystemTime}; +use std::time::Duration; use async_trait::async_trait; use serde::{Deserialize, Serialize}; @@ -9,7 +9,7 @@ use tokio::sync::watch; use garage_db as db; use garage_util::background::*; -use garage_util::config::{MerkleBackpressureEnum, MerkleFixedQueue}; +use garage_util::config::MerkleBackpressureEnum; use garage_util::data::*; use garage_util::encode::{nonversioned_decode, nonversioned_encode}; use garage_util::error::Error; @@ -135,6 +135,11 @@ impl MerkleUpdater { k ); } + + let maybe_bound = self.data.merkle_todo_bounded_queue.clone(); + if let Some(b) = &maybe_bound { + b.add_permits(1); + } Ok(()) } diff --git a/src/table/metrics.rs b/src/table/metrics.rs index c9e6d048..f5c029d9 100644 --- a/src/table/metrics.rs +++ b/src/table/metrics.rs @@ -89,7 +89,7 @@ impl TableMetrics { observer.observe(free, &[KeyValue::new("table_name", table_name)]) } ) - .with_description("Merkle tree updater TODO sleep backpressure to apply in ms") + .with_description("Merkle TODO queue free slots") .init(), _gc_todo_len: meter .u64_value_observer( diff --git a/src/table/sync.rs b/src/table/sync.rs index 0db440ba..2f83aca2 100644 --- a/src/table/sync.rs +++ b/src/table/sync.rs @@ -11,7 +11,6 @@ use serde::{Deserialize, Serialize}; use serde_bytes::ByteBuf; use tokio::select; use tokio::sync::{mpsc, watch, Notify}; -use tokio::time::sleep; use garage_util::background::*; use garage_util::data::*;