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.

Author: aitap

R/frollapply.R

@@ -411,7 +411,7 @@ 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))
+                lapply(jobs[.Call(Cis_direct_child, jobs)], function(pid) try(tools::pskill(pid), silent = TRUE))
                 parallel::mccollect(jobs)
               })
               # nocov end

src/data.table.h

@@ -322,6 +322,7 @@ bool perhapsDataTable(SEXP x);
 SEXP perhapsDataTableR(SEXP x);
 SEXP frev(SEXP x, SEXP copyArg);
 NORET void internal_error(const char *call_name, const char *format, ...);
+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},
 {"Csetgrowable", (DL_FUNC)&setgrowable, -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) {
@@ -659,3 +663,21 @@ void internal_error(const char *call_name, const char *format, ...) {
 
   error("%s %s: %s. %s", _("Internal error in"), call_name, buff, _("Please report to the data.table issues tracker."));
 }
+
+#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);
+  R_xlen_t 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
+}