Upgrade to mio 0.7

Cargo.toml

@@ -31,7 +31,10 @@ default = [
   "kv-log-macro",
   "log",
   "mio",
-  "mio-uds",
+  "mio/os-poll",
+  "mio/tcp",
+  "mio/udp",
+  "mio/uds",
   "num_cpus",
   "pin-project-lite",
 ]
@@ -67,8 +70,7 @@ futures-timer = { version = "3.0.2", optional = true }
 kv-log-macro = { version = "1.0.4", optional = true }
 log = { version = "0.4.8", features = ["kv_unstable"], optional = true }
 memchr = { version = "2.3.3", optional = true }
-mio = { version = "0.6.19", optional = true }
-mio-uds = { version = "0.6.7", optional = true }
+mio = { version = "0.7.0", optional = true }
 num_cpus = { version = "1.12.0", optional = true }
 once_cell = { version = "1.3.1", optional = true }
 pin-project-lite = { version = "0.1.4", optional = true }

src/net/tcp/listener.rs

@@ -5,8 +5,8 @@ use std::sync::Arc;
 
 use crate::future;
 use crate::io;
-use crate::rt::Watcher;
 use crate::net::{TcpStream, ToSocketAddrs};
+use crate::rt::Watcher;
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
@@ -79,7 +79,7 @@ impl TcpListener {
         let addrs = addrs.to_socket_addrs().await?;
 
         for addr in addrs {
-            match mio::net::TcpListener::bind(&addr) {
+            match mio::net::TcpListener::bind(addr) {
                 Ok(mio_listener) => {
                     return Ok(TcpListener {
                         watcher: Watcher::new(mio_listener),
@@ -114,11 +114,9 @@ impl TcpListener {
     /// # Ok(()) }) }
     /// ```
     pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
-        let (io, addr) =
-            future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.accept_std()))
-                .await?;
+        let (mio_stream, addr) =
+            future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.accept())).await?;
 
-        let mio_stream = mio::net::TcpStream::from_stream(io)?;
         let stream = TcpStream {
             watcher: Arc::new(Watcher::new(mio_stream)),
         };
@@ -206,7 +204,12 @@ impl<'a> Stream for Incoming<'a> {
 impl From<std::net::TcpListener> for TcpListener {
     /// Converts a `std::net::TcpListener` into its asynchronous equivalent.
     fn from(listener: std::net::TcpListener) -> TcpListener {
-        let mio_listener = mio::net::TcpListener::from_std(listener).unwrap();
+        // Make sure we are in nonblocking mode.
+        listener
+            .set_nonblocking(true)
+            .expect("failed to set nonblocking mode");
+
+        let mio_listener = mio::net::TcpListener::from_std(listener);
         TcpListener {
             watcher: Watcher::new(mio_listener),
         }

src/net/tcp/stream.rs

@@ -5,8 +5,8 @@ use std::sync::Arc;
 
 use crate::future;
 use crate::io::{self, Read, Write};
-use crate::rt::Watcher;
 use crate::net::ToSocketAddrs;
+use crate::rt::Watcher;
 use crate::task::{Context, Poll};
 
 /// A TCP stream between a local and a remote socket.
@@ -79,7 +79,7 @@ impl TcpStream {
             // when it returns with `Ok`. We therefore wait for write readiness to
             // be sure the connection has either been established or there was an
             // error which we check for afterwards.
-            let watcher = match mio::net::TcpStream::connect(&addr) {
+            let watcher = match mio::net::TcpStream::connect(addr) {
                 Ok(s) => Watcher::new(s),
                 Err(e) => {
                     last_err = Some(e);
@@ -343,6 +343,7 @@ impl Write for TcpStream {
     }
 
     fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.shutdown(std::net::Shutdown::Write)?;
         Pin::new(&mut &*self).poll_close(cx)
     }
 }
@@ -370,7 +371,12 @@ impl Write for &TcpStream {
 impl From<std::net::TcpStream> for TcpStream {
     /// Converts a `std::net::TcpStream` into its asynchronous equivalent.
     fn from(stream: std::net::TcpStream) -> TcpStream {
-        let mio_stream = mio::net::TcpStream::from_stream(stream).unwrap();
+        // Make sure we are in nonblocking mode.
+        stream
+            .set_nonblocking(true)
+            .expect("failed to set nonblocking mode");
+
+        let mio_stream = mio::net::TcpStream::from_std(stream);
         TcpStream {
             watcher: Arc::new(Watcher::new(mio_stream)),
         }

src/net/udp/mod.rs

@@ -69,12 +69,10 @@ impl UdpSocket {
     /// ```
     pub async fn bind<A: ToSocketAddrs>(addrs: A) -> io::Result<UdpSocket> {
         let mut last_err = None;
-        let addrs = addrs
-            .to_socket_addrs()
-            .await?;
+        let addrs = addrs.to_socket_addrs().await?;
 
         for addr in addrs {
-            match mio::net::UdpSocket::bind(&addr) {
+            match mio::net::UdpSocket::bind(addr) {
                 Ok(mio_socket) => {
                     return Ok(UdpSocket {
                         watcher: Watcher::new(mio_socket),
@@ -155,7 +153,7 @@ impl UdpSocket {
 
         future::poll_fn(|cx| {
             self.watcher
-                .poll_write_with(cx, |inner| inner.send_to(buf, &addr))
+                .poll_write_with(cx, |inner| inner.send_to(buf, addr))
         })
         .await
         .context(|| format!("could not send packet to {}", addr))
@@ -498,7 +496,12 @@ impl UdpSocket {
 impl From<std::net::UdpSocket> for UdpSocket {
     /// Converts a `std::net::UdpSocket` into its asynchronous equivalent.
     fn from(socket: std::net::UdpSocket) -> UdpSocket {
-        let mio_socket = mio::net::UdpSocket::from_socket(socket).unwrap();
+        // Make sure we are in nonblocking mode.
+        socket
+            .set_nonblocking(true)
+            .expect("failed to set nonblocking mode");
+
+        let mio_socket = mio::net::UdpSocket::from_std(socket);
         UdpSocket {
             watcher: Watcher::new(mio_socket),
         }

src/os/unix/net/datagram.rs

@@ -3,15 +3,11 @@
 use std::fmt;
 use std::net::Shutdown;
 
-use mio_uds;
-
-use super::SocketAddr;
 use crate::future;
 use crate::io;
-use crate::rt::Watcher;
 use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use crate::path::Path;
-use crate::task::spawn_blocking;
+use crate::rt::Watcher;
 
 /// A Unix datagram socket.
 ///
@@ -42,11 +38,11 @@ use crate::task::spawn_blocking;
 /// # Ok(()) }) }
 /// ```
 pub struct UnixDatagram {
-    watcher: Watcher<mio_uds::UnixDatagram>,
+    watcher: Watcher<mio::net::UnixDatagram>,
 }
 
 impl UnixDatagram {
-    fn new(socket: mio_uds::UnixDatagram) -> UnixDatagram {
+    fn new(socket: mio::net::UnixDatagram) -> UnixDatagram {
         UnixDatagram {
             watcher: Watcher::new(socket),
         }
@@ -67,8 +63,8 @@ impl UnixDatagram {
     /// ```
     pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixDatagram> {
         let path = path.as_ref().to_owned();
-        let socket = spawn_blocking(move || mio_uds::UnixDatagram::bind(path)).await?;
-        Ok(UnixDatagram::new(socket))
+        let mio_socket = mio::net::UnixDatagram::bind(path)?;
+        Ok(UnixDatagram::new(mio_socket))
     }
 
     /// Creates a Unix datagram which is not bound to any address.
@@ -85,7 +81,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub fn unbound() -> io::Result<UnixDatagram> {
-        let socket = mio_uds::UnixDatagram::unbound()?;
+        let socket = mio::net::UnixDatagram::unbound()?;
         Ok(UnixDatagram::new(socket))
     }
 
@@ -105,7 +101,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub fn pair() -> io::Result<(UnixDatagram, UnixDatagram)> {
-        let (a, b) = mio_uds::UnixDatagram::pair()?;
+        let (a, b) = mio::net::UnixDatagram::pair()?;
         let a = UnixDatagram::new(a);
         let b = UnixDatagram::new(b);
         Ok((a, b))
@@ -152,7 +148,7 @@ impl UnixDatagram {
     /// #
     /// # Ok(()) }) }
     /// ```
-    pub fn local_addr(&self) -> io::Result<SocketAddr> {
+    pub fn local_addr(&self) -> io::Result<mio::net::SocketAddr> {
         self.watcher.get_ref().local_addr()
     }
 
@@ -175,7 +171,7 @@ impl UnixDatagram {
     /// #
     /// # Ok(()) }) }
     /// ```
-    pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+    pub fn peer_addr(&self) -> io::Result<mio::net::SocketAddr> {
         self.watcher.get_ref().peer_addr()
     }
 
@@ -196,7 +192,7 @@ impl UnixDatagram {
     /// #
     /// # Ok(()) }) }
     /// ```
-    pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
+    pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, mio::net::SocketAddr)> {
         future::poll_fn(|cx| {
             self.watcher
                 .poll_read_with(cx, |inner| inner.recv_from(buf))
@@ -315,7 +311,12 @@ impl fmt::Debug for UnixDatagram {
 impl From<std::os::unix::net::UnixDatagram> for UnixDatagram {
     /// Converts a `std::os::unix::net::UnixDatagram` into its asynchronous equivalent.
     fn from(datagram: std::os::unix::net::UnixDatagram) -> UnixDatagram {
-        let mio_datagram = mio_uds::UnixDatagram::from_datagram(datagram).unwrap();
+        // Make sure we are in nonblocking mode.
+        datagram
+            .set_nonblocking(true)
+            .expect("failed to set nonblocking mode");
+
+        let mio_datagram = mio::net::UnixDatagram::from_std(datagram);
         UnixDatagram {
             watcher: Watcher::new(mio_datagram),
         }

src/os/unix/net/listener.rs

@@ -1,20 +1,17 @@
 //! Unix-specific networking extensions.
 
 use std::fmt;
-use std::pin::Pin;
 use std::future::Future;
+use std::pin::Pin;
 
-use mio_uds;
-
-use super::SocketAddr;
 use super::UnixStream;
 use crate::future;
 use crate::io;
-use crate::rt::Watcher;
 use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use crate::path::Path;
+use crate::rt::Watcher;
 use crate::stream::Stream;
-use crate::task::{spawn_blocking, Context, Poll};
+use crate::task::{Context, Poll};
 
 /// A Unix domain socket server, listening for connections.
 ///
@@ -50,7 +47,7 @@ use crate::task::{spawn_blocking, Context, Poll};
 /// # Ok(()) }) }
 /// ```
 pub struct UnixListener {
-    watcher: Watcher<mio_uds::UnixListener>,
+    watcher: Watcher<mio::net::UnixListener>,
 }
 
 impl UnixListener {
@@ -69,10 +66,10 @@ impl UnixListener {
     /// ```
     pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixListener> {
         let path = path.as_ref().to_owned();
-        let listener = spawn_blocking(move || mio_uds::UnixListener::bind(path)).await?;
+        let mio_listener = mio::net::UnixListener::bind(path)?;
 
         Ok(UnixListener {
-            watcher: Watcher::new(listener),
+            watcher: Watcher::new(mio_listener),
         })
     }
 
@@ -92,28 +89,15 @@ impl UnixListener {
     /// #
     /// # Ok(()) }) }
     /// ```
-    pub async fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
+    pub async fn accept(&self) -> io::Result<(UnixStream, mio::net::SocketAddr)> {
         future::poll_fn(|cx| {
-            let res = futures_core::ready!(self.watcher.poll_read_with(cx, |inner| {
-                match inner.accept_std() {
-                    // Converting to `WouldBlock` so that the watcher will
-                    // add the waker of this task to a list of readers.
-                    Ok(None) => Err(io::ErrorKind::WouldBlock.into()),
-                    res => res,
-                }
-            }));
-
-            match res? {
-                Some((io, addr)) => {
-                    let mio_stream = mio_uds::UnixStream::from_stream(io)?;
-                    let stream = UnixStream {
-                        watcher: Watcher::new(mio_stream),
-                    };
-                    Poll::Ready(Ok((stream, addr)))
-                }
-                // This should never happen since `None` is converted to `WouldBlock`
-                None => unreachable!(),
-            }
+            let (mio_stream, addr) =
+                futures_core::ready!(self.watcher.poll_read_with(cx, |inner| { inner.accept() }))?;
+
+            let stream = UnixStream {
+                watcher: Watcher::new(mio_stream),
+            };
+            Poll::Ready(Ok((stream, addr)))
         })
         .await
     }
@@ -162,7 +146,7 @@ impl UnixListener {
     /// #
     /// # Ok(()) }) }
     /// ```
-    pub fn local_addr(&self) -> io::Result<SocketAddr> {
+    pub fn local_addr(&self) -> io::Result<mio::net::SocketAddr> {
         self.watcher.get_ref().local_addr()
     }
 }
@@ -209,7 +193,12 @@ impl Stream for Incoming<'_> {
 impl From<std::os::unix::net::UnixListener> for UnixListener {
     /// Converts a `std::os::unix::net::UnixListener` into its asynchronous equivalent.
     fn from(listener: std::os::unix::net::UnixListener) -> UnixListener {
-        let mio_listener = mio_uds::UnixListener::from_listener(listener).unwrap();
+        // Make sure we are in nonblocking mode.
+        listener
+            .set_nonblocking(true)
+            .expect("failed to set nonblocking mode");
+
+        let mio_listener = mio::net::UnixListener::from_std(listener);
         UnixListener {
             watcher: Watcher::new(mio_listener),
         }