Repository: tokio-rs/mio
Pull request #1215 (2019-12-11 15:47:13)

A partial fix for issue #1191. Do not panic mio on sockstate error, i…
…nstead keep the err sock states and just retry them later.

This is a partial fix for 0.7 because it just handles the errors, and makes the sockstates available next poll instead of panicking. Pushing the error to user is more complicated as it would require new api added. I'd suggest to leave this part until custom IOCP handlers are implemented and  #1195 is merged.

dtacalau: I've done a change in the logic of `update_sockets_events`. The following block should be done no matter if a sock update failed or not, because if we do it only on error then we end up with duplicates in the update_queue:
```
        update_queue.retain(|sock| {
            sock.lock().unwrap().has_error()
        });
```

I've also found another issue we need to address: `reregister done on a sock which has_error will result in having the same socked added twice in the update queue`. Will not result in a bug, just that we'll call update twice, 2nd call being useless. To fix this `self.add_socket_to_update_queue(socket);` has to be called only if `sock.has_error is false`. 
@Thomasdezeeuw any idea how can I achieve this? I got into mutability issues, cannot move out of a lock, so I cannot check for has_error and add it to the queue at the while sock.lock is being held. 
kleimkuhler: @dtacalau I still need to familiarize more with the Windows impl, but could `set_event` do the check and return `false` if `SockState.error` is `true`? We hold the lock right before going into that block of code so it could just return early
kleimkuhler: @dtacalau Also I would appreciate it for changes in general if you can not force push commits 🙂. It makes it difficult to review PRs. I like to be able to leave a review, and then when a review is re-requested I can see what *changed* since the last time.

When a force push happens, the "viewed already" cache resets and I don't have a way to tell what is new and what I've reviewed.

I wouldn't worry about the git log either because commits should be squashed in a PR.
dtacalau: > @dtacalau I still need to familiarize more with the Windows impl, but could `set_event` do the check and return `false` if `SockState.error` is `true`? We hold the lock right before going into that block of code so it could just return early

@kleimkuhler Yeah that would work, By changing `set_event` to return `has_error` we avoid double locking. Another way would be to check `has_error` under lock, like this `sock.lock().unwrap().has_error()`:
```
if (sock.lock().unwrap().set_event(event)) {
                self.add_socket_to_update_queue(socket);
}
self.update_sockets_events_if_polling()?;
```


dtacalau: > @dtacalau Also I would appreciate it for changes in general if you can not force push commits . 

Ok, got it, sorry for that.
kleimkuhler: @dtacalau Yea so we still want to call `socket.set_sock_state(Some(sock));` after `set_event` it looks like. In that case, `set_event` could return `false` if `error.is_some()`, and then we only update based on that result. So something like:

```
let should_update = sock.lock().unwrap().set_event(event);
socket.set_sock_state(Some(sock));
if should_update {
    unsafe {
        self.add_socket_to_update_queue(socket);
        self.update_sockets_events_if_polling()?;
    }
}
```
dtacalau: rebased on master

@kleimkuhler I'm not quite happy with the solution from https://github.com/tokio-rs/mio/pull/1215#issuecomment-566166194. It works but it feels like a hack `conveniently modifying the return of set_event for this fix`. What if we need it to return something else for another issue we might find :)? 

I'm going to leave it as it is until I figure something cleaner. WDYT?

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6c2fbfe..5132fa2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,10 +1,64 @@
-# 0.7.0 alpha.1
-
-Currently the changelog for v0.7.0 is incomplete, this tracked in issue #1217.
+# 0.7.0
+
+Version 0.7 of Mio contains various major changes compared to version 0.6.
+Overall a large number of API changes have been made to reduce the complexity of
+the implementation and remove overhead where possible.
+
+Please refer to the upgrade guide in docs/upgrade_guide.md for additional
+information.
+
+## Added
+
+* `Interest` structure that replaces `Ready` in registering event sources.
+* `Registry` structure that separates the registering and polling functionality.
+* `Waker` structure that allows another thread to wake a thread polling `Poll`.
+* Unix Domain Socket (UDS) types: `UnixDatagram`, `UnixListener` and
+  `UnixStream`.
+
+## Removed
+
+* All code deprecated in 0.6 was removed in 0.7.
+* Support for Fuchsia was removed as the code was unmaintained.
+* Support for Bitrig was removed, rustc dropped support for it also.
+* `UnixReady` was merged into `Ready`.
+* Custom user-space readiness queue was removed, this includes the public
+  `Registration` and `SetReadiness` types.
+* `PollOpt` was removed and all registrations use edge-triggers. See the upgrade
+  guide on how to process event using edge-triggers.
+* The network types (types in the `net` module) now support only the same API as
+  found in the standard library, various methods on the types were removed.
+* `TcpStream` now supports vectored I/O.
+* `Poll::poll_interruptible` was removed. Instead `Poll::poll` will now return
+  an error if one occurs.
+* `From<usize>` is removed from `Token`, the internal field is still public, so
+  `Token(my_token)` can still be used.
+
+## Changed
 
 * The `join_multicast_v4` and `leave_multicast_v4` methods now take their
   `Ipv4Addr` arguments by value rather than by reference.
-* Fix lazycell related compilation issues.
+* Various documentation improvements were made around correct usage of `Poll`
+  and registered event sources. It is recommended to reread the documentation of
+  at least `event::Source` and `Poll`.
+* Mio now uses Rust 2018 and rustfmt for all code.
+* `Event` was changed to be a wrapper around the OS event. This means it can be
+  significantly larger on some OSes.
+* `Ready` was removed and replaced with various `is_*` methods on `Event`. For
+  example instead checking for readable readiness using
+  `Event::ready().is_readble()`, you would call `Event::is_readble()`.
+* `Ready::is_hup` was removed in favour of `Event::is_read_closed` and
+  `Event::is_write_closed`.
+* The Iterator implementation of `Events` was changed to return `&Event`.
+* `Evented` was renamed to `event::Source` and now takes mutable reference to
+  the source.
+* Minimum supported Rust version was increased to 1.39.
+* By default Mio now uses a shim implementation. To enable the full
+  implementation, that uses the OS, enable the `os-oll` feature. To enable the
+  network types use `tcp`, `udp` and/or `uds`. For more documentation on the
+  features see docs/features.md.
+* The entire Windows implementation was rewritten.
+* Various optimisation were made to reduce the number of system calls in
+  creating and using sockets, e.g. making use of `accept4(2)`.
 
 # 0.6.19 (May 28, 2018)
 
diff --git a/examples/tcp_server.rs b/examples/tcp_server.rs
index d306964..b33140b 100644
--- a/examples/tcp_server.rs
+++ b/examples/tcp_server.rs
@@ -1,3 +1,5 @@
+// You can run this example from the root of the mio repo:
+// cargo run --example tcp_server --features="os-poll tcp"
 use mio::event::Event;
 use mio::net::{TcpListener, TcpStream};
 use mio::{Events, Interest, Poll, Registry, Token};
diff --git a/examples/udp_server.rs b/examples/udp_server.rs
index 330f988..febb662 100644
--- a/examples/udp_server.rs
+++ b/examples/udp_server.rs
@@ -1,3 +1,5 @@
+// You can run this example from the root of the mio repo:
+// cargo run --example udp_server --features="os-poll udp"
 use log::warn;
 use mio::net::UdpSocket;
 use mio::{Events, Interest, Poll, Token};
diff --git a/src/event/events.rs b/src/event/events.rs
index 8ca49e2..4d4407b 100644
--- a/src/event/events.rs
+++ b/src/event/events.rs
@@ -212,8 +212,6 @@ impl<'a> Iterator for Iter<'a> {
 
 impl fmt::Debug for Events {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Events")
-            .field("capacity", &self.capacity())
-            .finish()
+        f.debug_list().entries(self).finish()
     }
 }
diff --git a/src/interest.rs b/src/interest.rs
index 2ee37ff..6bc5929 100644
--- a/src/interest.rs
+++ b/src/interest.rs
@@ -96,7 +96,7 @@ impl ops::BitOr for Interest {
 
     #[inline]
     fn bitor(self, other: Self) -> Self {
-        Interest(unsafe { NonZeroU8::new_unchecked(self.0.get() | other.0.get()) })
+        self.add(other)
     }
 }
 
diff --git a/src/io_source.rs b/src/io_source.rs
index 73fd6e5..6939c0d 100644
--- a/src/io_source.rs
+++ b/src/io_source.rs
@@ -221,7 +221,7 @@ impl SelectorId {
     const UNASSOCIATED: usize = 0;
 
     /// Create a new `SelectorId`.
-    fn new() -> SelectorId {
+    const fn new() -> SelectorId {
         SelectorId {
             id: AtomicUsize::new(Self::UNASSOCIATED),
         }
diff --git a/src/net/udp.rs b/src/net/udp.rs
index 46d288e..164315a 100644
--- a/src/net/udp.rs
+++ b/src/net/udp.rs
@@ -465,8 +465,9 @@ impl UdpSocket {
     /// address of the local interface with which the system should join the
     /// multicast group. If it's equal to `INADDR_ANY` then an appropriate
     /// interface is chosen by the system.
-    pub fn join_multicast_v4(&self, multiaddr: Ipv4Addr, interface: Ipv4Addr) -> io::Result<()> {
-        self.inner.join_multicast_v4(&multiaddr, &interface)
+    #[allow(clippy::trivially_copy_pass_by_ref)]
+    pub fn join_multicast_v4(&self, multiaddr: &Ipv4Addr, interface: &Ipv4Addr) -> io::Result<()> {
+        self.inner.join_multicast_v4(multiaddr, interface)
     }
 
     /// Executes an operation of the `IPV6_ADD_MEMBERSHIP` type.
@@ -474,6 +475,7 @@ impl UdpSocket {
     /// This function specifies a new multicast group for this socket to join.
     /// The address must be a valid multicast address, and `interface` is the
     /// index of the interface to join/leave (or 0 to indicate any interface).
+    #[allow(clippy::trivially_copy_pass_by_ref)]
     pub fn join_multicast_v6(&self, multiaddr: &Ipv6Addr, interface: u32) -> io::Result<()> {
         self.inner.join_multicast_v6(multiaddr, interface)
     }
@@ -484,8 +486,9 @@ impl UdpSocket {
     /// [`join_multicast_v4`][link].
     ///
     /// [link]: #method.join_multicast_v4
-    pub fn leave_multicast_v4(&self, multiaddr: Ipv4Addr, interface: Ipv4Addr) -> io::Result<()> {
-        self.inner.leave_multicast_v4(&multiaddr, &interface)
+    #[allow(clippy::trivially_copy_pass_by_ref)]
+    pub fn leave_multicast_v4(&self, multiaddr: &Ipv4Addr, interface: &Ipv4Addr) -> io::Result<()> {
+        self.inner.leave_multicast_v4(multiaddr, interface)
     }
 
     /// Executes an operation of the `IPV6_DROP_MEMBERSHIP` type.
@@ -494,6 +497,7 @@ impl UdpSocket {
     /// [`join_multicast_v6`][link].
     ///
     /// [link]: #method.join_multicast_v6
+    #[allow(clippy::trivially_copy_pass_by_ref)]
     pub fn leave_multicast_v6(&self, multiaddr: &Ipv6Addr, interface: u32) -> io::Result<()> {
         self.inner.leave_multicast_v6(multiaddr, interface)
     }
diff --git a/src/poll.rs b/src/poll.rs
index 16528bd..efd5555 100644
--- a/src/poll.rs
+++ b/src/poll.rs
@@ -151,6 +151,16 @@ use std::{fmt, io};
 /// # }
 /// ```
 ///
+/// ### Dropping `Poll`
+///
+/// When the `Poll` instance is dropped it may cancel in-flight operations for
+/// the registered [event sources], meaning that no further events for them may
+/// be received. It also means operations on the registered event sources may no
+/// longer work. It is up to the user to keep the `Poll` instance alive while
+/// registered event sources are being used.
+///
+/// [event sources]: ./event/trait.Source.html
+///
 /// # Implementation notes
 ///
 /// `Poll` is backed by the selector provided by the operating system.
diff --git a/src/sys/windows/selector.rs b/src/sys/windows/selector.rs
index 96a8ecb..fd1481a 100644
--- a/src/sys/windows/selector.rs
+++ b/src/sys/windows/selector.rs
@@ -104,6 +104,9 @@ pub struct SockState {
     poll_status: SockPollStatus,
     delete_pending: bool,
 
+    // last raw os error
+    error: Option<i32>,
+
     pinned: PhantomPinned,
 }
 
@@ -111,6 +114,9 @@ impl SockState {
     fn update(&mut self, self_arc: &Pin<Arc<Mutex<SockState>>>) -> io::Result<()> {
         assert!(!self.delete_pending);
 
+        // make sure to reset previous error before a new update
+        self.error = None;
+
         if let SockPollStatus::Pending = self.poll_status {
             if (self.user_evts & afd::KNOWN_EVENTS & !self.pending_evts) == 0 {
                 /* All the events the user is interested in are already being monitored by
@@ -122,7 +128,11 @@ impl SockState {
                  * events that the user is interested in. Therefore, cancel the pending
                  * poll operation; when we receive it's completion package, a new poll
                  * operation will be submitted with the correct event mask. */
-                self.cancel()?;
+                 if let Err(e) = self.cancel() {
+                    self.error = e.raw_os_error();
+                    return Err(e);
+                }
+                return Ok(());
             }
         } else if let SockPollStatus::Cancelled = self.poll_status {
             /* The poll operation has already been cancelled, we're still waiting for
@@ -156,6 +166,7 @@ impl SockState {
                         self.mark_delete();
                         return Ok(());
                     } else {
+                        self.error = e.raw_os_error();
                         return Err(e);
                     }
                 }
@@ -166,6 +177,7 @@ impl SockState {
         } else {
             unreachable!("Invalid poll status during update, {:#?}", self)
         }
+
         Ok(())
     }
 
@@ -246,6 +258,10 @@ impl SockState {
             self.delete_pending = true;
         }
     }
+
+    fn has_error(&self) -> bool {
+        self.error.is_some()
+    }
 }
 
 cfg_net! {
@@ -262,6 +278,7 @@ cfg_net! {
                 user_data: 0,
                 poll_status: SockPollStatus::Idle,
                 delete_pending: false,
+                error: None,
                 pinned: PhantomPinned,
             })
         }
@@ -455,16 +472,18 @@ impl SelectorInner {
 
     unsafe fn update_sockets_events(&self) -> io::Result<()> {
         let mut update_queue = self.update_queue.lock().unwrap();
-        loop {
-            let sock = match update_queue.pop_front() {
-                Some(sock) => sock,
-                None => break,
-            };
+        for sock in update_queue.iter_mut() {
             let mut sock_internal = sock.lock().unwrap();
             if !sock_internal.is_pending_deletion() {
-                sock_internal.update(&sock).unwrap();
+                let _ = sock_internal.update(&sock);
             }
         }
+
+        // remove all sock which do not have error, they have afd op pending
+        update_queue.retain(|sock| {
+            sock.lock().unwrap().has_error()
+        });
+
         self.afd_group.release_unused_afd();
         Ok(())
     }
@@ -561,6 +580,8 @@ cfg_net! {
                 state.lock().unwrap().set_event(event);
             }
 
+            // FIXME: a sock which has_error true should not be re-added to
+            // the update queue because it's already there.
             self.queue_state(state);
             unsafe { self.update_sockets_events_if_polling() }
         }


Test patch:
diff --git a/tests/udp_socket.rs b/tests/udp_socket.rs
index ec74c0b..874654d 100644
--- a/tests/udp_socket.rs
+++ b/tests/udp_socket.rs
@@ -933,12 +933,12 @@ pub fn multicast() {
     let mut rx = UdpSocket::bind(any_local_address()).unwrap();
 
     info!("Joining group 227.1.1.100");
-    let any = "0.0.0.0".parse().unwrap();
-    rx.join_multicast_v4("227.1.1.100".parse().unwrap(), any)
+    let any = &"0.0.0.0".parse().unwrap();
+    rx.join_multicast_v4(&"227.1.1.100".parse().unwrap(), any)
         .unwrap();
 
     info!("Joining group 227.1.1.101");
-    rx.join_multicast_v4("227.1.1.101".parse().unwrap(), any)
+    rx.join_multicast_v4(&"227.1.1.101".parse().unwrap(), any)
         .unwrap();
 
     info!("Registering SENDER");