feat: Expose other kqueue filters

Author: notgull

Cargo.toml

@@ -25,7 +25,7 @@ concurrent-queue = "2"
 futures-lite = "1.11.0"
 log = "0.4.11"
 parking = "2.0.0"
-polling = "2.0.0"
+polling = { git = "https://github.com/smol-rs/polling.git" }
 rustix = { version = "0.36.0", default-features = false, features = ["std", "fs"] }
 slab = "0.4.2"
 socket2 = { version = "0.4.2", features = ["all"] }

examples/kqueue-process.rs

@@ -0,0 +1,55 @@
+//! Uses the `async_io::os::kqueue` module to wait for a process to terminate.
+//!
+//! Run with:
+//!
+//! ```
+//! cargo run --example kqueue-process
+//! ```
+
+#[cfg(any(
+    target_os = "macos",
+    target_os = "ios",
+    target_os = "tvos",
+    target_os = "watchos",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "dragonfly",
+))]
+fn main() -> std::io::Result<()> {
+    use std::process::Command;
+
+    use async_io::os::kqueue::{AsyncKqueueExt, Exit};
+    use async_io::Async;
+    use futures_lite::future;
+
+    future::block_on(async {
+        // Spawn a process.
+        let process = Command::new("sleep")
+            .arg("3")
+            .spawn()
+            .expect("failed to spawn process");
+
+        // Wrap the process in an `Async` object that waits for it to exit.
+        let process = Async::with_filter(Exit::new(process))?;
+
+        // Wait for the process to exit.
+        process.readable().await?;
+
+        Ok(())
+    })
+}
+
+#[cfg(not(any(
+    target_os = "macos",
+    target_os = "ios",
+    target_os = "tvos",
+    target_os = "watchos",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "dragonfly",
+)))]
+fn main() {
+    println!("This example only works for kqueue-enabled platforms.");
+}

src/lib.rs

@@ -88,6 +88,8 @@ use crate::reactor::{Reactor, Source};
 mod driver;
 mod reactor;
 
+pub mod os;
+
 pub use driver::block_on;
 pub use reactor::{Readable, ReadableOwned, Writable, WritableOwned};
 
@@ -664,7 +666,7 @@ impl<T: AsRawFd> Async<T> {
 #[cfg(unix)]
 impl<T: AsRawFd> AsRawFd for Async<T> {
     fn as_raw_fd(&self) -> RawFd {
-        self.source.raw
+        self.get_ref().as_raw_fd()
     }
 }
 
@@ -740,7 +742,7 @@ impl<T: AsRawSocket> Async<T> {
 #[cfg(windows)]
 impl<T: AsRawSocket> AsRawSocket for Async<T> {
     fn as_raw_socket(&self) -> RawSocket {
-        self.source.raw
+        self.get_ref().as_raw_socket()
     }
 }
 

src/os.rs

@@ -0,0 +1,20 @@
+//! Platform-specific functionality.
+
+#[cfg(any(
+    target_os = "macos",
+    target_os = "ios",
+    target_os = "tvos",
+    target_os = "watchos",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "dragonfly",
+))]
+pub mod kqueue;
+
+mod __private {
+    #[doc(hidden)]
+    pub trait AsyncSealed {}
+
+    impl<T> AsyncSealed for crate::Async<T> {}
+}

src/os/kqueue.rs

@@ -0,0 +1,100 @@
+//! Functionality that is only available for `kqueue`-based platforms.
+
+use super::__private::AsyncSealed;
+use __private::FilterSealed;
+
+use crate::reactor::{Reactor, Registration};
+use crate::Async;
+
+use std::io::Result;
+use std::process::Child;
+
+/// An extension trait for [`Async`](crate::Async) that provides the ability to register other
+/// queueable objects into the reactor.
+///
+/// The underlying `kqueue` implementation can be used to poll for events besides file descriptor
+/// read/write readiness. This API makes these faculties available to the user.
+///
+/// See the [`Filter`] trait and its implementors for objects that currently support being registered
+/// into the reactor.
+pub trait AsyncKqueueExt<T: Filter>: AsyncSealed {
+    /// Create a new [`Async`](crate::Async) around a [`Filter`].
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// use std::process::Command;
+    ///
+    /// use async_io::Async;
+    /// use async_io::os::kqueue::{AsyncKqueueExt, Exit};
+    ///
+    /// // Create a new process to wait for.
+    /// let mut child = Command::new("sleep").arg("5").spawn().unwrap();
+    ///
+    /// // Wrap the process in an `Async` object that waits for it to exit.
+    /// let process = Async::with_filter(Exit::new(child)).unwrap();
+    ///
+    /// // Wait for the process to exit.
+    /// # async_io::block_on(async {
+    /// process.readable().await.unwrap();
+    /// # });
+    /// ```
+    fn with_filter(filter: T) -> Result<Async<T>>;
+}
+
+impl<T: Filter> AsyncKqueueExt<T> for Async<T> {
+    fn with_filter(mut filter: T) -> Result<Async<T>> {
+        Ok(Async {
+            source: Reactor::get().insert_io(filter.registration())?,
+            io: Some(filter),
+        })
+    }
+}
+
+/// Objects that can be registered into the reactor via a [`Async`](crate::Async).
+pub trait Filter: FilterSealed {}
+
+/// An object representing a signal.
+///
+/// When registered into [`Async`](crate::Async) via [`with_filter`](AsyncKqueueExt::with_filter),
+/// it will return a [`readable`](crate::Async::readable) event when the signal is received.
+#[derive(Debug, Copy, Clone, Eq, PartialEq, PartialOrd, Ord, Hash)]
+pub struct Signal(pub i32);
+
+impl FilterSealed for Signal {
+    fn registration(&mut self) -> Registration {
+        (*self).into()
+    }
+}
+impl Filter for Signal {}
+
+/// Wait for a child process to exit.
+///
+/// When registered into [`Async`](crate::Async) via [`with_filter`](AsyncKqueueExt::with_filter),
+/// it will return a [`readable`](crate::Async::readable) event when the child process exits.
+#[derive(Debug)]
+pub struct Exit(Option<Child>);
+
+impl Exit {
+    /// Create a new `Exit` object.
+    pub fn new(child: Child) -> Self {
+        Self(Some(child))
+    }
+}
+
+impl FilterSealed for Exit {
+    fn registration(&mut self) -> Registration {
+        self.0.take().expect("Cannot reregister child").into()
+    }
+}
+impl Filter for Exit {}
+
+mod __private {
+    use crate::reactor::Registration;
+
+    #[doc(hidden)]
+    pub trait FilterSealed {
+        /// Get a registration object for this filter.
+        fn registration(&mut self) -> Registration;
+    }
+}

src/reactor.rs

@@ -5,10 +5,6 @@ use std::future::Future;
 use std::io;
 use std::marker::PhantomData;
 use std::mem;
-#[cfg(unix)]
-use std::os::unix::io::RawFd;
-#[cfg(windows)]
-use std::os::windows::io::RawSocket;
 use std::panic;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -22,6 +18,9 @@ use futures_lite::ready;
 use polling::{Event, Poller};
 use slab::Slab;
 
+mod registration;
+pub use registration::Registration;
+
 const READ: usize = 0;
 const WRITE: usize = 1;
 
@@ -88,17 +87,13 @@ impl Reactor {
     }
 
     /// Registers an I/O source in the reactor.
-    pub(crate) fn insert_io(
-        &self,
-        #[cfg(unix)] raw: RawFd,
-        #[cfg(windows)] raw: RawSocket,
-    ) -> io::Result<Arc<Source>> {
+    pub(crate) fn insert_io(&self, raw: impl Into<Registration>) -> io::Result<Arc<Source>> {
         // Create an I/O source for this file descriptor.
         let source = {
             let mut sources = self.sources.lock().unwrap();
             let key = sources.vacant_entry().key();
             let source = Arc::new(Source {
-                raw,
+                registration: raw.into(),
                 key,
                 state: Default::default(),
             });
@@ -107,7 +102,7 @@ impl Reactor {
         };
 
         // Register the file descriptor.
-        if let Err(err) = self.poller.add(raw, Event::none(source.key)) {
+        if let Err(err) = source.registration.add(&self.poller, source.key) {
             let mut sources = self.sources.lock().unwrap();
             sources.remove(source.key);
             return Err(err);
@@ -120,7 +115,7 @@ impl Reactor {
     pub(crate) fn remove_io(&self, source: &Source) -> io::Result<()> {
         let mut sources = self.sources.lock().unwrap();
         sources.remove(source.key);
-        self.poller.delete(source.raw)
+        source.registration.delete(&self.poller)
     }
 
     /// Registers a timer in the reactor.
@@ -299,8 +294,8 @@ impl ReactorLock<'_> {
                         // e.g. we were previously interested in both readability and writability,
                         // but only one of them was emitted.
                         if !state[READ].is_empty() || !state[WRITE].is_empty() {
-                            self.reactor.poller.modify(
-                                source.raw,
+                            source.registration.modify(
+                                &self.reactor.poller,
                                 Event {
                                     key: source.key,
                                     readable: !state[READ].is_empty(),
@@ -341,13 +336,8 @@ enum TimerOp {
 /// A registered source of I/O events.
 #[derive(Debug)]
 pub(crate) struct Source {
-    /// Raw file descriptor on Unix platforms.
-    #[cfg(unix)]
-    pub(crate) raw: RawFd,
-
-    /// Raw socket handle on Windows.
-    #[cfg(windows)]
-    pub(crate) raw: RawSocket,
+    /// This source's registration into the reactor.
+    registration: Registration,
 
     /// The key of this source obtained during registration.
     key: usize,
@@ -436,8 +426,8 @@ impl Source {
 
         // Update interest in this I/O handle.
         if was_empty {
-            Reactor::get().poller.modify(
-                self.raw,
+            self.registration.modify(
+                &Reactor::get().poller,
                 Event {
                     key: self.key,
                     readable: !state[READ].is_empty(),
@@ -490,7 +480,7 @@ impl<T> Future for Readable<'_, T> {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         ready!(Pin::new(&mut self.0).poll(cx))?;
-        log::trace!("readable: fd={}", self.0.handle.source.raw);
+        log::trace!("readable: fd={:?}", &self.0.handle.source.registration);
         Poll::Ready(Ok(()))
     }
 }
@@ -510,7 +500,10 @@ impl<T> Future for ReadableOwned<T> {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         ready!(Pin::new(&mut self.0).poll(cx))?;
-        log::trace!("readable_owned: fd={}", self.0.handle.source.raw);
+        log::trace!(
+            "readable_owned: fd={:?}",
+            &self.0.handle.source.registration
+        );
         Poll::Ready(Ok(()))
     }
 }
@@ -530,7 +523,7 @@ impl<T> Future for Writable<'_, T> {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         ready!(Pin::new(&mut self.0).poll(cx))?;
-        log::trace!("writable: fd={}", self.0.handle.source.raw);
+        log::trace!("writable: fd={:?}", &self.0.handle.source.registration);
         Poll::Ready(Ok(()))
     }
 }
@@ -550,7 +543,10 @@ impl<T> Future for WritableOwned<T> {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         ready!(Pin::new(&mut self.0).poll(cx))?;
-        log::trace!("writable_owned: fd={}", self.0.handle.source.raw);
+        log::trace!(
+            "writable_owned: fd={:?}",
+            &self.0.handle.source.registration
+        );
         Poll::Ready(Ok(()))
     }
 }
@@ -610,8 +606,8 @@ impl<H: Borrow<crate::Async<T>> + Clone, T> Future for Ready<H, T> {
 
         // Update interest in this I/O handle.
         if was_empty {
-            Reactor::get().poller.modify(
-                handle.borrow().source.raw,
+            handle.borrow().source.registration.modify(
+                &Reactor::get().poller,
                 Event {
                     key: handle.borrow().source.key,
                     readable: !state[READ].is_empty(),