diff --git a/Cargo.lock b/Cargo.lock index bd5a1f9f..25f3a068 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1470,6 +1470,7 @@ dependencies = [ "tokio", "tokio-stream", "tokio-util 0.7.14", + "tracing", ] [[package]] diff --git a/src/garage/main.rs b/src/garage/main.rs index ac95e854..04f34d72 100644 --- a/src/garage/main.rs +++ b/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 { diff --git a/src/net/Cargo.toml b/src/net/Cargo.toml index b48eb153..3a8986a3 100644 --- a/src/net/Cargo.toml +++ b/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 diff --git a/src/net/client.rs b/src/net/client.rs index 1b0b00b9..e0c09cf5 100644 --- a/src/net/client.rs +++ b/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>>, - rpc_table_write_inflight_limiter: Semaphore, + rpc_table_write_inflight_limiter: Option, } 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 { - RpcInFlightLimiter::NoLimit => None, - RpcInFlightLimiter::TableWrite => Some( - self.rpc_table_write_inflight_limiter - .acquire() - .await - .unwrap(), - ), + let _permit = match (limiter, &self.rpc_table_write_inflight_limiter) { + (RpcInFlightLimiter::TableWrite, Some(sem)) => { + info!( + "Available RPC table write slots: {}", + sem.available_permits() + ); + Some(sem.acquire().await.unwrap()) + } + _ => None, }; let query_send = self.query_send.load_full().ok_or(Error::ConnectionClosed)?; diff --git a/src/net/netapp.rs b/src/net/netapp.rs index 2842b4aa..928951fd 100644 --- a/src/net/netapp.rs +++ b/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, pub(crate) server_conns: RwLock>>, pub(crate) client_conns: RwLock>>, @@ -101,6 +103,7 @@ impl NetApp { netid: auth::Key, privkey: ed25519::SecretKey, bind_outgoing_to: Option, + max_in_flight_table_write: Option, ) -> Arc { 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()), diff --git a/src/rpc/system.rs b/src/rpc/system.rs index 2a52ae5d..6d6af809 100644 --- a/src/rpc/system.rs +++ b/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 ----