gsadmin/garage
1
0
Fork 0
mirror of https://github.com/deuxfleurs-org/garage.git synced 2026-08-07 05:13:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

add a rpc in-flight limiter

Browse Source
This commit is contained in:
Quentin Dufour
2025-05-01 08:56:57 +02:00
parent d78e5f8a1b
commit f34558af07
12 changed files with 173 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/block/manager.rs
+2
View File
@@ -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 => {
src/garage/admin/mod.rs
+12 -1
View File
@@ -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?
}))
src/garage/cli/cmd.rs
+17 -3
View File
@@ -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<Vec<KnownNodeInfo>, 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),
src/garage/cli/layout.rs
+14 -2
View File
@@ -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<LayoutHistory, Error> {
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(())
src/net/client.rs
+15 -1
View File
@@ -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<HashMap<RequestID, oneshot::Sender<ByteStream>>>,
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<T>,
path: &str,
prio: RequestPriority,
limiter: RpcInFlightLimiter,
) -> Result<Resp<T>, 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)
}
src/net/endpoint.rs
+17 -2
View File
@@ -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<Resp<M>, Error>
where
T: IntoReq<M>,
@@ -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<<M as Message>::Response, Error> {
Ok(self.call_streaming(target, req, prio).await?.into_msg())
Ok(self
.call_streaming(target, req, prio, limiter)
.await?
.into_msg())
}
}
src/net/netapp.rs
+1
View File
@@ -427,6 +427,7 @@ impl NetApp {
server_port,
},
PRIO_NORMAL,
RpcInFlightLimiter::NoLimit,
)
.await
.map(|_| ())
src/net/peering.rs
+7 -2
View File
@@ -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),
src/rpc/rpc_helper.rs
+12 -1
View File
@@ -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<T> {
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<T> RequestStrategy<T> {
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<T> RequestStrategy<T> {
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);
src/table/merkle.rs
+67 -56
View File
@@ -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<F: TableSchema, R: TableReplication> {
data: Arc<TableData<F, R>>,
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<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()[..]);
// @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<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
Arc::new(Self {
data,
empty_node_hash,
previous_merkle_len: AtomicUsize::new(0),
})
}
pub(crate) fn spawn_workers(self: &Arc<Self>, background: &BackgroundRunner) {
background.spawn_worker(MerkleWorker(self.clone()));
background.spawn_worker(MerkleWorker(self.clone(), MerkleWorkerStats::new()));
}
fn updater_loop_iter(&self) -> Result<WorkerState, Error> {
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<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
}
}
struct MerkleWorker<F: TableSchema, R: TableReplication>(Arc<MerkleUpdater<F, R>>);
struct MerkleWorker<F: TableSchema, R: TableReplication>(
Arc<MerkleUpdater<F, R>>,
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<F: TableSchema, R: TableReplication>(
&mut self,
updater: &MerkleUpdater<F, R>,
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<F: TableSchema, R: TableReplication> Worker for MerkleWorker<F, R> {
@@ -371,6 +378,10 @@ impl<F: TableSchema, R: TableReplication> Worker for MerkleWorker<F, R> {
async fn work(&mut self, _must_exit: &mut watch::Receiver<bool>) -> Result<WorkerState, Error> {
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();
src/table/table.rs
+2 -1
View File
@@ -139,7 +139,8 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
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?;
src/util/config.rs
+7 -1
View File
@@ -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<Option<i32>, D::Error>