cleaner

Author: Vasilyev Dmitriy Viktorovich

src/main.rs

@@ -53,7 +53,7 @@ use pg_doorman::core_affinity;
 use pg_doorman::daemon;
 use pg_doorman::format_duration;
 use pg_doorman::messages::configure_tcp_socket;
-use pg_doorman::pool::{ClientServerMap, ConnectionPool};
+use pg_doorman::pool::{clean_connections, ClientServerMap, ConnectionPool};
 use pg_doorman::rate_limit::RateLimiter;
 use pg_doorman::stats::{Collector, Reporter, REPORTER, TOTAL_CONNECTION_COUNTER};
 use pg_doorman::tls::load_identity;
@@ -194,6 +194,10 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             stats_collector.collect().await;
         });
 
+        tokio::task::spawn(async move {
+            clean_connections().await;
+        });
+
         #[cfg(windows)]
         let mut term_signal = win_signal::ctrl_close().unwrap();
         #[cfg(windows)]

src/mobc/lib.rs

@@ -14,6 +14,7 @@ use futures_util::lock::{Mutex, MutexGuard};
 use futures_util::select;
 use futures_util::FutureExt;
 use futures_util::StreamExt;
+use log::info;
 use std::fmt;
 use std::future::Future;
 use std::ops::{Deref, DerefMut};
@@ -209,7 +210,19 @@ impl<M: Manager> Pool<M> {
             && self.0.state.num_open.load(Ordering::Relaxed) > 0
             && internals.cleaner_ch.is_none()
         {
-            log::debug!("run connection cleaner");
+            let shared1 = Arc::downgrade(&self.0);
+            let clean_rate = self.0.config.clean_rate;
+            let (cleaner_ch_sender, cleaner_ch) = mpsc::channel(1);
+            internals.cleaner_ch = Some(cleaner_ch_sender);
+            self.0.manager.spawn_task(async move {
+                connection_cleaner(shared1, cleaner_ch, clean_rate).await;
+            });
+        }
+    }
+
+    pub async fn clean_connections(&self) {
+        let mut internals = self.0.internals.lock().await;
+        if self.0.state.num_open.load(Ordering::Relaxed) > 0 && internals.cleaner_ch.is_none() {
             let shared1 = Arc::downgrade(&self.0);
             let clean_rate = self.0.config.clean_rate;
             let (cleaner_ch_sender, cleaner_ch) = mpsc::channel(1);
@@ -459,13 +472,10 @@ async fn clean_connection<M: Manager>(shared: &Weak<SharedPool<M>>) -> bool {
     let shared = match shared.upgrade() {
         Some(shared) => shared,
         None => {
-            log::debug!("Failed to clean connections");
             return false;
         }
     };
 
-    log::debug!("Clean connections");
-
     let mut internals = shared.internals.lock().await;
     if shared.state.num_open.load(Ordering::Relaxed) == 0 || internals.config.max_lifetime.is_none()
     {
@@ -477,10 +487,6 @@ async fn clean_connection<M: Manager>(shared: &Weak<SharedPool<M>>) -> bool {
     let mut closing = vec![];
 
     let mut i = 0;
-    log::debug!(
-        "clean connections, idle conns {}",
-        internals.free_conns.len()
-    );
 
     loop {
         if i >= internals.free_conns.len() {

src/pool.rs

@@ -510,3 +510,13 @@ pub fn get_pool(db: &str, user: &str, virtual_pool_id: u16) -> Option<Connection
 pub fn get_all_pools() -> HashMap<PoolIdentifierVirtual, ConnectionPool> {
     (*(*POOLS.load())).clone()
 }
+
+pub async fn clean_connections() {
+    let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(10));
+    loop {
+        interval.tick().await;
+        for (_, pool) in get_all_pools() {
+            pool.database.clean_connections().await;
+        }
+    }
+}

tests/tests.toml

@@ -25,7 +25,7 @@ default_tcp_so_linger = 0
 # non-buffer streaming messages smaller than 1mb
 max_message_size = 1048576
 
-server_lifetime = 500000
+server_lifetime = 5000
 idle_timeout = 3000
 
 # admin user.