Improve parallel teardown

Close the input connection first, and then give the workers
some time to quit. This helps covr to write out the coverage
counters.

Author: gaborcsardi

R/parallel.R

@@ -314,12 +314,13 @@ queue_teardown <- function(queue) {
   num <- nrow(tasks)
 
   # calling quit() here creates a race condition, and the output of
-  # the deferred_run() might be lost.
+  # the deferred_run() might be lost. Instead we close the input
+  # connection in a separate task.
   clean_fn <- function() {
     withr::deferred_run(.GlobalEnv)
   }
 
-  topoll <- list()
+  topoll <- integer()
   for (i in seq_len(num)) {
     if (
       !is.null(tasks$worker[[i]]) && tasks$worker[[i]]$get_state() == "idle"
@@ -329,38 +330,57 @@ queue_teardown <- function(queue) {
       tryCatch(
         {
           tasks$worker[[i]]$call(clean_fn)
-          topoll <- c(topoll, tasks$worker[[i]])
+          topoll <- c(topoll, i)
         },
-        error = function(e) tasks$worker[i] <- list(NULL)
+        error = function(e) NULL
       )
     }
   }
 
-  # Give covr time to write out the coverage files
+  # Give covr a bit more time
   if (in_covr()) {
     grace <- 30L
   } else {
-    grace <- 3L
+    grace <- 1L
   }
   first_error <- NULL
   limit <- Sys.time() + grace
   while (length(topoll) > 0 && (timeout <- limit - Sys.time()) > 0) {
     timeout <- as.double(timeout, units = "secs") * 1000
-    conns <- lapply(topoll, function(x) x$get_poll_connection())
+    conns <- lapply(tasks$worker[topoll], function(x) x$get_poll_connection())
     pr <- unlist(processx::poll(conns, as.integer(timeout)))
     for (i in which(pr == "ready")) {
-      msg <- topoll[[i]]$read()
+      msg <- tasks$worker[[topoll[i]]]$read()
       first_error <- first_error %||% msg$error
     }
     topoll <- topoll[pr != "ready"]
   }
 
+  topoll <- integer()
   for (i in seq_len(num)) {
-    if (!is.null(tasks$worker[[i]])) {
+    if (
+      !is.null(tasks$worker[[i]]) && tasks$worker[[i]]$get_state() == "idle"
+    ) {
       tryCatch(
-        close(tasks$worker[[i]]$get_input_connection()),
+        {
+          close(tasks$worker[[i]]$get_input_connection())
+          topoll <- c(topoll, i)
+        },
         error = function(e) NULL
       )
+    }
+  }
+
+  limit <- Sys.time() + grace
+  while (length(topoll) > 0 && (timeout <- limit - Sys.time()) > 0) {
+    timeout <- as.double(timeout, units = "secs") * 1000
+    conns <- lapply(tasks$worker[topoll], function(x) x$get_poll_connection())
+    pr <- unlist(processx::poll(conns, as.integer(timeout)))
+    topoll <- topoll[pr != "ready"]
+  }
+
+  for (i in seq_len(num)) {
+    if (!is.null(tasks$worker[[i]])) {
       if (ps::ps_is_supported()) {
         tryCatch(tasks$worker[[i]]$kill_tree(), error = function(e) NULL)
       } else {