From 983cd53f1304c484d1612a412f49b0f835141123 Mon Sep 17 00:00:00 2001
From: Jake Hillion <jake@hillion.co.uk>
Date: Tue, 24 May 2022 21:46:24 +0100
Subject: [PATCH 1/2] added ability to be interrupted to spawners

---
 src/spawner/mod.rs | 32 +++++++++++++++++++++++++-------
 1 file changed, 25 insertions(+), 7 deletions(-)

diff --git a/src/spawner/mod.rs b/src/spawner/mod.rs
index 9359a0c..c241376 100644
--- a/src/spawner/mod.rs
+++ b/src/spawner/mod.rs
@@ -19,6 +19,7 @@ use std::path::{Path, PathBuf};
 use nix::sys::signal::{kill, Signal};
 use nix::sys::socket::{recvmsg, ControlMessageOwned, MsgFlags};
 use nix::unistd::{self, Pid};
+use nix::Error as NixError;
 
 const BUFFER_SIZE: usize = 1024;
 const MAX_FILE_DESCRIPTORS: usize = 16;
@@ -151,7 +152,16 @@ impl<'a> Spawner<'a> {
     fn pipe_trigger(&self, mut pipe: File, spec: &Entrypoint, name: &str) -> Result<()> {
         let mut buf = [0_u8; BUFFER_SIZE];
         loop {
-            let read_bytes = pipe.read(&mut buf)?;
+            let read_bytes = match pipe.read(&mut buf) {
+                Ok(n) => Ok(n),
+                Err(e) => {
+                    if e.kind() == std::io::ErrorKind::Interrupted {
+                        return Ok(());
+                    }
+
+                    Err(e)
+                }
+            }?;
             if read_bytes == 0 {
                 return Ok(());
             }
@@ -199,16 +209,24 @@ impl<'a> Spawner<'a> {
         let mut cmsg_buf = nix::cmsg_space!([RawFd; MAX_FILE_DESCRIPTORS]);
 
         loop {
-            let msg = recvmsg::<()>(
+            let msg = match recvmsg::<()>(
                 socket.as_raw_fd(),
                 &mut [],
                 Some(&mut cmsg_buf),
                 MsgFlags::empty(),
-            )
-            .map_err(|e| Error::Nix {
-                msg: "recvmsg",
-                src: e,
-            })?;
+            ) {
+                Ok(m) => Ok(m),
+                Err(e) => {
+                    if e == NixError::EINTR {
+                        return Ok(());
+                    }
+
+                    Err(Error::Nix {
+                        msg: "recvmsg",
+                        src: e,
+                    })
+                }
+            }?;
 
             debug!("triggering from socket recvmsg");
 
-- 
2.47.0


From efbf6be52076d6ece8a1fd318dea29bd3444d03c Mon Sep 17 00:00:00 2001
From: Jake Hillion <jake@hillion.co.uk>
Date: Tue, 24 May 2022 22:13:31 +0100
Subject: [PATCH 2/2] signal handling in spawners

---
 src/spawner/mod.rs | 38 +++++++++++++++++++++++++++++++++++++-
 1 file changed, 37 insertions(+), 1 deletion(-)

diff --git a/src/spawner/mod.rs b/src/spawner/mod.rs
index c241376..9a7302c 100644
--- a/src/spawner/mod.rs
+++ b/src/spawner/mod.rs
@@ -18,7 +18,8 @@ use std::path::{Path, PathBuf};
 
 use nix::sys::signal::{kill, Signal};
 use nix::sys::socket::{recvmsg, ControlMessageOwned, MsgFlags};
-use nix::unistd::{self, Pid};
+use nix::sys::wait::{waitid, Id, WaitPidFlag, WaitStatus};
+use nix::unistd::{self, fork, ForkResult, Pid};
 use nix::Error as NixError;
 
 const BUFFER_SIZE: usize = 1024;
@@ -150,6 +151,9 @@ impl<'a> Spawner<'a> {
     }
 
     fn pipe_trigger(&self, mut pipe: File, spec: &Entrypoint, name: &str) -> Result<()> {
+        // put the work in a forked process that can handle signals
+        Self::fork_for_trigger()?;
+
         let mut buf = [0_u8; BUFFER_SIZE];
         loop {
             let read_bytes = match pipe.read(&mut buf) {
@@ -206,6 +210,9 @@ impl<'a> Spawner<'a> {
     }
 
     fn file_socket_trigger(&self, socket: File, spec: &Entrypoint, name: &str) -> Result<()> {
+        // put the work in a forked process that can handle signals
+        Self::fork_for_trigger()?;
+
         let mut cmsg_buf = nix::cmsg_space!([RawFd; MAX_FILE_DESCRIPTORS]);
 
         loop {
@@ -281,6 +288,35 @@ impl<'a> Spawner<'a> {
         }
     }
 
+    fn fork_for_trigger() -> Result<()> {
+        // SAFETY: only unsafe in a multi-threaded program
+        if let ForkResult::Parent { child: _pid } = unsafe { fork() }.map_err(|e| Error::Nix {
+            msg: "fork",
+            src: e,
+        })? {
+            let status = waitid(Id::All, WaitPidFlag::WEXITED).map_err(|e| Error::Nix {
+                msg: "waitpid",
+                src: e,
+            })?;
+
+            match status {
+                WaitStatus::Exited(_pid, code) => {
+                    std::process::exit(code);
+                }
+                WaitStatus::Signaled(pid, sig, _coredump) => {
+                    debug!(
+                        "trigger: forked child {} was terminated with signal {}",
+                        pid, sig
+                    );
+                    std::process::exit(-1);
+                }
+                _ => unreachable!(),
+            }
+        }
+
+        Ok(())
+    }
+
     fn stop_self(name: &str) -> Result<()> {
         info!("stopping process `{}`", name);
 
-- 
2.47.0