Repository: blackbeam/mysql_async (language: rust)

Issue:
`Pool::disconnect` waits forever, how to disconnect correctly? 
Hi, i am beginner in Rust, trying to write some small app. I trying to create graceful shutdown for my app, and i ran into problem when `Pool::disconnect` waits forever. Docs states that it will waits until all taken conns are dropped, but i cannot understand why the `Conn` is not dropped. I've created a [MRE](https://github.com/heavycharged/mysql_async_bug). Can someone check and explain me where i went wrong? Thanks.







Discussion / hints:
@heavycharged Are you using BinLogStream, because this happens when you do acquire a connection using BinLogStream for example, I made a PR to fix that https://github.com/blackbeam/mysql_async/pull/200
Hey, @oussama. Thanks for reply, but i am not using `BinlogStream`. 
Here is [minimal reproducible example](https://github.com/heavycharged/mysql_async_bug), it simply calls `SELECT 42` from database.
The problem occurs only when i spawn `actix_web::HttpServer`. If i remove code with `HttpServer`, and simply make some requests, everything works fine and `Pool::disconnect` finishes in few moments.
In my example `#[actix_web::rt]` used, but switching to `tokio` runtime not helps to solve the problem (i guess `actix` uses `tokio` runtime in last versions).
I thought that `actix_web` shutdowns the runtime when handling the `Ctrl+C`, so i tried to disable signal handler with `disable_signals()` and handle it manually. This not helps, it still freezes.

Fix patch:
diff --git a/src/conn/pool/futures/disconnect_pool.rs b/src/conn/pool/futures/disconnect_pool.rs
index 4d6616b9..c409e18e 100644
--- a/src/conn/pool/futures/disconnect_pool.rs
+++ b/src/conn/pool/futures/disconnect_pool.rs
@@ -12,9 +12,13 @@ use std::{
     task::{Context, Poll},
 };
 
+use futures_core::ready;
+use tokio::sync::mpsc::UnboundedSender;
+
 use crate::{
     conn::pool::{Inner, Pool},
     error::Error,
+    Conn,
 };
 
 use std::sync::{atomic, Arc};
@@ -27,12 +31,14 @@ use std::sync::{atomic, Arc};
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct DisconnectPool {
     pool_inner: Arc<Inner>,
+    drop: Option<UnboundedSender<Option<Conn>>>,
 }
 
 impl DisconnectPool {
     pub(crate) fn new(pool: Pool) -> Self {
         Self {
             pool_inner: pool.inner,
+            drop: Some(pool.drop),
         }
     }
 }
@@ -40,7 +46,8 @@ impl DisconnectPool {
 impl Future for DisconnectPool {
     type Output = Result<(), Error>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.pool_inner.close.store(true, atomic::Ordering::Release);
         let mut exchange = self.pool_inner.exchange.lock().unwrap();
         exchange.spawn_futures_if_needed(&self.pool_inner);
         exchange.waiting.push_back(cx.waker().clone());
@@ -49,7 +56,19 @@ impl Future for DisconnectPool {
         if self.pool_inner.closed.load(atomic::Ordering::Acquire) {
             Poll::Ready(Ok(()))
         } else {
-            Poll::Pending
+            match self.drop.take() {
+                Some(drop) => match drop.send(None) {
+                    Ok(_) => {
+                        // Recycler is alive. Waiting for it to finish.
+                        Poll::Ready(Ok(ready!(Box::pin(drop.closed()).as_mut().poll(cx))))
+                    }
+                    Err(_) => {
+                        // Recycler seem dead. No one will wake us.
+                        Poll::Ready(Ok(()))
+                    }
+                },
+                None => Poll::Pending,
+            }
         }
     }
 }
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index ffbc2b89..d7a8ea10 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -162,15 +162,6 @@ impl Pool {
     /// **Note:** This Future won't resolve until all active connections, taken from it,
     /// are dropped or disonnected. Also all pending and new `GetConn`'s will resolve to error.
     pub fn disconnect(self) -> DisconnectPool {
-        let was_closed = self.inner.close.swap(true, atomic::Ordering::AcqRel);
-        if !was_closed {
-            // make sure we wake up the Recycler.
-            //
-            // note the lack of an .expect() here, because the Recycler may decide that there are
-            // no connections to wait for and exit quickly!
-            let _ = self.drop.send(None).is_ok();
-        }
-
         DisconnectPool::new(self)
     }
 


Test patch: