From f7c30befcd5bca7240684669e396a7570520ceb7 Mon Sep 17 00:00:00 2001 From: XuXiaoCheng <2762267080@qq.com> Date: Mon, 23 Feb 2026 02:15:44 +0800 Subject: [PATCH 1/4] Add drop_callback to SyncWrapper --- crates/deadpool-sync/src/lib.rs | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/crates/deadpool-sync/src/lib.rs b/crates/deadpool-sync/src/lib.rs index 75c339bb..525e9f77 100644 --- a/crates/deadpool-sync/src/lib.rs +++ b/crates/deadpool-sync/src/lib.rs @@ -75,6 +75,7 @@ where { obj: Arc>>, runtime: Runtime, + drop_callback: Option>, } // Implemented manually to avoid unnecessary trait bound on `E` type parameter. @@ -112,6 +113,7 @@ where result.map(|obj| Self { obj: Arc::new(Mutex::new(Some(obj))), runtime, + drop_callback: None, }) } @@ -157,6 +159,12 @@ where pub fn try_lock(&self) -> Result, TryLockError>>> { self.obj.try_lock().map(SyncGuard) } + + /// Set the callback when the actual obj is dropped. + // https://github.com/deadpool-rs/deadpool/issues/330 + pub fn set_drop_callback(&mut self, drop_callback: F) where F: FnOnce() + Send + 'static { + let _ = self.drop_callback.replace(Box::new(drop_callback)); + } } impl Drop for SyncWrapper @@ -165,11 +173,18 @@ where { fn drop(&mut self) { let arc = self.obj.clone(); + let drop_callback = self.drop_callback.take(); + #[cfg(feature = "tracing")] + let span = tracing::Span::current(); // Drop the `rusqlite::Connection` inside a `spawn_blocking` // as the `drop` function of it can block. - spawn_blocking_background(self.runtime, move || match arc.lock() { - Ok(mut guard) => drop(guard.take()), - Err(e) => drop(e.into_inner().take()), + spawn_blocking_background(self.runtime, move || { + drop(arc.lock().unwrap_or_else(PoisonError::into_inner).take()); + if let Some(callback) = drop_callback { + #[cfg(feature = "tracing")] + let _span = span.enter(); + callback(); + } }) .unwrap(); } From 21ac5fba604f4262caeef68fd050294c3086ba36 Mon Sep 17 00:00:00 2001 From: XuXiaoCheng <2762267080@qq.com> Date: Mon, 23 Feb 2026 02:17:49 +0800 Subject: [PATCH 2/4] Add wait_all_inactive method to deadpool_sqlite::Manager --- crates/deadpool-sqlite/Cargo.toml | 1 + crates/deadpool-sqlite/src/lib.rs | 19 ++++++++++++++++++- 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/crates/deadpool-sqlite/Cargo.toml b/crates/deadpool-sqlite/Cargo.toml index 630e2628..0da12cc5 100644 --- a/crates/deadpool-sqlite/Cargo.toml +++ b/crates/deadpool-sqlite/Cargo.toml @@ -71,6 +71,7 @@ rusqlite = "0.38.0" serde = { package = "serde", version = "1.0", features = [ "derive", ], optional = true } +tokio = { version = "1.5", features = ["sync"] } [dev-dependencies] config = { version = "0.15", features = ["json"] } diff --git a/crates/deadpool-sqlite/src/lib.rs b/crates/deadpool-sqlite/src/lib.rs index 34329969..c797b807 100644 --- a/crates/deadpool-sqlite/src/lib.rs +++ b/crates/deadpool-sqlite/src/lib.rs @@ -53,6 +53,7 @@ pub struct Manager { config: Config, recycle_count: AtomicIsize, runtime: Runtime, + active_count: tokio::sync::watch::Sender, } impl Manager { @@ -64,8 +65,20 @@ impl Manager { config: config.clone(), recycle_count: AtomicIsize::new(0), runtime, + active_count: tokio::sync::watch::Sender::new(0), } } + + /// Retrieve the active `rusqlite::Connection` count. + #[must_use] + pub fn active_count(&self) -> usize { + *self.active_count.borrow() + } + + /// Wait for all `rusqlite::Connection` is closed in background. + pub async fn wait_all_inactive(&self) { + let _ = self.active_count.subscribe().wait_for(|count| *count == 0).await; + } } impl managed::Manager for Manager { @@ -74,7 +87,11 @@ impl managed::Manager for Manager { async fn create(&self) -> Result { let path = self.config.path.clone(); - SyncWrapper::new(self.runtime, move || rusqlite::Connection::open(path)).await + let mut conn = SyncWrapper::new(self.runtime, move || rusqlite::Connection::open(path)).await?; + self.active_count.send_modify(|count| *count += 1); + let active_count = self.active_count.clone(); + conn.set_drop_callback(move || active_count.send_modify(|count| *count -= 1)); + Ok(conn) } async fn recycle( From b1aa3b72a1704fe92b162a37fd0875a41cbeab00 Mon Sep 17 00:00:00 2001 From: XuXiaoCheng <2762267080@qq.com> Date: Mon, 23 Feb 2026 02:33:34 +0800 Subject: [PATCH 3/4] Fix compilation error, ensure SyncWrapper Send and Sync --- crates/deadpool-sync/src/lib.rs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/crates/deadpool-sync/src/lib.rs b/crates/deadpool-sync/src/lib.rs index 525e9f77..c1d7d6cc 100644 --- a/crates/deadpool-sync/src/lib.rs +++ b/crates/deadpool-sync/src/lib.rs @@ -75,9 +75,14 @@ where { obj: Arc>>, runtime: Runtime, - drop_callback: Option>, + drop_callback: Option>>, } +const _: () = { + const fn assert_send_sync() {} + assert_send_sync::>(); +}; + // Implemented manually to avoid unnecessary trait bound on `E` type parameter. impl fmt::Debug for SyncWrapper where @@ -163,7 +168,7 @@ where /// Set the callback when the actual obj is dropped. // https://github.com/deadpool-rs/deadpool/issues/330 pub fn set_drop_callback(&mut self, drop_callback: F) where F: FnOnce() + Send + 'static { - let _ = self.drop_callback.replace(Box::new(drop_callback)); + let _ = self.drop_callback.replace(Mutex::new(Box::new(drop_callback))); } } @@ -183,7 +188,7 @@ where if let Some(callback) = drop_callback { #[cfg(feature = "tracing")] let _span = span.enter(); - callback(); + callback.into_inner().unwrap()(); } }) .unwrap(); From cdd8b5a80ca77faed370806407c5d22b051d7eb2 Mon Sep 17 00:00:00 2001 From: XuXiaoCheng <2762267080@qq.com> Date: Tue, 24 Feb 2026 01:21:31 +0800 Subject: [PATCH 4/4] Fix fmt and MSRV --- crates/deadpool-sqlite/Cargo.toml | 2 +- crates/deadpool-sqlite/src/lib.rs | 9 +++++++-- crates/deadpool-sync/src/lib.rs | 9 +++++++-- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/crates/deadpool-sqlite/Cargo.toml b/crates/deadpool-sqlite/Cargo.toml index 0da12cc5..a154f5fd 100644 --- a/crates/deadpool-sqlite/Cargo.toml +++ b/crates/deadpool-sqlite/Cargo.toml @@ -71,7 +71,7 @@ rusqlite = "0.38.0" serde = { package = "serde", version = "1.0", features = [ "derive", ], optional = true } -tokio = { version = "1.5", features = ["sync"] } +tokio = { version = "1.37", features = ["sync"] } [dev-dependencies] config = { version = "0.15", features = ["json"] } diff --git a/crates/deadpool-sqlite/src/lib.rs b/crates/deadpool-sqlite/src/lib.rs index c797b807..4472ea8c 100644 --- a/crates/deadpool-sqlite/src/lib.rs +++ b/crates/deadpool-sqlite/src/lib.rs @@ -77,7 +77,11 @@ impl Manager { /// Wait for all `rusqlite::Connection` is closed in background. pub async fn wait_all_inactive(&self) { - let _ = self.active_count.subscribe().wait_for(|count| *count == 0).await; + let _ = self + .active_count + .subscribe() + .wait_for(|count| *count == 0) + .await; } } @@ -87,7 +91,8 @@ impl managed::Manager for Manager { async fn create(&self) -> Result { let path = self.config.path.clone(); - let mut conn = SyncWrapper::new(self.runtime, move || rusqlite::Connection::open(path)).await?; + let mut conn = + SyncWrapper::new(self.runtime, move || rusqlite::Connection::open(path)).await?; self.active_count.send_modify(|count| *count += 1); let active_count = self.active_count.clone(); conn.set_drop_callback(move || active_count.send_modify(|count| *count -= 1)); diff --git a/crates/deadpool-sync/src/lib.rs b/crates/deadpool-sync/src/lib.rs index c1d7d6cc..6959034a 100644 --- a/crates/deadpool-sync/src/lib.rs +++ b/crates/deadpool-sync/src/lib.rs @@ -167,8 +167,13 @@ where /// Set the callback when the actual obj is dropped. // https://github.com/deadpool-rs/deadpool/issues/330 - pub fn set_drop_callback(&mut self, drop_callback: F) where F: FnOnce() + Send + 'static { - let _ = self.drop_callback.replace(Mutex::new(Box::new(drop_callback))); + pub fn set_drop_callback(&mut self, drop_callback: F) + where + F: FnOnce() + Send + 'static, + { + let _ = self + .drop_callback + .replace(Mutex::new(Box::new(drop_callback))); } }