diff --git a/src/lib.rs b/src/lib.rs index 1aaf136..a5b2939 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -325,11 +325,24 @@ impl Pool { gauge!(IDLE_CONNECTIONS).set(0.0); let (share_config, internal_config) = config.split(); + let max_lifetime = internal_config.max_lifetime; + let clean_rate = share_config.clean_rate; + + // Start the periodic cleaner up-front when a `max_lifetime` is + // configured, so idle connections are proactively reaped instead of + // only being checked lazily on checkout. + let (cleaner_ch_sender, cleaner_ch_receiver) = if max_lifetime.is_some() { + let (tx, rx) = mpsc::channel(1); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + let internals = Mutex::new(PoolInternals { config: internal_config, free_conns: Vec::new(), wait_duration: Duration::from_secs(0), - cleaner_ch: None, + cleaner_ch: cleaner_ch_sender, }); let pool_state = PoolState { @@ -347,6 +360,14 @@ impl Pool { state: pool_state, }); + if let Some(cleaner_ch) = cleaner_ch_receiver { + log::debug!("run connection cleaner"); + let shared1 = Arc::downgrade(&shared); + shared.manager.spawn_task(async move { + connection_cleaner(shared1, cleaner_ch, clean_rate).await; + }); + } + Pool(shared) } diff --git a/tests/mobc.rs b/tests/mobc.rs index 4dae16b..0ec6f47 100644 --- a/tests/mobc.rs +++ b/tests/mobc.rs @@ -573,7 +573,68 @@ fn test_max_lifetime_lazy() { } assert_eq!(5, DROPPED.load(Ordering::SeqCst)); pool.get().await.unwrap(); + // The connection above is returned to the pool as idle and then + // exceeds `max_lifetime` while sitting unused. The periodic cleaner + // now proactively reaps it even though nothing checks it out again. delay_for(Duration::from_secs(2)).await; + assert_eq!(6, DROPPED.load(Ordering::SeqCst)); + Ok::<(), Error>(()) + }) + .unwrap(); +} + +#[test] +fn test_max_lifetime_proactive_reap() { + static DROPPED: AtomicUsize = AtomicUsize::new(0); + let mut rt: Runtime = Runtime::new().unwrap(); + + struct Connection; + + impl Drop for Connection { + fn drop(&mut self) { + DROPPED.fetch_add(1, Ordering::SeqCst); + } + } + + struct Handler; + + #[async_trait] + impl Manager for Handler { + type Connection = Connection; + type Error = TestError; + + async fn connect(&self) -> Result { + Ok(Connection) + } + + async fn check(&self, conn: Self::Connection) -> Result { + Ok(conn) + } + } + let handler = Handler; + rt.block_on(async { + let pool = Pool::builder() + .max_open(5) + .max_idle(5) + .max_lifetime(Some(Duration::from_millis(300))) + .get_timeout(Some(Duration::from_secs(1))) + .clean_rate(Duration::from_secs(1)) + .build(handler); + + let mut v = vec![]; + for _ in 0..5 { + v.push(pool.get().await.unwrap()); + } + // Return all connections to the pool as idle, but never check any + // connection back out again. + drop(v); + assert_eq!(0, DROPPED.load(Ordering::SeqCst)); + + // Wait past both `max_lifetime` and `clean_rate` without performing + // any checkout. The periodic cleaner should proactively reap the + // now-expired idle connections on its own. + delay_for(Duration::from_secs(2)).await; + assert_eq!(5, DROPPED.load(Ordering::SeqCst)); Ok::<(), Error>(()) })