net: surface errors from SO_ERROR on recv for UDP sockets on Linux (#8001)

tokio/src/net/udp.rs

@@ -785,7 +785,7 @@ impl UdpSocket {
     pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
         self.io
             .registration()
-            .async_io(Interest::READABLE, || self.io.recv(buf))
+            .async_io(Interest::READABLE | Interest::ERROR, || self.io.recv(buf))
             .await
     }
 
@@ -986,7 +986,9 @@ impl UdpSocket {
         /// }
         /// ```
         pub async fn recv_buf<B: BufMut>(&self, buf: &mut B) -> io::Result<usize> {
-            self.io.registration().async_io(Interest::READABLE, || {
+            self.io
+                .registration()
+                .async_io(Interest::READABLE | Interest::ERROR, || {
                 let dst = buf.chunk_mut();
                 let dst =
                     unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
@@ -1000,7 +1002,8 @@ impl UdpSocket {
                 }
 
                 Ok(n)
-            }).await
+            })
+            .await
         }
 
         /// Tries to receive a single datagram message on the socket. On success,
@@ -1117,7 +1120,9 @@ impl UdpSocket {
         /// }
         /// ```
         pub async fn recv_buf_from<B: BufMut>(&self, buf: &mut B) -> io::Result<(usize, SocketAddr)> {
-            self.io.registration().async_io(Interest::READABLE, || {
+            self.io
+                .registration()
+                .async_io(Interest::READABLE | Interest::ERROR, || {
                 let dst = buf.chunk_mut();
                 let dst =
                     unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
@@ -1130,8 +1135,9 @@ impl UdpSocket {
                     buf.advance_mut(n);
                 }
 
-                Ok((n,addr))
-            }).await
+                Ok((n, addr))
+            })
+            .await
         }
     }
 
@@ -1315,7 +1321,9 @@ impl UdpSocket {
     pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
         self.io
             .registration()
-            .async_io(Interest::READABLE, || self.io.recv_from(buf))
+            .async_io(Interest::READABLE | Interest::ERROR, || {
+                self.io.recv_from(buf)
+            })
             .await
     }
 
@@ -1556,7 +1564,7 @@ impl UdpSocket {
     pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
         self.io
             .registration()
-            .async_io(Interest::READABLE, || self.io.peek(buf))
+            .async_io(Interest::READABLE | Interest::ERROR, || self.io.peek(buf))
             .await
     }
 
@@ -1698,7 +1706,9 @@ impl UdpSocket {
     pub async fn peek_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
         self.io
             .registration()
-            .async_io(Interest::READABLE, || self.io.peek_from(buf))
+            .async_io(Interest::READABLE | Interest::ERROR, || {
+                self.io.peek_from(buf)
+            })
             .await
     }
 
@@ -1815,7 +1825,9 @@ impl UdpSocket {
     pub async fn peek_sender(&self) -> io::Result<SocketAddr> {
         self.io
             .registration()
-            .async_io(Interest::READABLE, || self.peek_sender_inner())
+            .async_io(Interest::READABLE | Interest::ERROR, || {
+                self.peek_sender_inner()
+            })
             .await
     }
 

tokio/tests/udp.rs

@@ -13,7 +13,8 @@
 use std::future::poll_fn;
 use std::io;
 use std::sync::Arc;
-use tokio::{io::ReadBuf, net::UdpSocket};
+use std::time::Duration;
+use tokio::{io::ReadBuf, net::UdpSocket, time};
 use tokio_test::assert_ok;
 
 const MSG: &[u8] = b"hello";
@@ -54,6 +55,37 @@ async fn send_recv_poll() -> std::io::Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+#[cfg_attr(
+    target_os = "wasi",
+    ignore = "temporarily disabled for WASI pending https://github.com/WebAssembly/wasi-libc/pull/734"
+)]
+async fn send_to_recv_closed_returns_err() -> std::io::Result<()> {
+    let sender = UdpSocket::bind("127.0.0.1:0").await?;
+    let receiver = UdpSocket::bind("127.0.0.1:0").await?;
+
+    let receiver_addr = receiver.local_addr()?;
+    drop(receiver);
+    sender.connect(receiver_addr).await?;
+    sender.send(MSG).await?;
+
+    let mut recv_buf = [0u8; 32];
+    let err = time::timeout(Duration::from_secs(5), sender.recv(&mut recv_buf))
+        .await
+        .expect("timed out instead of returning error")
+        .unwrap_err();
+    let errno = err.kind();
+
+    assert!(
+        // Linux/BSD returns ECONNREFUSED, but Windows will usually return ECONNRESET instead.
+        matches!(
+            errno,
+            io::ErrorKind::ConnectionRefused | io::ErrorKind::ConnectionReset
+        )
+    );
+    Ok(())
+}
+
 #[tokio::test]
 #[cfg_attr(
     target_os = "wasi",