Update mio to 0.8 (#4270)

tokio/Cargo.toml

@@ -49,20 +49,19 @@ macros = ["tokio-macros"]
 stats = []
 net = [
   "libc",
+  "mio/net",
+  "mio/os-ext",
   "mio/os-poll",
-  "mio/os-util",
-  "mio/tcp",
-  "mio/udp",
-  "mio/uds",
+  "socket2/all",
   "winapi/namedpipeapi",
 ]
 process = [
   "bytes",
   "once_cell",
   "libc",
+  "mio/net",
+  "mio/os-ext",
   "mio/os-poll",
-  "mio/os-util",
-  "mio/uds",
   "signal-hook-registry",
   "winapi/threadpoollegacyapiset",
 ]
@@ -75,9 +74,9 @@ rt-multi-thread = [
 signal = [
   "once_cell",
   "libc",
+  "mio/net",
+  "mio/os-ext",
   "mio/os-poll",
-  "mio/uds",
-  "mio/os-util",
   "signal-hook-registry",
   "winapi/consoleapi",
 ]
@@ -94,9 +93,10 @@ pin-project-lite = "0.2.0"
 bytes = { version = "1.0.0", optional = true }
 once_cell = { version = "1.5.2", optional = true }
 memchr = { version = "2.2", optional = true }
-mio = { version = "0.7.6", optional = true }
+mio = { version = "0.8.0", optional = true }
 num_cpus = { version = "1.8.0", optional = true }
 parking_lot = { version = "0.11.0", optional = true }
+socket2 = { version = "0.4.2", optional = true }
 
 # Currently unstable. The API exposed by these features may be broken at any time.
 # Requires `--cfg tokio_unstable` to enable.
@@ -128,7 +128,6 @@ proptest = "1"
 rand = "0.8.0"
 tempfile = "3.1.0"
 async-stream = "0.3"
-socket2 = "0.4"
 
 [target.'cfg(target_os = "freebsd")'.dev-dependencies]
 mio-aio = { version = "0.6.0", features = ["tokio"] }

tokio/src/net/tcp/socket.rs

@@ -1,5 +1,6 @@
 use crate::net::{TcpListener, TcpStream};
 
+use std::convert::TryInto;
 use std::fmt;
 use std::io;
 use std::net::SocketAddr;
@@ -84,7 +85,7 @@ cfg_net! {
     /// [`socket2`]: https://docs.rs/socket2/
     #[cfg_attr(docsrs, doc(alias = "connect_std"))]
     pub struct TcpSocket {
-        inner: mio::net::TcpSocket,
+        inner: socket2::Socket,
     }
 }
 
@@ -119,7 +120,11 @@ impl TcpSocket {
     /// }
     /// ```
     pub fn new_v4() -> io::Result<TcpSocket> {
-        let inner = mio::net::TcpSocket::new_v4()?;
+        let inner = socket2::Socket::new(
+            socket2::Domain::IPV4,
+            socket2::Type::STREAM,
+            Some(socket2::Protocol::TCP),
+        )?;
         Ok(TcpSocket { inner })
     }
 
@@ -153,7 +158,11 @@ impl TcpSocket {
     /// }
     /// ```
     pub fn new_v6() -> io::Result<TcpSocket> {
-        let inner = mio::net::TcpSocket::new_v6()?;
+        let inner = socket2::Socket::new(
+            socket2::Domain::IPV6,
+            socket2::Type::STREAM,
+            Some(socket2::Protocol::TCP),
+        )?;
         Ok(TcpSocket { inner })
     }
 
@@ -184,7 +193,7 @@ impl TcpSocket {
     /// }
     /// ```
     pub fn set_reuseaddr(&self, reuseaddr: bool) -> io::Result<()> {
-        self.inner.set_reuseaddr(reuseaddr)
+        self.inner.set_reuse_address(reuseaddr)
     }
 
     /// Retrieves the value set for `SO_REUSEADDR` on this socket.
@@ -210,7 +219,7 @@ impl TcpSocket {
     /// }
     /// ```
     pub fn reuseaddr(&self) -> io::Result<bool> {
-        self.inner.get_reuseaddr()
+        self.inner.reuse_address()
     }
 
     /// Allows the socket to bind to an in-use port. Only available for unix systems
@@ -244,7 +253,7 @@ impl TcpSocket {
         doc(cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos"))))
     )]
     pub fn set_reuseport(&self, reuseport: bool) -> io::Result<()> {
-        self.inner.set_reuseport(reuseport)
+        self.inner.set_reuse_port(reuseport)
     }
 
     /// Allows the socket to bind to an in-use port. Only available for unix systems
@@ -279,14 +288,14 @@ impl TcpSocket {
         doc(cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos"))))
     )]
     pub fn reuseport(&self) -> io::Result<bool> {
-        self.inner.get_reuseport()
+        self.inner.reuse_port()
     }
 
     /// Sets the size of the TCP send buffer on this socket.
     ///
     /// On most operating systems, this sets the `SO_SNDBUF` socket option.
     pub fn set_send_buffer_size(&self, size: u32) -> io::Result<()> {
-        self.inner.set_send_buffer_size(size)
+        self.inner.set_send_buffer_size(size as usize)
     }
 
     /// Returns the size of the TCP send buffer for this socket.
@@ -313,14 +322,14 @@ impl TcpSocket {
     ///
     /// [`set_send_buffer_size`]: #method.set_send_buffer_size
     pub fn send_buffer_size(&self) -> io::Result<u32> {
-        self.inner.get_send_buffer_size()
+        self.inner.send_buffer_size().map(|n| n as u32)
     }
 
     /// Sets the size of the TCP receive buffer on this socket.
     ///
     /// On most operating systems, this sets the `SO_RCVBUF` socket option.
     pub fn set_recv_buffer_size(&self, size: u32) -> io::Result<()> {
-        self.inner.set_recv_buffer_size(size)
+        self.inner.set_recv_buffer_size(size as usize)
     }
 
     /// Returns the size of the TCP receive buffer for this socket.
@@ -347,7 +356,7 @@ impl TcpSocket {
     ///
     /// [`set_recv_buffer_size`]: #method.set_recv_buffer_size
     pub fn recv_buffer_size(&self) -> io::Result<u32> {
-        self.inner.get_recv_buffer_size()
+        self.inner.recv_buffer_size().map(|n| n as u32)
     }
 
     /// Sets the linger duration of this socket by setting the SO_LINGER option.
@@ -395,7 +404,9 @@ impl TcpSocket {
     /// }
     /// ```
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
-        self.inner.get_localaddr()
+        self.inner
+            .local_addr()
+            .map(|addr| addr.as_socket().unwrap())
     }
 
     /// Binds the socket to the given address.
@@ -427,7 +438,7 @@ impl TcpSocket {
     /// }
     /// ```
     pub fn bind(&self, addr: SocketAddr) -> io::Result<()> {
-        self.inner.bind(addr)
+        self.inner.bind(&addr.into())
     }
 
     /// Establishes a TCP connection with a peer at the specified socket address.
@@ -463,7 +474,13 @@ impl TcpSocket {
     /// }
     /// ```
     pub async fn connect(self, addr: SocketAddr) -> io::Result<TcpStream> {
-        let mio = self.inner.connect(addr)?;
+        self.inner.connect(&addr.into())?;
+
+        #[cfg(windows)]
+        let mio = unsafe { mio::net::TcpStream::from_raw_socket(self.inner.into_raw_socket()) };
+        #[cfg(unix)]
+        let mio = unsafe { mio::net::TcpStream::from_raw_fd(self.inner.into_raw_fd()) };
+
         TcpStream::connect_mio(mio).await
     }
 
@@ -503,7 +520,14 @@ impl TcpSocket {
     /// }
     /// ```
     pub fn listen(self, backlog: u32) -> io::Result<TcpListener> {
-        let mio = self.inner.listen(backlog)?;
+        let backlog = backlog.try_into().unwrap_or(i32::MAX);
+        self.inner.listen(backlog)?;
+
+        #[cfg(windows)]
+        let mio = unsafe { mio::net::TcpListener::from_raw_socket(self.inner.into_raw_socket()) };
+        #[cfg(unix)]
+        let mio = unsafe { mio::net::TcpListener::from_raw_fd(self.inner.into_raw_fd()) };
+
         TcpListener::new(mio)
     }
 
@@ -523,7 +547,7 @@ impl TcpSocket {
     ///
     /// #[tokio::main]
     /// async fn main() -> std::io::Result<()> {
-    ///     
+    ///
     ///     let socket2_socket = Socket::new(Domain::IPV4, Type::STREAM, None)?;
     ///
     ///     let socket = TcpSocket::from_std_stream(socket2_socket.into());
@@ -534,16 +558,12 @@ impl TcpSocket {
     pub fn from_std_stream(std_stream: std::net::TcpStream) -> TcpSocket {
         #[cfg(unix)]
         {
-            use std::os::unix::io::{FromRawFd, IntoRawFd};
-
             let raw_fd = std_stream.into_raw_fd();
             unsafe { TcpSocket::from_raw_fd(raw_fd) }
         }
 
         #[cfg(windows)]
         {
-            use std::os::windows::io::{FromRawSocket, IntoRawSocket};
-
             let raw_socket = std_stream.into_raw_socket();
             unsafe { TcpSocket::from_raw_socket(raw_socket) }
         }
@@ -572,7 +592,7 @@ impl FromRawFd for TcpSocket {
     /// The caller is responsible for ensuring that the socket is in
     /// non-blocking mode.
     unsafe fn from_raw_fd(fd: RawFd) -> TcpSocket {
-        let inner = mio::net::TcpSocket::from_raw_fd(fd);
+        let inner = socket2::Socket::from_raw_fd(fd);
         TcpSocket { inner }
     }
 }
@@ -607,7 +627,7 @@ impl FromRawSocket for TcpSocket {
     /// The caller is responsible for ensuring that the socket is in
     /// non-blocking mode.
     unsafe fn from_raw_socket(socket: RawSocket) -> TcpSocket {
-        let inner = mio::net::TcpSocket::from_raw_socket(socket);
+        let inner = socket2::Socket::from_raw_socket(socket);
         TcpSocket { inner }
     }
 }

tokio/src/net/tcp/stream.rs

@@ -387,7 +387,7 @@ impl TcpStream {
     ///             // if the readiness event is a false positive.
     ///             match stream.try_read(&mut data) {
     ///                 Ok(n) => {
-    ///                     println!("read {} bytes", n);        
+    ///                     println!("read {} bytes", n);
     ///                 }
     ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
     ///                     continue;
@@ -1090,9 +1090,8 @@ impl TcpStream {
     /// # }
     /// ```
     pub fn linger(&self) -> io::Result<Option<Duration>> {
-        let mio_socket = std::mem::ManuallyDrop::new(self.to_mio());
-
-        mio_socket.get_linger()
+        let socket = self.to_socket();
+        socket.linger()
     }
 
     /// Sets the linger duration of this socket by setting the SO_LINGER option.
@@ -1117,23 +1116,12 @@ impl TcpStream {
     /// # }
     /// ```
     pub fn set_linger(&self, dur: Option<Duration>) -> io::Result<()> {
-        let mio_socket = std::mem::ManuallyDrop::new(self.to_mio());
-
-        mio_socket.set_linger(dur)
+        let socket = self.to_socket();
+        socket.set_linger(dur)
     }
 
-    fn to_mio(&self) -> mio::net::TcpSocket {
-        #[cfg(windows)]
-        {
-            use std::os::windows::io::{AsRawSocket, FromRawSocket};
-            unsafe { mio::net::TcpSocket::from_raw_socket(self.as_raw_socket()) }
-        }
-
-        #[cfg(unix)]
-        {
-            use std::os::unix::io::{AsRawFd, FromRawFd};
-            unsafe { mio::net::TcpSocket::from_raw_fd(self.as_raw_fd()) }
-        }
+    fn to_socket(&self) -> socket2::SockRef<'_> {
+        socket2::SockRef::from(self)
     }
 
     /// Gets the value of the `IP_TTL` option for this socket.