Repository: blackbeam/mysql_async (language: rust)

Issue:
Pool::disconnect blocks indefinitely
I have a wrapper around a `Pool` object (using the new async update) which has a `disconnect` method:
```rust
pub async fn disconnect(&self) -> Result<()> {
    self.pool.clone().disconnect().await?;
    Ok(())
}
```
When I want to exit my program, I run this on my tokio runtime with `::block_on`. However, it never returns.

cc @jonhoo 


Discussion / hints:
So, from memory, this will wait until all currently-open connections are closed. Are you still holding on to connections elsewhere?
I get one connection from the pool to test the connection, and then drop it. That should move it back into the pool, right?

I'm confused why the disconnect method would wait for the connections in the pool to close. Isn't that what this method should do? How else am I supposed to clean up my pool?
Hmm, yes, that should do it. Do you have some example code to play with?

Ah, `disconnect` does not _force_ a disconnect, it is more like tokio's `shutdown_on_idle`. If you have currently active connections, then those are allowed to finish before the pool is terminated. Forcibly terminating connections can easily lead to weird interactions, and would also require significantly more synchronization internally.
I can't give an example atm but how should I be cleaning up a Pool properly? Is dropping enough? I could also drop my runtime first and then call disconnect on a new one but that's quite ugly. Is there an example of how this should be done? I couldn't understand the docs (I understand English is not the authors first language).
Dropping any connections you have received from the pool should be enough to make `Pool::disconnect` resolve. It'd be super helpful if you could produce a small reproducing example to dig into :)
It still reproduces for me when I don't even use the `Pool` once.

```rust
pub struct Database {
    pool: Pool,
}

impl Database {
    pub async fn disconnect(&self) -> Result<()> {
        self.pool.clone().disconnect().await?;
        Ok(())
    }
}

fn main() {
    /* create pool */
    /* obtain runtime */
    runtime.block_on(database.disconnect());
    /* wait forever */
}
```
Perhaps it's related to the clone I do? In my opinion that method should just take `&self`.

Fix patch:
diff --git a/src/conn/pool/futures/disconnect_pool.rs b/src/conn/pool/futures/disconnect_pool.rs
index d5198003..4b5a4d1c 100644
--- a/src/conn/pool/futures/disconnect_pool.rs
+++ b/src/conn/pool/futures/disconnect_pool.rs
@@ -35,6 +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.wake.push(cx.waker().clone());
 
         if self.pool_inner.closed.load(atomic::Ordering::Acquire) {
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index 6e9098b2..54cdb1be 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -174,8 +174,9 @@ impl Future for Recycler {
         self.inner.wake(readied);
 
         if self.inner.closed.load(atomic::Ordering::Acquire) {
-            // since there are no more Pools, we also know that no-one is waiting anymore,
-            // so we don't have to worry about calling wake more times
+            // `DisconnectPool` might still wait to be woken up.
+            self.inner.wake(usize::max_value());
+
             Poll::Ready(())
         } else {
             Poll::Pending
@@ -192,10 +193,25 @@ struct Inner {
     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>>>>,
+    maker: Mutex<Option<(mpsc::UnboundedReceiver<Option<Conn>>, usize)>>,
 }
 
 impl Inner {
+    fn spawn_recycler_if_needed(self: Arc<Self>) {
+        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,
+            });
+        }
+    }
+
     fn wake(&self, mut readied: usize) {
         if readied == 0 {
             return;
@@ -263,7 +279,7 @@ impl Pool {
                 wake: crossbeam::queue::SegQueue::new(),
                 exist: 0.into(),
                 extra_wakeups: 0.into(),
-                maker: Mutex::new(Some(rx)),
+                maker: Mutex::new(Some((rx, pool_constraints.min()))),
             }),
             drop: tx,
             pool_constraints,
@@ -302,6 +318,7 @@ impl Pool {
             // no connections to wait for and exit quickly!
             let _ = self.drop.try_send(None).is_ok();
         }
+
         new_disconnect_pool(self)
     }
 
@@ -380,19 +397,7 @@ impl Pool {
             let exist = self.inner.exist.fetch_add(1, atomic::Ordering::AcqRel);
             if exist == 0 {
                 // we may have to start the recycler.
-                let mut lock = self.inner.maker.lock().unwrap();
-                if let Some(dropped) = lock.take() {
-                    // we're the first connection!
-                    tokio::spawn(Recycler {
-                        inner: self.inner.clone(),
-                        discard: FuturesUnordered::new(),
-                        discarded: 0,
-                        cleaning: FuturesUnordered::new(),
-                        dropped,
-                        min: self.pool_constraints.min(),
-                        eof: false,
-                    });
-                }
+                self.inner.clone().spawn_recycler_if_needed();
             }
 
             if exist < self.pool_constraints.max() {
@@ -521,6 +526,24 @@ mod test {
         TransactionOptions,
     };
 
+    #[test]
+    fn should_not_hang() -> super::Result<()> {
+        pub struct Database {
+            pool: Pool,
+        }
+
+        impl Database {
+            pub async fn disconnect(self) -> super::Result<()> {
+                self.pool.disconnect().await?;
+                Ok(())
+            }
+        }
+
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        let database = Database { pool: Pool::new(get_opts()) };
+        runtime.block_on(database.disconnect())
+    }
+
     #[tokio::test]
     async fn should_connect() -> super::Result<()> {
         let pool = Pool::new(get_opts());


Test patch: