net: debug_assert on creating a tokio socket from a blocking one (#7166)

See #5595 and #7172.

This adds a debug assertion that checks that a supplied underlying std socket is set to nonblocking mode when constructing a tokio socket object from such an object.

This only works on unix.

Cargo.toml

@@ -23,6 +23,7 @@ unexpected_cfgs = { level = "warn", check-cfg = [
   'cfg(fuzzing)',
   'cfg(loom)',
   'cfg(mio_unsupported_force_poll_poll)',
+  'cfg(tokio_allow_from_blocking_fd)',
   'cfg(tokio_internal_mt_counters)',
   'cfg(tokio_no_parking_lot)',
   'cfg(tokio_no_tuning_tests)',

tokio/src/net/tcp/listener.rs

@@ -1,5 +1,6 @@
 use crate::io::{Interest, PollEvented};
 use crate::net::tcp::TcpStream;
+use crate::util::check_socket_for_blocking;
 
 cfg_not_wasi! {
     use crate::net::{to_socket_addrs, ToSocketAddrs};
@@ -209,6 +210,10 @@ impl TcpListener {
     /// will block the thread, which will cause unexpected behavior.
     /// Non-blocking mode can be set using [`set_nonblocking`].
     ///
+    /// Passing a listener in blocking mode is always erroneous,
+    /// and the behavior in that case may change in the future.
+    /// For example, it could panic.
+    ///
     /// [`set_nonblocking`]: std::net::TcpListener::set_nonblocking
     ///
     /// # Examples
@@ -236,6 +241,8 @@ impl TcpListener {
     /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
     #[track_caller]
     pub fn from_std(listener: net::TcpListener) -> io::Result<TcpListener> {
+        check_socket_for_blocking(&listener)?;
+
         let io = mio::net::TcpListener::from_std(listener);
         let io = PollEvented::new(io)?;
         Ok(TcpListener { io })

tokio/src/net/tcp/stream.rs

@@ -7,6 +7,7 @@ cfg_not_wasi! {
 use crate::io::{AsyncRead, AsyncWrite, Interest, PollEvented, ReadBuf, Ready};
 use crate::net::tcp::split::{split, ReadHalf, WriteHalf};
 use crate::net::tcp::split_owned::{split_owned, OwnedReadHalf, OwnedWriteHalf};
+use crate::util::check_socket_for_blocking;
 
 use std::fmt;
 use std::io;
@@ -173,6 +174,10 @@ impl TcpStream {
     /// will block the thread, which will cause unexpected behavior.
     /// Non-blocking mode can be set using [`set_nonblocking`].
     ///
+    /// Passing a listener in blocking mode is always erroneous,
+    /// and the behavior in that case may change in the future.
+    /// For example, it could panic.
+    ///
     /// [`set_nonblocking`]: std::net::TcpStream::set_nonblocking
     ///
     /// # Examples
@@ -200,6 +205,8 @@ impl TcpStream {
     /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
     #[track_caller]
     pub fn from_std(stream: std::net::TcpStream) -> io::Result<TcpStream> {
+        check_socket_for_blocking(&stream)?;
+
         let io = mio::net::TcpStream::from_std(stream);
         let io = PollEvented::new(io)?;
         Ok(TcpStream { io })

tokio/src/net/udp.rs

@@ -1,5 +1,6 @@
 use crate::io::{Interest, PollEvented, ReadBuf, Ready};
 use crate::net::{to_socket_addrs, ToSocketAddrs};
+use crate::util::check_socket_for_blocking;
 
 use std::fmt;
 use std::io;
@@ -192,6 +193,10 @@ impl UdpSocket {
     /// will block the thread, which will cause unexpected behavior.
     /// Non-blocking mode can be set using [`set_nonblocking`].
     ///
+    /// Passing a listener in blocking mode is always erroneous,
+    /// and the behavior in that case may change in the future.
+    /// For example, it could panic.
+    ///
     /// [`set_nonblocking`]: std::net::UdpSocket::set_nonblocking
     ///
     /// # Panics
@@ -220,6 +225,8 @@ impl UdpSocket {
     /// ```
     #[track_caller]
     pub fn from_std(socket: net::UdpSocket) -> io::Result<UdpSocket> {
+        check_socket_for_blocking(&socket)?;
+
         let io = mio::net::UdpSocket::from_std(socket);
         UdpSocket::new(io)
     }

tokio/src/net/unix/datagram/socket.rs

@@ -1,5 +1,6 @@
 use crate::io::{Interest, PollEvented, ReadBuf, Ready};
 use crate::net::unix::SocketAddr;
+use crate::util::check_socket_for_blocking;
 
 use std::fmt;
 use std::io;
@@ -449,6 +450,10 @@ impl UnixDatagram {
     /// will block the thread, which will cause unexpected behavior.
     /// Non-blocking mode can be set using [`set_nonblocking`].
     ///
+    /// Passing a listener in blocking mode is always erroneous,
+    /// and the behavior in that case may change in the future.
+    /// For example, it could panic.
+    ///
     /// [`set_nonblocking`]: std::os::unix::net::UnixDatagram::set_nonblocking
     ///
     /// # Panics
@@ -484,6 +489,8 @@ impl UnixDatagram {
     /// ```
     #[track_caller]
     pub fn from_std(datagram: net::UnixDatagram) -> io::Result<UnixDatagram> {
+        check_socket_for_blocking(&datagram)?;
+
         let socket = mio::net::UnixDatagram::from_std(datagram);
         let io = PollEvented::new(socket)?;
         Ok(UnixDatagram { io })

tokio/src/net/unix/listener.rs

@@ -1,5 +1,6 @@
 use crate::io::{Interest, PollEvented};
 use crate::net::unix::{SocketAddr, UnixStream};
+use crate::util::check_socket_for_blocking;
 
 use std::fmt;
 use std::io;
@@ -106,6 +107,10 @@ impl UnixListener {
     /// will block the thread, which will cause unexpected behavior.
     /// Non-blocking mode can be set using [`set_nonblocking`].
     ///
+    /// Passing a listener in blocking mode is always erroneous,
+    /// and the behavior in that case may change in the future.
+    /// For example, it could panic.
+    ///
     /// [`set_nonblocking`]: std::os::unix::net::UnixListener::set_nonblocking
     ///
     /// # Examples
@@ -133,6 +138,8 @@ impl UnixListener {
     /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
     #[track_caller]
     pub fn from_std(listener: net::UnixListener) -> io::Result<UnixListener> {
+        check_socket_for_blocking(&listener)?;
+
         let listener = mio::net::UnixListener::from_std(listener);
         let io = PollEvented::new(listener)?;
         Ok(UnixListener { io })

tokio/src/net/unix/stream.rs

@@ -3,6 +3,7 @@ use crate::net::unix::split::{split, ReadHalf, WriteHalf};
 use crate::net::unix::split_owned::{split_owned, OwnedReadHalf, OwnedWriteHalf};
 use crate::net::unix::ucred::{self, UCred};
 use crate::net::unix::SocketAddr;
+use crate::util::check_socket_for_blocking;
 
 use std::fmt;
 use std::future::poll_fn;
@@ -791,6 +792,10 @@ impl UnixStream {
     /// will block the thread, which will cause unexpected behavior.
     /// Non-blocking mode can be set using [`set_nonblocking`].
     ///
+    /// Passing a listener in blocking mode is always erroneous,
+    /// and the behavior in that case may change in the future.
+    /// For example, it could panic.
+    ///
     /// [`set_nonblocking`]: std::os::unix::net::UnixStream::set_nonblocking
     ///
     /// # Examples
@@ -818,6 +823,8 @@ impl UnixStream {
     /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
     #[track_caller]
     pub fn from_std(stream: net::UnixStream) -> io::Result<UnixStream> {
+        check_socket_for_blocking(&stream)?;
+
         let stream = mio::net::UnixStream::from_std(stream);
         let io = PollEvented::new(stream)?;
 

tokio/src/util/blocking_check.rs

@@ -0,0 +1,29 @@
+#[cfg(unix)]
+use std::os::fd::AsFd;
+
+#[cfg(unix)]
+#[allow(unused_variables)]
+#[track_caller]
+pub(crate) fn check_socket_for_blocking<S: AsFd>(s: &S) -> crate::io::Result<()> {
+    #[cfg(not(tokio_allow_from_blocking_fd))]
+    {
+        let sock = socket2::SockRef::from(s);
+
+        debug_assert!(
+            sock.nonblocking()?,
+            "Registering a blocking socket with the tokio runtime is unsupported. \
+            If you wish to do anyways, please add `--cfg tokio_allow_from_blocking_fd` to your \
+            RUSTFLAGS. See github.com/tokio-rs/tokio/issues/7172 for details."
+        );
+    }
+
+    Ok(())
+}
+
+#[cfg(not(unix))]
+#[allow(unused_variables)]
+pub(crate) fn check_socket_for_blocking<S>(s: &S) -> crate::io::Result<()> {
+    // we cannot retrieve the nonblocking status on windows
+    // and i dont know how to support wasi yet
+    Ok(())
+}

tokio/src/util/mod.rs

@@ -5,6 +5,12 @@ cfg_io_driver! {
 #[cfg(feature = "rt")]
 pub(crate) mod atomic_cell;
 
+#[cfg(feature = "net")]
+mod blocking_check;
+#[cfg(feature = "net")]
+#[allow(unused_imports)]
+pub(crate) use blocking_check::check_socket_for_blocking;
+
 pub(crate) mod metric_atomics;
 
 #[cfg(any(feature = "rt", feature = "signal", feature = "process"))]

tokio/tests/io_driver.rs

@@ -96,7 +96,10 @@ fn panics_when_io_disabled() {
     let rt = runtime::Builder::new_current_thread().build().unwrap();
 
     rt.block_on(async {
-        let _ =
-            tokio::net::TcpListener::from_std(std::net::TcpListener::bind("127.0.0.1:0").unwrap());
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+
+        listener.set_nonblocking(true).unwrap();
+
+        let _ = tokio::net::TcpListener::from_std(listener);
     });
 }

tokio/tests/io_driver_drop.rs

@@ -13,6 +13,9 @@ fn tcp_doesnt_block() {
     let listener = {
         let _enter = rt.enter();
         let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+
+        listener.set_nonblocking(true).unwrap();
+
         TcpListener::from_std(listener).unwrap()
     };
 
@@ -33,6 +36,9 @@ fn drop_wakes() {
     let listener = {
         let _enter = rt.enter();
         let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+
+        listener.set_nonblocking(true).unwrap();
+
         TcpListener::from_std(listener).unwrap()
     };
 

tokio/tests/no_rt.rs

@@ -39,5 +39,9 @@ async fn timeout_value() {
 )]
 #[cfg_attr(miri, ignore)] // No `socket` in miri.
 fn io_panics_when_no_tokio_context() {
-    let _ = tokio::net::TcpListener::from_std(std::net::TcpListener::bind("127.0.0.1:0").unwrap());
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+
+    listener.set_nonblocking(true).unwrap();
+
+    let _ = tokio::net::TcpListener::from_std(listener);
 }

tokio/tests/tcp_peek.rs

@@ -13,11 +13,18 @@ use std::{io::Write, net};
 #[tokio::test]
 async fn peek() {
     let listener = net::TcpListener::bind("127.0.0.1:0").unwrap();
+
     let addr = listener.local_addr().unwrap();
     let t = thread::spawn(move || assert_ok!(listener.accept()).0);
 
     let left = net::TcpStream::connect(addr).unwrap();
+
+    left.set_nonblocking(true).unwrap();
+
     let mut right = t.join().unwrap();
+
+    right.set_nonblocking(true).unwrap();
+
     let _ = right.write(&[1, 2, 3, 4]).unwrap();
 
     let mut left: TcpStream = left.try_into().unwrap();