From f34558af077d532f61bdd88218bca1fb07324781 Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Thu, 1 May 2025 08:56:57 +0200 Subject: [PATCH] add a rpc in-flight limiter --- src/block/manager.rs | 2 + src/garage/admin/mod.rs | 13 ++++- src/garage/cli/cmd.rs | 20 ++++++- src/garage/cli/layout.rs | 16 ++++- src/net/client.rs | 16 ++++- src/net/endpoint.rs | 19 +++++- src/net/netapp.rs | 1 + src/net/peering.rs | 9 ++- src/rpc/rpc_helper.rs | 13 ++++- src/table/merkle.rs | 123 +++++++++++++++++++++------------------ src/table/table.rs | 3 +- src/util/config.rs | 8 ++- 12 files changed, 173 insertions(+), 70 deletions(-) diff --git a/src/block/manager.rs b/src/block/manager.rs index 41b2f02a..18f9e198 100644 --- a/src/block/manager.rs +++ b/src/block/manager.rs @@ -17,6 +17,7 @@ use opentelemetry::{ Context, }; +use garage_net::endpoint::RpcInFlightLimiter; use garage_net::stream::{read_stream_to_end, stream_asyncread, ByteStream}; use garage_db as db; @@ -295,6 +296,7 @@ impl BlockManager { &node_id, BlockRpc::GetBlock(*hash, order_tag), priority, + RpcInFlightLimiter::TableWrite, ); tokio::select! { res = rpc => { diff --git a/src/garage/admin/mod.rs b/src/garage/admin/mod.rs index 3bbc2b86..321326e0 100644 --- a/src/garage/admin/mod.rs +++ b/src/garage/admin/mod.rs @@ -13,6 +13,8 @@ use serde::{Deserialize, Serialize}; use format_table::format_table_to_string; +use garage_net::endpoint::RpcInFlightLimiter; + use garage_util::background::BackgroundRunner; use garage_util::data::*; use garage_util::error::Error as GarageError; @@ -118,6 +120,7 @@ impl AdminRpcHandler { &node, AdminRpc::LaunchRepair(opt_to_send.clone()), PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, ) .await; if !matches!(resp, Ok(Ok(_))) { @@ -164,7 +167,12 @@ impl AdminRpcHandler { let node_id = (*node).into(); match self .endpoint - .call(&node_id, AdminRpc::Stats(opt), PRIO_NORMAL) + .call( + &node_id, + AdminRpc::Stats(opt), + PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, + ) .await { Ok(Ok(AdminRpc::Ok(s))) => writeln!(&mut ret, "{}", s).unwrap(), @@ -407,6 +415,7 @@ impl AdminRpcHandler { variable: variable.clone(), }), PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, ) .await?? { @@ -456,6 +465,7 @@ impl AdminRpcHandler { value: value.to_string(), }), PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, ) .await?? { @@ -488,6 +498,7 @@ impl AdminRpcHandler { &to, AdminRpc::MetaOperation(MetaOperation::Snapshot { all: false }), PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, ) .await? })) diff --git a/src/garage/cli/cmd.rs b/src/garage/cli/cmd.rs index 44d3d96c..22449b41 100644 --- a/src/garage/cli/cmd.rs +++ b/src/garage/cli/cmd.rs @@ -2,6 +2,7 @@ use std::collections::{HashMap, HashSet}; use std::time::Duration; use format_table::format_table; +use garage_net::endpoint::RpcInFlightLimiter; use garage_util::error::*; use garage_rpc::layout::*; @@ -200,7 +201,12 @@ pub async fn cmd_connect( args: ConnectNodeOpt, ) -> Result<(), Error> { match rpc_cli - .call(&rpc_host, SystemRpc::Connect(args.node), PRIO_NORMAL) + .call( + &rpc_host, + SystemRpc::Connect(args.node), + PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, + ) .await?? { SystemRpc::Ok => { @@ -216,7 +222,10 @@ pub async fn cmd_admin( rpc_host: NodeID, args: AdminRpc, ) -> Result<(), HelperError> { - match rpc_cli.call(&rpc_host, args, PRIO_NORMAL).await?? { + match rpc_cli + .call(&rpc_host, args, PRIO_NORMAL, RpcInFlightLimiter::NoLimit) + .await?? + { AdminRpc::Ok(msg) => { println!("{}", msg); } @@ -271,7 +280,12 @@ pub async fn fetch_status( rpc_host: NodeID, ) -> Result, Error> { match rpc_cli - .call(&rpc_host, SystemRpc::GetKnownNodes, PRIO_NORMAL) + .call( + &rpc_host, + SystemRpc::GetKnownNodes, + PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, + ) .await?? { SystemRpc::ReturnKnownNodes(nodes) => Ok(nodes), diff --git a/src/garage/cli/layout.rs b/src/garage/cli/layout.rs index f053eef4..554d7096 100644 --- a/src/garage/cli/layout.rs +++ b/src/garage/cli/layout.rs @@ -1,6 +1,7 @@ use bytesize::ByteSize; use format_table::format_table; +use garage_net::endpoint::RpcInFlightLimiter; use garage_util::crdt::Crdt; use garage_util::error::*; @@ -45,7 +46,12 @@ pub async fn cmd_assign_role( args: AssignRoleOpt, ) -> Result<(), Error> { let status = match rpc_cli - .call(&rpc_host, SystemRpc::GetKnownNodes, PRIO_NORMAL) + .call( + &rpc_host, + SystemRpc::GetKnownNodes, + PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, + ) .await?? { SystemRpc::ReturnKnownNodes(nodes) => nodes, @@ -475,7 +481,12 @@ pub async fn fetch_layout( rpc_host: NodeID, ) -> Result { match rpc_cli - .call(&rpc_host, SystemRpc::PullClusterLayout, PRIO_NORMAL) + .call( + &rpc_host, + SystemRpc::PullClusterLayout, + PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, + ) .await?? { SystemRpc::AdvertiseClusterLayout(t) => Ok(t), @@ -493,6 +504,7 @@ pub async fn send_layout( &rpc_host, SystemRpc::AdvertiseClusterLayout(layout), PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, ) .await??; Ok(()) diff --git a/src/net/client.rs b/src/net/client.rs index 20e1dacd..1b0b00b9 100644 --- a/src/net/client.rs +++ b/src/net/client.rs @@ -14,7 +14,7 @@ use futures::Stream; use kuska_handshake::async_std::{handshake_client, BoxStream}; use tokio::net::TcpStream; use tokio::select; -use tokio::sync::{mpsc, oneshot, watch}; +use tokio::sync::{mpsc, oneshot, watch, Semaphore}; use tokio_util::compat::*; #[cfg(feature = "telemetry")] @@ -25,6 +25,7 @@ use opentelemetry::{ #[cfg(feature = "telemetry")] use opentelemetry_contrib::trace::propagator::binary::*; +use crate::endpoint::RpcInFlightLimiter; use crate::error::*; use crate::message::*; use crate::netapp::*; @@ -41,6 +42,7 @@ pub(crate) struct ClientConn { next_query_number: AtomicU32, inflight: Mutex>>, + rpc_table_write_inflight_limiter: Semaphore, } impl ClientConn { @@ -98,6 +100,7 @@ impl ClientConn { next_query_number: AtomicU32::from(RequestID::default()), query_send: ArcSwapOption::new(Some(Arc::new(query_send))), inflight: Mutex::new(HashMap::new()), + rpc_table_write_inflight_limiter: Semaphore::new(64), }); netapp.connected_as_client(peer_id, conn.clone()); @@ -144,10 +147,20 @@ impl ClientConn { req: Req, path: &str, prio: RequestPriority, + limiter: RpcInFlightLimiter, ) -> Result, Error> where T: Message, { + let _permit = match limiter { + RpcInFlightLimiter::NoLimit => None, + RpcInFlightLimiter::TableWrite => Some( + self.rpc_table_write_inflight_limiter + .acquire() + .await + .unwrap(), + ), + }; let query_send = self.query_send.load_full().ok_or(Error::ConnectionClosed)?; let id = self @@ -212,6 +225,7 @@ impl ClientConn { let stream = Box::pin(canceller.for_stream(stream)); let resp_enc = RespEnc::decode(stream).await?; + drop(_permit); debug!("client: got response to request {} (path {})", id, path); Resp::from_enc(resp_enc) } diff --git a/src/net/endpoint.rs b/src/net/endpoint.rs index d46acc42..a4dee42f 100644 --- a/src/net/endpoint.rs +++ b/src/net/endpoint.rs @@ -57,6 +57,13 @@ where } } +#[derive(Debug, Copy, Clone, Default)] +pub enum RpcInFlightLimiter { + #[default] + NoLimit, + TableWrite, +} + // ---- /// This struct represents an endpoint for message of type `M`. @@ -114,6 +121,7 @@ where target: &NodeID, req: T, prio: RequestPriority, + limiter: RpcInFlightLimiter, ) -> Result, Error> where T: IntoReq, @@ -136,7 +144,10 @@ where "Not connected: {}", hex::encode(&target[..8]) ))), - Some(c) => c.call(req.into_req()?, self.path.as_str(), prio).await, + Some(c) => { + c.call(req.into_req()?, self.path.as_str(), prio, limiter) + .await + } } } } @@ -149,8 +160,12 @@ where target: &NodeID, req: M, prio: RequestPriority, + limiter: RpcInFlightLimiter, ) -> Result<::Response, Error> { - Ok(self.call_streaming(target, req, prio).await?.into_msg()) + Ok(self + .call_streaming(target, req, prio, limiter) + .await? + .into_msg()) } } diff --git a/src/net/netapp.rs b/src/net/netapp.rs index 36c6fc88..2842b4aa 100644 --- a/src/net/netapp.rs +++ b/src/net/netapp.rs @@ -427,6 +427,7 @@ impl NetApp { server_port, }, PRIO_NORMAL, + RpcInFlightLimiter::NoLimit, ) .await .map(|_| ()) diff --git a/src/net/peering.rs b/src/net/peering.rs index 08378a08..bca0e07f 100644 --- a/src/net/peering.rs +++ b/src/net/peering.rs @@ -406,7 +406,7 @@ impl PeeringManager { ping_time ); let ping_response = select! { - r = self.ping_endpoint.call(&id, ping_msg, PRIO_HIGH) => r, + r = self.ping_endpoint.call(&id, ping_msg, PRIO_HIGH, RpcInFlightLimiter::NoLimit) => r, _ = tokio::time::sleep(ping_timeout) => Err(Error::Message("Ping timeout".into())), }; @@ -458,7 +458,12 @@ impl PeeringManager { let pex_message = PeerListMessage { list: peer_list }; match self .peer_list_endpoint - .call(id, pex_message, PRIO_BACKGROUND) + .call( + id, + pex_message, + PRIO_BACKGROUND, + RpcInFlightLimiter::NoLimit, + ) .await { Err(e) => warn!("Error doing peer exchange: {}", e), diff --git a/src/rpc/rpc_helper.rs b/src/rpc/rpc_helper.rs index 2505c2ce..62efe15d 100644 --- a/src/rpc/rpc_helper.rs +++ b/src/rpc/rpc_helper.rs @@ -6,6 +6,7 @@ use std::time::Duration; use futures::future::join_all; use futures::stream::futures_unordered::FuturesUnordered; use futures::stream::StreamExt; +use garage_net::endpoint::RpcInFlightLimiter; use tokio::select; use opentelemetry::KeyValue; @@ -44,6 +45,8 @@ pub struct RequestStrategy { rs_timeout: Timeout, /// Data to drop when everything completes rs_drop_on_complete: T, + /// RPC In Flight Limiter + rs_inflight_limiter: RpcInFlightLimiter, } #[derive(Copy, Clone)] @@ -61,6 +64,7 @@ impl Clone for RequestStrategy<()> { rs_priority: self.rs_priority, rs_timeout: self.rs_timeout, rs_drop_on_complete: (), + rs_inflight_limiter: self.rs_inflight_limiter, } } } @@ -74,6 +78,7 @@ impl RequestStrategy<()> { rs_priority: prio, rs_timeout: Timeout::Default, rs_drop_on_complete: (), + rs_inflight_limiter: RpcInFlightLimiter::NoLimit, } } /// Add an item to be dropped on completion @@ -84,6 +89,7 @@ impl RequestStrategy<()> { rs_priority: self.rs_priority, rs_timeout: self.rs_timeout, rs_drop_on_complete: drop_on_complete, + rs_inflight_limiter: RpcInFlightLimiter::NoLimit, } } } @@ -109,6 +115,10 @@ impl RequestStrategy { self.rs_timeout = Timeout::Custom(timeout); self } + pub fn with_write_limiter(mut self) -> Self { + self.rs_inflight_limiter = RpcInFlightLimiter::TableWrite; + self + } /// Extract drop_on_complete item fn extract_drop_on_complete(self) -> (RequestStrategy<()>, T) { ( @@ -118,6 +128,7 @@ impl RequestStrategy { rs_priority: self.rs_priority, rs_timeout: self.rs_timeout, rs_drop_on_complete: (), + rs_inflight_limiter: self.rs_inflight_limiter, }, self.rs_drop_on_complete, ) @@ -185,7 +196,7 @@ impl RpcHelper { let node_id = to.into(); let rpc_call = endpoint - .call_streaming(&node_id, msg, strat.rs_priority) + .call_streaming(&node_id, msg, strat.rs_priority, strat.rs_inflight_limiter) .with_context(Context::current_with_span(span)) .record_duration(&self.0.metrics.rpc_duration, &metric_tags); diff --git a/src/table/merkle.rs b/src/table/merkle.rs index c8c29cec..bfec0696 100644 --- a/src/table/merkle.rs +++ b/src/table/merkle.rs @@ -1,8 +1,5 @@ -use std::sync::{ - atomic::{AtomicUsize, Ordering}, - Arc, -}; -use std::time::Duration; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; use async_trait::async_trait; use serde::{Deserialize, Serialize}; @@ -29,7 +26,6 @@ use crate::schema::*; pub struct MerkleUpdater { data: Arc>, - previous_merkle_len: AtomicUsize, // Content of the todo tree: items where // - key = the key of an item in the main table, ie hash(partition_key)+sort_key @@ -75,6 +71,7 @@ impl MerkleUpdater { pub(crate) fn new(data: Arc>) -> Arc { let empty_node_hash = blake2sum(&nonversioned_encode(&MerkleNode::Empty).unwrap()[..]); + // @FIXME: move in worker 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), @@ -83,70 +80,22 @@ impl MerkleUpdater { Arc::new(Self { data, empty_node_hash, - previous_merkle_len: AtomicUsize::new(0), }) } pub(crate) fn spawn_workers(self: &Arc, background: &BackgroundRunner) { - background.spawn_worker(MerkleWorker(self.clone())); + background.spawn_worker(MerkleWorker(self.clone(), MerkleWorkerStats::new())); } fn updater_loop_iter(&self) -> Result { if let Some((key, valhash)) = self.data.merkle_todo.first()? { self.update_item(&key, &valhash)?; - match &self.data.config { - MerkleBackpressureEnum::None => (), - MerkleBackpressureEnum::Aimd(a) => self.adapt_aimd_backpressure(a)?, - }; Ok(WorkerState::Busy) } else { Ok(WorkerState::Idle) } } - fn adapt_aimd_backpressure(&self, config: &MerkleBackpressureAimd) -> Result<(), Error> { - // Capture evolution of the merkle todo length - let prev_merkle_len = self.previous_merkle_len.load(Ordering::Relaxed); - let current_merkle_len = self.data.merkle_todo.len()?; - debug!( - "prev merkle len: {}, new merkle len: {}", - prev_merkle_len, current_merkle_len - ); - - // Algorithm inspired by Additive Increase Multiplicative Decrease (AIMD) - { - let a = self.data.merkle_todo_sleep.clone(); - let mut v = a.lock().unwrap(); - let mut b; - if current_merkle_len <= prev_merkle_len { - // If we decrease the queue size, we can decrease the sleep time - b = v.saturating_sub(Duration::from_micros(config.underload_us)); - } else { - // If we are late, we increase the queue size - b = v - .mul_f64(config.overload_mult) - .saturating_add(Duration::from_micros(1)); - } - debug!( - "raw sleep. before {} -> after {}", - v.as_micros(), - b.as_micros() - ); - b = b.min(Duration::from_micros(config.max_us)); - b = b.max(Duration::from_micros(config.initial_us)); - debug!( - "cut sleep. before {} -> after {}", - v.as_micros(), - b.as_micros() - ); - *v = b; - } - - self.previous_merkle_len - .store(current_merkle_len, Ordering::Relaxed); - Ok(()) - } - fn update_item(&self, k: &[u8], vhash_by: &[u8]) -> Result<(), Error> { let khash = blake2sum(k); @@ -354,7 +303,65 @@ impl MerkleUpdater { } } -struct MerkleWorker(Arc>); +struct MerkleWorker( + Arc>, + MerkleWorkerStats, +); +struct MerkleWorkerStats { + last_update: SystemTime, + previous_merkle_len: usize, +} +impl MerkleWorkerStats { + fn new() -> MerkleWorkerStats { + Self { + last_update: SystemTime::now(), + previous_merkle_len: 0, + } + } + + fn adapt_aimd_backpressure( + &mut self, + updater: &MerkleUpdater, + config: &MerkleBackpressureAimd, + ) -> Result<(), Error> { + // Must have some elapsed time between runs + if self.last_update.elapsed().unwrap() < Duration::from_micros(config.sample_us) { + return Ok(()); // skip update + } + + // Capture evolution of the merkle todo length + let prev_merkle_len = self.previous_merkle_len; + let current_merkle_len = updater.data.merkle_todo.len()?; + debug!( + "prev merkle len: {}, new merkle len: {}", + prev_merkle_len, current_merkle_len + ); + + // Algorithm inspired by Additive Increase Multiplicative Decrease (AIMD) + { + let a = updater.data.merkle_todo_sleep.clone(); + let mut v = a.lock().unwrap(); + let mut b; + if current_merkle_len <= prev_merkle_len { + // If we decrease the queue size, we can decrease the sleep time + b = v.saturating_sub(Duration::from_micros(config.underload_us)); + } else { + // If we are late, we increase the queue size + b = v + .mul_f64(config.overload_mult) + .saturating_add(Duration::from_micros(1)); + } + b = b.min(Duration::from_micros(config.max_us)); + b = b.max(Duration::from_micros(config.initial_us)); + debug!("sleep. before {} -> after {}", v.as_micros(), b.as_micros()); + *v = b; + } + + self.last_update = SystemTime::now(); + self.previous_merkle_len = current_merkle_len; + Ok(()) + } +} #[async_trait] impl Worker for MerkleWorker { @@ -371,6 +378,10 @@ impl Worker for MerkleWorker { async fn work(&mut self, _must_exit: &mut watch::Receiver) -> Result { let updater = self.0.clone(); + match &updater.data.config { + MerkleBackpressureEnum::None => (), + MerkleBackpressureEnum::Aimd(a) => self.1.adapt_aimd_backpressure(&updater, a)?, + }; tokio::task::spawn_blocking(move || { for _i in 0..100 { let s = updater.updater_loop_iter(); diff --git a/src/table/table.rs b/src/table/table.rs index 3a17c191..6d0e9216 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -139,7 +139,8 @@ impl Table { who.as_ref(), rpc, RequestStrategy::with_priority(PRIO_NORMAL) - .with_quorum(self.data.replication.write_quorum()), + .with_quorum(self.data.replication.write_quorum()) + .with_write_limiter(), ) .await?; diff --git a/src/util/config.rs b/src/util/config.rs index 4a9cf131..fa37111e 100644 --- a/src/util/config.rs +++ b/src/util/config.rs @@ -282,6 +282,8 @@ pub struct MerkleBackpressureAimd { pub underload_us: u64, #[serde(default = "default_overload_mult")] pub overload_mult: f64, + #[serde(default = "default_sample_us")] + pub sample_us: u64, } /// Read and parse configuration @@ -323,7 +325,11 @@ fn default_underload_us() -> u64 { } fn default_overload_mult() -> f64 { - 1.01 + 1.1 +} + +fn default_sample_us() -> u64 { + 100 * 1000 } fn deserialize_compression<'de, D>(deserializer: D) -> Result, D::Error>