From 4e4d795303eda4d448cf3cbba600c7892f6a1728 Mon Sep 17 00:00:00 2001 From: Lionel Henry Date: Thu, 27 Apr 2023 16:28:30 +0200 Subject: [PATCH 1/3] Merge process finalisers --- R/finalize.R | 17 +++++++++ R/initialize.R | 4 ++- R/process.R | 6 +++- src/Makevars | 2 +- src/init.c | 2 +- src/processx.h | 10 ++++-- src/unix/processx-unix.h | 2 +- src/unix/processx.c | 46 ++++++++++++++---------- src/utils.c | 78 ++++++++++++++++++++++++++++++++++++++++ src/win/processx-win.h | 1 - src/win/processx.c | 12 +++---- 11 files changed, 147 insertions(+), 33 deletions(-) create mode 100644 R/finalize.R create mode 100644 src/utils.c diff --git a/R/finalize.R b/R/finalize.R new file mode 100644 index 00000000..7d1b34b1 --- /dev/null +++ b/R/finalize.R @@ -0,0 +1,17 @@ +process_finalize <- function(private) { + ps <- list() + + if (private$cleanup) { + # Can't be created in advance because the ps finalizer might run first + handle <- ps::ps_handle( + private$pid, + format_unix_time(private$starttime_raw) + ) + ps <- c(ps, list(handle)) + } + if (private$cleanup_tree) { + ps <- c(ps, ps::ps_find_tree(private$tree_id)) + } + + ps::ps_kill_parallel(ps, private$cleanup_grace) +} diff --git a/R/initialize.R b/R/initialize.R index 30ddaa0a..9f1dac76 100644 --- a/R/initialize.R +++ b/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"), @@ -213,6 +212,9 @@ process_initialize <- function( } private$starttime <- max(private$starttime_raw, before_start) + # Needed for cleaning up + private$pid <- self$get_pid() + ## Need to close this, otherwise the child's end of the pipe ## will not be closed when the child exits, and then we cannot ## poll it. diff --git a/R/process.R b/R/process.R index f81d3e52..ee5e9fef 100644 --- a/R/process.R +++ b/R/process.R @@ -795,6 +795,7 @@ process <- R6::R6Class( windows_hide_window = NULL, status = NULL, # C file handle + pid = NULL, # pid for cleanup supervised = FALSE, # Whether process is tracked by supervisor @@ -816,8 +817,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() ) { diff --git a/src/Makevars b/src/Makevars index c634710d..47817553 100644 --- a/src/Makevars +++ b/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 diff --git a/src/init.c b/src/init.c index 15adb7ba..127bffbc 100644 --- a/src/init.c +++ b/src/init.c @@ -98,7 +98,7 @@ SEXP is_valgrind_(void) 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 }, diff --git a/src/processx.h b/src/processx.h index 6ceaf9f3..7716322e 100644 --- a/src/processx.h +++ b/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 diff --git a/src/unix/processx-unix.h b/src/unix/processx-unix.h index 504b4c1a..5f2aa838 100644 --- a/src/unix/processx-unix.h +++ b/src/unix/processx-unix.h @@ -23,7 +23,7 @@ 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 */ processx_connection_t *pipes[3]; diff --git a/src/unix/processx.c b/src/unix/processx.c index a75480ae..e4fbd857 100644 --- a/src/unix/processx.c +++ b/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,19 @@ 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); +} + void processx__finalizer(SEXP status) { processx_handle_t *handle = (processx_handle_t*) R_ExternalPtrAddr(status); @@ -373,15 +386,9 @@ void processx__finalizer(SEXP status) { if (!handle) return; - // FIXME: Do we need cleancall here? - 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)); + SEXP err = r_safe_eval(call, R_BaseEnv, NULL); + UNPROTECT(1); /* Note: if no cleanup is requested, then we still have a sigchld handler, to read out the exit code via waitpid, but no handle @@ -390,21 +397,25 @@ 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 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; result = PROTECT(R_MakeExternalPtr(handle, private, R_NilValue)); R_RegisterCFinalizerEx(result, processx__finalizer, 1); - handle->cleanup = cleanup; - handle->cleanup_grace = cleanup_grace; UNPROTECT(1); return result; @@ -452,14 +463,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 +504,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) { diff --git a/src/utils.c b/src/utils.c new file mode 100644 index 00000000..a765d986 --- /dev/null +++ b/src/utils.c @@ -0,0 +1,78 @@ +#include +#include "processx.h" + +// Need to jump out of the `R_UnwindProtect()` context +#include + +r_no_return +void r_unwind(SEXP x) { + if (inherits(x, "error")) { + SEXP call = PROTECT(lang2(install("stop"), x)); + eval(call, R_BaseEnv); + } else { + R_ContinueUnwind(x); + } + error("Unreachable"); +} + +static +void unwind_cleanup(void *payload, Rboolean jump) { + if (jump) { + jmp_buf *env = (jmp_buf *) payload; + longjmp(*env, 1); + } +} + +// Conversion of SEXP-returning callback to a void-returning one +struct callback_compat { + void (*fn)(void *data); + void *data; +}; + +static +SEXP callback_compat(void *payload) { + struct callback_compat *data = (struct callback_compat *) payload; + data->fn(data->data); + return R_NilValue; +} + +SEXP r_unwind_protect(void (*fn)(void *data), void *data) { + SEXP cont = PROTECT(R_MakeUnwindCont()); + jmp_buf env; + + struct callback_compat compat_data = { + .fn = fn, + .data = data + }; + + if (setjmp(env)) { + UNPROTECT(1); + return cont; + } + + R_UnwindProtect(&callback_compat, &compat_data, &unwind_cleanup, &env, cont); + + UNPROTECT(1); + return NULL; +} + +struct safe_eval { + SEXP expr; + SEXP env; + SEXP *out; +}; + +static +void safe_eval_callback(void *payload) { + struct safe_eval *data = (struct safe_eval *) payload; + SEXP out = eval(data->expr, data->env); + + if (data->out) { + *data->out = out; + } +} + +SEXP r_safe_eval(SEXP expr, SEXP env, SEXP *out) { + struct safe_eval data = { .expr = expr, .env = env, .out = out }; + return r_unwind_protect(&safe_eval_callback, &data); +} diff --git a/src/win/processx-win.h b/src/win/processx-win.h index d60179ab..b533c9be 100644 --- a/src/win/processx-win.h +++ b/src/win/processx-win.h @@ -13,7 +13,6 @@ typedef struct processx_handle_s { HANDLE waitObject; processx_connection_t *pipes[3]; int cleanup; - double cleanup_grace; double create_time; double end_time; /* 0.0 until the process exits */ void *ptycon; /* ConPTY handle (HPCON), NULL if not using PTY */ diff --git a/src/win/processx.c b/src/win/processx.c index 9ff64c78..6ce27d30 100644 --- a/src/win/processx.c +++ b/src/win/processx.c @@ -922,7 +922,7 @@ void processx__finalizer(SEXP status) { processx__handle_destroy(handle); } -SEXP processx__make_handle(SEXP private, int cleanup, double cleanup_grace) { +SEXP processx__make_handle(SEXP private, int cleanup) { processx_handle_t * handle; SEXP result; @@ -933,7 +933,6 @@ SEXP processx__make_handle(SEXP private, int cleanup, double cleanup_grace) { result = PROTECT(R_MakeExternalPtr(handle, private, R_NilValue)); R_RegisterCFinalizerEx(result, processx__finalizer, 1); handle->cleanup = cleanup; - handle->cleanup_grace = cleanup_grace; UNPROTECT(1); return result; @@ -952,8 +951,8 @@ void processx__handle_destroy(processx_handle_t *handle) { SEXP processx_exec(SEXP command, SEXP args, SEXP pty, SEXP pty_options, SEXP connections, SEXP env, SEXP windows_verbatim_args, SEXP windows_hide, 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) { const char *ccommand = CHAR(STRING_ELT(command, 0)); const char *cencoding = CHAR(STRING_ELT(encoding, 0)); @@ -971,8 +970,7 @@ SEXP processx_exec(SEXP command, SEXP args, SEXP pty, SEXP pty_options, DWORD process_flags; processx_handle_t *handle; - int ccleanup = INTEGER(cleanup)[0]; - double ccleanup_grace = REAL(cleanup_grace)[0]; + int ccleanup = LOGICAL(cleanup)[0]; SEXP result; DWORD dwerr; @@ -1040,7 +1038,7 @@ SEXP processx_exec(SEXP command, SEXP args, SEXP pty, SEXP pty_options, } } - result = PROTECT(processx__make_handle(private, ccleanup, ccleanup_grace)); + result = PROTECT(processx__make_handle(private, ccleanup)); handle = R_ExternalPtrAddr(result); application_path = processx__search_path(application, cwd, path); From a3ca45c23b9b39ea011ecd580014876b9aa41259 Mon Sep 17 00:00:00 2001 From: Lionel Henry Date: Tue, 2 May 2023 10:00:24 +0200 Subject: [PATCH 2/3] Implement session finalizer with parallel kill --- NEWS.md | 5 +++ R/finalize.R | 32 ++++++++++++++- src/Makevars.win | 2 +- src/init.c | 2 + src/unix/processx-unix.h | 1 + src/unix/processx.c | 75 ++++++++++++++++++++++++++++++++++- src/utils.c | 4 ++ tests/testthat/test-process.R | 50 +++++++++++++++++++++++ 8 files changed, 168 insertions(+), 3 deletions(-) diff --git a/NEWS.md b/NEWS.md index 0b903b6d..835283ce 100644 --- a/NEWS.md +++ b/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 diff --git a/R/finalize.R b/R/finalize.R index 7d1b34b1..c693de22 100644 --- a/R/finalize.R +++ b/R/finalize.R @@ -1,4 +1,9 @@ process_finalize <- function(private) { + ps <- process_cleanup_list(private) + ps::ps_kill_parallel(ps, private$cleanup_grace) +} + +process_cleanup_list <- function(private) { ps <- list() if (private$cleanup) { @@ -9,9 +14,34 @@ process_finalize <- function(private) { ) ps <- c(ps, list(handle)) } + if (private$cleanup_tree) { ps <- c(ps, ps::ps_find_tree(private$tree_id)) } - ps::ps_kill_parallel(ps, private$cleanup_grace) + ps +} + +session_finalize <- function(node) { + ps <- list() + grace <- 0 + + while (!node_is_root(node)) { + private <- wref_key(node_value(node)) + ps <- c(ps, process_cleanup_list(private)) + + if (!is.null(private$cleanup_grace)) { + grace <- max(grace, private$cleanup_grace) + } + + node <- node_next(node) + } + + ps::ps_kill_parallel(ps, grace) } + +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]] diff --git a/src/Makevars.win b/src/Makevars.win index 037ef13f..d11bcdf8 100644 --- a/src/Makevars.win +++ b/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 diff --git a/src/init.c b/src/init.c index 127bffbc..a6af6116 100644 --- a/src/init.c +++ b/src/init.c @@ -95,6 +95,7 @@ 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, @@ -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 }, diff --git a/src/unix/processx-unix.h b/src/unix/processx-unix.h index 5f2aa838..6a462c85 100644 --- a/src/unix/processx-unix.h +++ b/src/unix/processx-unix.h @@ -26,6 +26,7 @@ typedef struct processx_handle_s { 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 */ diff --git a/src/unix/processx.c b/src/unix/processx.c index e4fbd857..75f8e018 100644 --- a/src/unix/processx.c +++ b/src/unix/processx.c @@ -374,6 +374,27 @@ SEXP finalizer_call(SEXP private) { 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); @@ -390,6 +411,19 @@ void processx__finalizer(SEXP status) { SEXP 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 handler, to read out the exit code via waitpid, but no handle any more. */ @@ -403,6 +437,45 @@ void processx__finalizer(SEXP status) { } } +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; @@ -415,7 +488,7 @@ static SEXP processx__make_handle(SEXP private, int cleanup) { handle->r6_private = private; result = PROTECT(R_MakeExternalPtr(handle, private, R_NilValue)); - R_RegisterCFinalizerEx(result, processx__finalizer, 1); + processx__register_finalizer(result, handle); UNPROTECT(1); return result; diff --git a/src/utils.c b/src/utils.c index a765d986..bd3189b8 100644 --- a/src/utils.c +++ b/src/utils.c @@ -76,3 +76,7 @@ SEXP r_safe_eval(SEXP expr, SEXP env, SEXP *out) { struct safe_eval data = { .expr = expr, .env = env, .out = out }; return r_unwind_protect(&safe_eval_callback, &data); } + +SEXP processx__wref_key(SEXP x) { + return R_WeakRefKey(x); +} diff --git a/tests/testthat/test-process.R b/tests/testthat/test-process.R index 8bf7ab87..fab49425 100644 --- a/tests/testthat/test-process.R +++ b/tests/testthat/test-process.R @@ -441,3 +441,53 @@ test_that("can kill with SIGTERM when ignored", { Sys.sleep(0.05) expect_true(p$is_alive()) }) + +test_that("clean up in parallel", { + opts <- callr::r_process_options(extra = list( + cleanup_tree = TRUE, + cleanup_grace = 0.2 + )) + + fn <- function(load_sigtermignore, cleanup) { + sub_opts <- callr::r_process_options(extra = list( + cleanup = cleanup + )) + s <- callr::r_session$new(sub_opts) + + # Ensure grace delay kicks in + s$run(load_sigtermignore) + + # Make sure sessions are not closed by stdin EOF + s$call(function() Sys.sleep(60)) + + # Keep alive through options + sessions <- c(getOption("sessions"), list(s)) + options(sessions = sessions) + } + + # On GC + s <- callr::r_session$new(opts) + for (i in 1:5) s$run(fn, list(load_sigtermignore, cleanup = FALSE)) + + tree <- ps::ps_find_tree(s$.__enclos_env__$private$tree_id) + expect_true(all(sapply(tree, ps::ps_is_running))) + + rm(s) + gc() + + # Why is this needed? + Sys.sleep(0.05) + + expect_false(any(sapply(tree, ps::ps_is_running))) + + # On session quit + s <- callr::r_session$new(opts) + for (i in 1:5) s$run(fn, list(load_sigtermignore, cleanup = TRUE)) + + tree <- ps::ps_find_tree(s$.__enclos_env__$private$tree_id) + expect_true(all(sapply(tree, ps::ps_is_running))) + + s$run(function() q()) + + expect_false(any(sapply(tree, ps::ps_is_running))) +}) From 94be1edb8c271a611d539e0911a0d8e4759d769a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?G=C3=A1bor=20Cs=C3=A1rdi?= Date: Fri, 13 Nov 2026 15:38:27 +0100 Subject: [PATCH 3/3] Use ps::ps_kill() and kill process groups on cleanup ps::ps_kill_parallel() was never released, so cleanup on GC and on session quit failed. Use ps::ps_kill() for the process tree instead, with the grace period converted to milliseconds. Kill the process group of each process via processx's own handle, as the old C finalizer did, so cleanup = TRUE kills the whole group again and works without ps support. ps is only used for cleanup_tree = TRUE, and only if ps supports the platform. Errors are caught for each process, so one failure does not stop the others from being cleaned up. Only call into R from the C finalizer if cleanup = TRUE. Replace the parallel cleanup test, which used an undefined helper, with px-based tests for group kill, grace period and parallel kill on session quit. Update C line numbers in error snapshots. Co-Authored-By: Claude Opus 5.5 --- R/finalize.R | 105 +++++++++++++++----- R/initialize.R | 3 - R/process.R | 1 - src/unix/processx.c | 11 ++- tests/testthat/_snaps/Darwin/process.md | 4 +- tests/testthat/_snaps/Darwin/run.md | 2 +- tests/testthat/_snaps/Linux/process.md | 4 +- tests/testthat/_snaps/Linux/run.md | 2 +- tests/testthat/test-cleanup.R | 122 ++++++++++++++++++++++++ tests/testthat/test-process.R | 50 ---------- 10 files changed, 215 insertions(+), 89 deletions(-) diff --git a/R/finalize.R b/R/finalize.R index c693de22..7d2b5ddd 100644 --- a/R/finalize.R +++ b/R/finalize.R @@ -1,43 +1,96 @@ +# Called from the C finalizer of the process handle (Unix only), when +# the process object is garbage collected. process_finalize <- function(private) { - ps <- process_cleanup_list(private) - ps::ps_kill_parallel(ps, private$cleanup_grace) + cleanup_kill(list(private), private$cleanup_grace) } -process_cleanup_list <- function(private) { - ps <- list() - - if (private$cleanup) { - # Can't be created in advance because the ps finalizer might run first - handle <- ps::ps_handle( - private$pid, - format_unix_time(private$starttime_raw) - ) - ps <- c(ps, list(handle)) - } +# 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 - if (private$cleanup_tree) { - ps <- c(ps, ps::ps_find_tree(private$tree_id)) + 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) } - ps + cleanup_kill(privates, grace) } -session_finalize <- function(node) { - ps <- list() - grace <- 0 +# 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()) + } - while (!node_is_root(node)) { - private <- wref_key(node_value(node)) - ps <- c(ps, process_cleanup_list(private)) + 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 + } - if (!is.null(private$cleanup_grace)) { - grace <- max(grace, private$cleanup_grace) + # 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 + ) } + } - node <- node_next(node) + # 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 + ) + } } - ps::ps_kill_parallel(ps, grace) + 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) diff --git a/R/initialize.R b/R/initialize.R index 9f1dac76..c75ddb48 100644 --- a/R/initialize.R +++ b/R/initialize.R @@ -212,9 +212,6 @@ process_initialize <- function( } private$starttime <- max(private$starttime_raw, before_start) - # Needed for cleaning up - private$pid <- self$get_pid() - ## Need to close this, otherwise the child's end of the pipe ## will not be closed when the child exits, and then we cannot ## poll it. diff --git a/R/process.R b/R/process.R index ee5e9fef..762edbb4 100644 --- a/R/process.R +++ b/R/process.R @@ -795,7 +795,6 @@ process <- R6::R6Class( windows_hide_window = NULL, status = NULL, # C file handle - pid = NULL, # pid for cleanup supervised = FALSE, # Whether process is tracked by supervisor diff --git a/src/unix/processx.c b/src/unix/processx.c index 75f8e018..32bb8d1f 100644 --- a/src/unix/processx.c +++ b/src/unix/processx.c @@ -407,9 +407,13 @@ void processx__finalizer(SEXP status) { if (!handle) return; - SEXP call = PROTECT(finalizer_call(handle->r6_private)); - SEXP err = r_safe_eval(call, R_BaseEnv, NULL); - UNPROTECT(1); + /* Only call into R if there is something to clean up */ + SEXP err = NULL; + if (handle->cleanup) { + 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; @@ -486,6 +490,7 @@ static SEXP processx__make_handle(SEXP private, int cleanup) { handle->waitpipe[0] = handle->waitpipe[1] = -1; handle->r6_private = private; + handle->cleanup = cleanup; result = PROTECT(R_MakeExternalPtr(handle, private, R_NilValue)); processx__register_finalizer(result, handle); diff --git a/tests/testthat/_snaps/Darwin/process.md b/tests/testthat/_snaps/Darwin/process.md index 55153d60..68485d18 100644 --- a/tests/testthat/_snaps/Darwin/process.md +++ b/tests/testthat/_snaps/Darwin/process.md @@ -6,7 +6,7 @@ Error in `process_initialize()`: ! ! Native call to `processx_exec` failed Caused by error in `chain_call(...)` at initialize.R::: - ! cannot start processx process '/' (system error 2, No such file or directory) @unix/processx.c:650 (processx_exec) + ! cannot start processx process '/' (system error 2, No such file or directory) @unix/processx.c:738 (processx_exec) # working directory does not exist @@ -16,5 +16,5 @@ Error in `process_initialize()`: ! ! Native call to `processx_exec` failed Caused by error in `chain_call(...)` at initialize.R::: - ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:650 (processx_exec) + ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:738 (processx_exec) diff --git a/tests/testthat/_snaps/Darwin/run.md b/tests/testthat/_snaps/Darwin/run.md index d56145a8..b8aa8232 100644 --- a/tests/testthat/_snaps/Darwin/run.md +++ b/tests/testthat/_snaps/Darwin/run.md @@ -6,5 +6,5 @@ Error in `process_initialize()`: ! ! Native call to `processx_exec` failed Caused by error in `chain_call(...)` at initialize.R::: - ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:650 (processx_exec) + ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:738 (processx_exec) diff --git a/tests/testthat/_snaps/Linux/process.md b/tests/testthat/_snaps/Linux/process.md index 55153d60..68485d18 100644 --- a/tests/testthat/_snaps/Linux/process.md +++ b/tests/testthat/_snaps/Linux/process.md @@ -6,7 +6,7 @@ Error in `process_initialize()`: ! ! Native call to `processx_exec` failed Caused by error in `chain_call(...)` at initialize.R::: - ! cannot start processx process '/' (system error 2, No such file or directory) @unix/processx.c:650 (processx_exec) + ! cannot start processx process '/' (system error 2, No such file or directory) @unix/processx.c:738 (processx_exec) # working directory does not exist @@ -16,5 +16,5 @@ Error in `process_initialize()`: ! ! Native call to `processx_exec` failed Caused by error in `chain_call(...)` at initialize.R::: - ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:650 (processx_exec) + ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:738 (processx_exec) diff --git a/tests/testthat/_snaps/Linux/run.md b/tests/testthat/_snaps/Linux/run.md index d56145a8..b8aa8232 100644 --- a/tests/testthat/_snaps/Linux/run.md +++ b/tests/testthat/_snaps/Linux/run.md @@ -6,5 +6,5 @@ Error in `process_initialize()`: ! ! Native call to `processx_exec` failed Caused by error in `chain_call(...)` at initialize.R::: - ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:650 (processx_exec) + ! cannot start processx process '/px' (system error 2, No such file or directory) @unix/processx.c:738 (processx_exec) diff --git a/tests/testthat/test-cleanup.R b/tests/testthat/test-cleanup.R index 59682ffb..348eeaad 100644 --- a/tests/testthat/test-cleanup.R +++ b/tests/testthat/test-cleanup.R @@ -21,3 +21,125 @@ test_that("process can stay alive", { expect_true(process__exists(pid)) }) + +test_that("GC cleanup kills the process group", { + skip_on_os("windows") + px <- get_tool("px") + + # The background child is in the same process group as `sh`, but + # it is not in the process tree cleanup, because `cleanup_tree` is + # FALSE. So only killing the process group can clean it up. + p <- process$new( + "sh", + c("-c", paste(shQuote(px), "sleep 60 & echo $!; wait")), + stdout = "|" + ) + p$poll_io(5000) + child <- as.integer(p$read_output_lines()) + on.exit(tools::pskill(child, tools::SIGKILL), add = TRUE) + expect_true(process__exists(child)) + + rm(p) + gc() + + retry_until(function() !process__exists(child)) +}) + +test_that("GC cleanup waits for the grace period", { + skip_on_os("windows") + px <- get_tool("px") + + p <- process$new( + px, + c("sigterm", "ignore", "outln", "ready", "sleep", "60"), + stdout = "|", + cleanup_grace = 0.5 + ) + p$poll_io(5000) + expect_equal(p$read_output_lines(), "ready") + pid <- p$get_pid() + on.exit(tools::pskill(pid, tools::SIGKILL), add = TRUE) + + # Finalizers run right after gc() returns, so system.time(gc()) + # would not include them. + start <- Sys.time() + rm(p) + gc() + elapsed <- as.double(Sys.time() - start, units = "secs") + + # SIGTERM is ignored, so we need to wait the whole grace period + expect_gte(elapsed, 0.4) + retry_until(function() !process__exists(pid)) +}) + +test_that("GC cleanup does not wait if the process exits on SIGTERM", { + skip_on_os("windows") + px <- get_tool("px") + + p <- process$new(px, c("sleep", "60"), cleanup_grace = 5) + pid <- p$get_pid() + on.exit(tools::pskill(pid, tools::SIGKILL), add = TRUE) + + # Finalizers run right after gc() returns, so system.time(gc()) + # would not include them. + start <- Sys.time() + rm(p) + gc() + elapsed <- as.double(Sys.time() - start, units = "secs") + + expect_lt(elapsed, 2) + retry_until(function() !process__exists(pid)) +}) + +test_that("session quit kills processes in parallel", { + skip_on_os("windows") + skip_on_cran() + skip_if_not_installed("callr") + px <- get_tool("px") + + rp <- callr::r_bg( + function(px) { + ps <- lapply(1:3, function(i) { + processx::process$new( + px, + c("sigterm", "ignore", "outln", "ready", "sleep", "60"), + stdout = "|", + cleanup_grace = 1 + ) + }) + for (p in ps) p$poll_io(5000) + keep <- processx::process$new(px, c("sleep", "60"), cleanup = FALSE) + # Keep them alive until the session quits + assign(".processes", c(ps, list(keep)), envir = globalenv()) + cat(sapply(ps, function(p) p$get_pid()), sep = "\n") + cat(keep$get_pid(), "\n", sep = "") + invisible() + }, + list(px = px), + stdout = "|" + ) + on.exit(rp$kill(), add = TRUE) + + pids <- character() + deadline <- Sys.time() + 10 + while (length(pids) < 4 && Sys.time() < deadline) { + rp$poll_io(1000) + pids <- c(pids, rp$read_output_lines()) + } + pids <- as.integer(pids) + expect_length(pids, 4) + killed <- pids[1:3] + kept <- pids[4] + on.exit(tools::pskill(pids, tools::SIGKILL), add = TRUE) + + # The three processes ignore SIGTERM and have a 1s grace period. + # Killed one by one this would take 3s, in parallel about 1s. + start <- Sys.time() + rp$wait(10000) + elapsed <- as.double(Sys.time() - start, units = "secs") + expect_false(rp$is_alive()) + expect_lt(elapsed, 2.5) + + retry_until(function() !any(sapply(killed, process__exists))) + expect_true(process__exists(kept)) +}) diff --git a/tests/testthat/test-process.R b/tests/testthat/test-process.R index fab49425..8bf7ab87 100644 --- a/tests/testthat/test-process.R +++ b/tests/testthat/test-process.R @@ -441,53 +441,3 @@ test_that("can kill with SIGTERM when ignored", { Sys.sleep(0.05) expect_true(p$is_alive()) }) - -test_that("clean up in parallel", { - opts <- callr::r_process_options(extra = list( - cleanup_tree = TRUE, - cleanup_grace = 0.2 - )) - - fn <- function(load_sigtermignore, cleanup) { - sub_opts <- callr::r_process_options(extra = list( - cleanup = cleanup - )) - s <- callr::r_session$new(sub_opts) - - # Ensure grace delay kicks in - s$run(load_sigtermignore) - - # Make sure sessions are not closed by stdin EOF - s$call(function() Sys.sleep(60)) - - # Keep alive through options - sessions <- c(getOption("sessions"), list(s)) - options(sessions = sessions) - } - - # On GC - s <- callr::r_session$new(opts) - for (i in 1:5) s$run(fn, list(load_sigtermignore, cleanup = FALSE)) - - tree <- ps::ps_find_tree(s$.__enclos_env__$private$tree_id) - expect_true(all(sapply(tree, ps::ps_is_running))) - - rm(s) - gc() - - # Why is this needed? - Sys.sleep(0.05) - - expect_false(any(sapply(tree, ps::ps_is_running))) - - # On session quit - s <- callr::r_session$new(opts) - for (i in 1:5) s$run(fn, list(load_sigtermignore, cleanup = TRUE)) - - tree <- ps::ps_find_tree(s$.__enclos_env__$private$tree_id) - expect_true(all(sapply(tree, ps::ps_is_running))) - - s$run(function() q()) - - expect_false(any(sapply(tree, ps::ps_is_running))) -})