set config

Cargo.lock

@@ -1470,6 +1470,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tokio-util 0.7.14",
+ "tracing",
 ]
 
 [[package]]

src/garage/main.rs

@@ -244,7 +244,7 @@ async fn cli_command(opt: Opt) -> Result<(), Error> {
 	// Generate a temporary keypair for our RPC client
 	let (_pk, sk) = sodiumoxide::crypto::sign::ed25519::gen_keypair();
 
-	let netapp = NetApp::new(GARAGE_VERSION_TAG, network_key, sk, None);
+	let netapp = NetApp::new(GARAGE_VERSION_TAG, network_key, sk, None, None);
 
 	// Find and parse the address of the target host
 	let (id, addr, is_default_addr) = if let Some(h) = opt.rpc_host {

src/net/Cargo.toml

@@ -39,6 +39,7 @@ kuska-handshake.workspace = true
 
 opentelemetry = { workspace = true, optional = true }
 opentelemetry-contrib = { workspace = true, optional = true }
+tracing.workspace = true
 
 [dev-dependencies]
 pretty_env_logger.workspace = true

src/net/client.rs

@@ -4,6 +4,7 @@ use std::pin::Pin;
 use std::sync::atomic::{self, AtomicU32};
 use std::sync::{Arc, Mutex};
 use std::task::Poll;
+use tracing::*;
 
 use arc_swap::ArcSwapOption;
 use bytes::Bytes;
@@ -42,7 +43,7 @@ pub(crate) struct ClientConn {
 
 	next_query_number: AtomicU32,
 	inflight: Mutex<HashMap<RequestID, oneshot::Sender<ByteStream>>>,
-	rpc_table_write_inflight_limiter: Semaphore,
+	rpc_table_write_inflight_limiter: Option<Semaphore>,
 }
 
 impl ClientConn {
@@ -100,9 +101,14 @@ 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),
+			rpc_table_write_inflight_limiter: netapp.max_in_flight_table_write.map(Semaphore::new),
 		});
 
+		info!(
+			"Created conn with table write limit set to {}",
+			netapp.max_in_flight_table_write.unwrap_or(0)
+		);
+
 		netapp.connected_as_client(peer_id, conn.clone());
 
 		let debug_name = format!("CLI {}", hex::encode(&peer_id[..8]));
@@ -152,14 +158,15 @@ impl ClientConn {
 	where
 		T: Message,
 	{
-		let _permit = match limiter {
+		let _permit = match (limiter, &self.rpc_table_write_inflight_limiter) {
-			RpcInFlightLimiter::NoLimit => None,
+			(RpcInFlightLimiter::TableWrite, Some(sem)) => {
-			RpcInFlightLimiter::TableWrite => Some(
+				info!(
-				self.rpc_table_write_inflight_limiter
+					"Available RPC table write slots: {}",
-					.acquire()
+					sem.available_permits()
-					.await
+				);
-					.unwrap(),
+				Some(sem.acquire().await.unwrap())
-			),
+			}
+			_ => None,
 		};
 		let query_send = self.query_send.load_full().ok_or(Error::ConnectionClosed)?;
 

src/net/netapp.rs

@@ -74,6 +74,8 @@ pub struct NetApp {
 	pub id: NodeID,
 	/// Private key associated with our peer ID
 	pub privkey: ed25519::SecretKey,
+	/// Config related to netapp
+	pub(crate) max_in_flight_table_write: Option<usize>,
 
 	pub(crate) server_conns: RwLock<HashMap<NodeID, Arc<ServerConn>>>,
 	pub(crate) client_conns: RwLock<HashMap<NodeID, Arc<ClientConn>>>,
@@ -101,6 +103,7 @@ impl NetApp {
 		netid: auth::Key,
 		privkey: ed25519::SecretKey,
 		bind_outgoing_to: Option<IpAddr>,
+		max_in_flight_table_write: Option<usize>,
 	) -> Arc<Self> {
 		let mut version_tag = [0u8; 16];
 		version_tag[0..8].copy_from_slice(&u64::to_be_bytes(NETAPP_VERSION_TAG)[..]);
@@ -114,6 +117,7 @@ impl NetApp {
 			netid,
 			id,
 			privkey,
+			max_in_flight_table_write,
 			server_conns: RwLock::new(HashMap::new()),
 			client_conns: RwLock::new(HashMap::new()),
 			endpoints: RwLock::new(HashMap::new()),

src/rpc/system.rs

@@ -21,7 +21,7 @@ use garage_net::{NetApp, NetworkKey, NodeID, NodeKey};
 
 #[cfg(feature = "kubernetes-discovery")]
 use garage_util::config::KubernetesDiscoveryConfig;
-use garage_util::config::{Config, DataDirEnum};
+use garage_util::config::{Config, DataDirEnum, RpcInFlightLimiterEnum};
 use garage_util::data::*;
 use garage_util::error::*;
 use garage_util::persister::Persister;
@@ -256,7 +256,17 @@ impl System {
 		let bind_outgoing_to = Some(config)
 			.filter(|x| x.rpc_bind_outgoing)
 			.map(|x| x.rpc_bind_addr.ip());
-		let netapp = NetApp::new(GARAGE_VERSION_TAG, network_key, node_key, bind_outgoing_to);
+		let maybe_max_table_write = match &config.experimental.rpc_in_flight_limiters {
+			RpcInFlightLimiterEnum::None => None,
+			RpcInFlightLimiterEnum::FixedSize(v) => Some(v.max_table_write),
+		};
+		let netapp = NetApp::new(
+			GARAGE_VERSION_TAG,
+			network_key,
+			node_key,
+			bind_outgoing_to,
+			maybe_max_table_write,
+		);
 		let system_endpoint = netapp.endpoint(SYSTEM_RPC_PATH.into());
 
 		// ---- setup netapp public listener and full mesh peering strategy ----