Skip to content

Commit

Permalink
feat: Allow fallback to signal backend
Browse files Browse the repository at this point in the history
As pidfd isn't available in older versions of Linux that Rust still
supports, this is necessary for running on older Linux. In addition,
signals tests are still kept in CI.

Signed-off-by: John Nunley <dev@notgull.net>
  • Loading branch information
notgull committed Mar 30, 2024
1 parent 1e0751f commit bbd42b5
Show file tree
Hide file tree
Showing 6 changed files with 271 additions and 60 deletions.
3 changes: 3 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,9 @@ jobs:
if: startsWith(matrix.rust, 'nightly')
run: cargo check -Z features=dev_dep
- run: cargo test
- run: cargo test
env:
RUSTFLAGS: ${{ env.RUSTFLAGS }} --cfg async_process_force_signal_backend

msrv:
runs-on: ${{ matrix.os }}
Expand Down
2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,14 @@ tracing = { version = "0.1.40", default-features = false }

[target.'cfg(unix)'.dependencies]
async-io = "2.1.0"
async-signal = "0.2.3"
rustix = { version = "0.38", default-features = false, features = ["std", "fs"] }

[target.'cfg(any(target_os = "linux", target_os = "android"))'.dependencies]
async-channel = "2.0.0"
async-task = "4.7.0"

[target.'cfg(all(unix, not(any(target_os = "linux", target_os = "android"))))'.dependencies]
async-signal = "0.2.3"
rustix = { version = "0.38", default-features = false, features = ["std", "fs", "process"] }

[target.'cfg(windows)'.dependencies]
Expand Down
39 changes: 12 additions & 27 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -80,24 +80,13 @@ use futures_lite::{future, io, prelude::*};
#[doc(no_inline)]
pub use std::process::{ExitStatus, Output, Stdio};

cfg_if::cfg_if! {
if #[cfg(any(
target_os = "linux",
target_os = "android"
))] {
#[path = "reaper/wait.rs"]
mod reaper;
} else {
#[path = "reaper/signal.rs"]
mod reaper;
}
}

#[cfg(unix)]
pub mod unix;
#[cfg(windows)]
pub mod windows;

mod reaper;

mod sealed {
pub trait Sealed {}
}
Expand Down Expand Up @@ -177,11 +166,6 @@ impl Reaper {
.expect("cannot spawn async-process thread");
}

/// Reap zombie processes forever.
async fn reap(&'static self, driver_guard: reaper::Lock) -> ! {
self.sys.reap(driver_guard).await
}

/// Register a process with this reaper.
fn register(&'static self, child: std::process::Child) -> io::Result<reaper::ChildGuard> {
self.ensure_driven();
Expand Down Expand Up @@ -728,16 +712,9 @@ impl TryFrom<ChildStderr> for OwnedFd {
/// }).await;
/// # });
/// ```
#[allow(clippy::manual_async_fn)]
#[inline]
pub fn driver() -> impl Future<Output = Infallible> + Send + 'static {
struct CallOnDrop<F: FnMut()>(F);

impl<F: FnMut()> Drop for CallOnDrop<F> {
fn drop(&mut self) {
(self.0)();
}
}

async {
// Get the reaper.
let reaper = Reaper::get();
Expand All @@ -760,7 +737,7 @@ pub fn driver() -> impl Future<Output = Infallible> + Send + 'static {

// Acquire the reaper lock and start polling the SIGCHLD event.
let guard = reaper.sys.lock().await;
reaper.reap(guard).await
reaper.sys.reap(guard).await
}
}

Expand Down Expand Up @@ -1158,6 +1135,14 @@ fn blocking_fd(fd: rustix::fd::BorrowedFd<'_>) -> io::Result<()> {
Ok(())
}

struct CallOnDrop<F: FnMut()>(F);

impl<F: FnMut()> Drop for CallOnDrop<F> {
fn drop(&mut self) {
(self.0)();
}
}

#[cfg(test)]
mod test {
#[test]
Expand Down
221 changes: 221 additions & 0 deletions src/reaper/mod.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,221 @@
//! The underlying system reaper.
//!
//! There are two backends:
//!
//! - signal, which waits for SIGCHLD.
//! - wait, which waits directly on a process handle.
//!
//! "wait" is preferred, but is not available on all supported Linuxes. So we
//! test to see if pidfd is supported first. If it is, we use wait. If not, we use
//! signal.

#![allow(irrefutable_let_patterns)]

/// Enable the waiting reaper.
#[cfg(any(target_os = "linux", target_os = "android"))]
macro_rules! cfg_wait {
($($tt:tt)*) => {$($tt)*};
}

/// Enable the waiting reaper.
#[cfg(not(any(target_os = "linux", target_os = "android")))]
macro_rules! cfg_wait {
($($tt:tt)*) => {};
}

/// Enable signals.
macro_rules! cfg_signal {
($($tt:tt)*) => {$($tt)*};
}

cfg_wait! {
mod wait;
}

cfg_signal! {
mod signal;
}

use std::io;
use std::sync::Mutex;

/// The underlying system reaper.
pub(crate) enum Reaper {
#[cfg(any(target_os = "linux", target_os = "android"))]
/// The reaper based on the wait backend.
Wait(wait::Reaper),

/// The reaper based on the signal backend.
Signal(signal::Reaper),
}

/// The wrapper around a child.
pub(crate) enum ChildGuard {
#[cfg(any(target_os = "linux", target_os = "android"))]
/// The child guard based on the wait backend.
Wait(wait::ChildGuard),

/// The child guard based on the signal backend.
Signal(signal::ChildGuard),
}

/// A lock on the reaper.
pub(crate) enum Lock {
#[cfg(any(target_os = "linux", target_os = "android"))]
/// The wait-based reaper needs no lock.
Wait,

/// The lock for the signal-based reaper.
Signal(signal::Lock),
}

impl Reaper {
/// Create a new reaper.
pub(crate) fn new() -> Self {
cfg_wait! {
if wait::available() && !cfg!(async_process_force_signal_backend) {
return Self::Wait(wait::Reaper::new());
}
}

// Return the signal-based reaper.
cfg_signal! {
return Self::Signal(signal::Reaper::new());
}

#[allow(unreachable_code)]
{
panic!("neither the signal backend nor the waiter backend is available")
}
}

/// Lock the driver thread.
///
/// This makes it so only one thread can reap at once.
pub(crate) async fn lock(&'static self) -> Lock {
cfg_wait! {
if let Self::Wait(_this) = self {
// No locking needed.
return Lock::Wait;
}
}

cfg_signal! {
if let Self::Signal(this) = self {
// We need to lock.
return Lock::Signal(this.lock().await);
}
}

unreachable!()
}

/// Reap zombie processes forever.
pub(crate) async fn reap(&'static self, lock: Lock) -> ! {
cfg_wait! {
if let (Self::Wait(this), Lock::Wait) = (self, &lock) {
return this.reap().await;
}
}

cfg_signal! {
if let (Self::Signal(this), Lock::Signal(lock)) = (self, lock) {
return this.reap(lock).await;
}
}

unreachable!()
}

/// Register a child into this reaper.
pub(crate) fn register(&'static self, child: std::process::Child) -> io::Result<ChildGuard> {
cfg_wait! {
if let Self::Wait(this) = self {
return this.register(child).map(ChildGuard::Wait);
}
}

cfg_signal! {
if let Self::Signal(this) = self {
return this.register(child).map(ChildGuard::Signal);
}
}

unreachable!()
}

/// Wait for the inner child to complete.
pub(crate) async fn status(
&'static self,
child: &Mutex<crate::ChildGuard>,
) -> io::Result<std::process::ExitStatus> {
cfg_wait! {
if let Self::Wait(this) = self {
return this.status(child).await;
}
}

cfg_signal! {
if let Self::Signal(this) = self {
return this.status(child).await;
}
}

unreachable!()
}

/// Do we have any registered zombie processes?
pub(crate) fn has_zombies(&'static self) -> bool {
cfg_wait! {
if let Self::Wait(this) = self {
return this.has_zombies();
}
}

cfg_signal! {
if let Self::Signal(this) = self {
return this.has_zombies();
}
}

unreachable!()
}
}

impl ChildGuard {
/// Get a reference to the inner process.
pub(crate) fn get_mut(&mut self) -> &mut std::process::Child {
cfg_wait! {
if let Self::Wait(this) = self {
return this.get_mut();
}
}

cfg_signal! {
if let Self::Signal(this) = self {
return this.get_mut();
}
}

unreachable!()
}

/// Start reaping this child process.
pub(crate) fn reap(&mut self, reaper: &'static Reaper) {
cfg_wait! {
if let (Self::Wait(this), Reaper::Wait(reaper)) = (&mut *self, reaper) {
this.reap(reaper);
return;
}
}

cfg_signal! {
if let (Self::Signal(this), Reaper::Signal(reaper)) = (self, reaper) {
this.reap(reaper);
return;
}
}

unreachable!()
}
}
10 changes: 7 additions & 3 deletions src/reaper/signal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -131,9 +131,8 @@ impl ChildGuard {

/// Begin the reaping process for this child.
pub(crate) fn reap(&mut self, reaper: &'static Reaper) {
let mut zombies = reaper.zombies.lock().unwrap();
if let Ok(None) = self.get_mut().try_wait() {
zombies.push(self.inner.take().unwrap());
reaper.zombies.lock().unwrap().push(self.inner.take().unwrap());
}
}
}
Expand Down Expand Up @@ -178,8 +177,13 @@ cfg_if::cfg_if! {
/// Register a process object into this pipe.
fn register(&self, child: &std::process::Child) -> io::Result<()> {
// Called when a child exits.
#[allow(clippy::infallible_destructuring_match)]
unsafe extern "system" fn callback(_: *mut c_void, _: BOOLEAN) {
crate::Reaper::get().sys.pipe.sender.try_send(()).ok();
let reaper = match &crate::Reaper::get().sys {
super::Reaper::Signal(reaper) => reaper,
};

reaper.pipe.sender.try_send(()).ok();
}

// Register this child process to invoke `callback` on exit.
Expand Down
Loading

0 comments on commit bbd42b5

Please sign in to comment.