Merge branch 'release/0.10.1'

Author: HenrikBengtsson

.github/workflows/R-CMD-check.yaml

@@ -19,10 +19,10 @@ jobs:
           - {os: windows-latest, r: 'devel'    }
           - {os: windows-latest, r: 'release'  }
           - {os: windows-latest, r: 'oldrel'   }
-#          - {os: macOS-latest,   r: 'devel'    }
+          - {os: macOS-latest,   r: 'devel'    }
           - {os: macOS-latest,   r: 'release'  }
-#          - {os: macOS-latest,   r: 'oldrel'   }
-#          - {os: ubuntu-latest,  r: 'devel'    }
+          - {os: macOS-latest,   r: 'oldrel'   }
+          - {os: ubuntu-latest,  r: 'devel'    }
           - {os: ubuntu-latest,  r: 'release'  }
           - {os: ubuntu-latest,  r: 'oldrel'   }
           - {os: ubuntu-latest,  r: 'oldrel-1' }
@@ -34,6 +34,7 @@ jobs:
           - {os: ubuntu-latest,  r: 'release'  , language: ko,    label: ko    }
           - {os: ubuntu-latest,  r: 'release'  , language: zh_CN, label: zh_CN }
           - {os: ubuntu-latest,  r: 'release'  , language: zh_TW, label: zh_TW }
+          - {os: windows-latest, r: 'devel', future_version: develop, label: 'w/ future-develop' }
 
     env:
       GITHUB_PAT: ${{ secrets.GITHUB_TOKEN }}
@@ -53,6 +54,7 @@ jobs:
       ## Specific to futures
       R_PARALLELLY_AVAILABLECORES_SYSTEM: ${{ matrix.config.availablecores }}
       R_FUTURE_RNG_ONMISUSE: error
+      R_FUTURE_VERSION: ${{ matrix.config.future_version }}
 
     steps:
       - uses: actions/checkout@v4
@@ -77,6 +79,15 @@ jobs:
           install.packages(".", repos = NULL, type = "source")  ## needed by parallel workers
         shell: Rscript {0}
 
+      - name: Test with specific future version?
+        run: |
+          future_version <- Sys.getenv("R_FUTURE_VERSION")
+          if (nzchar(future_version)) {
+            install.packages("remotes")
+            remotes::install_github("futureverse/future", ref=future_version)
+          }
+        shell: Rscript {0}
+        
       - name: Session info
         run: |
           options(width = 100)

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future.mirai
-Version: 0.10.0
+Version: 0.10.1
 Depends:
     future (>= 1.49.0)
 Imports:

NEWS.md

@@ -1,3 +1,23 @@
+# Version 0.10.1
+
+## Bug Fix
+
+ * If a mirai future that was terminated abruptly (e.g. via
+   `tools::pskill()` or by the operating system), then it was not
+   detected as such. Instead it resulted in an unexpected error that
+   could not be recovered from. Now it is detected and a
+   `FutureInterruptError` is signaled, which can then be handled and
+   the future may be `reset()`.
+
+ * `result()` on an interrupted mirai future would only throw
+   FutureInterruptError the first time. Succeeding calls would result
+   in other errors.
+ 
+ * `resolved()` on a mirai future already known to be interrupted
+   would requery the mirai object, instead of returning TRUE
+   immediately.
+ 
+
 # Version 0.10.0
 
 ## New Features

R/MiraiFutureBackend-class.R

@@ -92,6 +92,8 @@ launchFuture.MiraiFutureBackend <- local({
     reg <- backend[["reg"]]
     FutureRegistry(reg, action = "add", future = future, earlySignal = FALSE)
 
+    stopifnot(inherits(future[["mirai"]], "mirai"))
+
     invisible(future)
   }
 })
@@ -238,24 +240,27 @@ resolved.MiraiFuture <- function(x, ...) {
 
   resolved <- NextMethod()
   if(resolved) {
-    if (debug) mdebug("already resolved", debug = debug)
+    if (debug) mdebug("already resolved")
     return(TRUE)
   }
 
-  if(x[["state"]] == "finished") {
-    if (debug) mdebug("already resolved (state == finished)", debug = debug)
+  if(x[["state"]] %in% c("finished", "interrupted")) {
+    if (debug) mdebugf("already resolved (state == %s)", sQuote(x[["state"]]))
+    stopifnot(inherits(x[["mirai"]], "mirai"))
     return(TRUE)
   } else if(x[["state"]] == "created") { # Not yet submitted to queue (iff lazy)
     if (debug) mdebug("just created; launching")
     x <- run(x)
+    stopifnot(inherits(x[["mirai"]], "mirai"))
     return(FALSE)
   }
 
   if (debug) mdebug_push("mirai::unresolved() ...")
   mirai <- x[["mirai"]]
+  stopifnot(inherits(mirai, "mirai"))
   res <- unresolved(mirai)
   if (debug) {
-    mstr(res, debug = debug)
+    mstr(res)
     mdebug_pop()
   }
 
@@ -266,16 +271,21 @@ resolved.MiraiFuture <- function(x, ...) {
 #' @importFrom future result FutureInterruptError
 #' @export
 result.MiraiFuture <- function(future, ...) {
-  if(isTRUE(future[["state"]] == "finished")) {
-    return(future[["result"]])
-  }
-
   debug <- isTRUE(getOption("future.mirai.debug"))
   if (debug) {
     mdebugf_push("result() for %s ...", class(future)[1])
+    mdebugf("Future state: %s", sQuote(future[["state"]]))
     on.exit(mdebug_pop())
   }
 
+
+  state <- future[["state"]]
+  if (state == "finished") {
+    return(future[["result"]])
+  } else if (state == "interrupted") {
+    stop(future[["result"]])
+  }
+
   backend <- future[["backend"]]
   reg <- backend[["reg"]]
 
@@ -286,22 +296,34 @@ result.MiraiFuture <- function(future, ...) {
     dt <- dt[dt > 0]
     dt_str <- paste(sprintf("%s=%gs", names(dt), dt), collapse = ", ")
     mdebugf("collected mirai in %s", dt_str)
+    mdebugf("mirai result class: %s", class(result)[1])
   }
 
   if (inherits(result, "errorValue")) {
-    label <- sQuoteLabel(future[["label"]])
+    if (debug) mdebugf("mirai error value: %s", result)
+    
+    label <- sQuoteLabel(future)
 
-    if (result == 20L) {
-      state <- future[["state"]]
+    if (result %in% c(19L, 20L)) {
       stop_if_not(state %in% c("canceled", "interrupted", "running"))
-    
-      event <- if (state %in% "running") {
-        event <- sprintf("failed for unknown reason while %s", state)
-        future[["state"]] <- "interrupted"
-      } else {
-        event <- sprintf("was %s", state)
+      
+      if (result == 19L) {
+        ## "If a daemon crashes or terminates unexpectedly during evaluation,
+        ##  an 'errorValue' 19 (Connection reset) is returned."
+        ## Source: help("collect_mirai", package = "mirai")
+        event <- sprintf("was terminated while %s", state)
+      } else if (result == 20L) {
+        ## "Stops a 'mirai' if still in progress, causing it to resolve
+        ##  immediately to an 'errorValue' 20 (Operation canceled)."
+        ## Source: help("stop_mirai", package = "mirai")
+        event <- if (state %in% "running") {
+          event <- sprintf("failed for unknown reason while %s", state)
+        } else {
+          event <- sprintf("was %s", state)
+        }
       }
 
+      future[["state"]] <- "interrupted"
       msg <- sprintf("Future (%s) of class %s %s, while running on localhost (error code %d)", label, class(future)[1], event, result)
       if (debug) mdebug(msg)
       result <- FutureInterruptError(msg, future = future)
@@ -353,6 +375,7 @@ mirai_version <- local({
 #' @importFrom mirai call_mirai
 mirai_collect_future <- function(future) {
   mirai <- future[["mirai"]]
+  stopifnot(inherits(mirai, "mirai"))
   call_mirai(mirai)$data
 }
 
@@ -362,6 +385,7 @@ mirai_collect_future <- function(future) {
 #' @export
 interruptFuture.MiraiFutureBackend <- function(backend, future, ...) {
   mirai <- future[["mirai"]]
+  stopifnot(inherits(mirai, "mirai"))
   stop_mirai(mirai)
   future[["state"]] <- "interrupted"
   future

R/utils-debug.R

@@ -16,63 +16,58 @@ debug_indent <- local({
 if (!exists(".debug", inherits = FALSE)) .debug <- new.env(parent = emptyenv())
 if (!"stack" %in% names(".debug")) .debug$stack <- list()
 
-mdebug_push <- function(..., debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
-  msg <- mdebug(..., debug = debug)
+mdebug_push <- function(...) {
+  msg <- mdebug(...)
   .debug$stack <- c(.debug$stack, msg)
   invisible(msg)
 }
 
-mdebugf_push <- function(..., debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
-  msg <- mdebugf(..., debug = debug)
+mdebugf_push <- function(...) {
+  msg <- mdebugf(...)
   .debug$stack <- c(.debug$stack, msg)
   invisible(msg)
 }
 
-mdebug_pop <- function(..., debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
+mdebug_pop <- function(...) {
   n <- length(.debug$stack)
-  msg <- .debug$stack[n]
+  msg <- c(...)
+  if (length(msg) == 0) {
+    msg <- .debug$stack[n]
+    msg <- sprintf("%s done", msg)
+  }
   .debug$stack <- .debug$stack[-n]
-  mdebug(sprintf("%s done", msg), debug = debug)
+  if (length(msg) == 0 || !is.na(msg)) mdebug(msg)
 }
 
-mdebugf_pop <- function(..., debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
+mdebugf_pop <- function(...) {
   n <- length(.debug$stack)
   msg <- .debug$stack[n]
   .debug$stack <- .debug$stack[-n]
-  mdebug(sprintf("%s done", msg), debug = debug)
+  mdebug(sprintf("%s done", msg))
 }
 
-mdebug <- function(..., prefix = now(), debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
+mdebug <- function(..., prefix = now()) {
   prefix <- paste(prefix, debug_indent(), sep = "")
   msg <- paste(..., sep = "")
   message(sprintf("%s%s", prefix, msg))
   invisible(msg)
 }
 
-mdebugf <- function(..., appendLF = TRUE,
-                    prefix = now(), debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
+mdebugf <- function(..., appendLF = TRUE, prefix = now()) {
   prefix <- paste(prefix, debug_indent(), sep = "")
   msg <- sprintf(...)
   message(sprintf("%s%s", prefix, msg), appendLF = appendLF)
   invisible(msg)
 }
 
 #' @importFrom utils capture.output
-mprint <- function(..., appendLF = TRUE, prefix = now(), debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
+mprint <- function(..., appendLF = TRUE, prefix = now()) {
   prefix <- paste(prefix, debug_indent(), sep = "")
   message(paste(prefix, capture.output(print(...)), sep = "", collapse = "\n"), appendLF = appendLF)
 }
 
 #' @importFrom utils capture.output str
-mstr <- function(..., appendLF = TRUE, prefix = now(), debug = isTRUE(getOption("future.debug"))) {
-  if (!debug) return()
+mstr <- function(..., appendLF = TRUE, prefix = now()) {
   prefix <- paste(prefix, debug_indent(), sep = "")
   message(paste(prefix, capture.output(str(...)), sep = "", collapse = "\n"), appendLF = appendLF)
 }

R/zzz.R

@@ -21,13 +21,21 @@ with_assert <- function(expr, ...) { invisible(expr) }
   }
 
   ## Set 'debug' option by environment variable
-  value <- Sys.getenv("R_FUTURE_MIRAI_DEBUG", "FALSE")
-  value <- isTRUE(suppressWarnings(as.logical(value)))
-  options(future.mirai.debug = value)
+  value <- getOption("future.mirai.debug")
+  if (is.null(value)) {
+    value <- Sys.getenv("R_FUTURE_MIRAI_DEBUG", NA_character_)
+    if (!is.na(value)) {
+      value <- isTRUE(suppressWarnings(as.logical(value)))
+      options(future.mirai.debug = value)
+    }
+  }
 
   ## Set 'queue' option by environment variable
-  value <- Sys.getenv("R_FUTURE_MIRAI_QUEUE", NA_character_)
-  if (!is.na(value)) {
-    options(future.mirai.queue = value)
+  value <- getOption("future.mirai.queue")
+  if (is.null(value)) {
+    value <- Sys.getenv("R_FUTURE_MIRAI_QUEUE", NA_character_)
+    if (!is.na(value)) {
+      options(future.mirai.queue = value)
+    }
   }
 }