Interruption problems in frollapply (#7428)

* frollapply: let the interrupt continue

Instead of calling invokeRestart("abort") in the interrupt handler,
return from it. This continues the dispatch of the interrupt and lets an
outer handler catch it:

tryCatch(
  frollapply(1:1e6, 1, \(.) { Sys.sleep(ret <- sum(.)); ret}),
  interrupt = \(e) 'interrupted'
)
^C[1] "interrupted"

With invokeRestart("abort"), the interrupt cannot be handled further.

* frollapply: wait for terminated parallel processes

While handling an interrupt, ask mccollect() to wait for the child
process to exit (with a warning) in order to avoid producing zombies.
Otherwise a process that is too slow to react to SIGTERM will remain a
zombie until the parent process exits.

* Use waitid() to choose processes to terminate

Since the PIDs of our worker processes could have been reused, first
test them using waitid(NOWAIT) to make sure they are still ours.

* Interruption handling is not covered

Author: aitap

R/frollapply.R

@@ -395,11 +395,11 @@ frollapply = function(X, N, FUN, ..., by.column=TRUE, fill=NA, align=c("right","
             interrupt = function(e) {
               # nocov start
               suspendInterrupts({
-                lapply(jobs, function(pid) try(tools::pskill(pid), silent = TRUE))
-                parallel::mccollect(jobs, wait = FALSE)
+                lapply(jobs[.Call(Cis_direct_child, jobs)], function(pid) try(tools::pskill(pid), silent = TRUE))
+                parallel::mccollect(jobs)
               })
-              invokeRestart("abort") ## raise SIGINT
               # nocov end
+              # Let the interrupt continue without invoking restarts
             }
           )
           ## check for any errors in FUN, warnings are silently ignored

src/data.table.h

@@ -343,6 +343,7 @@ SEXP R_allocResizableVector_(SEXPTYPE type, R_xlen_t maxlen);
 SEXP R_duplicateAsResizable_(SEXP x);
 void R_resizeVector_(SEXP x, R_xlen_t newlen);
 #endif
+SEXP is_direct_child(SEXP pids);
 
 // types.c
 char *end(char *start);

src/init.c

@@ -160,6 +160,7 @@ R_CallMethodDef callMethods[] = {
 {"CmemcpyDTadaptive", (DL_FUNC)&memcpyDTadaptive, -1},
 {"CcopyAsGrowable", (DL_FUNC)&copyAsGrowable, -1},
 {"Cfrolladapt", (DL_FUNC)&frolladapt, -1},
+{"Cis_direct_child", (DL_FUNC)&is_direct_child, -1},
 {NULL, NULL, 0}
 };
 

src/utils.c

@@ -1,3 +1,7 @@
+#ifndef _WIN32
+#  include <sys/wait.h>
+#endif
+
 #include "data.table.h"
 
 bool within_int32_repres(double x) {
@@ -672,3 +676,23 @@ void R_resizeVector_(SEXP x, R_xlen_t newlen) {
   SETLENGTH(x, newlen);
 }
 #endif
+
+// # nocov start
+#ifdef _WIN32
+NORET
+#endif
+SEXP is_direct_child(SEXP pids) {
+#ifdef _WIN32
+  internal_error(__func__, "not implemented on Windows");
+#else
+  int *ppids = INTEGER(pids);
+  R_xlen_t len = xlength(pids);
+  SEXP ret = allocVector(LGLSXP, len);
+  int *pret = LOGICAL(ret);
+  siginfo_t info;
+  for (R_xlen_t i = 0; i < len; ++i)
+    pret[i] = waitid(P_PID, ppids[i], &info, WCONTINUED | WEXITED | WNOHANG | WNOWAIT | WSTOPPED) == 0;
+  return ret;
+#endif
+}
+// # nocov end