Add Interest::PRIORITY

To trigger Event::is_priority on Linux and Android.

In the future we might want to include EV_OOBAND for kqueue, but that
seems to be Apple only (FreeBSD and OpenBSD don't implement it at
least).

Co-authored-by: Lars Pöschel <[email protected]>

src/interest.rs

@@ -20,18 +20,9 @@ pub struct Interest(NonZeroU8);
 const READABLE: u8 = 0b0001;
 const WRITABLE: u8 = 0b0010;
 // The following are not available on all platforms.
-#[cfg_attr(
-    not(any(
-        target_os = "dragonfly",
-        target_os = "freebsd",
-        target_os = "ios",
-        target_os = "macos",
-    )),
-    allow(dead_code)
-)]
 const AIO: u8 = 0b0100;
-#[cfg_attr(not(target_os = "freebsd"), allow(dead_code))]
 const LIO: u8 = 0b1000;
+const PRIORITY: u8 = 0b10000;
 
 impl Interest {
     /// Returns a `Interest` set representing readable interests.
@@ -53,6 +44,10 @@ impl Interest {
     #[cfg(target_os = "freebsd")]
     pub const LIO: Interest = Interest(unsafe { NonZeroU8::new_unchecked(LIO) });
 
+    /// Returns a `Interest` set representing priority completion interests.
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    pub const PRIORITY: Interest = Interest(unsafe { NonZeroU8::new_unchecked(PRIORITY) });
+
     /// Add together two `Interest`.
     ///
     /// This does the same thing as the `BitOr` implementation, but is a
@@ -104,15 +99,20 @@ impl Interest {
         (self.0.get() & WRITABLE) != 0
     }
 
-    /// Returns true if `Interest` contains AIO readiness
+    /// Returns true if `Interest` contains AIO readiness.
     pub const fn is_aio(self) -> bool {
         (self.0.get() & AIO) != 0
     }
 
-    /// Returns true if `Interest` contains LIO readiness
+    /// Returns true if `Interest` contains LIO readiness.
     pub const fn is_lio(self) -> bool {
         (self.0.get() & LIO) != 0
     }
+
+    /// Returns true if `Interest` contains priority readiness.
+    pub const fn is_priority(self) -> bool {
+        (self.0.get() & PRIORITY) != 0
+    }
 }
 
 impl ops::BitOr for Interest {
@@ -173,6 +173,16 @@ impl fmt::Debug for Interest {
                 one = true
             }
         }
+        #[cfg(any(target_os = "linux", target_os = "android"))]
+        {
+            if self.is_priority() {
+                if one {
+                    write!(fmt, " | ")?
+                }
+                write!(fmt, "PRIORITY")?;
+                one = true
+            }
+        }
         debug_assert!(one, "printing empty interests");
         Ok(())
     }

src/sys/unix/selector/epoll.rs

@@ -1,6 +1,6 @@
 use crate::{Interest, Token};
 
-use libc::{EPOLLET, EPOLLIN, EPOLLOUT, EPOLLRDHUP};
+use libc::{EPOLLET, EPOLLIN, EPOLLOUT, EPOLLPRI, EPOLLRDHUP};
 use log::error;
 use std::os::unix::io::{AsRawFd, RawFd};
 #[cfg(debug_assertions)]
@@ -177,6 +177,10 @@ fn interests_to_epoll(interests: Interest) -> u32 {
         kind |= EPOLLOUT;
     }
 
+    if interests.is_priority() {
+        kind |= EPOLLPRI;
+    }
+
     kind as u32
 }
 

tests/tcp_stream.rs

@@ -793,3 +793,52 @@ fn hup_event_on_disconnect() {
         vec![ExpectEvent::new(Token(1), Interest::READABLE)],
     );
 }
+
+#[test]
+#[cfg(any(target_os = "linux", target_os = "android"))]
+fn priority_event_on_oob_data() {
+    let (mut poll, mut events) = init_with_poll();
+    let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
+
+    let listener = std::net::TcpListener::bind(addr).unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let mut client = TcpStream::connect(addr).unwrap();
+    poll.registry()
+        .register(
+            &mut client,
+            Token(0),
+            Interest::READABLE | Interest::PRIORITY,
+        )
+        .unwrap();
+
+    let (stream, _) = listener.accept().unwrap();
+
+    // Sending out of bound data should trigger priority event.
+    send_oob_data(&stream, DATA1).unwrap();
+    expect_events(
+        &mut poll,
+        &mut events,
+        vec![ExpectEvent::new(
+            Token(0),
+            Readiness::READABLE | Readiness::PRIORITY,
+        )],
+    );
+}
+
+#[cfg(any(target_os = "linux", target_os = "android"))]
+fn send_oob_data<S: AsRawFd>(stream: &S, data: &[u8]) -> io::Result<usize> {
+    unsafe {
+        let res = libc::send(
+            stream.as_raw_fd(),
+            data.as_ptr().cast(),
+            data.len(),
+            libc::MSG_OOB,
+        );
+        if res == -1 {
+            Err(io::Error::last_os_error())
+        } else {
+            Ok(res as usize)
+        }
+    }
+}