Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Expose other kqueue filters #83

Merged
merged 9 commits into from
Feb 3, 2023
Merged

feat: Expose other kqueue filters #83

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
fea85d3
feat: Expose other kqueue filters
notgull Feb 1, 2023
324b480
Fix netbsd/openbsd compilation
notgull Feb 1, 2023
1b18e14
Build MSRV for FreeBsd/OpenBsd in CI
notgull Feb 1, 2023
6d8fa23
Only run MSRV BSD builds on Linux
notgull Feb 1, 2023
7497fcf
Change API a little + fix netbsd timer
notgull Feb 3, 2023
5987fa8
Add inlines + move PollerSealed
notgull Feb 3, 2023
9a1bf97
rustfmt
notgull Feb 3, 2023
9d09ad3
Make filter fields public
notgull Feb 3, 2023
2a433eb
Fix examples
notgull Feb 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
357 changes: 1 addition & 356 deletions src/os.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,359 +13,4 @@
),
not(polling_test_poll_backend),
))]
/// Functionality that is only available for `kqueue`-based platforms.
pub mod kqueue {
use crate::sys::{mode_to_flags, FilterFlags};
use crate::{PollMode, Poller};
use __private::{FilterSealed, PollerSealed};
use std::os::unix::prelude::*;
use std::process::Child;
use std::{io, mem};

#[cfg(not(any(target_os = "dragonfly", target_os = "netbsd")))]
use std::convert::TryInto;
#[cfg(not(any(target_os = "dragonfly", target_os = "netbsd")))]
use std::time::Duration;

/// Functionality that is only available for `kqueue`-based platforms.
///
/// `kqueue` is able to monitor much more than just read/write readiness on file descriptors. Using
/// this extension trait, you can monitor for signals, process exits, and more. See the implementors
/// of the [`Filter`] trait for more information.
pub trait PollerKqueueExt: PollerSealed {
/// Add a filter to the poller.
///
/// This is similar to [`add`][Poller::add], but it allows you to specify a filter instead of
/// a socket. See the implementors of the [`Filter`] trait for more information.
///
/// # Examples
///
/// ```no_run
/// use polling::{Poller, PollMode};
/// use polling::os::kqueue::{Filter, PollerKqueueExt, Signal};
///
/// let poller = Poller::new().unwrap();
///
/// // Register the SIGINT signal.
/// poller.add_filter(Signal::new(libc::SIGINT), 0, PollMode::Oneshot).unwrap();
///
/// // Wait for the signal.
/// let mut events = vec![];
/// poller.wait(&mut events, None).unwrap();
/// # let _ = events;
/// ```
fn add_filter(&self, filter: impl Filter, key: usize, mode: PollMode) -> io::Result<()>;

/// Modify a filter in the poller.
///
/// This is similar to [`modify`][Poller::modify], but it allows you to specify a filter
/// instead of a socket. See the implementors of the [`Filter`] trait for more information.
///
/// # Examples
///
/// ```no_run
/// use polling::{Poller, PollMode};
/// use polling::os::kqueue::{Filter, PollerKqueueExt, Signal};
///
/// let poller = Poller::new().unwrap();
///
/// // Register the SIGINT signal.
/// poller.add_filter(Signal::new(libc::SIGINT), 0, PollMode::Oneshot).unwrap();
///
/// // Re-register with a different key.
/// poller.modify_filter(Signal::new(libc::SIGINT), 1, PollMode::Oneshot).unwrap();
///
/// // Wait for the signal.
/// let mut events = vec![];
/// poller.wait(&mut events, None).unwrap();
/// # let _ = events;
/// ```
fn modify_filter(&self, filter: impl Filter, key: usize, mode: PollMode) -> io::Result<()>;

/// Remove a filter from the poller.
///
/// This is used to remove filters that were previously added with
/// [`add_filter`](PollerKqueueExt::add_filter).
///
/// # Examples
///
/// ```no_run
/// use polling::{Poller, PollMode};
/// use polling::os::kqueue::{Filter, PollerKqueueExt, Signal};
///
/// let poller = Poller::new().unwrap();
///
/// // Register the SIGINT signal.
/// poller.add_filter(Signal::new(libc::SIGINT), 0, PollMode::Oneshot).unwrap();
///
/// // Remove the filter.
/// poller.delete_filter(Signal::new(libc::SIGINT)).unwrap();
/// ```
fn delete_filter(&self, filter: impl Filter) -> io::Result<()>;
}

impl PollerSealed for Poller {}

impl PollerKqueueExt for Poller {
fn add_filter(&self, filter: impl Filter, key: usize, mode: PollMode) -> io::Result<()> {
// No difference between adding and modifying in kqueue.
self.modify_filter(filter, key, mode)
}

fn modify_filter(&self, filter: impl Filter, key: usize, mode: PollMode) -> io::Result<()> {
// Convert the filter into a kevent.
let event = filter.filter(libc::EV_ADD | mode_to_flags(mode), key);

// Modify the filter.
self.poller.submit_changes([event])
}

fn delete_filter(&self, filter: impl Filter) -> io::Result<()> {
// Convert the filter into a kevent.
let event = filter.filter(libc::EV_DELETE, 0);

// Delete the filter.
self.poller.submit_changes([event])
}
}

/// A filter that can be registered into a `kqueue`.
pub trait Filter: FilterSealed {}

unsafe impl<T: FilterSealed + ?Sized> FilterSealed for &T {
fn filter(&self, flags: FilterFlags, key: usize) -> libc::kevent {
(**self).filter(flags, key)
}
}

impl<T: Filter + ?Sized> Filter for &T {}

/// Monitor this signal number.
///
/// No matter what `PollMode` is specified, this filter will always be
/// oneshot-only.
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Signal(c_int);

impl Signal {
/// Monitor this signal number.
pub fn new(signal: c_int) -> Self {
Self(signal)
}

/// Get the signal number.
pub fn signal(self) -> c_int {
self.0
}
}

/// Alias for `libc::c_int`.
#[allow(non_camel_case_types)]
pub type c_int = i32;

unsafe impl FilterSealed for Signal {
fn filter(&self, flags: FilterFlags, key: usize) -> libc::kevent {
libc::kevent {
ident: self.0 as _,
filter: libc::EVFILT_SIGNAL,
flags: flags | libc::EV_RECEIPT,
udata: key as _,
..unsafe { mem::zeroed() }
}
}
}

impl Filter for Signal {}

/// Monitor a child process.
#[derive(Debug)]
pub struct Process<'a> {
/// The child process to monitor.
child: &'a Child,

/// The operation to monitor.
ops: ProcessOps,
}

/// The operations that a monitored process can perform.
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[non_exhaustive]
pub enum ProcessOps {
/// The process exited.
Exit,

/// The process was forked.
Fork,

/// The process executed a new process.
Exec,
}

impl<'a> Process<'a> {
/// Monitor a child process.
pub fn new(child: &'a Child, ops: ProcessOps) -> Self {
Self { child, ops }
}
}

unsafe impl FilterSealed for Process<'_> {
fn filter(&self, flags: FilterFlags, key: usize) -> libc::kevent {
let fflags = match self.ops {
ProcessOps::Exit => libc::NOTE_EXIT,
ProcessOps::Fork => libc::NOTE_FORK,
ProcessOps::Exec => libc::NOTE_EXEC,
};

libc::kevent {
ident: self.child.id() as _,
filter: libc::EVFILT_PROC,
flags: flags | libc::EV_RECEIPT,
fflags,
udata: key as _,
..unsafe { mem::zeroed() }
}
}
}

impl Filter for Process<'_> {}

/// Monitor a file on disk.
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Vnode(RawFd);

impl Vnode {
/// Monitor a file on disk.
pub fn new(fd: RawFd) -> Self {
Self(fd)
}

/// Get the file descriptor.
pub fn fd(self) -> RawFd {
self.0
}
}

unsafe impl FilterSealed for Vnode {
fn filter(&self, flags: FilterFlags, key: usize) -> libc::kevent {
libc::kevent {
ident: self.0 as _,
filter: libc::EVFILT_VNODE,
flags: flags | libc::EV_RECEIPT,
udata: key as _,
..unsafe { mem::zeroed() }
}
}
}

impl Filter for Vnode {}

/// Wait for a timeout to expire.
///
/// Modifying the timeout after it has been added to the poller will reset it.
#[cfg(not(any(target_os = "dragonfly", target_os = "netbsd")))]
#[cfg_attr(
docsrs,
doc(cfg(not(any(target_os = "dragonfly", target_os = "netbsd"))))
)]
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Timer {
/// Identifier for the timer.
id: usize,

/// The timeout to wait for.
timeout: Duration,
}

#[cfg(not(any(target_os = "dragonfly", target_os = "netbsd")))]
impl Timer {
/// Create a new timer.
pub fn new(id: usize, timeout: Duration) -> Self {
Self { id, timeout }
}

/// Get the identifier for the timer.
pub fn id(self) -> usize {
self.id
}

/// Get the timeout to wait for.
pub fn timeout(self) -> Duration {
self.timeout
}
}

#[cfg(not(any(target_os = "dragonfly", target_os = "netbsd")))]
unsafe impl FilterSealed for Timer {
fn filter(&self, flags: FilterFlags, key: usize) -> libc::kevent {
// Figure out the granularity of the timer.
let (fflags, data) = {
#[cfg(not(target_os = "openbsd"))]
{
let subsec_nanos = self.timeout.subsec_nanos();

match (subsec_nanos % 1_000, subsec_nanos) {
(_, 0) => (
libc::NOTE_SECONDS,
self.timeout.as_secs().try_into().expect("too many seconds"),
),
(0, _) => (
libc::NOTE_USECONDS,
self.timeout
.as_micros()
.try_into()
.expect("too many microseconds"),
),
(_, _) => (
libc::NOTE_NSECONDS,
self.timeout
.as_nanos()
.try_into()
.expect("too many nanoseconds"),
),
}
}

#[cfg(target_os = "openbsd")]
{
// OpenBSD only supports milliseconds.
(
0,
self.timeout
.as_millis()
.try_into()
.expect("too many milliseconds"),
)
}
};

#[allow(clippy::needless_update)]
libc::kevent {
ident: self.id as _,
filter: libc::EVFILT_TIMER,
flags: flags | libc::EV_RECEIPT,
fflags,
data,
udata: key as _,
..unsafe { mem::zeroed() }
}
}
}

#[cfg(not(any(target_os = "dragonfly", target_os = "netbsd")))]
impl Filter for Timer {}

mod __private {
use crate::sys::FilterFlags;

#[doc(hidden)]
pub trait PollerSealed {}

#[doc(hidden)]
pub unsafe trait FilterSealed {
/// Get the filter for the given event.
///
/// This filter's flags must have `EV_RECEIPT`.
fn filter(&self, flags: FilterFlags, key: usize) -> libc::kevent;
}
}
}
pub mod kqueue;
Loading