Skip to content

Commit 4d7cab2

Browse files
committed
Reap expired connections on drop
The reaper only runs against the connections in its idle pool. This is fine for reaping idle connections, but for hotly contested connections beyond their maximum lifetime this can prove problematic. Consider an active connection beyond its lifetime and a reaper that runs every 3 seconds: - [t0] Connection is idle - [t1] Connection is active - [t2] Reaper runs, does not see connection - [t3] Connection is idle This pattern can repeat infinitely with the connection never being reaped. By checking the max lifetime on drop, we can ensure that expired connections are reaped in a reason amount of time (assuming they eventually do get dropped).
1 parent 4528b77 commit 4d7cab2

File tree

2 files changed

+46
-3
lines changed

2 files changed

+46
-3
lines changed

bb8/src/inner.rs

+11-3
Original file line numberDiff line numberDiff line change
@@ -149,12 +149,20 @@ where
149149
"handled in caller"
150150
);
151151

152+
let max_lifetime = self.inner.statics.max_lifetime;
153+
let is_expired = max_lifetime.map_or(false, |lt| conn.is_expired(Instant::now() - lt));
154+
let is_broken = self.inner.manager.has_broken(&mut conn.conn);
155+
152156
let mut locked = self.inner.internals.lock();
153-
match (state, self.inner.manager.has_broken(&mut conn.conn)) {
154-
(ConnectionState::Present, false) => locked.put(conn, None, self.inner.clone()),
155-
(_, is_broken) => {
157+
match (state, is_broken || is_expired) {
158+
(ConnectionState::Present, false) if !is_expired => {
159+
locked.put(conn, None, self.inner.clone())
160+
}
161+
_ => {
156162
if is_broken {
157163
self.inner.statistics.record(StatsKind::ClosedBroken);
164+
} else if is_expired {
165+
self.inner.statistics.record_connections_reaped(0, 1);
158166
}
159167
let approvals = locked.dropped(1, &self.inner.statics);
160168
self.spawn_replenishing_approvals(approvals);

bb8/tests/test.rs

+35
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
use bb8::*;
2+
use tokio::time::sleep;
23

34
use std::future::Future;
45
use std::marker::PhantomData;
@@ -585,6 +586,40 @@ async fn test_max_lifetime() {
585586
assert_eq!(pool.state().statistics.connections_closed_max_lifetime, 5);
586587
}
587588

589+
#[tokio::test]
590+
async fn test_max_lifetime_reap_on_drop() {
591+
static DROPPED: AtomicUsize = AtomicUsize::new(0);
592+
593+
#[derive(Default)]
594+
struct Connection;
595+
596+
impl Drop for Connection {
597+
fn drop(&mut self) {
598+
DROPPED.fetch_add(1, Ordering::SeqCst);
599+
}
600+
}
601+
602+
let manager = OkManager::<Connection>::new();
603+
let pool = Pool::builder()
604+
.max_lifetime(Some(Duration::from_secs(1)))
605+
.connection_timeout(Duration::from_secs(1))
606+
.reaper_rate(Duration::from_secs(999))
607+
.build(manager)
608+
.await
609+
.unwrap();
610+
611+
let conn = pool.get().await;
612+
613+
// And wait.
614+
sleep(Duration::from_secs(2)).await;
615+
assert_eq!(DROPPED.load(Ordering::SeqCst), 0);
616+
617+
// Connection is reaped on drop.
618+
drop(conn);
619+
assert_eq!(DROPPED.load(Ordering::SeqCst), 1);
620+
assert_eq!(pool.state().statistics.connections_closed_max_lifetime, 1);
621+
}
622+
588623
#[tokio::test]
589624
async fn test_min_idle() {
590625
let pool = Pool::builder()

0 commit comments

Comments
 (0)