Parallel kill on session quit

NEWS.md

@@ -9,6 +9,11 @@
   set to `FALSE` to send the signal to the child process only. The
   argument is ignored on Windows.
 
+* Processes are now killed in parallel with a period of grace on
+  garbage collection (if `cleanup_tree` is `TRUE`) and on session
+  quit. The delay can be controlled with the new `cleanup_grace`
+  argument.
+
 * The `grace` argument of the `kill()` method is now active on Unix
   platforms. processx first tries to kill with `SIGTERM` with a
   timeout of `grace` seconds. After the timeout, `SIGKILL` is sent as

R/finalize.R

@@ -0,0 +1,100 @@
+# Called from the C finalizer of the process handle (Unix only), when
+# the process object is garbage collected.
+process_finalize <- function(private) {
+  cleanup_kill(list(private), private$cleanup_grace)
+}
+
+# Called on session exit (Unix only), with the head of the list of
+# processes that are still alive.
+session_finalize <- function(node) {
+  privates <- list()
+  grace <- 0
+
+  while (!node_is_root(node)) {
+    private <- wref_key(node_value(node))
+    if (!is.null(private)) {
+      privates <- c(privates, list(private))
+      if (!is.null(private$cleanup_grace)) {
+        grace <- max(grace, private$cleanup_grace)
+      }
+    }
+    node <- node_next(node)
+  }
+
+  cleanup_kill(privates, grace)
+}
+
+# Kill all processes (and their process groups) that were started with
+# `cleanup = TRUE`, and the process trees of the ones started with
+# `cleanup_tree = TRUE`. All processes are killed in parallel: first
+# they get a SIGTERM, then we wait at most `grace` seconds for them to
+# exit, then the survivors get a SIGKILL.
+#
+# The process groups are signalled via processx's own handle, so this
+# works without ps support. The handle skips processes whose exit
+# status was already collected, so we never signal a reused pid.
+cleanup_kill <- function(privates, grace) {
+  privates <- Filter(function(p) isTRUE(p$cleanup), privates)
+  if (length(privates) == 0) {
+    return(invisible())
+  }
+
+  grace <- if (is.null(grace)) 0 else grace
+  deadline <- Sys.time() + grace
+
+  if (grace > 0) {
+    for (p in privates) cleanup_signal_group(p, 15L) # SIGTERM
+  }
+
+  # Process trees, ps waits for them in parallel, and sends SIGKILL
+  # to the ones that are still alive after the grace period.
+  trees <- Filter(function(p) isTRUE(p$cleanup_tree), privates)
+  if (length(trees) > 0 && ps::ps_is_supported()) {
+    tree <- unlist(
+      lapply(trees, function(p) {
+        tryCatch(ps::ps_find_tree(p$tree_id), error = function(e) list())
+      }),
+      recursive = FALSE
+    )
+    if (length(tree) > 0) {
+      tryCatch(
+        ps::ps_kill(tree, grace = grace * 1000),
+        error = function(e) NULL
+      )
+    }
+  }
+
+  # Wait for the rest of the grace period for the process group leaders
+  if (grace > 0) {
+    for (p in privates) {
+      left <- as.double(deadline - Sys.time(), units = "secs")
+      if (left <= 0) break
+      tryCatch(
+        call_with_cleanup(
+          c_processx_wait,
+          p$status,
+          as.integer(left * 1000),
+          NULL
+        ),
+        error = function(e) NULL
+      )
+    }
+  }
+
+  for (p in privates) cleanup_signal_group(p, 9L) # SIGKILL
+
+  invisible()
+}
+
+cleanup_signal_group <- function(private, signal) {
+  tryCatch(
+    .Call(c_processx_signal, private$status, signal, NULL, TRUE),
+    error = function(e) FALSE
+  )
+}
+
+wref_key <- function(x) .Call(c_processx__wref_key, x)
+node_is_root <- function(x) is.null(node_next(x))
+node_prev <- function(x) x[[1]]
+node_next <- function(x) x[[2]]
+node_value <- function(x) x[[3]]

R/initialize.R

@@ -193,7 +193,6 @@ process_initialize <- function(
     windows_detached_process,
     private,
     cleanup,
-    cleanup_grace,
     wd,
     encoding,
     paste0("PROCESSX_", private$tree_id, "=YES"),

R/process.R

@@ -816,8 +816,11 @@ process <- R6::R6Class(
     tree_id = NULL,
 
     finalize = function() {
+      # On Unix the C finalizer of the process handle calls
+      # `process_finalize()`, which also cleans up the process tree.
       if (
-        !is.null(private$tree_id) &&
+        os_type() == "windows" &&
+          !is.null(private$tree_id) &&
           private$cleanup_tree &&
           ps::ps_is_supported()
       ) {

src/Makevars

@@ -4,7 +4,7 @@ OBJECTS = init.o poll.o errors.o processx-connection.o   \
           processx-vector.o create-time.o base64.o       \
 	  unix/childlist.o unix/connection.o             \
           unix/processx.o unix/sigchld.o unix/utils.o    \
-	  unix/named_pipe.o cleancall.o
+	  unix/named_pipe.o cleancall.o utils.o
 
 all: tools/px tools/sock supervisor/supervisor client$(SHLIB_EXT) $(SHLIB) strip
 

src/Makevars.win

@@ -3,7 +3,7 @@
 OBJECTS = init.o poll.o errors.o processx-connection.o		     \
           processx-vector.o create-time.o base64.o                   \
           win/processx.o win/stdio.o win/named_pipe.o                \
-	  win/utils.o win/thread.o cleancall.o
+	  win/utils.o win/thread.o cleancall.o utils.o
 
 .PHONY: all clean
 

src/init.c

@@ -95,10 +95,11 @@ SEXP is_valgrind_(void)
 #endif
   return Rf_ScalarLogical(PROCESSX_RUNNING_ON_VALGRIND() != 0);
 }
+SEXP processx__wref_key(SEXP);
 
 static const R_CallMethodDef callMethods[]  = {
   CLEANCALL_METHOD_RECORD,
-  { "processx_exec",               (DL_FUNC) &processx_exec,              16 },
+  { "processx_exec",               (DL_FUNC) &processx_exec,              15 },
   { "processx_wait",               (DL_FUNC) &processx_wait,               3 },
   { "processx_is_alive",           (DL_FUNC) &processx_is_alive,           2 },
   { "processx_pty_close",          (DL_FUNC) &processx_pty_close,          2 },
@@ -118,6 +119,7 @@ static const R_CallMethodDef callMethods[]  = {
   { "processx__proc_start_time",   (DL_FUNC) &processx__proc_start_time,   1 },
   { "processx__proc_end_time",     (DL_FUNC) &processx__proc_end_time,     1 },
   { "processx__set_boot_time",     (DL_FUNC) &processx__set_boot_time,     1 },
+  { "processx__wref_key",          (DL_FUNC) &processx__wref_key,          1 },
 
   { "processx_connection_create",     (DL_FUNC) &processx_connection_create,     2 },
   { "processx_connection_read_chars", (DL_FUNC) &processx_connection_read_chars, 2 },

src/processx.h

@@ -45,8 +45,8 @@ extern "C" {
 SEXP processx_exec(SEXP command, SEXP args, SEXP pty, SEXP pty_options,
 		   SEXP connections, SEXP env, SEXP windows_verbatim_args,
 		   SEXP windows_hide_window, SEXP windows_detached_process,
-		   SEXP private_, SEXP cleanup, SEXP cleanup_grace,
-		   SEXP wd, SEXP encoding, SEXP tree_id, SEXP linux_pdeathsig);
+		   SEXP private_, SEXP cleanup, SEXP wd, SEXP encoding,
+		   SEXP tree_id, SEXP linux_pdeathsig);
 SEXP processx_wait(SEXP status, SEXP timeout, SEXP name);
 SEXP processx_is_alive(SEXP status, SEXP name);
 SEXP processx_pty_close(SEXP status, SEXP name);
@@ -122,4 +122,10 @@ typedef struct {
 }
 #endif
 
+#define r_no_return __attribute__ ((noreturn))
+
+r_no_return void r_unwind(SEXP x);
+SEXP r_unwind_protect(void (*fn)(void *data), void *data);
+SEXP r_safe_eval(SEXP expr, SEXP env, SEXP *out);
+
 #endif

src/unix/processx-unix.h

@@ -23,9 +23,10 @@ typedef struct processx_handle_s {
   int fd2;			/* readable */
   int waitpipe[2];		/* use it for wait() with timeout */
   int cleanup;
-  double cleanup_grace;
+  SEXP r6_private;
   double create_time;
   double end_time;          /* 0.0 until the process exits */
+  SEXP finalizer_node;
   processx_connection_t *pipes[3];
   int ptyfd;
   int pty_child_fd; /* parent holds child PTY fd open to prevent data loss on macOS */

src/unix/processx.c

@@ -22,7 +22,7 @@ static void processx__child_init(processx_handle_t *handle, SEXP connections,
                                  processx_options_t *options,
 				 const char *tree_id);
 
-static SEXP processx__make_handle(SEXP private, int cleanup, double cleanup_grace);
+static SEXP processx__make_handle(SEXP private, int cleanup);
 static void processx__handle_destroy(processx_handle_t *handle);
 void processx__create_connections(processx_handle_t *handle, SEXP private,
 				  const char *encoding);
@@ -361,6 +361,40 @@ SEXP c_processx_kill_data(void *payload) {
   return R_NilValue;
 }
 
+static
+SEXP finalizer_call(SEXP private) {
+  static SEXP finalize_fn = NULL;
+  if (!finalize_fn) {
+    finalize_fn = lang3(install(":::"),
+                        install("processx"),
+                        install("process_finalize"));
+    R_PreserveObject(finalize_fn);
+  }
+
+  return lang2(finalize_fn, private);
+}
+
+static SEXP session_finalizer_list = NULL;
+
+// These need to be macros to be considered protectors by rchk
+#define node_poke_prev(NODE, PREV) SET_VECTOR_ELT((NODE), 0, (PREV))
+#define node_poke_next(NODE, NEXT) SET_VECTOR_ELT((NODE), 1, (NEXT))
+#define node_poke_value(NODE, VALUE) SET_VECTOR_ELT((NODE), 2, (VALUE))
+
+static SEXP node_prev(SEXP node) { return VECTOR_ELT(node, 0); }
+static SEXP node_next(SEXP node) { return VECTOR_ELT(node, 1); }
+
+static
+SEXP new_node(SEXP prev, SEXP next, SEXP value) {
+  SEXP out = allocVector(VECSXP, 3);
+
+  node_poke_prev(out, prev);
+  node_poke_next(out, next);
+  node_poke_value(out, value);
+
+  return out;
+}
+
 void processx__finalizer(SEXP status) {
   processx_handle_t *handle = (processx_handle_t*) R_ExternalPtrAddr(status);
 
@@ -373,14 +407,25 @@ void processx__finalizer(SEXP status) {
   if (!handle)
     return;
 
-  // FIXME: Do we need cleancall here?
+  /* Only call into R if there is something to clean up */
+  SEXP err = NULL;
   if (handle->cleanup) {
-    struct cleanup_kill_data data = {
-      .status = status,
-      .grace = handle->cleanup_grace,
-      .name = R_NilValue
-    };
-    r_with_cleanup_context(c_processx_kill_data, &data);
+    SEXP call = PROTECT(finalizer_call(handle->r6_private));
+    err = r_safe_eval(call, R_BaseEnv, NULL);
+    UNPROTECT(1);
+  }
+
+  /* Remove node from session finalizer list */
+  SEXP node = handle->finalizer_node;
+  SEXP prev = node_prev(node);
+  SEXP next = node_next(node);
+
+  if (prev != R_NilValue) {
+    node_poke_next(prev, next);
+  }
+  node_poke_prev(next, prev);
+  if (node == session_finalizer_list) {
+    session_finalizer_list = next;
   }
 
   /* Note: if no cleanup is requested, then we still have a sigchld
@@ -390,21 +435,65 @@ void processx__finalizer(SEXP status) {
   /* Deallocate memory */
   R_ClearExternalPtr(status);
   processx__handle_destroy(handle);
+
+  if (err) {
+    r_unwind(err);
+  }
 }
 
-static SEXP processx__make_handle(SEXP private, int cleanup, double cleanup_grace) {
+static
+void processx__session_finalizer(SEXP _) {
+  static SEXP finalize_fn = NULL;
+  if (!finalize_fn) {
+    finalize_fn = lang3(install(":::"),
+                        install("processx"),
+                        install("session_finalize"));
+    R_PreserveObject(finalize_fn);
+  }
+
+  SEXP call = PROTECT(lang2(finalize_fn, session_finalizer_list));
+  eval(call, R_BaseEnv);
+  UNPROTECT(1);
+}
+
+static
+void processx__register_finalizer(SEXP status, processx_handle_t *handle) {
+  if (!session_finalizer_list) {
+    // This root node is never popped and protects the rest of the list
+    session_finalizer_list = new_node(R_NilValue, R_NilValue, R_NilValue);
+    R_PreserveObject(session_finalizer_list);
+    R_RegisterCFinalizerEx(R_BaseEnv, &processx__session_finalizer, 1);
+  }
+
+  // GC finalizer
+  R_RegisterCFinalizerEx(status, &processx__finalizer, 0);
+
+  // Session finalizer
+  SEXP private_weakref = R_MakeWeakRef(handle->r6_private, R_NilValue, R_NilValue, 0);
+  PROTECT(private_weakref);
+
+  SEXP node = new_node(R_NilValue, session_finalizer_list, private_weakref);
+  node_poke_prev(session_finalizer_list, node);
+  session_finalizer_list = node;
+  handle->finalizer_node = node;
+
+  UNPROTECT(1);
+}
+
+static SEXP processx__make_handle(SEXP private, int cleanup) {
   processx_handle_t * handle;
   SEXP result;
 
   handle = (processx_handle_t*) malloc(sizeof(processx_handle_t));
   if (!handle) { R_THROW_ERROR("Cannot make processx handle, out of memory"); }
   memset(handle, 0, sizeof(processx_handle_t));
+
   handle->waitpipe[0] = handle->waitpipe[1] = -1;
+  handle->r6_private = private;
+  handle->cleanup = cleanup;
 
   result = PROTECT(R_MakeExternalPtr(handle, private, R_NilValue));
-  R_RegisterCFinalizerEx(result, processx__finalizer, 1);
-  handle->cleanup = cleanup;
-  handle->cleanup_grace = cleanup_grace;
+  processx__register_finalizer(result, handle);
 
   UNPROTECT(1);
   return result;
@@ -452,14 +541,12 @@ void processx__make_socketpair(int pipe[2], const char *exe) {
 SEXP processx_exec(SEXP command, SEXP args, SEXP pty, SEXP pty_options,
                    SEXP connections, SEXP env, SEXP windows_verbatim_args,
                    SEXP windows_hide_window, SEXP windows_detached_process,
-                   SEXP private, SEXP cleanup, SEXP cleanup_grace, SEXP wd,
-                   SEXP encoding, SEXP tree_id, SEXP linux_pdeathsig) {
+                   SEXP private, SEXP cleanup, SEXP wd, SEXP encoding,
+                   SEXP tree_id, SEXP linux_pdeathsig) {
 
   char *ccommand = processx__tmp_string(command, 0);
   char **cargs = processx__tmp_character(args);
   char **cenv = isNull(env) ? 0 : processx__tmp_character(env);
-  int ccleanup = INTEGER(cleanup)[0];
-  double ccleanup_grace = REAL(cleanup_grace)[0];
 
   const int cpty = LOGICAL(pty)[0];
   const char *cencoding = CHAR(STRING_ELT(encoding, 0));
@@ -495,7 +582,8 @@ SEXP processx_exec(SEXP command, SEXP args, SEXP pty, SEXP pty_options,
 
   processx__setup_sigchld();
 
-  result = PROTECT(processx__make_handle(private, ccleanup, ccleanup_grace));
+  int ccleanup = LOGICAL(cleanup)[0];
+  result = PROTECT(processx__make_handle(private, ccleanup));
   handle = R_ExternalPtrAddr(result);
 
   if (cpty) {