Make Event thread safe for platforms using kqueue (#1906)

Co-authored-by: Thomas de Zeeuw <[email protected]>

Author: hulxv

src/sys/unix/selector/kqueue.rs

@@ -114,7 +114,7 @@ impl Selector {
             self.kq.as_raw_fd(),
             ptr::null(),
             0,
-            events.as_mut_ptr(),
+            events.as_mut_ptr().cast(),
             events.capacity() as Count,
             timeout,
         ))
@@ -330,17 +330,31 @@ impl AsRawFd for Selector {
     }
 }
 
-pub type Event = libc::kevent;
-pub struct Events(Vec<libc::kevent>);
+#[repr(transparent)]
+#[derive(Clone)]
+pub struct Event(libc::kevent);
 
-impl Events {
-    pub fn with_capacity(capacity: usize) -> Events {
-        Events(Vec::with_capacity(capacity))
+unsafe impl Send for Event {}
+unsafe impl Sync for Event {}
+
+impl Deref for Event {
+    type Target = libc::kevent;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
     }
 }
 
+impl DerefMut for Event {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+pub struct Events(Vec<Event>);
+
 impl Deref for Events {
-    type Target = Vec<libc::kevent>;
+    type Target = Vec<Event>;
 
     fn deref(&self) -> &Self::Target {
         &self.0
@@ -353,6 +367,12 @@ impl DerefMut for Events {
     }
 }
 
+impl Events {
+    pub fn with_capacity(capacity: usize) -> Events {
+        Events(Vec::with_capacity(capacity))
+    }
+}
+
 // `Events` cannot derive `Send` or `Sync` because of the
 // `udata: *mut ::c_void` field in `libc::kevent`. However, `Events`'s public
 // API treats the `udata` field as a `uintptr_t` which is `Send`. `Sync` is
@@ -370,11 +390,11 @@ pub mod event {
     use super::{Filter, Flags};
 
     pub fn token(event: &Event) -> Token {
-        Token(event.udata as usize)
+        Token(event.0.udata as usize)
     }
 
     pub fn is_readable(event: &Event) -> bool {
-        event.filter == libc::EVFILT_READ || {
+        event.0.filter == libc::EVFILT_READ || {
             #[cfg(any(
                 target_os = "freebsd",
                 target_os = "ios",
@@ -404,22 +424,22 @@ pub mod event {
     }
 
     pub fn is_writable(event: &Event) -> bool {
-        event.filter == libc::EVFILT_WRITE
+        event.0.filter == libc::EVFILT_WRITE
     }
 
     pub fn is_error(event: &Event) -> bool {
-        (event.flags & libc::EV_ERROR) != 0 ||
+        (event.0.flags & libc::EV_ERROR) != 0 ||
             // When the read end of the socket is closed, EV_EOF is set on
             // flags, and fflags contains the error if there is one.
-            (event.flags & libc::EV_EOF) != 0 && event.fflags != 0
+            (event.0.flags & libc::EV_EOF) != 0 && event.0.fflags != 0
     }
 
     pub fn is_read_closed(event: &Event) -> bool {
-        event.filter == libc::EVFILT_READ && event.flags & libc::EV_EOF != 0
+        event.0.filter == libc::EVFILT_READ && event.0.flags & libc::EV_EOF != 0
     }
 
     pub fn is_write_closed(event: &Event) -> bool {
-        event.filter == libc::EVFILT_WRITE && event.flags & libc::EV_EOF != 0
+        event.0.filter == libc::EVFILT_WRITE && event.0.flags & libc::EV_EOF != 0
     }
 
     pub fn is_priority(_: &Event) -> bool {
@@ -439,7 +459,7 @@ pub mod event {
             target_os = "watchos",
         ))]
         {
-            event.filter == libc::EVFILT_AIO
+            event.0.filter == libc::EVFILT_AIO
         }
         #[cfg(not(any(
             target_os = "dragonfly",
@@ -459,7 +479,7 @@ pub mod event {
     pub fn is_lio(event: &Event) -> bool {
         #[cfg(target_os = "freebsd")]
         {
-            event.filter == libc::EVFILT_LIO
+            event.0.filter == libc::EVFILT_LIO
         }
         #[cfg(not(target_os = "freebsd"))]
         {
@@ -861,14 +881,14 @@ pub mod event {
         );
 
         // Can't reference fields in packed structures.
-        let ident = event.ident;
-        let data = event.data;
-        let udata = event.udata;
+        let ident = event.0.ident;
+        let data = event.0.data;
+        let udata = event.0.udata;
         f.debug_struct("kevent")
             .field("ident", &ident)
-            .field("filter", &FilterDetails(event.filter))
-            .field("flags", &FlagsDetails(event.flags))
-            .field("fflags", &FflagsDetails(event.fflags))
+            .field("filter", &FilterDetails(event.0.filter))
+            .field("flags", &FlagsDetails(event.0.flags))
+            .field("fflags", &FflagsDetails(event.0.fflags))
             .field("data", &data)
             .field("udata", &udata)
             .finish()

tests/events.rs

@@ -3,11 +3,12 @@
 
 use std::time::Duration;
 
+use mio::event::{self, Event, Events};
 use mio::net::TcpStream;
-use mio::{event, Token, Waker};
+use mio::{Token, Waker};
 
 mod util;
-use util::init_with_poll;
+use util::{assert_send, assert_sync, init_with_poll};
 
 const WAKE_TOKEN: Token = Token(10);
 
@@ -41,3 +42,12 @@ fn events_all() {
     events.clear();
     assert!(events.is_empty());
 }
+
+#[test]
+fn is_event_send_sync() {
+    assert_send::<Event>();
+    assert_sync::<Event>();
+
+    assert_send::<Events>();
+    assert_sync::<Events>();
+}