Repository: blackbeam/mysql_async (language: rust)

Issue:
Panic while getting connection from pool (support for tokio-0.3?)
With the tokio-0.3.3, I am seeing panic error while getting connection from the pool. This was working fine with tokio-2.x version.

tokio version:0.3.3
mysql_asybc: 0.25.0

tokio runtime created using: Builder::new_multi_thread()

```
  0:        0x10f5a9394 - std::backtrace_rs::backtrace::libunwind::trace::hb4ff9014ec5817b3
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/../../backtrace/src/backtrace/libunwind.rs:100:5
   1:        0x10f5a9394 - std::backtrace_rs::backtrace::trace_unsynchronized::h0d03cf0f7c5d8d01
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/../../backtrace/src/backtrace/mod.rs:66:5
   2:        0x10f5a9394 - std::sys_common::backtrace::_print_fmt::h5b2bf2921eea85ae
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/sys_common/backtrace.rs:67:5
   3:        0x10f5a9394 - <std::sys_common::backtrace::_print::DisplayBacktrace as core::fmt::Display>::fmt::h808cc0fc691f5922
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/sys_common/backtrace.rs:46:22
   4:        0x10f5ceb30 - core::fmt::write::h0ce880d33cd2a300
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/core/src/fmt/mod.rs:1076:17
   5:        0x10f5a1b46 - std::io::Write::write_fmt::h057e6b5365b25cdb
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/io/mod.rs:1516:15
   6:        0x10f5ab4d9 - std::sys_common::backtrace::_print::ha1932a5f36d2b146
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/sys_common/backtrace.rs:49:5
   7:        0x10f5ab4d9 - std::sys_common::backtrace::print::had07999cb13bdf77
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/sys_common/backtrace.rs:36:9
   8:        0x10f5ab4d9 - std::panicking::default_hook::{{closure}}::hcbe1ecfff77305a1
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/panicking.rs:208:50
   9:        0x10f5ab19d - std::panicking::default_hook::h8422621c02c35b35
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/panicking.rs:227:9
  10:        0x10f5abaab - std::panicking::rust_panic_with_hook::h82b76d3adf4cbc3a
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/panicking.rs:577:17
  11:        0x10f5ab62b - std::panicking::begin_panic_handler::{{closure}}::h1e3858c0f26acd6d
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/panicking.rs:484:9
  12:        0x10f5a9838 - std::sys_common::backtrace::__rust_end_short_backtrace::hb43bfc309d1c5e68
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/sys_common/backtrace.rs:141:18
  13:        0x10f5ab5ea - rust_begin_unwind
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/std/src/panicking.rs:483:5
  14:        0x10f5f4f2f - core::panicking::panic_fmt::h33aee6f32da9c3a4
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/core/src/panicking.rs:85:14
  15:        0x10f5f4dba - core::option::expect_failed::he41d9b6b925c99b1
                               at /rustc/ffa2e7ae8fbf9badc035740db949b9dae271c29f/library/core/src/option.rs:1260:5
  16:        0x10eeb0dde - core::option::Option<T>::expect::ha8ead4dbfc53f3b9
                               at /Users/xxx/.rustup/toolchains/nightly-x86_64-apple-darwin/lib/rustlib/src/rust/library/core/src/option.rs:349:21
  17:        0x10ee93b45 - tokio::task::spawn::spawn::h35bd0066fc875c3c
                               at /Users/xxx/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.22/src/task/spawn.rs:130:28
  18:        0x10ee44c64 - mysql_async::conn::pool::Exchange::spawn_futures_if_needed::h7b820e5753962bfb
                               at /Users/xxx/.cargo/registry/src/github.com-1ecc6299db9ec823/mysql_async-0.25.0/src/conn/pool/mod.rs:79:13
  19:        0x10ee458e6 - mysql_async::conn::pool::Pool::poll_new_conn_inner::h80344d2cc2281d13
                               at /Users/xxx/.cargo/registry/src/github.com-1ecc6299db9ec823/mysql_async-0.25.0/src/conn/pool/mod.rs:236:9
  20:        0x10ee4568c - mysql_async::conn::pool::Pool::poll_new_conn::h3266b761607811a0
                               at /Users/xxx/.cargo/registry/src/github.com-1ecc6299db9ec823/mysql_async-0.25.0/src/conn/pool/mod.rs:223:9
  21:        0x10ee7efa2 - <mysql_async::conn::pool::futures::get_conn::GetConn as core::future::future::Future>::poll::hab19b271f37ed892
                               at /Users/xxx/.cargo/registry/src/github.com-1ecc6299db9ec823/mysql_async-0.25.0/src/conn/pool/futures/get_conn.rs:77:51
```


Discussion / hints:
I believe issue is due to application using tokio-0.3.3 but mysql_async relies on tokio-2.17
Hi.

Looks like an update to tokio-0.3.3 is blocked by tokio-rs/tokio#2965.
We have to wait for tokio-rs/tokio#3076.
@blackbeam thank you. [ tokio-rs/tokio#3076](https://github.com/tokio-rs/tokio/pull/3076) is merged. 
Any plan to release a version to support tokio-0.3.4? 

thread 'xxx-xx' panicked at 'must be called from the context of Tokio runtime configured with either `basic_scheduler` or `threaded_scheduler`', /Users/xxx/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.23/src/task/spawn.rs:131:10

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 6c968e45..a29cb851 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -12,30 +12,32 @@ exclude = ["test/*"]
 edition = "2018"
 
 [dependencies]
-bytes = "0.5.2"
-crossbeam = "0.7"
+bytes = "1.0"
 futures-core = "0.3"
 futures-util = "0.3"
 futures-sink = "0.3"
 lazy_static = "1"
 lru = "0.6.0"
-mio-named-pipes = "0.1.6"
-mysql_common = "0.24.1"
+mio = "0.7.7"
+mysql_common = "0.26.0"
 native-tls = "0.2"
 pem = "0.8.1"
 percent-encoding = "2.1.0"
-pin-project = "0.4.17"
+pin-project = "1.0.2"
 serde = "1"
 serde_json = "1"
 thiserror = "1.0.4"
-tokio = { version = "0.2.17", features = ["io-util", "net", "sync", "fs", "rt-core", "time", "stream", "macros"] }
-tokio-util = { version = "0.3.1", features = ["codec"] }
-tokio-tls = "0.3"
+tokio = { version = "1.0", features = ["io-util", "fs", "net", "time"] }
+tokio-util = { version = "0.6.0", features = ["codec"] }
+tokio-native-tls = { git = "https://github.com/tokio-rs/tls.git" }
 twox-hash = "1"
 url = "2.1"
+uuid = { version = "0.8.1", features = ["v4"] }
 
 [dev-dependencies]
 tempfile = "3.1.0"
+socket2 = "0.3.17"
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread"] }
 rand = "0.8.0"
 
 [features]
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index 07125d90..5f5c037e 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -42,40 +42,39 @@ jobs:
           DATABASE_URL: mysql://root:root@127.0.0.1:3306/mysql
         displayName: Run tests
 
-  - job: "TestBasicMacOs"
-    pool:
-      vmImage: "macOS-10.15"
-    strategy:
-      maxParallel: 10
-      matrix:
-        stable:
-          RUST_TOOLCHAIN: stable
-    steps:
-      - bash: |
-          brew update
-          brew install mysql
-          brew services start mysql
-          brew services stop mysql
-          sleep 3
-          echo 'local_infile=1' >> /usr/local/etc/my.cnf
-          echo 'socket=/tmp/mysql.sock' >> /usr/local/etc/my.cnf
-          brew services start mysql
-          /usr/local/Cellar/mysql/*/bin/mysqld --verbose --help
-          sleep 20
-          /usr/local/Cellar/mysql/*/bin/mysql -e "SET GLOBAL max_allowed_packet = 36700160;" -uroot
-        displayName: Install MySql
-      - bash: |
-          curl https://sh.rustup.rs -sSf | sh -s -- -y --default-toolchain $RUST_TOOLCHAIN
-        displayName: Install rust (MacOs)
-      - bash: |
-          SSL=false COMPRESS=false cargo test
-          SSL=true COMPRESS=false cargo test
-          SSL=false COMPRESS=true cargo test
-          SSL=true COMPRESS=true cargo test
-        env:
-          RUST_BACKTRACE: 1
-          DATABASE_URL: mysql://root@127.0.0.1/mysql
-        displayName: Run tests
+  # - job: "TestBasicMacOs"
+  #   pool:
+  #     vmImage: "macOS-10.15"
+  #   strategy:
+  #     maxParallel: 10
+  #     matrix:
+  #       stable:
+  #         RUST_TOOLCHAIN: stable
+  #   steps:
+  #     - bash: |
+  #         brew update
+  #         brew install mysql
+  #         brew services start mysql
+  #         brew services stop mysql
+  #         sleep 3
+  #         echo 'local_infile=1' >> /usr/local/etc/my.cnf
+  #         echo 'socket=/tmp/mysql.sock' >> /usr/local/etc/my.cnf
+  #         brew services start mysql
+  #         sleep 5
+  #         /usr/local/Cellar/mysql/*/bin/mysql -e "SET GLOBAL max_allowed_packet = 36700160;" -uroot
+  #       displayName: Install MySql
+  #     - bash: |
+  #         curl https://sh.rustup.rs -sSf | sh -s -- -y --default-toolchain $RUST_TOOLCHAIN
+  #       displayName: Install rust (MacOs)
+  #     - bash: |
+  #         SSL=false COMPRESS=false cargo test
+  #         SSL=true COMPRESS=false cargo test
+  #         SSL=false COMPRESS=true cargo test
+  #         SSL=true COMPRESS=true cargo test
+  #       env:
+  #         RUST_BACKTRACE: 1
+  #         DATABASE_URL: mysql://root@127.0.0.1/mysql
+  #       displayName: Run tests
 
   - job: "TestBasicWindows"
     pool:
diff --git a/rustfmt.toml b/rustfmt.toml
deleted file mode 100644
index 32a9786f..00000000
--- a/rustfmt.toml
+++ /dev/null
@@ -1,1 +0,0 @@
-edition = "2018"
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
index b3d4f85b..5c3ed52a 100644
--- a/src/conn/mod.rs
+++ b/src/conn/mod.rs
@@ -143,7 +143,7 @@ impl ConnInner {
     fn stream_mut(&mut self) -> Result<&mut Stream> {
         self.stream
             .as_mut()
-            .ok_or(DriverError::ConnectionClosed.into())
+            .ok_or_else(|| DriverError::ConnectionClosed.into())
     }
 }
 
@@ -281,7 +281,9 @@ impl Conn {
                 result
             }
             Err(err) => {
-                self.take_stream().close().await?;
+                if self.inner.stream.is_some() {
+                    self.take_stream().close().await?;
+                }
                 Err(err)
             }
         }
@@ -344,7 +346,6 @@ impl Conn {
     fn setup_stream(&mut self) -> Result<()> {
         debug_assert!(self.inner.stream.is_some());
         if let Some(stream) = self.inner.stream.as_mut() {
-            stream.set_keepalive_ms(self.inner.opts.tcp_keepalive())?;
             stream.set_tcp_nodelay(self.inner.opts.tcp_nodelay())?;
         }
         Ok(())
@@ -450,7 +451,8 @@ impl Conn {
                 }
                 AuthPlugin::Other(ref name) => Err(DriverError::UnknownAuthPlugin {
                     name: String::from_utf8_lossy(name.as_ref()).to_string(),
-                })?,
+                }
+                .into()),
             }
         })
     }
@@ -500,20 +502,14 @@ impl Conn {
                     self.drop_packet().await?;
                     Ok(())
                 }
-                _ => Err(DriverError::UnexpectedPacket {
-                    payload: packet.into(),
-                }
-                .into()),
+                _ => Err(DriverError::UnexpectedPacket { payload: packet }.into()),
             },
             Some(0xfe) if !self.inner.auth_switched => {
                 let auth_switch_request = parse_auth_switch_request(&*packet)?.into_owned();
                 self.perform_auth_switch(auth_switch_request).await?;
                 Ok(())
             }
-            _ => Err(DriverError::UnexpectedPacket {
-                payload: packet.into(),
-            }
-            .into()),
+            _ => Err(DriverError::UnexpectedPacket { payload: packet }.into()),
         }
     }
 
@@ -541,7 +537,7 @@ impl Conn {
             self.handle_ok(ok_packet.into_owned());
         } else if let Ok(err_packet) = parse_err_packet(&*packet, self.capabilities()) {
             self.handle_err(err_packet.clone().into_owned());
-            return Err(err_packet.into()).into();
+            return Err(err_packet.into());
         }
 
         Ok(())
@@ -583,7 +579,7 @@ impl Conn {
 
     /// Returns future that sends full command body to a server.
     pub(crate) async fn write_command_raw(&mut self, body: Vec<u8>) -> Result<()> {
-        debug_assert!(body.len() > 0);
+        debug_assert!(!body.is_empty());
         self.clean_dirty().await?;
         self.reset_seq_id();
         self.write_packet(body).await
@@ -607,7 +603,7 @@ impl Conn {
     }
 
     async fn run_init_commands(&mut self) -> Result<()> {
-        let mut init: Vec<_> = self.inner.opts.init().iter().cloned().collect();
+        let mut init = self.inner.opts.init().to_vec();
 
         while let Some(query) = init.pop() {
             self.query_drop(query).await?;
@@ -622,10 +618,18 @@ impl Conn {
         let fut = Box::pin(async move {
             let mut conn = Conn::empty(opts.clone());
 
-            let stream = if let Some(path) = opts.socket() {
-                Stream::connect_socket(path.to_owned()).await?
+            let stream = if let Some(_path) = opts.socket() {
+                #[cfg(unix)]
+                {
+                    Stream::connect_socket(_path.to_owned()).await?
+                }
+                #[cfg(target_os = "windows")]
+                return Err(crate::DriverError::NamedPipesDisabled.into());
             } else {
-                Stream::connect_tcp(opts.hostport_or_url()).await?
+                let keepalive = opts
+                    .tcp_keepalive()
+                    .map(|x| std::time::Duration::from_millis(x.into()));
+                Stream::connect_tcp(opts.hostport_or_url(), keepalive).await?
             };
 
             conn.inner.stream = Some(stream);
@@ -659,13 +663,10 @@ impl Conn {
             let opts = self.inner.opts.clone();
             if opts.socket().is_none() {
                 let opts = OptsBuilder::from_opts(opts).socket(Some(&**socket));
-                match Conn::new(opts).await {
-                    Ok(conn) => {
-                        let old_conn = std::mem::replace(self, conn);
-                        // tidy up the old connection
-                        old_conn.close_conn().await?;
-                    }
-                    Err(_) => (),
+                if let Ok(conn) = Conn::new(opts).await {
+                    let old_conn = std::mem::replace(self, conn);
+                    // tidy up the old connection
+                    old_conn.close_conn().await?;
                 }
             }
         }
@@ -925,7 +926,7 @@ mod test {
     #[test]
     fn should_not_panic_if_dropped_without_tokio_runtime() {
         let fut = Conn::new(get_opts());
-        let mut runtime = tokio::runtime::Runtime::new().unwrap();
+        let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
             fut.await.unwrap();
         });
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index 5ab83d41..dda23740 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -235,21 +235,17 @@ impl Pool {
 
         exchange.spawn_futures_if_needed(&self.inner);
 
-        loop {
-            if let Some(IdlingConn { mut conn, .. }) = exchange.available.pop_back() {
-                if !conn.expired() {
-                    return Poll::Ready(Ok(GetConn {
-                        pool: Some(self.clone()),
-                        inner: GetConnInner::Checking(BoxFuture(Box::pin(async move {
-                            conn.stream_mut()?.check().await?;
-                            Ok(conn)
-                        }))),
-                    }));
-                } else {
-                    self.send_to_recycler(conn);
-                }
+        while let Some(IdlingConn { mut conn, .. }) = exchange.available.pop_back() {
+            if !conn.expired() {
+                return Poll::Ready(Ok(GetConn {
+                    pool: Some(self.clone()),
+                    inner: GetConnInner::Checking(BoxFuture(Box::pin(async move {
+                        conn.stream_mut()?.check().await?;
+                        Ok(conn)
+                    }))),
+                }));
             } else {
-                break;
+                self.send_to_recycler(conn);
             }
         }
 
@@ -289,11 +285,10 @@ impl Drop for Conn {
 mod test {
     use futures_util::{
         future::{join_all, select, select_all, try_join_all},
-        stream::StreamExt,
         try_join, FutureExt,
     };
     use mysql_common::row::Row;
-    use tokio::time::delay_for;
+    use tokio::time::sleep;
 
     use std::time::Duration;
 
@@ -326,7 +321,7 @@ mod test {
             }
         }
 
-        let mut runtime = tokio::runtime::Runtime::new().unwrap();
+        let runtime = tokio::runtime::Runtime::new().unwrap();
         let database = Database {
             pool: Pool::new(get_opts()),
         };
@@ -385,7 +380,7 @@ mod test {
             // now check, that they're still in the pool..
             assert_eq!(ex_field!(pool, available).len(), NUM_CONNS);
 
-            delay_for(Duration::from_millis(500)).await;
+            sleep(Duration::from_millis(500)).await;
 
             // now get new connection..
             let _conn = pool.get_conn().await?;
@@ -422,7 +417,7 @@ mod test {
             }
             drop(tx);
             // see that all the tx's eventually complete
-            while let Some(_) = rx.next().await {}
+            while let Some(_) = rx.recv().await {}
         }
         drop(pool);
     }
@@ -476,16 +471,16 @@ mod test {
         drop(conns);
 
         // wait for a bit to let the connections be reclaimed
-        delay_for(Duration::from_millis(100)).await;
+        sleep(Duration::from_millis(100)).await;
 
         // check that connections are still in the pool because of inactive_connection_ttl
         assert_eq!(ex_field!(pool_clone, available).len(), POOL_MAX);
 
         // then, wait for ttl_check_interval
-        delay_for(TTL_CHECK_INTERVAL).await;
+        sleep(TTL_CHECK_INTERVAL).await;
 
         // wait a bit more to let the connections be reclaimed by the ttl check
-        delay_for(Duration::from_millis(500)).await;
+        sleep(Duration::from_millis(500)).await;
 
         // check that we have the expected number of connections
         assert_eq!(ex_field!(pool_clone, available).len(), POOL_MIN);
@@ -518,7 +513,7 @@ mod test {
             let _ = conns.pop();
 
             // then, wait for a bit to let the connection be reclaimed
-            delay_for(Duration::from_millis(50)).await;
+            sleep(Duration::from_millis(50)).await;
 
             // now check that we have the expected # of connections
             // this may look a little funky, but think of it this way:
@@ -599,7 +594,7 @@ mod test {
         assert_eq!(wait_timeout, Some(3));
         assert_eq!(ex_field!(pool, exist), 1);
 
-        delay_for(Duration::from_secs(6)).await;
+        sleep(Duration::from_secs(6)).await;
 
         let mut conn = pool.get_conn().await?;
         let id2: Option<usize> = conn.query_first("SELECT CONNECTION_ID()").await?;
@@ -640,8 +635,7 @@ mod test {
         const PANIC_MESSAGE: &str = "ORIGINAL_PANIC";
 
         let result = std::panic::catch_unwind(|| {
-            runtime::Builder::new()
-                .basic_scheduler()
+            runtime::Builder::new_current_thread()
                 .enable_all()
                 .build()
                 .unwrap()
@@ -662,7 +656,7 @@ mod 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 rt = tokio::runtime::Runtime::new().unwrap();
             let (tx, rx) = tokio::sync::oneshot::channel();
             rt.block_on(async move {
                 let pool = Pool::new(get_opts());
@@ -682,7 +676,7 @@ mod 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 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());
@@ -765,10 +759,14 @@ mod test {
             // we need to cancel the op in the middle
             // this should not lead to the `packet out of order` error.
             let delay_micros = rand::random::<u128>() % max_delay;
-            select(delay_for(Duration::from_micros(delay_micros as u64)), fut).await;
+            select(
+                sleep(Duration::from_micros(delay_micros as u64)).boxed(),
+                fut,
+            )
+            .await;
 
             // give some time for connections to return to the pool
-            delay_for(Duration::from_millis(100)).await;
+            sleep(Duration::from_millis(100)).await;
         }
         Ok(())
     }
diff --git a/src/conn/pool/recycler.rs b/src/conn/pool/recycler.rs
index 6bc44ee3..da3104d6 100644
--- a/src/conn/pool/recycler.rs
+++ b/src/conn/pool/recycler.rs
@@ -92,7 +92,7 @@ impl Future for Recycler {
 
         while !self.eof {
             // see if there are more connections for us to recycle
-            match Pin::new(&mut self.dropped).poll_next(cx) {
+            match Pin::new(&mut self.dropped).poll_recv(cx) {
                 Poll::Ready(Some(Some(conn))) => {
                     assert!(conn.inner.pool.is_none());
                     conn_decision!(self, conn);
diff --git a/src/conn/pool/ttl_check_inerval.rs b/src/conn/pool/ttl_check_inerval.rs
index 8f8d2b7c..0cb4f5f4 100644
--- a/src/conn/pool/ttl_check_inerval.rs
+++ b/src/conn/pool/ttl_check_inerval.rs
@@ -6,10 +6,7 @@
 // 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::{ok, FutureExt},
-    stream::{StreamExt, StreamFuture},
-};
+use futures_util::future::{ok, FutureExt};
 use pin_project::pin_project;
 use tokio::time::{self, Interval};
 
@@ -32,14 +29,14 @@ use std::pin::Pin;
 pub(crate) struct TtlCheckInterval {
     inner: Arc<Inner>,
     #[pin]
-    interval: StreamFuture<Interval>,
+    interval: Interval,
     pool_opts: PoolOpts,
 }
 
 impl TtlCheckInterval {
     /// Creates new `TtlCheckInterval`.
     pub fn new(pool_opts: PoolOpts, inner: Arc<Inner>) -> Self {
-        let interval = time::interval(pool_opts.ttl_check_interval()).into_future();
+        let interval = time::interval(pool_opts.ttl_check_interval());
         Self {
             inner,
             interval,
@@ -76,12 +73,11 @@ impl Future for TtlCheckInterval {
 
     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 _ = futures_core::ready!(self.as_mut().project().interval.poll_tick(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/error.rs b/src/error.rs
index 376044ca..288c51a7 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -149,6 +149,9 @@ pub enum DriverError {
 
     #[error("Bad compressed packet header.")]
     BadCompressedPacketHeader,
+
+    #[error("Named pipe connections temporary disabled (see tokio-rs/tokio#3118)")]
+    NamedPipesDisabled,
 }
 
 impl From<DriverError> for Error {
diff --git a/src/io/mod.rs b/src/io/mod.rs
index 901d19d5..41912e1f 100644
--- a/src/io/mod.rs
+++ b/src/io/mod.rs
@@ -8,34 +8,43 @@
 
 pub use self::{read_packet::ReadPacket, write_packet::WritePacket};
 
-use bytes::{BufMut, BytesMut};
+use bytes::BytesMut;
 use futures_core::{ready, stream};
 use futures_util::stream::{FuturesUnordered, StreamExt};
+use mio::net::{TcpKeepalive, TcpSocket};
 use mysql_common::proto::codec::PacketCodec as PacketCodecInner;
 use native_tls::{Certificate, Identity, TlsConnector};
 use pin_project::pin_project;
-use tokio::{io::ErrorKind::Interrupted, net::TcpStream, prelude::*};
+#[cfg(unix)]
+use tokio::io::AsyncWriteExt;
+use tokio::{
+    io::{AsyncRead, AsyncWrite, ErrorKind::Interrupted, ReadBuf},
+    net::TcpStream,
+};
 use tokio_util::codec::{Decoder, Encoder, Framed, FramedParts};
 
+#[cfg(unix)]
+use std::path::Path;
 use std::{
     fmt,
     fs::File,
     future::Future,
     io::{
         self,
-        ErrorKind::{NotConnected, Other, UnexpectedEof},
+        ErrorKind::{BrokenPipe, NotConnected, Other},
         Read,
     },
-    mem::MaybeUninit,
-    net::ToSocketAddrs,
+    net::{SocketAddr, ToSocketAddrs},
     ops::{Deref, DerefMut},
-    path::Path,
     pin::Pin,
     task::{Context, Poll},
     time::Duration,
 };
 
-use crate::{error::IoError, io::socket::Socket, opts::SslOpts};
+use crate::{error::IoError, opts::SslOpts};
+
+#[cfg(unix)]
+use crate::io::socket::Socket;
 
 macro_rules! with_interrupted {
     ($e:expr) => {
@@ -90,7 +99,8 @@ impl Encoder<Vec<u8>> for PacketCodec {
 #[derive(Debug)]
 pub(crate) enum Endpoint {
     Plain(Option<TcpStream>),
-    Secure(#[pin] tokio_tls::TlsStream<TcpStream>),
+    Secure(#[pin] tokio_native_tls::TlsStream<TcpStream>),
+    #[cfg(unix)]
     Socket(#[pin] Socket),
 }
 
@@ -102,11 +112,18 @@ struct CheckTcpStream<'a>(&'a mut TcpStream);
 
 impl Future for CheckTcpStream<'_> {
     type Output = io::Result<()>;
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let buf = &mut [0_u8];
-        match self.0.poll_peek(cx, buf) {
-            Poll::Ready(Ok(0)) => Poll::Ready(Err(io::Error::new(UnexpectedEof, "broken pipe"))),
-            Poll::Ready(Ok(_)) => Poll::Ready(Err(io::Error::new(Other, "unexpected read"))),
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match self.0.poll_read_ready(cx) {
+            Poll::Ready(Ok(())) => {
+                // stream is readable
+                let mut buf = [0_u8; 1];
+                match self.0.try_read(&mut buf) {
+                    Ok(0) => Poll::Ready(Err(io::Error::new(BrokenPipe, "broken pipe"))),
+                    Ok(_) => Poll::Ready(Err(io::Error::new(Other, "stream should be empty"))),
+                    Err(err) if err.kind() == io::ErrorKind::WouldBlock => Poll::Ready(Ok(())),
+                    Err(err) => Poll::Ready(Err(err)),
+                }
+            }
             Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
             Poll::Pending => Poll::Ready(Ok(())),
         }
@@ -116,15 +133,17 @@ impl Future for CheckTcpStream<'_> {
 impl Endpoint {
     /// Checks, that connection is alive.
     async fn check(&mut self) -> std::result::Result<(), IoError> {
+        //return Ok(());
         match self {
             Endpoint::Plain(Some(stream)) => {
                 CheckTcpStream(stream).await?;
                 Ok(())
             }
             Endpoint::Secure(tls_stream) => {
-                CheckTcpStream(tls_stream.get_mut()).await?;
+                CheckTcpStream(tls_stream.get_mut().get_mut().get_mut()).await?;
                 Ok(())
             }
+            #[cfg(unix)]
             Endpoint::Socket(socket) => {
                 socket.write(&[]).await?;
                 Ok(())
@@ -134,29 +153,17 @@ impl Endpoint {
     }
 
     pub fn is_secure(&self) -> bool {
-        if let Endpoint::Secure(_) = self {
-            true
-        } else {
-            false
-        }
-    }
-
-    pub fn set_keepalive_ms(&self, ms: Option<u32>) -> io::Result<()> {
-        let ms = ms.map(|val| Duration::from_millis(u64::from(val)));
-        match *self {
-            Endpoint::Plain(Some(ref stream)) => stream.set_keepalive(ms)?,
-            Endpoint::Plain(None) => unreachable!(),
-            Endpoint::Secure(ref stream) => stream.get_ref().set_keepalive(ms)?,
-            Endpoint::Socket(_) => (/* inapplicable */),
-        }
-        Ok(())
+        matches!(self, Endpoint::Secure(_))
     }
 
     pub fn set_tcp_nodelay(&self, val: bool) -> io::Result<()> {
         match *self {
             Endpoint::Plain(Some(ref stream)) => stream.set_nodelay(val)?,
             Endpoint::Plain(None) => unreachable!(),
-            Endpoint::Secure(ref stream) => stream.get_ref().set_nodelay(val)?,
+            Endpoint::Secure(ref stream) => {
+                stream.get_ref().get_ref().get_ref().set_nodelay(val)?
+            }
+            #[cfg(unix)]
             Endpoint::Socket(_) => (/* inapplicable */),
         }
         Ok(())
@@ -167,33 +174,31 @@ impl Endpoint {
         domain: String,
         ssl_opts: SslOpts,
     ) -> std::result::Result<(), IoError> {
+        #[cfg(unix)]
         if let Endpoint::Socket(_) = self {
             // inapplicable
             return Ok(());
         }
 
         let mut builder = TlsConnector::builder();
-        match ssl_opts.root_cert_path() {
-            Some(root_cert_path) => {
-                let mut root_cert_data = vec![];
-                let mut root_cert_file = File::open(root_cert_path)?;
-                root_cert_file.read_to_end(&mut root_cert_data)?;
-
-                let root_certs = Certificate::from_der(&*root_cert_data)
-                    .map(|x| vec![x])
-                    .or_else(|_| {
-                        pem::parse_many(&*root_cert_data)
-                            .iter()
-                            .map(pem::encode)
-                            .map(|s| Certificate::from_pem(s.as_bytes()))
-                            .collect()
-                    })?;
-
-                for root_cert in root_certs {
-                    builder.add_root_certificate(root_cert);
-                }
+        if let Some(root_cert_path) = ssl_opts.root_cert_path() {
+            let mut root_cert_data = vec![];
+            let mut root_cert_file = File::open(root_cert_path)?;
+            root_cert_file.read_to_end(&mut root_cert_data)?;
+
+            let root_certs = Certificate::from_der(&*root_cert_data)
+                .map(|x| vec![x])
+                .or_else(|_| {
+                    pem::parse_many(&*root_cert_data)
+                        .iter()
+                        .map(pem::encode)
+                        .map(|s| Certificate::from_pem(s.as_bytes()))
+                        .collect()
+                })?;
+
+            for root_cert in root_certs {
+                builder.add_root_certificate(root_cert);
             }
-            None => (),
         }
         if let Some(pkcs12_path) = ssl_opts.pkcs12_path() {
             let der = std::fs::read(pkcs12_path)?;
@@ -202,7 +207,7 @@ impl Endpoint {
         }
         builder.danger_accept_invalid_hostnames(ssl_opts.skip_domain_validation());
         builder.danger_accept_invalid_certs(ssl_opts.accept_invalid_certs());
-        let tls_connector: tokio_tls::TlsConnector = builder.build()?.into();
+        let tls_connector: tokio_native_tls::TlsConnector = builder.build()?.into();
 
         *self = match self {
             Endpoint::Plain(stream) => {
@@ -210,7 +215,9 @@ impl Endpoint {
                 let tls_stream = tls_connector.connect(&*domain, stream).await?;
                 Endpoint::Secure(tls_stream)
             }
-            Endpoint::Secure(_) | Endpoint::Socket(_) => unreachable!(),
+            Endpoint::Secure(_) => unreachable!(),
+            #[cfg(unix)]
+            Endpoint::Socket(_) => unreachable!(),
         };
 
         Ok(())
@@ -223,14 +230,15 @@ impl From<TcpStream> for Endpoint {
     }
 }
 
+#[cfg(unix)]
 impl From<Socket> for Endpoint {
     fn from(socket: Socket) -> Self {
         Endpoint::Socket(socket)
     }
 }
 
-impl From<tokio_tls::TlsStream<TcpStream>> for Endpoint {
-    fn from(stream: tokio_tls::TlsStream<TcpStream>) -> Self {
+impl From<tokio_native_tls::TlsStream<TcpStream>> for Endpoint {
+    fn from(stream: tokio_native_tls::TlsStream<TcpStream>) -> Self {
         Endpoint::Secure(stream)
     }
 }
@@ -238,45 +246,19 @@ impl From<tokio_tls::TlsStream<TcpStream>> for Endpoint {
 impl AsyncRead for Endpoint {
     fn poll_read(
         self: Pin<&mut Self>,
-        cx: &mut Context,
-        buf: &mut [u8],
-    ) -> Poll<std::result::Result<usize, tokio::io::Error>> {
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::result::Result<(), tokio::io::Error>> {
         let mut this = self.project();
         with_interrupted!(match this {
             EndpointProj::Plain(ref mut stream) => {
                 Pin::new(stream.as_mut().unwrap()).poll_read(cx, buf)
             }
             EndpointProj::Secure(ref mut stream) => stream.as_mut().poll_read(cx, buf),
+            #[cfg(unix)]
             EndpointProj::Socket(ref mut stream) => stream.as_mut().poll_read(cx, buf),
         })
     }
-
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
-        match self {
-            Endpoint::Plain(Some(stream)) => stream.prepare_uninitialized_buffer(buf),
-            Endpoint::Plain(None) => unreachable!(),
-            Endpoint::Secure(stream) => stream.prepare_uninitialized_buffer(buf),
-            Endpoint::Socket(stream) => stream.prepare_uninitialized_buffer(buf),
-        }
-    }
-
-    fn poll_read_buf<B>(
-        self: Pin<&mut Self>,
-        cx: &mut Context,
-        buf: &mut B,
-    ) -> Poll<std::result::Result<usize, tokio::io::Error>>
-    where
-        B: BufMut,
-    {
-        let mut this = self.project();
-        with_interrupted!(match this {
-            EndpointProj::Plain(ref mut stream) => {
-                Pin::new(stream.as_mut().unwrap()).poll_read_buf(cx, buf)
-            }
-            EndpointProj::Secure(ref mut stream) => stream.as_mut().poll_read_buf(cx, buf),
-            EndpointProj::Socket(ref mut stream) => stream.as_mut().poll_read_buf(cx, buf),
-        })
-    }
 }
 
 impl AsyncWrite for Endpoint {
@@ -291,6 +273,7 @@ impl AsyncWrite for Endpoint {
                 Pin::new(stream.as_mut().unwrap()).poll_write(cx, buf)
             }
             EndpointProj::Secure(ref mut stream) => stream.as_mut().poll_write(cx, buf),
+            #[cfg(unix)]
             EndpointProj::Socket(ref mut stream) => stream.as_mut().poll_write(cx, buf),
         })
     }
@@ -305,6 +288,7 @@ impl AsyncWrite for Endpoint {
                 Pin::new(stream.as_mut().unwrap()).poll_flush(cx)
             }
             EndpointProj::Secure(ref mut stream) => stream.as_mut().poll_flush(cx),
+            #[cfg(unix)]
             EndpointProj::Socket(ref mut stream) => stream.as_mut().poll_flush(cx),
         })
     }
@@ -319,6 +303,7 @@ impl AsyncWrite for Endpoint {
                 Pin::new(stream.as_mut().unwrap()).poll_shutdown(cx)
             }
             EndpointProj::Secure(ref mut stream) => stream.as_mut().poll_shutdown(cx),
+            #[cfg(unix)]
             EndpointProj::Socket(ref mut stream) => stream.as_mut().poll_shutdown(cx),
         })
     }
@@ -341,6 +326,7 @@ impl fmt::Debug for Stream {
 }
 
 impl Stream {
+    #[cfg(unix)]
     fn new<T: Into<Endpoint>>(endpoint: T) -> Self {
         let endpoint = endpoint.into();
 
@@ -350,16 +336,70 @@ impl Stream {
         }
     }
 
-    pub(crate) async fn connect_tcp<S>(addr: S) -> io::Result<Stream>
+    pub(crate) async fn connect_tcp<S>(addr: S, keepalive: Option<Duration>) -> io::Result<Stream>
     where
         S: ToSocketAddrs,
     {
+        // TODO: Use tokio to setup keepalive (see tokio-rs/tokio#3082)
+        async fn connect_stream(
+            addr: SocketAddr,
+            keepalive_opts: Option<TcpKeepalive>,
+        ) -> io::Result<TcpStream> {
+            let socket = if addr.is_ipv6() {
+                TcpSocket::new_v6()?
+            } else {
+                TcpSocket::new_v4()?
+            };
+
+            if let Some(keepalive_opts) = keepalive_opts {
+                socket.set_keepalive_params(keepalive_opts)?;
+            }
+
+            let stream = tokio::task::spawn_blocking(move || {
+                let mut stream = socket.connect(addr)?;
+                let mut poll = mio::Poll::new()?;
+                let mut events = mio::Events::with_capacity(1024);
+                poll.registry()
+                    .register(&mut stream, mio::Token(0), mio::Interest::WRITABLE)?;
+                loop {
+                    poll.poll(&mut events, None)?;
+
+                    for event in &events {
+                        if event.token() == mio::Token(0) && event.is_writable() {
+                            // The socket connected (probably, it could still be a spurious
+                            // wakeup)
+                            return Ok::<_, io::Error>(stream);
+                        }
+                    }
+                }
+            })
+            .await??;
+
+            #[cfg(unix)]
+            let std_stream = unsafe {
+                use std::os::unix::prelude::*;
+                let fd = stream.into_raw_fd();
+                std::net::TcpStream::from_raw_fd(fd)
+            };
+
+            #[cfg(windows)]
+            let std_stream = unsafe {
+                use std::os::windows::prelude::*;
+                let fd = stream.into_raw_socket();
+                std::net::TcpStream::from_raw_socket(fd)
+            };
+
+            Ok(TcpStream::from_std(std_stream)?)
+        }
+
+        let keepalive_opts = keepalive.map(|time| TcpKeepalive::new().with_time(time));
+
         match addr.to_socket_addrs() {
             Ok(addresses) => {
                 let mut streams = FuturesUnordered::new();
 
                 for address in addresses {
-                    streams.push(TcpStream::connect(address));
+                    streams.push(connect_stream(address, keepalive_opts.clone()));
                 }
 
                 let mut err = None;
@@ -379,27 +419,23 @@ impl Stream {
                 }
 
                 if let Some(e) = err {
-                    Err(e.into())
+                    Err(e)
                 } else {
                     Err(io::Error::new(
                         io::ErrorKind::InvalidInput,
                         "could not resolve to any address",
-                    )
-                    .into())
+                    ))
                 }
             }
-            Err(err) => Err(err.into()),
+            Err(err) => Err(err),
         }
     }
 
+    #[cfg(unix)]
     pub(crate) async fn connect_socket<P: AsRef<Path>>(path: P) -> io::Result<Stream> {
         Ok(Stream::new(Socket::new(path).await?))
     }
 
-    pub(crate) fn set_keepalive_ms(&self, ms: Option<u32>) -> io::Result<()> {
-        self.codec.as_ref().unwrap().get_ref().set_keepalive_ms(ms)
-    }
-
     pub(crate) fn set_tcp_nodelay(&self, val: bool) -> io::Result<()> {
         self.codec.as_ref().unwrap().get_ref().set_tcp_nodelay(val)
     }
@@ -481,3 +517,38 @@ impl stream::Stream for Stream {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    #[cfg(unix)] // no sane way to retrieve current keepalive value on windows
+    #[tokio::test]
+    async fn should_connect_with_keepalive() {
+        use crate::{test_misc::get_opts, Conn};
+
+        let opts = get_opts()
+            .tcp_keepalive(Some(42_000_u32))
+            .prefer_socket(false);
+        let mut conn: Conn = Conn::new(opts).await.unwrap();
+        let stream = conn.stream_mut().unwrap();
+        let endpoint = stream.codec.as_mut().unwrap().get_ref();
+        let stream = match endpoint {
+            super::Endpoint::Plain(Some(stream)) => stream,
+            super::Endpoint::Secure(tls_stream) => tls_stream.get_ref().get_ref().get_ref(),
+            _ => unreachable!(),
+        };
+        let sock = unsafe {
+            use std::os::unix::prelude::*;
+            let raw = stream.as_raw_fd();
+            socket2::Socket::from_raw_fd(raw)
+        };
+
+        assert_eq!(
+            sock.keepalive().unwrap(),
+            Some(std::time::Duration::from_millis(42_000)),
+        );
+
+        std::mem::forget(sock);
+
+        conn.disconnect().await.unwrap();
+    }
+}
diff --git a/src/io/read_packet.rs b/src/io/read_packet.rs
index 4580dd89..558dbae0 100644
--- a/src/io/read_packet.rs
+++ b/src/io/read_packet.rs
@@ -41,15 +41,13 @@ impl Future for ReadPacket<'_, '_> {
         match packet_opt {
             Some(packet) => {
                 self.0.touch();
-                return Poll::Ready(Ok(packet));
-            }
-            None => {
-                return Poll::Ready(Err(Error::new(
-                    ErrorKind::UnexpectedEof,
-                    "connection closed",
-                )
-                .into()));
+                Poll::Ready(Ok(packet))
             }
+            None => Poll::Ready(Err(Error::new(
+                ErrorKind::UnexpectedEof,
+                "connection closed",
+            )
+            .into())),
         }
     }
 }
diff --git a/src/io/socket.rs b/src/io/socket.rs
index f1a96ff5..42aae58f 100644
--- a/src/io/socket.rs
+++ b/src/io/socket.rs
@@ -6,20 +6,18 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use bytes::BufMut;
+#![cfg(unix)]
+
 use pin_project::pin_project;
-use tokio::{
-    io::{Error, ErrorKind::Interrupted},
-    prelude::*,
-};
+use tokio::io::{Error, ErrorKind::Interrupted, ReadBuf};
 
 use std::{
     io,
-    mem::MaybeUninit,
     path::Path,
     pin::Pin,
     task::{Context, Poll},
 };
+use tokio::io::{AsyncRead, AsyncWrite};
 
 /// Unix domain socket connection on unix, or named pipe connection on windows.
 #[pin_project]
@@ -28,9 +26,6 @@ pub(crate) struct Socket {
     #[pin]
     #[cfg(unix)]
     inner: tokio::net::UnixStream,
-    #[pin]
-    #[cfg(windows)]
-    inner: tokio::io::PollEvented<mio_named_pipes::NamedPipe>,
 }
 
 impl Socket {
@@ -41,43 +36,17 @@ impl Socket {
             inner: tokio::net::UnixStream::connect(path).await?,
         })
     }
-
-    /// Connects a new socket.
-    #[cfg(windows)]
-    pub async fn new<P: AsRef<Path>>(path: P) -> Result<Socket, io::Error> {
-        let pipe = mio_named_pipes::NamedPipe::new(path.as_ref())?;
-        pipe.connect()?;
-        Ok(Socket {
-            inner: tokio::io::PollEvented::new(pipe)?,
-        })
-    }
 }
 
 impl AsyncRead for Socket {
     fn poll_read(
         self: Pin<&mut Self>,
-        cx: &mut Context,
-        buf: &mut [u8],
-    ) -> Poll<Result<usize, Error>> {
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<Result<(), Error>> {
         let mut this = self.project();
         with_interrupted!(this.inner.as_mut().poll_read(cx, buf))
     }
-
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
-        self.inner.prepare_uninitialized_buffer(buf)
-    }
-
-    fn poll_read_buf<B>(
-        self: Pin<&mut Self>,
-        cx: &mut Context,
-        buf: &mut B,
-    ) -> Poll<Result<usize, Error>>
-    where
-        B: BufMut,
-    {
-        let mut this = self.project();
-        with_interrupted!(this.inner.as_mut().poll_read_buf(cx, buf))
-    }
 }
 
 impl AsyncWrite for Socket {
diff --git a/src/lib.rs b/src/lib.rs
index bd9daf59..0ec9c13e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -284,10 +284,10 @@ pub mod test_misc {
     }
 
     pub fn test_compression() -> bool {
-        ["true", "1"].contains(&&*env::var("COMPRESS").unwrap_or("".into()))
+        ["true", "1"].contains(&&*env::var("COMPRESS").unwrap_or_default())
     }
 
     pub fn test_ssl() -> bool {
-        ["true", "1"].contains(&&*env::var("SSL").unwrap_or("".into()))
+        ["true", "1"].contains(&&*env::var("SSL").unwrap_or_default())
     }
 }
diff --git a/src/local_infile_handler/mod.rs b/src/local_infile_handler/mod.rs
index 348ffed7..eec396b8 100644
--- a/src/local_infile_handler/mod.rs
+++ b/src/local_infile_handler/mod.rs
@@ -7,9 +7,10 @@
 // modified, or distributed except according to those terms.
 
 use crate::error;
-use tokio::prelude::*;
+use mysql_common::uuid::Uuid;
 
 use std::{fmt, future::Future, marker::Unpin, pin::Pin, sync::Arc};
+use tokio::io::AsyncRead;
 
 pub mod builtin;
 
@@ -22,7 +23,6 @@ pub mod builtin;
 ///
 /// ```rust
 /// # use mysql_async::{prelude::*, test_misc::get_opts, OptsBuilder, Result, Error};
-/// # use tokio::prelude::*;
 /// # use std::env;
 /// # #[tokio::main]
 /// # async fn main() -> Result<()> {
@@ -88,22 +88,21 @@ pub type InfileHandlerFuture = Pin<
 
 /// Object used to wrap `T: LocalInfileHandler` inside of Opts.
 #[derive(Clone)]
-pub struct LocalInfileHandlerObject(Arc<dyn LocalInfileHandler>);
+pub struct LocalInfileHandlerObject(Uuid, Arc<dyn LocalInfileHandler>);
 
 impl LocalInfileHandlerObject {
     pub fn new<T: LocalInfileHandler + 'static>(handler: T) -> Self {
-        LocalInfileHandlerObject(Arc::new(handler))
+        LocalInfileHandlerObject(Uuid::new_v4(), Arc::new(handler))
     }
 
     pub fn clone_inner(&self) -> Arc<dyn LocalInfileHandler> {
-        self.0.clone()
+        self.1.clone()
     }
 }
 
 impl PartialEq for LocalInfileHandlerObject {
     fn eq(&self, other: &LocalInfileHandlerObject) -> bool {
-        self.0.as_ref() as *const dyn LocalInfileHandler
-            == other.0.as_ref() as *const dyn LocalInfileHandler
+        self.0.eq(&other.0)
     }
 }
 
diff --git a/src/opts.rs b/src/opts.rs
index 43a9ffd2..a4dac7f3 100644
--- a/src/opts.rs
+++ b/src/opts.rs
@@ -408,7 +408,7 @@ impl Opts {
 
         // We use the URL for socket address resolution later, so make
         // sure it has a port set.
-        if !url.port().is_some() {
+        if url.port().is_none() {
             url.set_port(Some(DEFAULT_PORT))
                 .map_err(|_| UrlError::Invalid)?;
         }
@@ -596,7 +596,7 @@ impl Opts {
         self.inner.mysql_opts.ssl_opts.as_ref()
     }
 
-    /// Prefer socket connection (defaults to `true`).
+    /// Prefer socket connection (defaults to `true` **temporary `false` on Windows platform**).
     ///
     /// Will reconnect via socket (or named pipe on Windows) after TCP connection to `127.0.0.1`
     /// if `true`.
@@ -637,7 +637,7 @@ impl Opts {
     /// # Ok(()) }
     /// ```
     pub fn socket(&self) -> Option<&str> {
-        self.inner.mysql_opts.socket.as_ref().map(|x| &**x)
+        self.inner.mysql_opts.socket.as_deref()
     }
 
     /// If not `None`, then client will ask for compression if server supports it
@@ -697,7 +697,7 @@ impl Default for MysqlOpts {
             conn_ttl: None,
             stmt_cache_size: DEFAULT_STMT_CACHE_SIZE,
             ssl_opts: None,
-            prefer_socket: true,
+            prefer_socket: cfg!(not(target_os = "windows")),
             socket: None,
             compression: None,
         }
@@ -926,7 +926,7 @@ impl From<OptsBuilder> for Opts {
 
 fn get_opts_user_from_url(url: &Url) -> Option<String> {
     let user = url.username();
-    if user != "" {
+    if !user.is_empty() {
         Some(
             percent_decode(user.as_ref())
                 .decode_utf8_lossy()
@@ -1110,7 +1110,7 @@ fn mysqlopts_from_url(url: &Url) -> std::result::Result<MysqlOpts, UrlError> {
             } else {
                 return Err(UrlError::InvalidParamValue {
                     param: "compression".into(),
-                    value: value.to_string(),
+                    value,
                 });
             }
         } else {
diff --git a/src/queryable/mod.rs b/src/queryable/mod.rs
index 8713869a..28cb02e0 100644
--- a/src/queryable/mod.rs
+++ b/src/queryable/mod.rs
@@ -155,7 +155,7 @@ pub trait Queryable: Send {
         Q: AsRef<str> + Send + Sync + 'a,
         T: FromRow + Send + 'static;
 
-    /// Performs the given query and returns the first row of the first result set.
+    /// Performs the given query and maps each row of the first result set.
     fn query_map<'a, T, F, Q, U>(&'a mut self, query: Q, f: F) -> BoxFuture<'a, Vec<U>>
     where
         Q: AsRef<str> + Send + Sync + 'a,
@@ -212,7 +212,7 @@ pub trait Queryable: Send {
         P: Into<Params> + Send + 'b,
         T: FromRow + Send + 'static;
 
-    /// Exectues the given stmt and folds the first result set to a signel value.
+    /// Exectues the given stmt and maps each row of the first result set.
     ///
     /// It'll prepare `stmt`, if necessary.
     fn exec_map<'a: 'b, 'b, T, S, P, U, F>(
diff --git a/src/queryable/query_result.rs b/src/queryable/query_result.rs
index 1571bd60..806e5715 100644
--- a/src/queryable/query_result.rs
+++ b/src/queryable/query_result.rs
@@ -323,7 +323,7 @@ where
     /// Drops this query result.
     pub async fn drop_result(mut self) -> Result<()> {
         loop {
-            while let Some(_) = self.next().await? {}
+            while self.next().await?.is_some() {}
             if self.conn.get_pending_result().is_none() {
                 break Ok(());
             }


Test patch: