Handle crashed mirai workers

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future.mirai
-Version: 0.1.1-9015
+Version: 0.1.1-9016
 Depends:
     future
 Imports:

R/MiraiFuture-class.R

@@ -58,7 +58,7 @@ MiraiFuture <- function(expr = NULL,
       daemons(n = 0L)
     } else if (workers != nworkers) {
       daemons(n = 0L)  ## reset is required
-      daemons(n = workers, dispatcher = dispatcher)
+      daemons(n = workers, dispatcher = dispatcher, resilience = FALSE)
     }
   } else if (!is.null(workers)) {
     stop("Argument 'workers' should be a numeric scalar or NULL: ", mode(workers))
@@ -167,6 +167,14 @@ result.MiraiFuture <- function(future, ...) {
 
   mirai <- future[["mirai"]]
   result <- call_mirai_(mirai)$data
+
+  if (inherits(result, "errorValue")) {
+    label <- future$label
+    if (is.null(label)) label <- "<none>"
+    msg <- sprintf("Failed to retrieve results from %s (%s). The mirai framework reports on error value %s", class(future)[1], label, result)
+    stop(FutureError(msg))
+  }
+
   future[["result"]] <- result
   future[["state"]] <- "finished"
 
@@ -182,7 +190,8 @@ mirai_daemons_nworkers <- function() {
   if (is.data.frame(workers)) return(nrow(workers))
 
   if (length(workers) != 1L) {
-    stop(FutureError(sprintf("Length of mirai::status()$daemons is not one: %d", length(workers))))
+    msg <- sprintf("Length of mirai::status()$daemons is not one: %d", length(workers))
+    stop(FutureError(msg))
   }
 
   if (workers == 0L) return(Inf)

R/nbrOfWorkers.R

@@ -33,7 +33,10 @@ nbrOfWorkers.mirai <- function(evaluator) {
 nbrOfFreeWorkers.mirai <- function(evaluator, background = FALSE, ...) {
   res <- status()
   workers <- res[["daemons"]]
-  if (!is.numeric(workers)) {
+  if (is.character(workers)) {
+     workers <- res[["connections"]]
+     stopifnot(is.numeric(workers))
+  } else if (!is.numeric(workers)) {
     stop(FutureError(sprintf("Unknown type of mirai::daemons()$daemons: %s", typeof(workers))))
   }
 

tests/mirai_cluster,worker-termination.R.HIDE → tests/mirai_cluster,worker-termination.R

@@ -5,11 +5,12 @@ message("*** mirai_multisession() - terminating workers ...")
 plan(mirai_multisession, workers = 2L)
 
 all <- nbrOfWorkers()
+message("Number of workers: ", all)
+stopifnot(all == 2L)
 free <- nbrOfFreeWorkers()
-stopifnot(
-  nbrOfWorkers() == 2L,
-  nbrOfFreeWorkers() == 2L
-)
+message("Number of free workers: ", all)
+stopifnot(free == 2L)
+
 
 ## Force R worker to quit
 f <- future({ tools::pskill(pid = Sys.getpid()) })
@@ -18,8 +19,8 @@ print(res)
 stopifnot(inherits(res, "FutureError"))
 
 stopifnot(
-  nbrOfWorkers() == all,
-  nbrOfFreeWorkers() == free
+  nbrOfWorkers() == all - 1L,
+  nbrOfFreeWorkers() == free - 1L
 )
 
 message("*** mirai_multisession() - terminating workers ... DONE")