Repository: blackbeam/mysql_async (language: rust)

Issue:
Strange behavior when `PoolConstraints::min` is 0
I accidentally set `PoolConstraints::min` to `0` (and max to 1), and this causes the library to continuously drop and start connections as new queries are issued. That seems counterintuitive. Should this be asserted against in `PoolConstraints::new`?


Discussion / hints:
This reminds me an old issue in rust where the value was pushed to the stack in a loop when the stack was on it's boundary.

Thanks for report. I'll introduce a kind of "inactive connection timeout" for this.

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 06feed08..0323c1ca 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -12,18 +12,16 @@ exclude = ["test/*"]
 edition = "2018"
 
 [dependencies]
-byteorder = "1"
 bytes = "0.4"
 crossbeam = "0.7"
-fnv = "1"
 futures-core-preview = "=0.3.0-alpha.19"
 futures-util-preview = "=0.3.0-alpha.19"
 futures-sink-preview = "=0.3.0-alpha.19"
+mio-named-pipes = "0.1.6"
 mysql_common = "0.19.2"
 native-tls = "0.2"
 percent-encoding = "2.1.0"
 pin-project = "0.4.0"
-regex = "1"
 serde = "1"
 serde_json = "1"
 thiserror = "1.0.4"
diff --git a/appveyor.yml b/appveyor.yml
index 215435f2..0096b83a 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -1,3 +1,6 @@
+branches:
+  only:
+    - master
 version: 1.0.{build}
 clone_folder: c:\clone
 environment:
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
index b3baf5a0..166d7cc0 100644
--- a/src/conn/mod.rs
+++ b/src/conn/mod.rs
@@ -17,7 +17,15 @@ use mysql_common::{
     },
 };
 
-use std::{fmt, future::Future, mem, pin::Pin, str::FromStr, sync::Arc};
+use std::{
+    fmt,
+    future::Future,
+    mem,
+    pin::Pin,
+    str::FromStr,
+    sync::Arc,
+    time::{Duration, Instant},
+};
 
 use crate::{
     conn::{pool::Pool, stmt_cache::StmtCache},
@@ -28,7 +36,6 @@ use crate::{
     local_infile_handler::LocalInfileHandler,
     opts::Opts,
     queryable::{query_result, BinaryProtocol, Queryable, TextProtocol},
-    time::SteadyTime,
     Column, OptsBuilder,
 };
 
@@ -70,8 +77,8 @@ struct ConnInner {
     has_result: Option<(Arc<Vec<Column>>, Option<StmtCacheResult>)>,
     in_transaction: bool,
     opts: Opts,
-    last_io: SteadyTime,
-    wait_timeout: u32,
+    last_io: Instant,
+    wait_timeout: Duration,
     stmt_cache: StmtCache,
     nonce: Vec<u8>,
     auth_plugin: AuthPlugin<'static>,
@@ -110,8 +117,8 @@ impl ConnInner {
             has_result: None,
             pool: None,
             in_transaction: false,
-            last_io: SteadyTime::now(),
-            wait_timeout: 0,
+            last_io: Instant::now(),
+            wait_timeout: Duration::from_secs(0),
             stmt_cache: StmtCache::new(opts.get_stmt_cache_size()),
             socket: opts.get_socket().map(Into::into),
             opts,
@@ -371,7 +378,7 @@ impl Conn {
         Ok(conn)
     }
 
-    /// Returns future that resolves to `Conn`.
+    /// Returns future that resolves to [`Conn`].
     pub async fn new<T: Into<Opts>>(opts: T) -> Result<Conn> {
         let opts = opts.into();
         let mut conn = Conn::empty(opts.clone());
@@ -405,7 +412,7 @@ impl Conn {
             .await
     }
 
-    /// Returns future that resolves to `Conn`.
+    /// Returns future that resolves to [`Conn`].
     pub async fn from_url<T: AsRef<str>>(url: T) -> Result<Conn> {
         Conn::new(Opts::from_str(url.as_ref())?).await
     }
@@ -414,7 +421,7 @@ impl Conn {
     ///
     /// Returns new connection on success or self on error.
     ///
-    /// Won't try to reconnect if socket connection is already enforced in `Opts`.
+    /// Won't try to reconnect if socket connection is already enforced in [`Opts`].
     fn reconnect_via_socket_if_needed(self) -> Pin<Box<dyn Future<Output = Result<Conn>> + Send>> {
         // NOTE: we need to box this since it may recurse
         // see https://github.com/rust-lang/rust/issues/46415#issuecomment-528099782
@@ -434,9 +441,9 @@ impl Conn {
         })
     }
 
-    /// Returns future that resolves to `Conn` with socket address stored in it.
+    /// Returns future that resolves to [`Conn`] with socket address stored in it.
     ///
-    /// Do nothing if socket address is already in `Opts` or if `prefer_socket` is `false`.
+    /// Do nothing if socket address is already in [`Opts`] or if `prefer_socket` is `false`.
     async fn read_socket(self) -> Result<Self> {
         if self.inner.opts.get_prefer_socket() && self.inner.socket.is_none() {
             let (mut this, row_opt) = self.first("SELECT @@socket").await?;
@@ -447,7 +454,7 @@ impl Conn {
         }
     }
 
-    /// Returns future that resolves to `Conn` with `max_allowed_packet` stored in it.
+    /// Returns future that resolves to [`Conn`] with `max_allowed_packet` stored in it.
     async fn read_max_allowed_packet(self) -> Result<Self> {
         let (mut this, row_opt): (Self, _) = self.first("SELECT @@max_allowed_packet").await?;
         if let Some(stream) = this.inner.stream.as_mut() {
@@ -456,25 +463,30 @@ impl Conn {
         Ok(this)
     }
 
-    /// Returns future that resolves to `Conn` with `wait_timeout` stored in it.
+    /// Returns future that resolves to [`Conn`] with `wait_timeout` stored in it.
     async fn read_wait_timeout(self) -> Result<Self> {
         let (mut this, row_opt) = self.first("SELECT @@wait_timeout").await?;
-        this.inner.wait_timeout = row_opt.unwrap_or((28800,)).0;
+        let wait_timeout_secs = row_opt.unwrap_or((28800,)).0;
+        this.inner.wait_timeout = Duration::from_secs(wait_timeout_secs);
         Ok(this)
     }
 
     /// Returns true if time since last io exceeds wait_timeout (or conn_ttl if specified in opts).
     fn expired(&self) -> bool {
-        let idle_duration = SteadyTime::now() - self.inner.last_io;
         let ttl = self
             .inner
             .opts
             .get_conn_ttl()
             .unwrap_or(self.inner.wait_timeout);
-        idle_duration.num_milliseconds() > i64::from(ttl) * 1000
+        self.idling() > ttl
+    }
+
+    /// Returns duration since last io.
+    fn idling(&self) -> Duration {
+        self.inner.last_io.elapsed()
     }
 
-    /// Returns future that resolves to a `Conn` with `COM_RESET_CONNECTION` executed on it.
+    /// Returns future that resolves to a [`Conn`] with `COM_RESET_CONNECTION` executed on it.
     pub async fn reset(self) -> Result<Conn> {
         let pool = self.inner.pool.clone();
         let mut conn = if self.inner.version > (5, 7, 2) {
@@ -627,7 +639,7 @@ impl ConnectionLike for Conn {
     }
 
     fn touch(&mut self) {
-        self.inner.last_io = SteadyTime::now();
+        self.inner.last_io = Instant::now();
     }
 
     fn on_disconnect(&mut self) {
@@ -1177,30 +1189,30 @@ mod test {
 
     #[tokio::test]
     async fn should_handle_local_infile() -> super::Result<()> {
-        use std::io::Write;
+        use std::{io::Write, path::Path};
 
-        let tempdir = tempfile::TempDir::new().unwrap();
-        let file_path = tempdir.path().join("local_infile.txt");
+        let file_path = tempfile::Builder::new().tempfile_in("").unwrap();
+        let file_path = file_path.path();
+        let file_name = Path::new(file_path.file_name().unwrap());
 
         let mut opts = OptsBuilder::from_opts(get_opts());
-        opts.local_infile_handler(Some(WhiteListFsLocalInfileHandler::new(
-            &[file_path.as_path()][..],
-        )));
+        opts.local_infile_handler(Some(WhiteListFsLocalInfileHandler::new(&[file_name][..])));
 
-        let conn = Conn::new(opts).await?;
+        let conn = Conn::new(opts).await.unwrap();
         let conn = conn
             .drop_query("CREATE TEMPORARY TABLE tmp (a TEXT);")
-            .await?;
+            .await
+            .unwrap();
 
-        let mut file = ::std::fs::File::create(file_path.as_path()).unwrap();
+        let mut file = ::std::fs::File::create(file_name).unwrap();
         let _ = file.write(b"AAAAAA\n");
         let _ = file.write(b"BBBBBB\n");
         let _ = file.write(b"CCCCCC\n");
         let conn = match conn
-            .drop_query(format!(
-                "LOAD DATA LOCAL INFILE '{}' INTO TABLE tmp;",
-                file_path.as_path().display()
-            ))
+            .drop_query(dbg!(format!(
+                r#"LOAD DATA LOCAL INFILE "{}" INTO TABLE tmp;"#,
+                file_name.display()
+            )))
             .await
         {
             Ok(conn) => conn,
@@ -1212,9 +1224,11 @@ mod test {
         };
         let (conn, result) = conn
             .prep_exec("SELECT * FROM tmp;", ())
-            .await?
+            .await
+            .unwrap()
             .map_and_drop(|row| from_row::<(String,)>(row).0)
-            .await?;
+            .await
+            .unwrap();
         assert_eq!(result.len(), 3);
         assert_eq!(result[0], "AAAAAA");
         assert_eq!(result[1], "BBBBBB");
diff --git a/src/conn/pool/futures/disconnect_pool.rs b/src/conn/pool/futures/disconnect_pool.rs
index 4b5a4d1c..521bd811 100644
--- a/src/conn/pool/futures/disconnect_pool.rs
+++ b/src/conn/pool/futures/disconnect_pool.rs
@@ -35,7 +35,7 @@ impl Future for DisconnectPool {
     type Output = Result<(), Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.pool_inner.clone().spawn_recycler_if_needed();
+        self.pool_inner.clone().spawn_futures_if_needed();
         self.pool_inner.wake.push(cx.waker().clone());
 
         if self.pool_inner.closed.load(atomic::Ordering::Acquire) {
diff --git a/src/conn/pool/futures/get_conn.rs b/src/conn/pool/futures/get_conn.rs
index ffdfa9e3..dd1a7a82 100644
--- a/src/conn/pool/futures/get_conn.rs
+++ b/src/conn/pool/futures/get_conn.rs
@@ -31,7 +31,7 @@ impl GetConnInner {
     }
 }
 
-/// This future will take connection from a pool and resolve to `Conn`.
+/// This future will take connection from a pool and resolve to [`Conn`].
 pub struct GetConn {
     pub(crate) pool: Option<Pool>,
     pub(crate) inner: GetConnInner,
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index 57a380a4..8247cacb 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -6,212 +6,95 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures_core::{ready, stream::Stream};
-use futures_util::stream::futures_unordered::FuturesUnordered;
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll, Waker};
-
+use futures_core::ready;
 use tokio::sync::mpsc;
 
 use std::{
     fmt,
+    pin::Pin,
     str::FromStr,
     sync::{atomic, Arc, Mutex},
+    task::{Context, Poll, Waker},
+    time::{Duration, Instant},
 };
 
 use crate::{
     conn::{pool::futures::*, Conn},
     error::*,
-    opts::{Opts, PoolConstraints},
+    opts::{Opts, PoolOptions},
     queryable::{
         transaction::{Transaction, TransactionOptions},
         Queryable,
     },
-    BoxFuture,
 };
 
+mod recycler;
 // this is a really unfortunate name for a module
 pub mod futures;
-
-struct Recycler {
-    inner: Arc<Inner>,
-    discard: FuturesUnordered<BoxFuture<()>>,
-    discarded: usize,
-    cleaning: FuturesUnordered<BoxFuture<Conn>>,
-
-    // Option<Conn> so that we have a way to send a "I didn't make a Conn after all" signal
-    dropped: mpsc::UnboundedReceiver<Option<Conn>>,
-    min: usize,
-    eof: bool,
+mod ttl_check_inerval;
+
+/// Connection that is idling in the pool.
+#[derive(Debug)]
+struct IdlingConn {
+    /// The connection is idling since this `Instant`.
+    since: Instant,
+    /// Idling connection.
+    conn: Conn,
 }
 
-impl Future for Recycler {
-    type Output = ();
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut readied = 0;
-        let mut close = self.inner.close.load(atomic::Ordering::Acquire);
-
-        macro_rules! conn_decision {
-            ($self:ident, $readied:ident, $conn:ident) => {
-                if $conn.inner.stream.is_none() || $conn.inner.disconnected {
-                    // drop unestablished connection
-                    $self.discard.push(Box::pin(::futures_util::future::ok(())));
-                } else if $conn.inner.in_transaction || $conn.inner.has_result.is_some() {
-                    $self.cleaning.push(Box::pin($conn.cleanup()));
-                } else if $conn.expired() || $self.inner.idle.len() >= $self.min || close {
-                    $self.discard.push(Box::pin($conn.close()));
-                } else {
-                    $self
-                        .inner
-                        .idle
-                        .push($conn)
-                        .expect("more connections than max");
-                    $readied += 1;
-                }
-            };
-        }
-
-        while !self.eof {
-            // see if there are more connections for us to recycle
-            match Pin::new(&mut self.dropped).poll_next(cx) {
-                Poll::Ready(Some(Some(conn))) => {
-                    conn_decision!(self, readied, conn);
-                }
-                Poll::Ready(Some(None)) => {
-                    // someone signaled us that it's exit time
-                    close = self.inner.close.load(atomic::Ordering::Acquire);
-                    assert!(close);
-                    continue;
-                }
-                Poll::Ready(None) => {
-                    // no more connections are coming -- time to exit!
-                    self.inner.close.store(true, atomic::Ordering::Release);
-                    self.eof = true;
-                    close = true;
-                }
-                Poll::Pending => {
-                    // nope -- but let's still make progress on the ones we have
-                    break;
-                }
-            }
-        }
-
-        // if we've been asked to close, reclaim any idle connections
-        if close {
-            while let Ok(conn) = self.inner.idle.pop() {
-                conn_decision!(self, readied, conn);
-            }
-        }
-
-        // are any dirty connections ready for us to reclaim?
-        loop {
-            match Pin::new(&mut self.cleaning).poll_next(cx) {
-                Poll::Pending | Poll::Ready(None) => break,
-                Poll::Ready(Some(Ok(conn))) => conn_decision!(self, readied, conn),
-                Poll::Ready(Some(Err(e))) => {
-                    // an error occurred while cleaning a connection.
-                    // what do we do? replace it with a new connection?
-                    self.discarded += 1;
-                    // NOTE: we're discarding the error here
-                    let _ = e;
-                }
-            }
-        }
-
-        // are there any torn-down connections for us to deal with?
-        loop {
-            match Pin::new(&mut self.discard).poll_next(cx) {
-                Poll::Pending | Poll::Ready(None) => break,
-                Poll::Ready(Some(Ok(()))) => {
-                    // yes! count it.
-                    self.discarded += 1
-                }
-                Poll::Ready(Some(Err(e))) => {
-                    // an error occurred while closing a connection.
-                    // what do we do? we still replace it with a new connection..
-                    self.discarded += 1;
-                    // NOTE: we're discarding the error here
-                    let _ = e;
-                }
-            }
-        }
-
-        if self.discarded != 0 {
-            // we need to open up slots for new connctions to be established!
-            self.inner
-                .exist
-                .fetch_sub(self.discarded, atomic::Ordering::AcqRel);
-            readied += self.discarded;
-            self.discarded = 0;
-        }
-
-        // NOTE: we are asserting here that no more connections will ever be returned to
-        // us. see the explanation in Pool::poll_new_conn for why this is okay, even during
-        // races on .exist
-        let effectively_eof = close && self.inner.exist.load(atomic::Ordering::Acquire) == 0;
-
-        if (self.eof || effectively_eof) && self.cleaning.is_empty() && self.discard.is_empty() {
-            // we know that all Pool handles have been dropped (self.dropped.poll returned None).
-
-            // if this assertion fails, where are the remaining connections?
-            assert_eq!(self.inner.idle.len(), 0);
-            assert_eq!(self.inner.exist.load(atomic::Ordering::Acquire), 0);
+impl IdlingConn {
+    /// Returns duration elapsed since this connection is idling.
+    fn elapsed(&self) -> Duration {
+        self.since.elapsed()
+    }
+}
 
-            // NOTE: it is _necessary_ that we set this _before_ we call .wake
-            // otherwise, the following may happen to the DisconnectPool future:
-            //
-            //  - We wake all in .wake
-            //  - DisconnectPool::poll adds to .wake
-            //  - DisconnectPool::poll reads .closed == false
-            //  - We set .closed = true
-            //
-            // At this point, DisconnectPool::poll will never be notified again.
-            self.inner.closed.store(true, atomic::Ordering::Release);
-        }
-
-        self.inner.wake(readied);
-
-        if self.inner.closed.load(atomic::Ordering::Acquire) {
-            // `DisconnectPool` might still wait to be woken up.
-            self.inner.wake(usize::max_value());
-
-            Poll::Ready(())
-        } else {
-            Poll::Pending
+impl From<Conn> for IdlingConn {
+    fn from(conn: Conn) -> Self {
+        Self {
+            since: Instant::now(),
+            conn,
         }
     }
 }
 
-struct Inner {
+pub struct Inner {
     close: atomic::AtomicBool,
     closed: atomic::AtomicBool,
-    idle: crossbeam::queue::ArrayQueue<Conn>,
+    idle: crossbeam::queue::ArrayQueue<IdlingConn>,
     wake: crossbeam::queue::SegQueue<Waker>,
     exist: atomic::AtomicUsize,
     extra_wakeups: atomic::AtomicUsize,
 
     // only used to spawn the recycler the first time we're in async context
-    maker: Mutex<Option<(mpsc::UnboundedReceiver<Option<Conn>>, usize)>>,
+    maker: Mutex<Option<(mpsc::UnboundedReceiver<Option<Conn>>, PoolOptions)>>,
 }
 
 impl Inner {
-    fn spawn_recycler_if_needed(self: Arc<Self>) {
+    /// This function will spawn the recycler for this pool
+    /// as well as the ttl check interval if `inactive_connection_ttl` isn't `0`.
+    fn spawn_futures_if_needed(self: Arc<Self>) {
+        use recycler::Recycler;
+        use ttl_check_inerval::TtlCheckInterval;
+
         let mut lock = self.maker.lock().unwrap();
-        if let Some((dropped, min)) = lock.take() {
-            tokio::spawn(Recycler {
-                inner: self.clone(),
-                discard: FuturesUnordered::new(),
-                discarded: 0,
-                cleaning: FuturesUnordered::new(),
-                dropped,
-                min,
-                eof: false,
-            });
+        if let Some((dropped, pool_options)) = lock.take() {
+            // Spawn the Recycler.
+            tokio::spawn(Recycler::new(pool_options.clone(), self.clone(), dropped));
+
+            // Spawn the ttl check interval if `inactive_connection_ttl` isn't `0`
+            if pool_options.inactive_connection_ttl() > Duration::from_secs(0) {
+                tokio::spawn(TtlCheckInterval::new(pool_options, self.clone()));
+            }
         }
     }
 
+    fn push_to_idle<T: Into<IdlingConn>>(&self, conn: T) {
+        self.idle
+            .push(conn.into())
+            .expect("more connections than max")
+    }
+
     fn wake(&self, mut readied: usize) {
         if readied == 0 {
             return;
@@ -251,16 +134,12 @@ impl Inner {
 pub struct Pool {
     opts: Opts,
     inner: Arc<Inner>,
-    pool_constraints: PoolConstraints,
     drop: mpsc::UnboundedSender<Option<Conn>>,
 }
 
 impl fmt::Debug for Pool {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Pool")
-            .field("opts", &self.opts)
-            .field("pool_constraints", &self.pool_constraints)
-            .finish()
+        f.debug_struct("Pool").field("opts", &self.opts).finish()
     }
 }
 
@@ -268,21 +147,20 @@ impl Pool {
     /// Creates new pool of connections.
     pub fn new<O: Into<Opts>>(opts: O) -> Pool {
         let opts = opts.into();
-        let pool_constraints = opts.get_pool_constraints().clone();
+        let pool_options = opts.get_pool_options().clone();
         let (tx, rx) = mpsc::unbounded_channel();
         Pool {
             opts,
             inner: Arc::new(Inner {
                 close: false.into(),
                 closed: false.into(),
-                idle: crossbeam::queue::ArrayQueue::new(pool_constraints.max()),
+                idle: crossbeam::queue::ArrayQueue::new(pool_options.constraints().max()),
                 wake: crossbeam::queue::SegQueue::new(),
                 exist: 0.into(),
                 extra_wakeups: 0.into(),
-                maker: Mutex::new(Some((rx, pool_constraints.min()))),
+                maker: Mutex::new(Some((rx, pool_options))),
             }),
             drop: tx,
-            pool_constraints,
         }
     }
 
@@ -334,12 +212,9 @@ impl Pool {
             && !conn.inner.in_transaction
             && conn.inner.has_result.is_none()
             && !self.inner.close.load(atomic::Ordering::Acquire)
-            && self.inner.idle.len() < self.pool_constraints.min()
+            && self.inner.idle.len() < self.opts.get_pool_options().active_bound()
         {
-            self.inner
-                .idle
-                .push(conn)
-                .expect("more connections than max");
+            self.inner.push_to_idle(conn);
             self.inner.wake(1);
         } else {
             self.drop
@@ -375,7 +250,7 @@ impl Pool {
         loop {
             match self.inner.idle.pop() {
                 Err(crossbeam::queue::PopError) => break,
-                Ok(conn) => {
+                Ok(IdlingConn { conn, .. }) => {
                     if conn.expired() {
                         self.return_conn(conn);
                         continue;
@@ -392,15 +267,15 @@ impl Pool {
         // we didn't _immediately_ get one -- try to make one
         // we first try to just do a load so we don't do an unnecessary add then sub
         let exist = self.inner.exist.load(atomic::Ordering::Acquire);
-        if exist < self.pool_constraints.max() {
+        if exist < self.opts.get_pool_options().constraints().max() {
             // we may be allowed to make a new one!
             let exist = self.inner.exist.fetch_add(1, atomic::Ordering::AcqRel);
             if exist == 0 {
                 // we may have to start the recycler.
-                self.inner.clone().spawn_recycler_if_needed();
+                self.inner.clone().spawn_futures_if_needed();
             }
 
-            if exist < self.pool_constraints.max() {
+            if exist < self.opts.get_pool_options().constraints().max() {
                 // we're allowed to make a new connection
 
                 // note, however, that there is a race here:
@@ -429,7 +304,7 @@ impl Pool {
             }
 
             let exist = self.inner.exist.fetch_sub(1, atomic::Ordering::AcqRel);
-            if exist < self.pool_constraints.max() {
+            if exist < self.opts.get_pool_options().constraints().max() {
                 // we'd _now_ be allowed to make a connection
                 return self.poll_new_conn_inner(retrying, cx);
             }
@@ -519,11 +394,13 @@ impl Drop for Conn {
 #[cfg(test)]
 mod test {
     use futures_util::stream::StreamExt;
+
     use std::sync::atomic;
+    use std::time::Duration;
 
     use crate::{
-        conn::pool::Pool, queryable::Queryable, test_misc::get_opts, PoolConstraints,
-        TransactionOptions,
+        conn::pool::Pool, opts::PoolOptions, queryable::Queryable, test_misc::get_opts,
+        PoolConstraints, TransactionOptions,
     };
 
     #[test]
@@ -587,7 +464,9 @@ mod test {
     #[tokio::test]
     async fn should_start_transaction() -> super::Result<()> {
         let mut opts = get_opts();
-        opts.pool_constraints(PoolConstraints::new(1, 1));
+        opts.pool_options(PoolOptions::with_constraints(
+            PoolConstraints::new(1, 1).unwrap(),
+        ));
         let pool = Pool::new(opts);
         pool.get_conn()
             .await?
@@ -618,10 +497,17 @@ mod test {
 
         const POOL_MIN: usize = 5;
         const POOL_MAX: usize = 10;
+        const INACTIVE_CONNECTION_TTL: Duration = Duration::from_millis(500);
+        const TTL_CHECK_INTERVAL: Duration = Duration::from_secs(1);
+
+        let constraints = PoolConstraints::new(POOL_MIN, POOL_MAX).unwrap();
+        let pool_options =
+            PoolOptions::new(constraints, INACTIVE_CONNECTION_TTL, TTL_CHECK_INTERVAL);
 
         // Clean
         let mut opts = get_opts();
-        opts.pool_constraints(PoolConstraints::new(POOL_MIN, POOL_MAX));
+        opts.pool_options(pool_options);
+
         let pool = Pool::new(opts);
         let pool_clone = pool.clone();
         let conns = (0..POOL_MAX).map(|_| pool.get_conn()).collect::<Vec<_>>();
@@ -640,7 +526,7 @@ mod test {
             let _ = conns.pop();
 
             // then, wait for a bit to let the connection be reclaimed
-            tokio::timer::delay(std::time::Instant::now() + std::time::Duration::from_millis(100))
+            tokio::timer::delay(std::time::Instant::now() + std::time::Duration::from_millis(50))
                 .await;
 
             // now check that we have the expected # of connections
@@ -656,6 +542,18 @@ mod test {
             let dropped = POOL_MAX - conns.len();
             let idle = min(dropped, POOL_MIN);
             let expected = conns.len() + idle;
+
+            if dropped > POOL_MIN {
+                // check that connection is still in the pool because of inactive_connection_ttl
+                let have = pool_clone.inner.exist.load(atomic::Ordering::SeqCst);
+                assert_eq!(have, expected + 1);
+
+                // then, wait for ttl_check_interval
+                tokio::timer::delay(std::time::Instant::now() + std::time::Duration::from_secs(1))
+                    .await;
+            }
+
+            // check that we have the expected number of connections
             let have = pool_clone.inner.exist.load(atomic::Ordering::SeqCst);
             assert_eq!(have, expected);
         }
@@ -667,7 +565,9 @@ mod test {
     #[tokio::test]
     async fn should_hold_bounds1() -> super::Result<()> {
         let mut opts = get_opts();
-        opts.pool_constraints(PoolConstraints::new(1, 2));
+        opts.pool_options(PoolOptions::with_constraints(
+            PoolConstraints::new(1, 2).unwrap(),
+        ));
         let pool = Pool::new(opts);
         let pool_clone = pool.clone();
 
@@ -781,21 +681,42 @@ mod test {
 
     #[cfg(feature = "nightly")]
     mod bench {
+        use futures_util::future::FutureExt;
         use futures_util::try_future::TryFutureExt;
         use tokio::runtime::Runtime;
 
         use crate::prelude::Queryable;
         use crate::test_misc::get_opts;
-        use crate::Pool;
+        use crate::{Pool, PoolConstraints, PoolOptions};
+        use std::time::Duration;
 
         #[bench]
         fn connect(bencher: &mut test::Bencher) {
-            let runtime = Runtime::new().expect("3");
+            let runtime = Runtime::new().unwrap();
             let pool = Pool::new(get_opts());
 
             bencher.iter(|| {
                 let fut = pool.get_conn().and_then(|conn| conn.ping());
-                runtime.block_on(fut).expect("1");
+                runtime.block_on(fut).unwrap();
+            });
+
+            runtime.block_on(pool.disconnect()).unwrap();
+        }
+
+        #[bench]
+        fn new_conn_on_pool_soft_boundary(bencher: &mut test::Bencher) {
+            let runtime = Runtime::new().unwrap();
+
+            let mut opts = get_opts();
+            let mut pool_opts = PoolOptions::with_constraints(PoolConstraints::new(0, 1).unwrap());
+            pool_opts.set_inactive_connection_ttl(Duration::from_secs(1));
+            opts.pool_options(pool_opts);
+
+            let pool = Pool::new(opts);
+
+            bencher.iter(|| {
+                let fut = pool.get_conn().map(drop);
+                runtime.block_on(fut);
             });
 
             runtime.block_on(pool.disconnect()).unwrap();
diff --git a/src/conn/pool/recycler.rs b/src/conn/pool/recycler.rs
new file mode 100644
index 00000000..3f136ad8
--- /dev/null
+++ b/src/conn/pool/recycler.rs
@@ -0,0 +1,187 @@
+// Copyright (c) 2019 mysql_async contributors
+//
+// Licensed under the Apache License, Version 2.0
+// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
+// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. All files in the project carrying such notice may not be copied,
+// modified, or distributed except according to those terms.
+
+use futures_core::stream::Stream;
+use futures_util::stream::futures_unordered::FuturesUnordered;
+use tokio::sync::mpsc;
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use super::{IdlingConn, Inner};
+use crate::{BoxFuture, Conn, PoolOptions};
+use tokio::sync::mpsc::UnboundedReceiver;
+
+pub struct Recycler {
+    inner: Arc<Inner>,
+    discard: FuturesUnordered<BoxFuture<()>>,
+    discarded: usize,
+    cleaning: FuturesUnordered<BoxFuture<Conn>>,
+
+    // Option<Conn> so that we have a way to send a "I didn't make a Conn after all" signal
+    dropped: mpsc::UnboundedReceiver<Option<Conn>>,
+    /// Pool options.
+    pool_opts: PoolOptions,
+    eof: bool,
+}
+
+impl Recycler {
+    pub fn new(
+        pool_opts: PoolOptions,
+        inner: Arc<Inner>,
+        dropped: UnboundedReceiver<Option<Conn>>,
+    ) -> Self {
+        Self {
+            inner,
+            discard: FuturesUnordered::new(),
+            discarded: 0,
+            cleaning: FuturesUnordered::new(),
+            dropped,
+            pool_opts,
+            eof: false,
+        }
+    }
+}
+
+impl Future for Recycler {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut readied = 0;
+        let mut close = self.inner.close.load(Ordering::Acquire);
+
+        macro_rules! conn_decision {
+            ($self:ident, $readied:ident, $conn:ident) => {
+                if $conn.inner.stream.is_none() || $conn.inner.disconnected {
+                    // drop unestablished connection
+                    $self.discard.push(Box::pin(::futures_util::future::ok(())));
+                } else if $conn.inner.in_transaction || $conn.inner.has_result.is_some() {
+                    $self.cleaning.push(Box::pin($conn.cleanup()));
+                } else if $conn.expired()
+                    || close
+                    || $self.inner.idle.len() >= $self.pool_opts.active_bound()
+                {
+                    $self.discard.push(Box::pin($conn.close()));
+                } else {
+                    $self.inner.push_to_idle($conn);
+                    $readied += 1;
+                }
+            };
+        }
+
+        while !self.eof {
+            // see if there are more connections for us to recycle
+            match Pin::new(&mut self.dropped).poll_next(cx) {
+                Poll::Ready(Some(Some(conn))) => {
+                    conn_decision!(self, readied, conn);
+                }
+                Poll::Ready(Some(None)) => {
+                    // someone signaled us that it's exit time
+                    close = self.inner.close.load(Ordering::Acquire);
+                    assert!(close);
+                    continue;
+                }
+                Poll::Ready(None) => {
+                    // no more connections are coming -- time to exit!
+                    self.inner.close.store(true, Ordering::Release);
+                    self.eof = true;
+                    close = true;
+                }
+                Poll::Pending => {
+                    // nope -- but let's still make progress on the ones we have
+                    break;
+                }
+            }
+        }
+
+        // if we've been asked to close, reclaim any idle connections
+        if close {
+            while let Ok(IdlingConn { conn, .. }) = self.inner.idle.pop() {
+                conn_decision!(self, readied, conn);
+            }
+        }
+
+        // are any dirty connections ready for us to reclaim?
+        loop {
+            match Pin::new(&mut self.cleaning).poll_next(cx) {
+                Poll::Pending | Poll::Ready(None) => break,
+                Poll::Ready(Some(Ok(conn))) => conn_decision!(self, readied, conn),
+                Poll::Ready(Some(Err(e))) => {
+                    // an error occurred while cleaning a connection.
+                    // what do we do? replace it with a new connection?
+                    self.discarded += 1;
+                    // NOTE: we're discarding the error here
+                    let _ = e;
+                }
+            }
+        }
+
+        // are there any torn-down connections for us to deal with?
+        loop {
+            match Pin::new(&mut self.discard).poll_next(cx) {
+                Poll::Pending | Poll::Ready(None) => break,
+                Poll::Ready(Some(Ok(()))) => {
+                    // yes! count it.
+                    self.discarded += 1
+                }
+                Poll::Ready(Some(Err(e))) => {
+                    // an error occurred while closing a connection.
+                    // what do we do? we still replace it with a new connection..
+                    self.discarded += 1;
+                    // NOTE: we're discarding the error here
+                    let _ = e;
+                }
+            }
+        }
+
+        if self.discarded != 0 {
+            // we need to open up slots for new connctions to be established!
+            self.inner.exist.fetch_sub(self.discarded, Ordering::AcqRel);
+            readied += self.discarded;
+            self.discarded = 0;
+        }
+
+        // NOTE: we are asserting here that no more connections will ever be returned to
+        // us. see the explanation in Pool::poll_new_conn for why this is okay, even during
+        // races on .exist
+        let effectively_eof = close && self.inner.exist.load(Ordering::Acquire) == 0;
+
+        if (self.eof || effectively_eof) && self.cleaning.is_empty() && self.discard.is_empty() {
+            // we know that all Pool handles have been dropped (self.dropped.poll returned None).
+
+            // if this assertion fails, where are the remaining connections?
+            assert_eq!(self.inner.idle.len(), 0);
+            assert_eq!(self.inner.exist.load(Ordering::Acquire), 0);
+
+            // NOTE: it is _necessary_ that we set this _before_ we call .wake
+            // otherwise, the following may happen to the DisconnectPool future:
+            //
+            //  - We wake all in .wake
+            //  - DisconnectPool::poll adds to .wake
+            //  - DisconnectPool::poll reads .closed == false
+            //  - We set .closed = true
+            //
+            // At this point, DisconnectPool::poll will never be notified again.
+            self.inner.closed.store(true, Ordering::Release);
+        }
+
+        self.inner.wake(readied);
+
+        if self.inner.closed.load(Ordering::Acquire) {
+            // `DisconnectPool` might still wait to be woken up.
+            self.inner.wake(usize::max_value());
+
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+}
diff --git a/src/conn/pool/ttl_check_inerval.rs b/src/conn/pool/ttl_check_inerval.rs
new file mode 100644
index 00000000..5571df13
--- /dev/null
+++ b/src/conn/pool/ttl_check_inerval.rs
@@ -0,0 +1,96 @@
+// Copyright (c) 2019 Anatoly Ikorsky
+//
+// Licensed under the Apache License, Version 2.0
+// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
+// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. All files in the project carrying such notice may not be copied,
+// modified, or distributed except according to those terms.
+
+use futures_util::future::FutureExt;
+use futures_util::stream::{StreamExt, StreamFuture};
+use pin_project::pin_project;
+use tokio::timer::Interval;
+
+use std::future::Future;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+use super::Inner;
+use crate::prelude::Queryable;
+use crate::PoolOptions;
+use futures_core::task::Context;
+use futures_core::Poll;
+use std::pin::Pin;
+
+/// Idling connections TTL check interval.
+///
+/// The purpose of this interval is to remove idling connections that both:
+/// * overflows min bound of the pool;
+/// * idles longer then `inactive_connection_ttl`.
+#[pin_project]
+pub struct TtlCheckInterval {
+    inner: Arc<Inner>,
+    #[pin]
+    interval: StreamFuture<Interval>,
+    pool_options: PoolOptions,
+}
+
+impl TtlCheckInterval {
+    /// Creates new `TtlCheckInterval`.
+    pub fn new(pool_options: PoolOptions, inner: Arc<Inner>) -> Self {
+        let interval = Interval::new_interval(pool_options.ttl_check_interval()).into_future();
+        Self {
+            inner,
+            interval,
+            pool_options,
+        }
+    }
+
+    /// Perform the check.
+    pub fn check_ttl(&self) {
+        let num_idling = self.inner.idle.len();
+        let mut num_to_drop = num_idling.saturating_sub(self.pool_options.constraints().min());
+        let mut num_returned = 0;
+
+        for _ in 0..num_idling {
+            match self.inner.idle.pop() {
+                Ok(idling_conn) => {
+                    if idling_conn.elapsed() > self.pool_options.inactive_connection_ttl() {
+                        tokio::spawn(idling_conn.conn.disconnect().map(drop));
+                        self.inner.exist.fetch_sub(1, Ordering::AcqRel);
+                        num_to_drop -= 1;
+                        if num_to_drop == 0 {
+                            break;
+                        }
+                    } else {
+                        self.inner.push_to_idle(idling_conn);
+                        num_returned += 1;
+                    }
+                }
+                Err(_) => break,
+            }
+        }
+
+        if num_returned > 0 {
+            self.inner.wake(num_returned);
+        }
+    }
+}
+
+impl Future for TtlCheckInterval {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        loop {
+            let (_, interval) = futures_core::ready!(self.as_mut().project().interval.poll(cx));
+            let close = self.inner.close.load(Ordering::Acquire);
+
+            if !close {
+                self.check_ttl();
+                self.interval = interval.into_future();
+            } else {
+                return Poll::Ready(());
+            }
+        }
+    }
+}
diff --git a/src/io/futures/write_packet.rs b/src/io/futures/write_packet.rs
index 61041c88..e3ac8adb 100644
--- a/src/io/futures/write_packet.rs
+++ b/src/io/futures/write_packet.rs
@@ -14,7 +14,7 @@ use std::task::{Context, Poll};
 
 use crate::{error::*, io::Stream};
 
-/// Future that writes packet to a `Stream` and resolves to a pair of `Stream` and MySql's sequence
+/// Future that writes packet to a [`Stream`] and resolves to a pair of [`Stream`] and MySql's sequence
 /// id.
 pub struct WritePacket {
     data: Option<Vec<u8>>,
diff --git a/src/lib.rs b/src/lib.rs
index 9debb02f..7e3e2047 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -134,7 +134,10 @@ pub use self::conn::pool::Pool;
 pub use self::queryable::transaction::IsolationLevel;
 
 #[doc(inline)]
-pub use self::opts::{Opts, OptsBuilder, PoolConstraints, SslOpts};
+pub use self::opts::{
+    Opts, OptsBuilder, PoolConstraints, PoolOptions, SslOpts, DEFAULT_INACTIVE_CONNECTION_TTL,
+    DEFAULT_TTL_CHECK_INTERVAL,
+};
 
 #[doc(inline)]
 pub use self::local_infile_handler::{builtin::WhiteListFsLocalInfileHandler, InfileHandlerFuture};
@@ -195,7 +198,7 @@ pub mod prelude {
     pub trait ConnectionLike: crate::connection_like::ConnectionLike {}
     impl<T: crate::connection_like::ConnectionLike> ConnectionLike for T {}
 
-    /// Trait for protocol markers [`TextProtocol`] and [`BinaryProtocol`].
+    /// Trait for protocol markers [`crate::TextProtocol`] and [`crate::BinaryProtocol`].
     pub trait Protocol: crate::queryable::Protocol {}
     impl<T: crate::queryable::Protocol> Protocol for T {}
 
diff --git a/src/local_infile_handler/mod.rs b/src/local_infile_handler/mod.rs
index 0b418d0e..5fc4c991 100644
--- a/src/local_infile_handler/mod.rs
+++ b/src/local_infile_handler/mod.rs
@@ -19,7 +19,7 @@ pub mod builtin;
 /// Trait used to handle local infile requests.
 ///
 /// Be aware of security issues with [LOAD DATA LOCAL](https://dev.mysql.com/doc/refman/8.0/en/load-data-local.html).
-/// Using [`WhiteListFsLocalInfileHandler`] is advised.
+/// Using [`crate::WhiteListFsLocalInfileHandler`] is advised.
 ///
 /// Simple handler example:
 ///
@@ -76,7 +76,6 @@ pub mod builtin;
 /// # }
 /// ```
 ///
-/// [`WhiteListFsLocalInfileHandler`]: ../struct.WhiteListFsLocalInfileHandler.html
 pub trait LocalInfileHandler: Sync + Send {
     /// `file_name` is the file name in `LOAD DATA LOCAL INFILE '<file name>' INTO TABLE ...;`
     /// query.
diff --git a/src/opts.rs b/src/opts.rs
index 41f92eb7..1370d209 100644
--- a/src/opts.rs
+++ b/src/opts.rs
@@ -15,6 +15,7 @@ use std::{
     path::Path,
     str::FromStr,
     sync::Arc,
+    time::Duration,
 };
 
 use crate::{
@@ -30,6 +31,17 @@ const_assert!(
 );
 const DEFAULT_STMT_CACHE_SIZE: usize = 10;
 
+/// Default `inactive_connection_ttl` of a pool.
+///
+/// `0` value means, that connection will be dropped immediately
+/// if it is outside of the pool's lower bound.
+pub const DEFAULT_INACTIVE_CONNECTION_TTL: Duration = Duration::from_secs(0);
+
+/// Default `ttl_check_interval` of a pool.
+///
+/// It isn't used if `inactive_connection_ttl` is `0`.
+pub const DEFAULT_TTL_CHECK_INTERVAL: Duration = Duration::from_secs(30);
+
 /// Ssl Options.
 #[derive(Debug, Clone, Eq, PartialEq, Hash, Default)]
 pub struct SslOpts {
@@ -100,9 +112,111 @@ impl SslOpts {
     }
 }
 
+/// Connection pool options.
+#[derive(Debug, Clone, Eq, PartialEq, Hash)]
+pub struct PoolOptions {
+    constraints: PoolConstraints,
+    inactive_connection_ttl: Duration,
+    ttl_check_interval: Duration,
+}
+
+impl PoolOptions {
+    /// Creates [`PoolOptions`].
+    pub const fn new(
+        constraints: PoolConstraints,
+        inactive_connection_ttl: Duration,
+        ttl_check_interval: Duration,
+    ) -> Self {
+        Self {
+            constraints,
+            inactive_connection_ttl,
+            ttl_check_interval,
+        }
+    }
+
+    /// Creates default [`PoolOptions`] with given constraints.
+    pub const fn with_constraints(constraints: PoolConstraints) -> Self {
+        Self {
+            constraints,
+            inactive_connection_ttl: DEFAULT_INACTIVE_CONNECTION_TTL,
+            ttl_check_interval: DEFAULT_TTL_CHECK_INTERVAL,
+        }
+    }
+
+    /// Sets pool constraints.
+    pub fn set_constraints(&mut self, constraints: PoolConstraints) {
+        self.constraints = constraints;
+    }
+
+    /// Returns `constrains` value.
+    pub fn constraints(&self) -> PoolConstraints {
+        self.constraints
+    }
+
+    /// Pool will recycle inactive connection if it outside of the lower bound of a pool
+    /// and if it is idling longer than this value (defaults to [`DEFAULT_INACTIVE_CONNECTION_TTL`]).
+    ///
+    /// Note that it may, actually, idle longer because of [`PoolOptions::ttl_check_interval`].
+    pub fn set_inactive_connection_ttl(&mut self, ttl: Duration) {
+        self.inactive_connection_ttl = ttl;
+    }
+
+    /// Returns `inactive_connection_ttl` value.
+    pub fn inactive_connection_ttl(&self) -> Duration {
+        self.inactive_connection_ttl
+    }
+
+    /// Pool will check idling connection for expiration with this interval
+    /// (defaults to [`DEFAULT_TTL_CHECK_INTERVAL`]).
+    ///
+    /// If `interval` is less than one second, then [`DEFAULT_TTL_CHECK_INTERVAL`] will be used.
+    pub fn set_ttl_check_interval(&mut self, interval: Duration) {
+        if interval < Duration::from_secs(1) {
+            self.ttl_check_interval = DEFAULT_TTL_CHECK_INTERVAL
+        } else {
+            self.ttl_check_interval = interval;
+        }
+    }
+
+    /// Returns `ttl_check_interval` value.
+    pub fn ttl_check_interval(&self) -> Duration {
+        self.ttl_check_interval
+    }
+
+    /// Returns active bound for this `PoolOptions`.
+    ///
+    /// This value controls how many connections will be returned to an idle queue of a pool.
+    ///
+    /// Active bound is either:
+    /// * `min` bound of the pool constraints, if this [`PoolOptions`] defines
+    ///   `inactive_connection_ttl` to be `0`. This means, that pool will hold no more than `min`
+    ///   number of idling connection and other connection will be immediately disconnected.
+    /// * `max` bound of the pool constraints, if this [`PoolOptions`] defines
+    ///   `inactive_connection_ttl` to be non-zero. This means, that pool will hold up to `max`
+    ///   number of idling connections and this number will be eventually reduced to `min`
+    ///   by a handler of `ttl_check_interval`.
+    pub(crate) fn active_bound(&self) -> usize {
+        if self.inactive_connection_ttl > Duration::from_secs(0) {
+            self.constraints.max
+        } else {
+            self.constraints.min
+        }
+    }
+}
+
+impl Default for PoolOptions {
+    fn default() -> Self {
+        Self {
+            constraints: DEFAULT_POOL_CONSTRAINTS,
+            inactive_connection_ttl: DEFAULT_INACTIVE_CONNECTION_TTL,
+            ttl_check_interval: DEFAULT_TTL_CHECK_INTERVAL,
+        }
+    }
+}
+
 /// Mysql connection options.
 ///
-/// Build one with [`OptsBuilder`](struct.OptsBuilder.html).
+/// Build one with [`OptsBuilder`].
 #[derive(Clone, Eq, PartialEq, Debug)]
 pub struct InnerOpts {
     /// Address of mysql server (defaults to `127.0.0.1`). Host names should also work.
@@ -132,12 +246,12 @@ pub struct InnerOpts {
     /// Local infile handler
     local_infile_handler: Option<LocalInfileHandlerObject>,
 
-    /// Bounds for the number of opened connections in `Pool` (defaults to `min: 10, max: 100`).
-    pool_constraints: PoolConstraints,
+    /// Connection pool options (defaults to [`PoolOptions::default`]).
+    pool_options: PoolOptions,
 
-    /// Pool will close connection if time since last IO exceeds this value
+    /// Pool will close connection if time since last IO exceeds this number of seconds
     /// (defaults to `wait_timeout`).
-    conn_ttl: Option<u32>,
+    conn_ttl: Option<Duration>,
 
     /// Commands to execute on each new database connection.
     init: Vec<String>,
@@ -179,7 +293,7 @@ pub struct InnerOpts {
 
 /// Mysql connection options.
 ///
-/// Build one with [`OptsBuilder`](struct.OptsBuilder.html).
+/// Build one with [`OptsBuilder`].
 #[derive(Clone, Eq, PartialEq, Debug, Default)]
 pub struct Opts {
     inner: Arc<InnerOpts>,
@@ -253,14 +367,14 @@ impl Opts {
             .map(|x| x.clone_inner())
     }
 
-    /// /// Bounds for the number of opened connections in `Pool` (defaults to `min: 10, max: 100`).
-    pub fn get_pool_constraints(&self) -> &PoolConstraints {
-        &self.inner.pool_constraints
+    /// Connection pool options (defaults to [`Default::default`]).
+    pub fn get_pool_options(&self) -> &PoolOptions {
+        &self.inner.pool_options
     }
 
-    /// Pool will close connection if time since last IO exceeds this value
+    /// Pool will close connection if time since last IO exceeds this number of seconds
     /// (defaults to `wait_timeout`).
-    pub fn get_conn_ttl(&self) -> Option<u32> {
+    pub fn get_conn_ttl(&self) -> Option<Duration> {
         self.inner.conn_ttl
     }
 
@@ -351,7 +465,7 @@ impl Default for InnerOpts {
             tcp_keepalive: None,
             tcp_nodelay: true,
             local_infile_handler: None,
-            pool_constraints: Default::default(),
+            pool_options: Default::default(),
             conn_ttl: None,
             stmt_cache_size: DEFAULT_STMT_CACHE_SIZE,
             ssl_opts: None,
@@ -364,15 +478,17 @@ impl Default for InnerOpts {
 
 /// Connection pool constraints.
 ///
-/// This type stores `min` and `max` constraints for `Pool` and ensures that `min <= max`.
-#[derive(Debug, Clone, Eq, PartialEq, Hash)]
+/// This type stores `min` and `max` constraints for [`crate::Pool`] and ensures that `min <= max`.
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
 pub struct PoolConstraints {
     min: usize,
     max: usize,
 }
 
 impl PoolConstraints {
-    /// Creates new `PoolConstraints` if constraints are valid (`min <= max`).
+    /// Creates new [`PoolConstraints`] if constraints are valid (`min <= max`).
+    ///
+    /// `inactive_connection_ttl` will have the default value.
     pub fn new(min: usize, max: usize) -> Option<PoolConstraints> {
         if min <= max {
             Some(PoolConstraints { min, max })
@@ -405,7 +521,7 @@ impl From<PoolConstraints> for (usize, usize) {
     }
 }
 
-/// Provides a way to build [`Opts`](struct.Opts.html).
+/// Provides a way to build [`Opts`].
 ///
 /// ```ignore
 /// // You can create new default builder
@@ -495,16 +611,16 @@ impl OptsBuilder {
         self
     }
 
-    /// Pool constraints. (defaults to `min: 10, max: 100`).
-    pub fn pool_constraints(&mut self, pool_constraints: Option<PoolConstraints>) -> &mut Self {
-        self.opts.pool_constraints = pool_constraints.unwrap_or(DEFAULT_POOL_CONSTRAINTS);
+    /// Connection pool options (defaults to `PoolOptions::default()`).
+    pub fn pool_options<T: Into<Option<PoolOptions>>>(&mut self, pool_options: T) -> &mut Self {
+        self.opts.pool_options = pool_options.into().unwrap_or_default();
         self
     }
 
-    /// Pool will close connection if time since last IO exceeds this value
+    /// Pool will close connection if time since last IO exceeds this number of seconds
     /// (defaults to `wait_timeout`. `None` to reset to default).
-    pub fn conn_ttl<T: Into<u32>>(&mut self, conn_ttl: Option<T>) -> &mut Self {
-        self.opts.conn_ttl = conn_ttl.map(Into::into);
+    pub fn conn_ttl<T: Into<Option<Duration>>>(&mut self, conn_ttl: T) -> &mut Self {
+        self.opts.conn_ttl = conn_ttl.into();
         self
     }
 
@@ -670,9 +786,33 @@ fn from_url(url: &str) -> std::result::Result<InnerOpts, UrlError> {
                     });
                 }
             }
+        } else if key == "inactive_connection_ttl" {
+            match u64::from_str(&*value) {
+                Ok(value) => opts
+                    .pool_options
+                    .set_inactive_connection_ttl(Duration::from_secs(value)),
+                _ => {
+                    return Err(UrlError::InvalidParamValue {
+                        param: "inactive_connection_ttl".into(),
+                        value,
+                    });
+                }
+            }
+        } else if key == "ttl_check_interval" {
+            match u64::from_str(&*value) {
+                Ok(value) => opts
+                    .pool_options
+                    .set_ttl_check_interval(Duration::from_secs(value)),
+                _ => {
+                    return Err(UrlError::InvalidParamValue {
+                        param: "ttl_check_interval".into(),
+                        value,
+                    });
+                }
+            }
         } else if key == "conn_ttl" {
             match u32::from_str(&*value) {
-                Ok(value) => opts.conn_ttl = Some(value),
+                Ok(value) => opts.conn_ttl = Some(Duration::from_secs(value as u64)),
                 _ => {
                     return Err(UrlError::InvalidParamValue {
                         param: "conn_ttl".into(),
@@ -749,7 +889,7 @@ fn from_url(url: &str) -> std::result::Result<InnerOpts, UrlError> {
     }
 
     if let Some(pool_constraints) = PoolConstraints::new(pool_min, pool_max) {
-        opts.pool_constraints = pool_constraints;
+        opts.pool_options.set_constraints(pool_constraints);
     } else {
         return Err(UrlError::InvalidPoolConstraints {
             min: pool_min,
diff --git a/src/queryable/query_result/mod.rs b/src/queryable/query_result/mod.rs
index 678661bd..90cf3739 100644
--- a/src/queryable/query_result/mod.rs
+++ b/src/queryable/query_result/mod.rs
@@ -214,7 +214,7 @@ where
     /// # Panic
     ///
     /// It'll panic if any row isn't convertible to `R` (i.e. programmer error or unknown schema).
-    /// * In case of programmer error see `FromRow` docs;
+    /// * In case of programmer error see [`FromRow`] docs;
     /// * In case of unknown schema use [`QueryResult::try_collect`].
     pub async fn collect<R>(self) -> Result<(Self, Vec<R>)>
     where
@@ -245,7 +245,7 @@ where
     }
 
     /// Returns future that collects result set of a query result and drops everything else.
-    /// It will resolve to a pair of wrapped `Queryable` and collected result set.
+    /// It will resolve to a pair of wrapped [`crate::prelude::Queryable`] and collected result set.
     ///
     /// # Panic
     ///
@@ -263,7 +263,7 @@ where
     }
 
     /// Returns future that collects result set of a query result and drops everything else.
-    /// It will resolve to a pair of wrapped `Queryable` and collected result set.
+    /// It will resolve to a pair of wrapped [`crate::prelude::Queryable`] and collected result set.
     ///
     /// It works the same way as [`QueryResult::collect_and_drop`] but won't panic
     /// if row isn't convertible to `R`.


Test patch:
diff --git a/tests/exports.rs b/tests/exports.rs
index d89899aa..81980ed5 100644
--- a/tests/exports.rs
+++ b/tests/exports.rs
@@ -8,7 +8,8 @@ use mysql_async::{
         ToValue,
     },
     time, uuid, BinaryProtocol, BoxFuture, Column, Conn, Deserialized, FromRowError,
-    FromValueError, IsolationLevel, Opts, OptsBuilder, Params, Pool, PoolConstraints, QueryResult,
-    Row, Serialized, SslOpts, Stmt, TextProtocol, Transaction, TransactionOptions, Value,
-    WhiteListFsLocalInfileHandler,
+    FromValueError, IsolationLevel, Opts, OptsBuilder, Params, Pool, PoolConstraints, PoolOptions,
+    QueryResult, Row, Serialized, SslOpts, Stmt, TextProtocol, Transaction, TransactionOptions,
+    Value, WhiteListFsLocalInfileHandler, DEFAULT_INACTIVE_CONNECTION_TTL,
+    DEFAULT_TTL_CHECK_INTERVAL,
 };