Replace complex pool with mutex pool

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"

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

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(()))