fixed issue tokio-rs#1080, re-enable failing udp tests on windows (tokio-rs#1167)

* fixed assert arguments order

Signed-off-by: Daniel Tacalau <[email protected]>

* fix issue tokio-rs#1080, re-enabled and fixed UDP tests failing on Windows. One tests needed to have interests reregistered by calling a blocking recv. Another tests needed a Windows specific assert error message.

Signed-off-by: Daniel Tacalau <[email protected]>

* Added a note in the read/peek(_from) docs that on Windows if the data is larger than the buffer specified, the buffer is filled with the first part of the data, and peek returns the error WSAEMSGSIZE(10040). The excess data is lost. Make sure to always use a sufficiently large buffer to hold the maximum UDP packet size, which can be up to 64 kilobytes in size.

Signed-off-by: Daniel Tacalau <[email protected]>

* Added two new tests verifying ET behavior for UDP sockets events. These tests rely on having a blocking recv(_from) reregistering the socket and reseting the interests.

Signed-off-by: Daniel Tacalau <[email protected]>

* replaces 64 kilobytes with 65536 bytes in the notes

Signed-off-by: Daniel Tacalau <[email protected]>

* changes after review for new et tests: removed unneeded block, expect a writable before actual write

Signed-off-by: Daniel Tacalau <[email protected]>

Author: dtacalau

src/net/udp.rs

@@ -228,6 +228,14 @@ impl UdpSocket {
     /// Receives data from the socket. On success, returns the number of bytes
     /// read and the address from whence the data came.
     ///
+    /// # Notes
+    ///
+    /// On Windows, if the data is larger than the buffer specified, the buffer
+    /// is filled with the first part of the data, and recv_from returns the error
+    /// WSAEMSGSIZE(10040). The excess data is lost.
+    /// Make sure to always use a sufficiently large buffer to hold the
+    /// maximum UDP packet size, which can be up to 65536 bytes in size.
+    ///
     /// # Examples
     ///
     /// ```no_run
@@ -256,6 +264,14 @@ impl UdpSocket {
     /// On success, returns the number of bytes read and the address from whence
     /// the data came.
     ///
+    /// # Notes
+    ///
+    /// On Windows, if the data is larger than the buffer specified, the buffer
+    /// is filled with the first part of the data, and peek_from returns the error
+    /// WSAEMSGSIZE(10040). The excess data is lost.
+    /// Make sure to always use a sufficiently large buffer to hold the
+    /// maximum UDP packet size, which can be up to 65536 bytes in size.
+    ///
     /// # Examples
     ///
     /// ```no_run
@@ -288,12 +304,28 @@ impl UdpSocket {
 
     /// Receives data from the socket previously bound with connect(). On success, returns
     /// the number of bytes read.
+    ///
+    /// # Notes
+    ///
+    /// On Windows, if the data is larger than the buffer specified, the buffer
+    /// is filled with the first part of the data, and recv returns the error
+    /// WSAEMSGSIZE(10040). The excess data is lost.
+    /// Make sure to always use a sufficiently large buffer to hold the
+    /// maximum UDP packet size, which can be up to 65536 bytes in size.
     pub fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
         self.sys.recv(buf)
     }
 
     /// Receives data from the socket, without removing it from the input queue.
     /// On success, returns the number of bytes read.
+    ///
+    /// # Notes
+    ///
+    /// On Windows, if the data is larger than the buffer specified, the buffer
+    /// is filled with the first part of the data, and peek returns the error
+    /// WSAEMSGSIZE(10040). The excess data is lost.
+    /// Make sure to always use a sufficiently large buffer to hold the
+    /// maximum UDP packet size, which can be up to 65536 bytes in size.
     pub fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
         self.sys.peek(buf)
     }

tests/udp_socket.rs

@@ -288,7 +288,6 @@ fn reconnect_udp_socket_sending() {
 }
 
 #[test]
-#[cfg_attr(windows, ignore = "fails on Windows, see #1080")]
 fn reconnect_udp_socket_receiving() {
     let (mut poll, mut events) = init_with_poll();
 
@@ -334,7 +333,11 @@ fn reconnect_udp_socket_receiving() {
     let mut buf = [0; 20];
     expect_read!(socket1.recv(&mut buf), DATA1);
 
+    //this will reregister socket1 resetting the interests
+    assert_would_block(socket1.recv(&mut buf));
+
     socket1.connect(address3).unwrap();
+
     checked_write!(socket3.send(DATA2));
 
     expect_events(
@@ -343,12 +346,16 @@ fn reconnect_udp_socket_receiving() {
         vec![ExpectEvent::new(ID1, Interests::READABLE)],
     );
 
-    // Read only a part of the data.
-    let max = 4;
-    expect_read!(socket1.recv(&mut buf[..max]), &DATA2[..max]);
+    // Read all data.
+    // On Windows, reading part of data returns error WSAEMSGSIZE (10040).
+    expect_read!(socket1.recv(&mut buf), DATA2);
 
-    // Now connect back to socket 2, dropping the unread data.
+    //this will reregister socket1 resetting the interests
+    assert_would_block(socket1.recv(&mut buf));
+
+    // Now connect back to socket 2.
     socket1.connect(address2).unwrap();
+
     checked_write!(socket2.send(DATA2));
 
     expect_events(
@@ -365,7 +372,6 @@ fn reconnect_udp_socket_receiving() {
 }
 
 #[test]
-#[cfg_attr(windows, ignore = "fails on Windows, see #1080")]
 fn unconnected_udp_socket_connected_methods() {
     let (mut poll, mut events) = init_with_poll();
 
@@ -387,7 +393,15 @@ fn unconnected_udp_socket_connected_methods() {
     );
 
     // Socket is unconnected, but we're using an connected method.
-    assert_error(socket1.send(DATA1), "address required");
+    if cfg!(not(target_os = "windows")) {
+        assert_error(socket1.send(DATA1), "address required");
+    }
+    if cfg!(target_os = "windows") {
+        assert_error(
+            socket1.send(DATA1),
+            "no address was supplied. (os error 10057)",
+        );
+    }
 
     // Now send some actual data.
     checked_write!(socket1.send_to(DATA1, address2));
@@ -848,3 +862,105 @@ pub fn multicast() {
         }
     }
 }
+
+#[test]
+fn et_behavior_recv() {
+    let (mut poll, mut events) = init_with_poll();
+
+    let socket1 = UdpSocket::bind(any_local_address()).unwrap();
+    let socket2 = UdpSocket::bind(any_local_address()).unwrap();
+
+    let address2 = socket2.local_addr().unwrap();
+
+    poll.registry()
+        .register(&socket1, ID1, Interests::WRITABLE)
+        .expect("unable to register UDP socket");
+    poll.registry()
+        .register(&socket2, ID2, Interests::READABLE.add(Interests::WRITABLE))
+        .expect("unable to register UDP socket");
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![
+            ExpectEvent::new(ID1, Interests::WRITABLE),
+            ExpectEvent::new(ID2, Interests::WRITABLE),
+        ],
+    );
+
+    socket1.connect(address2).unwrap();
+
+    let mut buf = [0; 20];
+    checked_write!(socket1.send(DATA1));
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID2, Interests::READABLE)],
+    );
+
+    expect_read!(socket2.recv(&mut buf), DATA1);
+
+    // this will reregister the socket2, resetting the interests
+    assert_would_block(socket2.recv(&mut buf));
+    checked_write!(socket1.send(DATA1));
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID2, Interests::READABLE)],
+    );
+
+    let mut buf = [0; 20];
+    expect_read!(socket2.recv(&mut buf), DATA1);
+}
+
+#[test]
+fn et_behavior_recv_from() {
+    let (mut poll, mut events) = init_with_poll();
+
+    let socket1 = UdpSocket::bind(any_local_address()).unwrap();
+    let socket2 = UdpSocket::bind(any_local_address()).unwrap();
+
+    let address1 = socket1.local_addr().unwrap();
+    let address2 = socket2.local_addr().unwrap();
+
+    poll.registry()
+        .register(&socket1, ID1, Interests::READABLE.add(Interests::WRITABLE))
+        .expect("unable to register UDP socket");
+    poll.registry()
+        .register(&socket2, ID2, Interests::READABLE.add(Interests::WRITABLE))
+        .expect("unable to register UDP socket");
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![
+            ExpectEvent::new(ID1, Interests::WRITABLE),
+            ExpectEvent::new(ID2, Interests::WRITABLE),
+        ],
+    );
+
+    checked_write!(socket1.send_to(DATA1, address2));
+
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID2, Interests::READABLE)],
+    );
+
+    let mut buf = [0; 20];
+    expect_read!(socket2.recv_from(&mut buf), DATA1, address1);
+
+    // this will reregister the socket2, resetting the interests
+    assert_would_block(socket2.recv_from(&mut buf));
+    checked_write!(socket1.send_to(DATA1, address2));
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(ID2, Interests::READABLE)],
+    );
+
+    expect_read!(socket2.recv_from(&mut buf), DATA1, address1);
+
+    assert!(socket1.take_error().unwrap().is_none());
+    assert!(socket2.take_error().unwrap().is_none());
+}

tests/util/mod.rs

@@ -172,8 +172,8 @@ pub fn assert_error<T, E: fmt::Display>(result: Result<T, E>, expected_msg: &str
         Err(err) => assert!(
             err.to_string().contains(expected_msg),
             "wanted: {}, got: {}",
+            expected_msg,
             err,
-            expected_msg
         ),
     }
 }