From 5e0fee77672447ec9e96024545385ad33bf53a04 Mon Sep 17 00:00:00 2001 From: Eric Zhang Date: Tue, 11 Feb 2025 10:54:22 -0500 Subject: [PATCH] Add RecyclingMethod to `deadpool-redis` This change adds a new `RecyclingMethod` enum to `deadpool-redis`, which can be used to clean up subscriptions when recycling connections. Also removing the support for configuring Redis connections via `AsyncConnectionConfig`. Part of the plan in https://github.com/deadpool-rs/deadpool/issues/226 Let me know if there's any other way I could help! (Not sure if you had these changes in another branch also, if that's the case happy to look at that and try to work off of your existing changes.) --- redis/src/cluster/config.rs | 7 +++++ redis/src/config.rs | 37 ++++++++++++++++++++++++ redis/src/lib.rs | 56 +++++++++++++++++------------------- redis/src/sentinel/config.rs | 18 ++++++++---- 4 files changed, 83 insertions(+), 35 deletions(-) diff --git a/redis/src/cluster/config.rs b/redis/src/cluster/config.rs index 076925f0..f89633fa 100644 --- a/redis/src/cluster/config.rs +++ b/redis/src/cluster/config.rs @@ -47,6 +47,11 @@ pub struct Config { /// [`redis::ConnectionInfo`] structures. pub connections: Option>, + /// [`Manager`] configuration. + /// + /// [`Manager`]: super::Manager + pub manager: Option, + /// Pool configuration. pub pool: Option, @@ -114,6 +119,7 @@ impl Config { Config { urls: Some(urls.into()), connections: None, + manager: None, pool: None, read_from_replicas: false, } @@ -125,6 +131,7 @@ impl Default for Config { Self { urls: None, connections: Some(vec![ConnectionInfo::default()]), + manager: None, pool: None, read_from_replicas: false, } diff --git a/redis/src/config.rs b/redis/src/config.rs index 821fc681..bf4a5526 100644 --- a/redis/src/config.rs +++ b/redis/src/config.rs @@ -45,6 +45,11 @@ pub struct Config { /// [`redis::ConnectionInfo`] structure. pub connection: Option, + /// [`Manager`] configuration. + /// + /// [`Manager`]: super::Manager + pub manager: Option, + /// Pool configuration. pub pool: Option, } @@ -93,6 +98,7 @@ impl Config { Config { url: Some(url.into()), connection: None, + manager: None, pool: None, } } @@ -104,6 +110,7 @@ impl Config { Config { url: None, connection: Some(connection_info.into()), + manager: None, pool: None, } } @@ -114,6 +121,7 @@ impl Default for Config { Self { url: None, connection: Some(ConnectionInfo::default()), + manager: None, pool: None, } } @@ -328,3 +336,32 @@ impl fmt::Display for ConfigError { } impl std::error::Error for ConfigError {} + +/// Configuration object for a [`Manager`]. +/// +/// This currently only makes it possible to specify which [`RecyclingMethod`] +/// should be used when retrieving existing objects from the [`Pool`]. +/// +/// [`Manager`]: super::Manager +#[derive(Clone, Copy, Debug, Default)] +#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))] +pub struct ManagerConfig { + /// Method of how a connection is recycled. See [`RecyclingMethod`]. + pub recycling_method: RecyclingMethod, +} + +/// Possible methods of how a connection is recycled. +/// +/// The default is [`Fast`] which does not check the connection health or +/// perform any clean-up queries. +/// +/// [`Fast`]: RecyclingMethod::Fast +/// [`Verified`]: RecyclingMethod::Verified +#[derive(Clone, Copy, Debug, Eq, PartialEq, Default)] +#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))] +pub enum RecyclingMethod { + /// Run `UNWATCH` to discard transaction state, then send a ping to verify + /// the connection is still alive. + #[default] + Clean, +} diff --git a/redis/src/lib.rs b/redis/src/lib.rs index b995be34..96cd4bf1 100644 --- a/redis/src/lib.rs +++ b/redis/src/lib.rs @@ -42,7 +42,8 @@ use redis::{ pub use redis; pub use self::config::{ - Config, ConfigError, ConnectionAddr, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, + Config, ConfigError, ConnectionAddr, ConnectionInfo, ManagerConfig, ProtocolVersion, + RecyclingMethod, RedisConnectionInfo, }; pub use deadpool::managed::reexports::*; @@ -127,20 +128,11 @@ impl ConnectionLike for Connection { /// [`Manager`] for creating and recycling [`redis`] connections. /// /// [`Manager`]: managed::Manager +#[derive(Debug)] pub struct Manager { + config: ManagerConfig, client: Client, ping_number: AtomicUsize, - connection_config: AsyncConnectionConfig, -} - -// `redis::AsyncConnectionConfig: !Debug` -impl std::fmt::Debug for Manager { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("Manager") - .field("client", &self.client) - .field("ping_number", &self.ping_number) - .finish() - } } impl Manager { @@ -150,22 +142,22 @@ impl Manager { /// /// If establishing a new [`Client`] fails. pub fn new(params: T) -> RedisResult { - Self::from_config(params, AsyncConnectionConfig::default()) + Self::from_config(params, ManagerConfig::default()) } - /// Creates a new [`Manager`] from the given `params` and [`AsyncConnectionConfig`]. + /// Creates a new [`Manager`] from the given `params` and [`ManagerConfig`]. /// /// # Errors /// /// If establishing a new [`Client`] fails. pub fn from_config( params: T, - connection_config: AsyncConnectionConfig, + config: ManagerConfig, ) -> RedisResult { Ok(Self { + config, client: Client::open(params)?, ping_number: AtomicUsize::new(0), - connection_config, }) } } @@ -177,25 +169,29 @@ impl managed::Manager for Manager { async fn create(&self) -> Result { let conn = self .client - .get_multiplexed_async_connection_with_config(&self.connection_config) + .get_multiplexed_async_connection_with_config(&AsyncConnectionConfig::default()) .await?; Ok(conn) } async fn recycle(&self, conn: &mut MultiplexedConnection, _: &Metrics) -> RecycleResult { - let ping_number = self.ping_number.fetch_add(1, Ordering::Relaxed).to_string(); - // Using pipeline to avoid roundtrip for UNWATCH - let (n,) = redis::Pipeline::with_capacity(2) - .cmd("UNWATCH") - .ignore() - .cmd("PING") - .arg(&ping_number) - .query_async::<(String,)>(conn) - .await?; - if n == ping_number { - Ok(()) - } else { - Err(managed::RecycleError::message("Invalid PING response")) + match self.config.recycling_method { + RecyclingMethod::Clean => { + let ping_number = self.ping_number.fetch_add(1, Ordering::Relaxed).to_string(); + // Using pipeline to avoid roundtrip for UNWATCH + let (n,) = redis::Pipeline::with_capacity(2) + .cmd("UNWATCH") + .ignore() + .cmd("PING") + .arg(&ping_number) + .query_async::<(String,)>(conn) + .await?; + if n == ping_number { + Ok(()) + } else { + Err(managed::RecycleError::message("Invalid PING response")) + } + } } } } diff --git a/redis/src/sentinel/config.rs b/redis/src/sentinel/config.rs index 13a4cb4e..922d966f 100644 --- a/redis/src/sentinel/config.rs +++ b/redis/src/sentinel/config.rs @@ -57,6 +57,12 @@ pub struct Config { pub connections: Option>, /// [`redis::sentinel::SentinelNodeConnectionInfo`] structures. pub node_connection_info: Option, + + /// [`Manager`] configuration. + /// + /// [`Manager`]: super::Manager + pub manager: Option, + /// Pool configuration. pub pool: Option, } @@ -123,11 +129,12 @@ impl Config { ) -> Config { Config { urls: Some(urls.into()), - connections: None, - master_name, server_type, - pool: None, + master_name, + connections: None, node_connection_info: None, + manager: None, + pool: None, } } @@ -150,11 +157,12 @@ impl Default for Config { Self { urls: None, - connections: Some(vec![default_connection_info.clone()]), server_type: SentinelServerType::Master, master_name: default_master_name(), - pool: None, + connections: Some(vec![default_connection_info.clone()]), node_connection_info: None, + manager: None, + pool: None, } } }