Repository: blackbeam/mysql_async (language: rust)

Issue:
Pool::get_conn hangs under contention
The following (new) test case occasionally hangs for me:
```rust
#[test]
fn can_handle_the_pressure() {
    let mut runtime = tokio::runtime::Runtime::new().unwrap();
    let pool = Pool::new(&**DATABASE_URL);
    for _ in 0..100 {
        use futures::{Sink, Stream};
        let (tx, rx) = futures::sync::mpsc::unbounded();
        for i in 0..10_000 {
            let pool = pool.clone();
            let tx = tx.clone();
            runtime.spawn(futures::future::lazy(move || {
                pool.get_conn()
                    .map_err(|e| unreachable!("{:?}", e))
                    .and_then(move |_| {
                        tx.send(i).map_err(|e| unreachable!("{:?}", e))
                    }).map(|_| ())
            }));
        }
        drop(tx);
        runtime.block_on(rx.fold(0, |_, _i| {
            Ok(0)
        })).unwrap();
    }
    drop(pool);
    runtime.shutdown_on_idle().wait().unwrap();
}
```
It specifically tries to set up contention over the `Pool` (notice that it doesn't actually issue any queries, it just repeatedly "gets" connections and then return them immediately), and it seems like occasionally it hits a deadlock. This _may_ be related to #64, but I'm not sure.


Discussion / hints:
Interestingly enough, I _still_ see this with a lot of concurrent connection churn, even after #66. I might take a stab at greatly simplifying the connection pooling tomorrow using an `mpsc` channel + an async mutex around the receiver. Or maybe even _just_ an async mutex with a `VecDeque`. It'll be worse scalability wise, but my guess is that you will never really notice in any real workload since most of the work will be happening _on_ the connection, and the overhead of _getting_ the connection will be negligible.

Thoughts @blackbeam ?
Hi!

Well, i have nothing against it, especially if it'll lead to a more stable behavior. Obviously, scalability worth nothing if pool simply hangs.

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 06865499..2cc078c0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -25,7 +25,7 @@ pin-project = "0.4.6"
 serde = "1"
 serde_json = "1"
 thiserror = "1.0.4"
-tokio = { version = "0.2", features = ["io-util", "net", "sync", "fs", "rt-core", "time", "stream", "macros"] }
+tokio = { version = "0.2.10", features = ["io-util", "net", "sync", "fs", "rt-core", "time", "stream", "macros"] }
 tokio-util = { version = "0.2", features = ["codec"] }
 tokio-tls = "0.3"
 twox-hash = "1"
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
index ae372e38..a86c35cc 100644
--- a/src/conn/mod.rs
+++ b/src/conn/mod.rs
@@ -57,11 +57,13 @@ fn disconnect(mut conn: Conn) {
 
         // Server will report broken connection if spawn fails.
         // this might fail if, say, the runtime is shutting down, but we've done what we could
-        tokio::spawn(async move {
-            if let Ok(conn) = conn.cleanup().await {
-                let _ = conn.disconnect().await;
-            }
-        });
+        if let Ok(handle) = tokio::runtime::Handle::try_current() {
+            handle.spawn(async move {
+                if let Ok(conn) = conn.cleanup().await {
+                    let _ = conn.disconnect().await;
+                }
+            });
+        }
     }
 }
 
@@ -1239,6 +1241,11 @@ mod test {
                 // The used command is not allowed with this MySQL version
                 return Ok(());
             }
+            Err(super::Error::Server(ref err)) if err.code == 3948 => {
+                // Loading local data is disabled;
+                // this must be enabled on both the client and server sides
+                return Ok(());
+            }
             e @ Err(_) => e.unwrap(),
         };
         let (conn, result) = conn
diff --git a/src/conn/pool/futures/disconnect_pool.rs b/src/conn/pool/futures/disconnect_pool.rs
index 521bd811..67d82612 100644
--- a/src/conn/pool/futures/disconnect_pool.rs
+++ b/src/conn/pool/futures/disconnect_pool.rs
@@ -35,8 +35,10 @@ 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_futures_if_needed();
-        self.pool_inner.wake.push(cx.waker().clone());
+        let mut exchange = self.pool_inner.exchange.lock().unwrap();
+        exchange.spawn_futures_if_needed(&self.pool_inner);
+        exchange.waiting.push_back(cx.waker().clone());
+        drop(exchange);
 
         if self.pool_inner.closed.load(atomic::Ordering::Acquire) {
             Poll::Ready(Ok(()))
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index 2ad14b3c..23de3b89 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -6,10 +6,10 @@
 // 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;
 use tokio::sync::mpsc;
 
 use std::{
+    collections::VecDeque,
     fmt,
     pin::Pin,
     str::FromStr,
@@ -58,79 +58,47 @@ impl From<Conn> for IdlingConn {
     }
 }
 
-pub struct Inner {
-    close: atomic::AtomicBool,
-    closed: atomic::AtomicBool,
-    idle: crossbeam::queue::ArrayQueue<IdlingConn>,
-    wake: crossbeam::queue::SegQueue<Waker>,
-    exist: atomic::AtomicUsize,
-    extra_wakeups: atomic::AtomicUsize,
-
+/// The exchange is where we track all connections as they come and go.
+///
+/// It is held under a single, non-asynchronous lock.
+/// This is fine as long as we never do expensive work while holding the lock!
+struct Exchange {
+    waiting: VecDeque<Waker>,
+    available: VecDeque<IdlingConn>,
+    exist: usize,
     // only used to spawn the recycler the first time we're in async context
-    maker: Mutex<Option<(mpsc::UnboundedReceiver<Option<Conn>>, PoolOptions)>>,
+    recycler: Option<(mpsc::UnboundedReceiver<Option<Conn>>, PoolOptions)>,
 }
 
-impl Inner {
+impl Exchange {
     /// 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>) {
+    fn spawn_futures_if_needed(&mut self, inner: &Arc<Inner>) {
         use recycler::Recycler;
         use ttl_check_inerval::TtlCheckInterval;
-
-        let mut lock = self.maker.lock().unwrap();
-        if let Some((dropped, pool_options)) = lock.take() {
+        if let Some((dropped, pool_options)) = self.recycler.take() {
             // Spawn the Recycler.
-            tokio::spawn(Recycler::new(pool_options.clone(), self.clone(), dropped));
+            tokio::spawn(Recycler::new(pool_options.clone(), inner.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()));
+                tokio::spawn(TtlCheckInterval::new(pool_options, inner.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;
-        }
-
-        while let Ok(task) = self.wake.pop() {
-            task.wake();
-            readied -= 1;
-            if readied == 0 {
-                if self.close.load(atomic::Ordering::Acquire) {
-                    // wake up as many as we can -- they should all error
-                    readied = usize::max_value();
-                    continue;
-                }
-
-                // no point in waking up more, since we don't have anything for them
-                // there _may_ be some tasks that weren't _really_ waiting though, and we need to
-                // make sure that those notifications go to someone who cares about them.
-                let extra = self.extra_wakeups.swap(0, atomic::Ordering::AcqRel);
-                if extra == 0 {
-                    break;
-                }
-
-                // one thing is worth noting here -- if there aren't enough waiting tasks in .wake
-                // to account for the value in extra, that is _okay_. those extra tasks we "would
-                // have" notified will instead see that they can proceed directly when they call
-                // .poll_new_conn(), or alternatively will be woken up directly by the place that
-                // increments .extra_wakeups in the first place
-                readied = extra;
-            }
-        }
-    }
+pub struct Inner {
+    close: atomic::AtomicBool,
+    closed: atomic::AtomicBool,
+    exchange: Mutex<Exchange>,
 }
 
 #[derive(Clone)]
 /// Asynchronous pool of MySql connections.
+///
+/// Note that you will probably want to await `Pool::disconnect` before dropping the runtime, as
+/// otherwise you may end up with a number of connections that are not cleanly terminated.
 pub struct Pool {
     opts: Opts,
     inner: Arc<Inner>,
@@ -154,11 +122,12 @@ impl Pool {
             inner: Arc::new(Inner {
                 close: false.into(),
                 closed: false.into(),
-                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_options))),
+                exchange: Mutex::new(Exchange {
+                    available: VecDeque::with_capacity(pool_options.constraints().max()),
+                    waiting: VecDeque::new(),
+                    exist: 0,
+                    recycler: Some((rx, pool_options)),
+                }),
             }),
             drop: tx,
         }
@@ -212,14 +181,30 @@ impl Pool {
             && !conn.inner.in_transaction
             && conn.inner.has_result.is_none()
             && !self.inner.close.load(atomic::Ordering::Acquire)
-            && self.inner.idle.len() < self.opts.get_pool_options().active_bound()
         {
-            self.inner.push_to_idle(conn);
-            self.inner.wake(1);
-        } else {
-            self.drop
-                .send(Some(conn))
-                .expect("recycler is active as long as any Pool is");
+            let mut exchange = self.inner.exchange.lock().unwrap();
+            if exchange.available.len() < self.opts.get_pool_options().active_bound() {
+                exchange.available.push_back(conn.into());
+                if let Some(w) = exchange.waiting.pop_front() {
+                    w.wake();
+                }
+                return;
+            }
+        }
+
+        if let Err(conn) = self.drop.send(Some(conn)) {
+            let conn = conn.0.unwrap();
+
+            // This _probably_ means that the Runtime is shutting down, and that the Recycler was
+            // dropped rather than allowed to exit cleanly.
+            if !self.inner.closed.load(atomic::Ordering::SeqCst) {
+                // Yup, Recycler was forcibly dropped!
+                // All we can do here is try the non-pool drop path for Conn.
+                assert!(conn.inner.pool.is_none());
+                drop(conn);
+            } else {
+                unreachable!("Recycler exited while connections still exist");
+            }
         }
     }
 
@@ -228,155 +213,52 @@ impl Pool {
     /// Decreases the exist counter since a broken or dropped connection should not count towards
     /// the total.
     fn cancel_connection(&self) {
-        let prev = self.inner.exist.fetch_sub(1, atomic::Ordering::AcqRel);
-        // NB: Wrapping around here would only be due to a programming error.
-        assert!(prev > 0, "exist must not wrap around");
+        let mut exchange = self.inner.exchange.lock().unwrap();
+        exchange.exist -= 1;
+        // we just enabled the creation of a new connection!
+        if let Some(w) = exchange.waiting.pop_front() {
+            w.wake();
+        }
     }
 
     /// Poll the pool for an available connection.
     fn poll_new_conn(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<GetConn>> {
-        self.poll_new_conn_inner(false, cx)
+        self.poll_new_conn_inner(cx)
     }
 
-    fn poll_new_conn_inner(
-        mut self: Pin<&mut Self>,
-        retrying: bool,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<GetConn>> {
+    fn poll_new_conn_inner(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<GetConn>> {
+        let mut exchange = self.inner.exchange.lock().unwrap();
+
+        // NOTE: this load must happen while we hold the lock,
+        // otherwise the recycler may choose to exit, see that .exist == 0, and then exit,
+        // and then we decide to create a new connection, which would then never be torn down.
         if self.inner.close.load(atomic::Ordering::Acquire) {
             return Err(Error::Driver(DriverError::PoolDisconnected)).into();
         }
 
-        loop {
-            match self.inner.idle.pop() {
-                Err(crossbeam::queue::PopError) => break,
-                Ok(IdlingConn { conn, .. }) => {
-                    if conn.expired() {
-                        self.return_conn(conn);
-                        continue;
-                    }
-
-                    return Poll::Ready(Ok(GetConn {
-                        pool: Some(self.clone()),
-                        inner: GetConnInner::Done(Some(conn)),
-                    }));
-                }
-            }
+        exchange.spawn_futures_if_needed(&self.inner);
+
+        if let Some(IdlingConn { conn, .. }) = exchange.available.pop_back() {
+            return Poll::Ready(Ok(GetConn {
+                pool: Some(self.clone()),
+                inner: GetConnInner::Done(Some(conn)),
+            }));
         }
 
         // 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.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_futures_if_needed();
-            }
-
-            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:
-                // imagine that Pool::disconnect was _just_ called. that is, after we checked at
-                // the start of this method. the Recycler checks .exist right _before_ we increment
-                // it, and notices that it is 0, and thus believes it is allowed to exit. if we
-                // continue to make a new connection here, that connection would not have a way to
-                // be dropped as part of the pool.
-                //
-                // so, we check .close again here (after the increment). if it is now true, we know
-                // that the Recycler may have exited, and we give up. if it is false, we _know_
-                // that the Recyler _must_ see our +1 before it decides to exit.
-                if self.inner.close.load(atomic::Ordering::Acquire) {
-                    self.inner.exist.fetch_sub(1, atomic::Ordering::AcqRel);
-                    // make sure we notify the Recycler in case it was waiting for our +1
-                    self.drop
-                        .send(None)
-                        .expect("recycler is active as long as any Pool is");
-                    return Err(Error::Driver(DriverError::PoolDisconnected)).into();
-                }
-
-                return Poll::Ready(Ok(GetConn {
-                    pool: Some(self.clone()),
-                    inner: GetConnInner::Connecting(Box::pin(Conn::new(self.opts.clone()))),
-                }));
-            }
-
-            let exist = self.inner.exist.fetch_sub(1, atomic::Ordering::AcqRel);
-            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);
-            }
-        }
-
-        if !retrying {
-            // no go -- we have to wait
-            self.inner.wake.push(cx.waker().clone());
-
-            // there's a potential race here -- imagine another task releases a connection after we
-            // try to poll .idle or check .exist, but before we push our task onto .wake. In that
-            // case, we might never be woken up again! so, we need to make those checks again here
-            // after we've scheduled ourselves for wakeup.
-            //
-            // an alternative strategy would be to _always_ push to .wake and then do the checks,
-            // but that would lead to a large number of spurious notifications/wakeups, as well as
-            // needless contention on .wake.
-            let conn = ready!(self.as_mut().poll_new_conn_inner(true, cx))?;
-
-            // this is a tricky case. we already registered ourselves as wanting to be woken up,
-            // but we now have a connection, so we won't be waiting. this means that _if_ we were
-            // to be woken up, that notification _really_ should have gone to some _other_ task,
-            // which now _won't_ be woken up.
-            //
-            // thew way we're going to fix that is to deal with both possible cases:
-            //
-            //  - someone _will_ try to wake us up
-            //  - someone has _already_ tried to wake us up
-            //
-            // we do this by requesting an "extra" wakeup next time someone is waking people up,
-            // and also waking someone up (perhaps spuriously) in case we have already been
-            // notified.
-            if let Ok(task) = self.inner.wake.pop() {
-                if task.will_wake(cx.waker()) {
-                    // phew -- we got out of that one easy!
-                    return Poll::Ready(Ok(conn));
-                }
-
-                // if we _haven't_ been notified yet, someone else may be deciding who to wake up
-                // _right now_. if they choose us, that's wasted. so, let's make sure they wake up
-                // at least one other task.
-                self.inner
-                    .extra_wakeups
-                    .fetch_add(1, atomic::Ordering::AcqRel);
-
-                // if someone has not yet notified us, the +1 above will make sure that they wake
-                // up at least one task that's not us. that candidate set has to include the task
-                // we just pulled off the queue.
-                self.inner.wake.push(task.clone());
-
-                // if someone _did_ already choose to notify us, we want to pass that on.
-                // but we also need to notify the task we took for a more subtle reason.
-                // consider this task0, and two other tasks, task1 and task2:
-                //
-                //  - task1 pushed to wake queue
-                //  - task0 pushed to wake queue
-                //  - task0 pops task1 from wake queue
-                //  - task0 increments extra_wakeups
-                //  - task2 tries to do a wakeup -- wakes only task0 (task1 not on the queue yet)
-                //  - task0 pushes task1 onto the queue
-                //
-                // in this case, task1 might never be awoken again, which is not okay.
-                // hence:
-                task.wake();
-            } else {
-                // someone tried to notify us, but also, no-one else is waiting,
-                // so there's no-one to "forward" that wake-up to.
-            }
-
-            return Poll::Ready(Ok(conn));
+        if exchange.exist < self.opts.get_pool_options().constraints().max() {
+            // we are allowed to make a new connection, so we will!
+            exchange.exist += 1;
+
+            return Poll::Ready(Ok(GetConn {
+                pool: Some(self.clone()),
+                inner: GetConnInner::Connecting(Box::pin(Conn::new(self.opts.clone()))),
+            }));
         }
 
+        // no go -- we have to wait
+        exchange.waiting.push_back(cx.waker().clone());
         Poll::Pending
     }
 }
@@ -396,7 +278,6 @@ mod test {
     use futures_util::stream::StreamExt;
     use futures_util::{future::try_join_all, try_join};
 
-    use std::sync::atomic;
     use std::time::Duration;
 
     use crate::{
@@ -404,6 +285,18 @@ mod test {
         PoolConstraints, TransactionOptions,
     };
 
+    macro_rules! conn_ex_field {
+        ($conn:expr, $field:tt) => {
+            ex_field!($conn.inner.pool.as_ref().unwrap(), $field)
+        };
+    }
+
+    macro_rules! ex_field {
+        ($pool:expr, $field:tt) => {
+            $pool.inner.exchange.lock().unwrap().$field
+        };
+    }
+
     #[test]
     fn should_not_hang() -> super::Result<()> {
         pub struct Database {
@@ -517,10 +410,7 @@ mod test {
 
         // we want to continuously drop connections
         // and check that they are _actually_ dropped until we reach POOL_MIN
-        assert_eq!(
-            pool_clone.inner.exist.load(atomic::Ordering::SeqCst),
-            POOL_MAX
-        );
+        assert_eq!(ex_field!(pool_clone, exist), POOL_MAX);
 
         while !conns.is_empty() {
             // first, drop a connection
@@ -545,7 +435,7 @@ mod test {
 
             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);
+                let have = ex_field!(pool_clone, exist);
                 assert_eq!(have, expected + 1);
 
                 // then, wait for ttl_check_interval
@@ -553,7 +443,7 @@ mod test {
             }
 
             // check that we have the expected number of connections
-            let have = pool_clone.inner.exist.load(atomic::Ordering::SeqCst);
+            let have = ex_field!(pool_clone, exist);
             assert_eq!(have, expected);
         }
 
@@ -572,31 +462,21 @@ mod test {
 
         let (conn1, conn2) = try_join!(pool.get_conn(), pool.get_conn()).unwrap();
 
-        assert_eq!(
-            conn1
-                .inner
-                .pool
-                .as_ref()
-                .unwrap()
-                .inner
-                .exist
-                .load(atomic::Ordering::SeqCst),
-            2
-        );
-        assert_eq!(conn1.inner.pool.as_ref().unwrap().inner.idle.len(), 0);
+        assert_eq!(conn_ex_field!(conn1, exist), 2);
+        assert_eq!(conn_ex_field!(conn1, available).len(), 0);
 
         drop(conn1);
         drop(conn2);
         // only one of conn1 and conn2 should have gone to idle,
         // and should have immediately been picked up by new_conn (now conn1)
         let conn1 = pool_clone.get_conn().await?;
-        assert_eq!(conn1.inner.pool.as_ref().unwrap().inner.idle.len(), 0);
+        assert_eq!(conn_ex_field!(conn1, available).len(), 0);
 
         drop(conn1);
 
         // the connection should be returned to idle
         // (but may not have been returned _yet_)
-        assert!(pool.inner.idle.len() <= 1);
+        assert!(ex_field!(pool, available).len() <= 1);
         pool.disconnect().await?;
         Ok(())
     }
@@ -608,7 +488,7 @@ mod test {
         let pool = Pool::new(String::from("mysql://255.255.255.255"));
 
         assert!(try_join!(pool.get_conn(), pool.get_conn()).is_err());
-        assert_eq!(pool.inner.exist.load(atomic::Ordering::SeqCst), 0);
+        assert_eq!(ex_field!(pool, exist), 0);
         Ok(())
     }
 
@@ -677,6 +557,46 @@ mod test {
         );
     }
 
+    #[test]
+    fn should_not_panic_on_unclean_shutdown() {
+        // run more than once to trigger different drop orders
+        for _ in 0..10 {
+            let mut rt = tokio::runtime::Runtime::new().unwrap();
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            rt.block_on(async move {
+                let pool = Pool::new(get_opts());
+                let c = pool.get_conn().await.unwrap();
+                tokio::spawn(async move {
+                    let _ = rx.await;
+                    let _ = c.drop_query("SELECT 1").await;
+                });
+            });
+            drop(rt);
+            // c is still active here, so if anything it's been forcibly dropped
+            let _ = tx.send(());
+        }
+    }
+
+    #[test]
+    fn should_perform_clean_shutdown() {
+        // run more than once to trigger different drop orders
+        for _ in 0..10 {
+            let mut rt = tokio::runtime::Runtime::new().unwrap();
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            let jh = rt.spawn(async move {
+                let pool = Pool::new(get_opts());
+                let c = pool.get_conn().await.unwrap();
+                tokio::spawn(async move {
+                    let _ = rx.await;
+                    let _ = c.drop_query("SELECT 1").await;
+                });
+                let _ = pool.disconnect().await;
+            });
+            let _ = tx.send(());
+            rt.block_on(jh).unwrap();
+        }
+    }
+
     /*
     #[test]
     #[ignore]
diff --git a/src/conn/pool/recycler.rs b/src/conn/pool/recycler.rs
index 3f136ad8..8f6b04d2 100644
--- a/src/conn/pool/recycler.rs
+++ b/src/conn/pool/recycler.rs
@@ -55,24 +55,28 @@ 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) => {
+            ($self: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()
-                {
+                } else if $conn.expired() || close {
                     $self.discard.push(Box::pin($conn.close()));
                 } else {
-                    $self.inner.push_to_idle($conn);
-                    $readied += 1;
+                    let mut exchange = $self.inner.exchange.lock().unwrap();
+                    if exchange.available.len() >= $self.pool_opts.active_bound() {
+                        drop(exchange);
+                        $self.discard.push(Box::pin($conn.close()));
+                    } else {
+                        exchange.available.push_back($conn.into());
+                        if let Some(w) = exchange.waiting.pop_front() {
+                            w.wake();
+                        }
+                    }
                 }
             };
         }
@@ -81,7 +85,8 @@ impl Future for Recycler {
             // 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);
+                    assert!(conn.inner.pool.is_none());
+                    conn_decision!(self, conn);
                 }
                 Poll::Ready(Some(None)) => {
                     // someone signaled us that it's exit time
@@ -103,9 +108,12 @@ impl Future for Recycler {
         }
 
         // 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);
+        if close || self.eof {
+            while let Some(IdlingConn { conn, .. }) =
+                self.inner.exchange.lock().unwrap().available.pop_front()
+            {
+                assert!(conn.inner.pool.is_none());
+                conn_decision!(self, conn);
             }
         }
 
@@ -113,10 +121,13 @@ impl Future for Recycler {
         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(Ok(conn))) => conn_decision!(self, conn),
                 Poll::Ready(Some(Err(e))) => {
                     // an error occurred while cleaning a connection.
                     // what do we do? replace it with a new connection?
+                    // for a conn to end up in cleaning, it must have come through .dropped.
+                    // anything that comes through .dropped we know has .pool.is_none().
+                    // therefore, dropping the conn won't decrement .exist, so we need to do that.
                     self.discarded += 1;
                     // NOTE: we're discarding the error here
                     let _ = e;
@@ -130,6 +141,8 @@ impl Future for Recycler {
                 Poll::Pending | Poll::Ready(None) => break,
                 Poll::Ready(Some(Ok(()))) => {
                     // yes! count it.
+                    // note that we must decrement .exist since the connection does not have a
+                    // .pool, and therefore won't do anything useful when it is dropped.
                     self.discarded += 1
                 }
                 Poll::Ready(Some(Err(e))) => {
@@ -144,22 +157,27 @@ impl Future for Recycler {
 
         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;
+            let mut exchange = self.inner.exchange.lock().unwrap();
+            exchange.exist -= self.discarded;
+            for _ in 0..self.discarded {
+                if let Some(w) = exchange.waiting.pop_front() {
+                    w.wake();
+                }
+            }
+            drop(exchange);
             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;
+        let effectively_eof = close && self.inner.exchange.lock().unwrap().exist == 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);
+            assert_eq!(self.inner.exchange.lock().unwrap().available.len(), 0);
 
             // NOTE: it is _necessary_ that we set this _before_ we call .wake
             // otherwise, the following may happen to the DisconnectPool future:
@@ -173,11 +191,16 @@ impl Future for Recycler {
             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());
+            let mut exchange = self.inner.exchange.lock().unwrap();
+            while let Some(w) = exchange.waiting.pop_front() {
+                w.wake();
+            }
+            // we're about to exit, so there better be no outstanding connections
+            assert_eq!(exchange.exist, 0);
+            assert_eq!(exchange.available.len(), 0);
+            drop(exchange);
 
             Poll::Ready(())
         } else {
@@ -185,3 +208,14 @@ impl Future for Recycler {
         }
     }
 }
+
+impl Drop for Recycler {
+    fn drop(&mut self) {
+        if !self.inner.closed.load(Ordering::Acquire) {
+            // user did not wait for outstanding connections to finish!
+            // this is not good -- we won't be able to shut down our connections cleanly
+            // all we can do is try to ensure a clean shutdown
+            self.inner.close.store(true, Ordering::SeqCst);
+        }
+    }
+}
diff --git a/src/conn/pool/ttl_check_inerval.rs b/src/conn/pool/ttl_check_inerval.rs
index fecba1ac..8c2a2765 100644
--- a/src/conn/pool/ttl_check_inerval.rs
+++ b/src/conn/pool/ttl_check_inerval.rs
@@ -47,31 +47,25 @@ impl TtlCheckInterval {
 
     /// 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;
+        let mut exchange = self.inner.exchange.lock().unwrap();
 
-        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,
+        let num_idling = exchange.available.len();
+        let num_to_drop = num_idling.saturating_sub(self.pool_options.constraints().min());
+
+        for _ in 0..num_to_drop {
+            let idling_conn = exchange.available.pop_front().unwrap();
+            if idling_conn.elapsed() > self.pool_options.inactive_connection_ttl() {
+                assert!(idling_conn.conn.inner.pool.is_none());
+                tokio::spawn(idling_conn.conn.disconnect().map(drop));
+                exchange.exist -= 1;
+            } else {
+                exchange.available.push_back(idling_conn);
             }
         }
 
-        if num_returned > 0 {
-            self.inner.wake(num_returned);
+        if num_to_drop > 0 {
+            // there were available connections, so no tasks should be waiting
+            assert_eq!(exchange.waiting.len(), 0);
         }
     }
 }
diff --git a/src/local_infile_handler/mod.rs b/src/local_infile_handler/mod.rs
index 5fc4c991..a471a4c5 100644
--- a/src/local_infile_handler/mod.rs
+++ b/src/local_infile_handler/mod.rs
@@ -62,6 +62,11 @@ pub mod builtin;
 ///         // The used command is not allowed with this MySQL version
 ///         return Ok(());
 ///     },
+///     Err(mysql_async::error::Error::Server(ref err)) if err.code == 3948 => {
+///         // Loading local data is disabled;
+///         // this must be enabled on both the client and server sides
+///         return Ok(());
+///     }
 ///     e@Err(_) => e.unwrap(),
 /// };
 /// let result = conn.prep_exec("SELECT * FROM tmp;", ()).await?;


Test patch: