From aadf22a22d369a52002830df3c951ebe63dcfd4d Mon Sep 17 00:00:00 2001 From: Lukasz Klimek <842586+lklimek@users.noreply.github.com> Date: Fri, 12 Apr 2024 16:34:52 +0200 Subject: [PATCH 1/2] feat(dapi-client): reuse dapi connections --- Cargo.lock | 2 + packages/rs-dapi-client/Cargo.toml | 3 +- packages/rs-dapi-client/src/address_list.rs | 28 ++++- .../rs-dapi-client/src/connection_pool.rs | 110 ++++++++++++++++++ packages/rs-dapi-client/src/dapi_client.rs | 14 ++- packages/rs-dapi-client/src/lib.rs | 1 + packages/rs-dapi-client/src/transport.rs | 9 +- packages/rs-dapi-client/src/transport/grpc.rs | 49 +++++--- 8 files changed, 194 insertions(+), 22 deletions(-) create mode 100644 packages/rs-dapi-client/src/connection_pool.rs diff --git a/Cargo.lock b/Cargo.lock index 339360ad000..09360bc6d0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3537,6 +3537,8 @@ dependencies = [ "futures", "hex", "http", + "lazy_static", + "lru", "rand", "sha2", "thiserror", diff --git a/packages/rs-dapi-client/Cargo.toml b/packages/rs-dapi-client/Cargo.toml index 32e87c7ce6c..26480eb9345 100644 --- a/packages/rs-dapi-client/Cargo.toml +++ b/packages/rs-dapi-client/Cargo.toml @@ -24,6 +24,7 @@ tokio = { version = "1.32.0", default-features = false } sha2 = { version = "0.10", optional = true } chrono = { version = "0.4.31", optional = true } hex = { version = "0.4.3", optional = true } - +lru = { version = "0.12.3" } +lazy_static = { version = "1.4.0" } [dev-dependencies] tokio = { version = "1.32.0", features = ["macros"] } diff --git a/packages/rs-dapi-client/src/address_list.rs b/packages/rs-dapi-client/src/address_list.rs index 6f1d2e8cf9c..a56820af2d6 100644 --- a/packages/rs-dapi-client/src/address_list.rs +++ b/packages/rs-dapi-client/src/address_list.rs @@ -160,9 +160,15 @@ impl AddressList { /// Randomly select a not banned address. pub fn get_live_address(&self) -> Option<&Address> { - let now = time::Instant::now(); let mut rng = SmallRng::from_entropy(); + self.unbanned().into_iter().choose(&mut rng) + } + + /// Get all addresses that are not banned. + fn unbanned(&self) -> Vec<&Address> { + let now = time::Instant::now(); + self.addresses .iter() .filter(|addr| { @@ -170,7 +176,25 @@ impl AddressList { .map(|banned_until| banned_until < now) .unwrap_or(true) }) - .choose(&mut rng) + .collect() + } + + /// Get number of available, not banned addresses. + pub fn available(&self) -> usize { + self.unbanned().len() + } + + /// Get number of all addresses, both banned and not banned. + pub fn len(&self) -> usize { + self.addresses.len() + } + + /// Check if the list is empty. + /// Returns true if there are no addresses in the list. + /// Returns false if there is at least one address in the list. + /// Banned addresses are also counted. + pub fn is_empty(&self) -> bool { + self.addresses.is_empty() } } diff --git a/packages/rs-dapi-client/src/connection_pool.rs b/packages/rs-dapi-client/src/connection_pool.rs new file mode 100644 index 00000000000..098dd243696 --- /dev/null +++ b/packages/rs-dapi-client/src/connection_pool.rs @@ -0,0 +1,110 @@ +use std::sync::{Arc, Mutex}; + +use http::Uri; +use lru::LruCache; + +use crate::{ + request_settings::AppliedRequestSettings, + transport::{CoreGrpcClient, PlatformGrpcClient}, +}; + +/// ConnectionPool represents pool of connections to DAPI nodes. +/// +/// It can be cloned and shared between threads. +/// Cloning the pool will create a new reference to the same pool. +#[derive(Debug, Clone)] +pub struct ConnectionPool { + inner: Arc>>, +} + +impl ConnectionPool { + /// Create a new pool with a given capacity. + /// The pool will evict the least recently used item when the capacity is reached. + /// + /// # Panics + /// + /// Panics if the capacity is zero. + pub fn new(capacity: usize) -> Self { + Self { + inner: Arc::new(Mutex::new(LruCache::new( + capacity.try_into().expect("must be non-zero"), + ))), + } + } +} + +impl Default for ConnectionPool { + fn default() -> Self { + Self::new(50) + } +} + +impl ConnectionPool { + /// Get item from the pool for the given uri and settings. + pub fn get(&self, uri: &Uri, settings: Option<&AppliedRequestSettings>) -> Option { + let key = format!("{}{:?}", uri, settings); + self.inner.lock().expect("must lock").get(&key).cloned() + } + + /// Get value from cache or create it using provided closure. + /// If value is already in the cache, it will be returned. + /// If value is not in the cache, it will be created by calling `create()` and stored in the cache. + pub fn get_or_create( + &self, + uri: &Uri, + settings: Option<&AppliedRequestSettings>, + create: impl FnOnce() -> PoolItem, + ) -> PoolItem { + if let Some(cli) = self.get(uri, settings) { + return cli; + } + + let cli = create(); + self.put(uri, settings, cli.clone()); + cli + } + + /// Put item into the pool for the given uri and settings. + pub fn put(&self, uri: &Uri, settings: Option<&AppliedRequestSettings>, value: PoolItem) { + let key = format!("{}{:?}", uri, settings); + self.inner.lock().expect("must lock").put(key, value); + } +} + +/// Item stored in the pool. +/// +/// We use an enum as we need to represent two different types of clients. +#[derive(Clone, Debug)] +pub enum PoolItem { + Core(CoreGrpcClient), + Platform(PlatformGrpcClient), +} + +impl From for PoolItem { + fn from(client: PlatformGrpcClient) -> Self { + Self::Platform(client) + } +} +impl From for PoolItem { + fn from(client: CoreGrpcClient) -> Self { + Self::Core(client) + } +} + +impl From for PlatformGrpcClient { + fn from(client: PoolItem) -> Self { + match client { + PoolItem::Platform(client) => client, + _ => panic!("ClientType is not Platform"), + } + } +} + +impl From for CoreGrpcClient { + fn from(client: PoolItem) -> Self { + match client { + PoolItem::Core(client) => client, + _ => panic!("ClientType is not Core"), + } + } +} diff --git a/packages/rs-dapi-client/src/dapi_client.rs b/packages/rs-dapi-client/src/dapi_client.rs index 7486b8f5a9c..f6980503044 100644 --- a/packages/rs-dapi-client/src/dapi_client.rs +++ b/packages/rs-dapi-client/src/dapi_client.rs @@ -8,6 +8,7 @@ use std::time::Duration; use tracing::Instrument; use crate::address_list::AddressListError; +use crate::connection_pool::ConnectionPool; use crate::{ transport::{TransportClient, TransportRequest}, Address, AddressList, CanRetry, RequestSettings, @@ -64,6 +65,7 @@ pub trait DapiRequestExecutor { pub struct DapiClient { address_list: RwLock, settings: RequestSettings, + pool: ConnectionPool, #[cfg(feature = "dump")] pub(crate) dump_dir: Option, } @@ -71,9 +73,13 @@ pub struct DapiClient { impl DapiClient { /// Initialize new [DapiClient] and optionally override default settings. pub fn new(address_list: AddressList, settings: RequestSettings) -> Self { + // multiply by 3 as we need to store core and platform addresses, and we want some spare capacity just in case + let address_count = 3 * address_list.len(); + Self { address_list: RwLock::new(address_list), settings, + pool: ConnectionPool::new(address_count), #[cfg(feature = "dump")] dump_dir: None, } @@ -153,9 +159,13 @@ impl DapiRequestExecutor for DapiClient { // It stays wrapped in `Result` since we want to return // `impl Future`, not a `Result` itself. let address = address_result?; + let pool = self.pool.clone(); - let mut transport_client = - R::Client::with_uri_and_settings(address.uri().clone(), &applied_settings); + let mut transport_client = R::Client::with_uri_and_settings( + address.uri().clone(), + &applied_settings, + &pool, + ); let response = transport_request .execute_transport(&mut transport_client, &applied_settings) diff --git a/packages/rs-dapi-client/src/lib.rs b/packages/rs-dapi-client/src/lib.rs index 597bc28acce..6fe633b2142 100644 --- a/packages/rs-dapi-client/src/lib.rs +++ b/packages/rs-dapi-client/src/lib.rs @@ -3,6 +3,7 @@ #![deny(missing_docs)] mod address_list; +mod connection_pool; mod dapi_client; #[cfg(feature = "dump")] pub mod dump; diff --git a/packages/rs-dapi-client/src/transport.rs b/packages/rs-dapi-client/src/transport.rs index 81f343c7777..bf514bf3cc9 100644 --- a/packages/rs-dapi-client/src/transport.rs +++ b/packages/rs-dapi-client/src/transport.rs @@ -2,6 +2,7 @@ pub(crate) mod grpc; +use crate::connection_pool::ConnectionPool; pub use crate::request_settings::AppliedRequestSettings; use crate::{CanRetry, RequestSettings}; use dapi_grpc::mock::Mockable; @@ -50,8 +51,12 @@ pub trait TransportClient: Send + Sized { type Error: CanRetry + Send + Debug; /// Build client using node's url. - fn with_uri(uri: Uri) -> Self; + fn with_uri(uri: Uri, pool: &ConnectionPool) -> Self; /// Build client using node's url and [AppliedRequestSettings]. - fn with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Self; + fn with_uri_and_settings( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Self; } diff --git a/packages/rs-dapi-client/src/transport/grpc.rs b/packages/rs-dapi-client/src/transport/grpc.rs index ea0d8946343..eb685303a4a 100644 --- a/packages/rs-dapi-client/src/transport/grpc.rs +++ b/packages/rs-dapi-client/src/transport/grpc.rs @@ -3,6 +3,7 @@ use std::time::Duration; use super::{CanRetry, TransportClient, TransportRequest}; +use crate::connection_pool::ConnectionPool; use crate::{request_settings::AppliedRequestSettings, RequestSettings}; use dapi_grpc::core::v0::core_client::CoreClient; use dapi_grpc::core::v0::{self as core_proto}; @@ -17,15 +18,13 @@ pub type PlatformGrpcClient = PlatformClient; /// Core Client using gRPC transport. pub type CoreGrpcClient = CoreClient; -fn channel_with_uri(uri: Uri) -> Channel { - Channel::builder(uri).connect_lazy() -} - -fn channel_with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Channel { +fn create_channel(uri: Uri, settings: Option<&AppliedRequestSettings>) -> Channel { let mut builder = Channel::builder(uri); - if let Some(timeout) = settings.connect_timeout { - builder = builder.connect_timeout(timeout); + if let Some(settings) = settings { + if let Some(timeout) = settings.connect_timeout { + builder = builder.connect_timeout(timeout); + } } builder.connect_lazy() @@ -34,24 +33,44 @@ fn channel_with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> impl TransportClient for PlatformGrpcClient { type Error = dapi_grpc::tonic::Status; - fn with_uri(uri: Uri) -> Self { - Self::new(channel_with_uri(uri)) + fn with_uri(uri: Uri, pool: &ConnectionPool) -> Self { + pool.get_or_create(&uri, None, || { + Self::new(create_channel(uri.clone(), None)).into() + }) + .into() } - fn with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Self { - Self::new(channel_with_uri_and_settings(uri, settings)) + fn with_uri_and_settings( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Self { + pool.get_or_create(&uri, Some(settings), || { + Self::new(create_channel(uri.clone(), Some(settings))).into() + }) + .into() } } impl TransportClient for CoreGrpcClient { type Error = dapi_grpc::tonic::Status; - fn with_uri(uri: Uri) -> Self { - Self::new(channel_with_uri(uri)) + fn with_uri(uri: Uri, pool: &ConnectionPool) -> Self { + pool.get_or_create(&uri, None, || { + Self::new(create_channel(uri.clone(), None)).into() + }) + .into() } - fn with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Self { - Self::new(channel_with_uri_and_settings(uri, settings)) + fn with_uri_and_settings( + uri: Uri, + settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Self { + pool.get_or_create(&uri, Some(settings), || { + Self::new(create_channel(uri.clone(), Some(settings))).into() + }) + .into() } } From 96f47e7ef885bcc8b14aab2d049b768e604450e9 Mon Sep 17 00:00:00 2001 From: Lukasz Klimek <842586+lklimek@users.noreply.github.com> Date: Fri, 12 Apr 2024 17:05:03 +0200 Subject: [PATCH 2/2] chore: self-review --- Cargo.lock | 1 - packages/rs-dapi-client/Cargo.toml | 2 +- packages/rs-dapi-client/src/connection_pool.rs | 4 ++-- 3 files changed, 3 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 09360bc6d0f..e761dabf554 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3537,7 +3537,6 @@ dependencies = [ "futures", "hex", "http", - "lazy_static", "lru", "rand", "sha2", diff --git a/packages/rs-dapi-client/Cargo.toml b/packages/rs-dapi-client/Cargo.toml index 26480eb9345..c7af5325529 100644 --- a/packages/rs-dapi-client/Cargo.toml +++ b/packages/rs-dapi-client/Cargo.toml @@ -25,6 +25,6 @@ sha2 = { version = "0.10", optional = true } chrono = { version = "0.4.31", optional = true } hex = { version = "0.4.3", optional = true } lru = { version = "0.12.3" } -lazy_static = { version = "1.4.0" } + [dev-dependencies] tokio = { version = "1.32.0", features = ["macros"] } diff --git a/packages/rs-dapi-client/src/connection_pool.rs b/packages/rs-dapi-client/src/connection_pool.rs index 098dd243696..49216e38658 100644 --- a/packages/rs-dapi-client/src/connection_pool.rs +++ b/packages/rs-dapi-client/src/connection_pool.rs @@ -95,7 +95,7 @@ impl From for PlatformGrpcClient { fn from(client: PoolItem) -> Self { match client { PoolItem::Platform(client) => client, - _ => panic!("ClientType is not Platform"), + _ => panic!("ClientType is not Platform: {:?}", client), } } } @@ -104,7 +104,7 @@ impl From for CoreGrpcClient { fn from(client: PoolItem) -> Self { match client { PoolItem::Core(client) => client, - _ => panic!("ClientType is not Core"), + _ => panic!("ClientType is not Core: {:?}", client), } } }