Interrupted futures are now reset to vanilla, lazy futures + all launched futures has a reference to their FutureBackend

Author: HenrikBengtsson

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future
-Version: 1.34.0-9302
+Version: 1.34.0-9303
 Title: Unified Parallel and Distributed Processing in R for Everyone
 Imports:
     digest,

NAMESPACE

@@ -53,6 +53,7 @@ S3method(print,FutureStrategy)
 S3method(print,FutureStrategyList)
 S3method(print,future)
 S3method(print,sessionDetails)
+S3method(resetFuture,FutureBackend)
 S3method(resetWorkers,default)
 S3method(resetWorkers,multicore)
 S3method(resolve,Future)
@@ -149,6 +150,7 @@ export(multisession)
 export(nbrOfFreeWorkers)
 export(nbrOfWorkers)
 export(plan)
+export(resetFuture)
 export(resetWorkers)
 export(resolve)
 export(resolved)

NEWS.md

@@ -20,6 +20,10 @@
 
  * Interrupted futures are now handled and produce an informative warning.
 
+ * Interrupted futures are reset of vanilla, lazy futures. This way
+   interrupted futures can be re-evaluated again, for instance, on
+   another future backend.
+
 ## Bug Fixes
 
  * Now 'interrupt' conditions are captured during the evaluation of

R/backend_api-Future-class.R

@@ -444,6 +444,7 @@ run.Future <- function(future, ...) {
     class(future) <- backend[["futureClasses"]]
 
     if (debug) mdebug(" - Launching futures ...")
+    future[["backend"]] <- backend
     future2 <- launchFuture(backend, future = future)
     if (debug) mdebug(" - Launching futures ... done")
     if (debug) mdebug(" - Future launched: ", commaq(class(future2)))

R/backend_api-FutureBackend-class.R

@@ -47,6 +47,49 @@ launchFuture.FutureBackend <- function(backend, future, ...) {
 }
 
 
+#' @rdname FutureBackend
+#' @export
+resetFuture <- function(backend, future, ...) {
+  UseMethod("resetFuture")
+}
+
+
+#' @export
+resetFuture.FutureBackend <- function(backend, future, ...) {
+  core_fields <- c(
+    "version",
+    "expr",
+    "envir",
+    "stdout",
+    "conditions",
+    "globals",
+    "packages",
+    "seed",
+    "lazy",
+    "asynchronous",
+    "local",
+    "reset",
+    "label",
+    "earlySignal",
+    "gc",
+    "onReference",
+    "calls",
+    "state"
+  )
+  class(future) <- "Future"
+  
+  drop <- setdiff(names(future), core_fields)
+  for (name in drop) {
+    future[[name]] <- NULL
+  }
+
+  future[["state"]] <- "created"
+  future[["lazy"]] <- TRUE
+
+  future
+}
+
+
 makeFutureBackend <- function(evaluator, ...) {
   backend <- attr(evaluator, "backend")
 

R/backend_api-SequentialFutureBackend-class.R

@@ -72,6 +72,8 @@ launchFuture.SequentialFutureBackend <- function(backend, future, ...) {
   split <- backend[["split"]]
   if (!is.null(split)) data[["capture"]][["split"]] <- split
 
+  future[["backend"]] <- backend
+
   ## Inherit 'earlySignal' from backend?
   earlySignal <- backend[["earlySignal"]]
   if (!is.null(earlySignal)) future[["earlySignal"]] <- earlySignal

R/protected_api-signalConditions.R

@@ -102,7 +102,9 @@ signalConditions <- function(future, include = "condition", exclude = NULL, resi
       source <- attr(session_uuid, "source")
       host <- source[["host"]]
       pid <- source[["pid"]]
-      msg <- sprintf("A future ('%s') of class %s was interrupted at %s, while running on %s (pid %s)", label, class(future)[1], format(when, format = "%FT%T"), sQuote(host), pid)
+      msg <- sprintf("A future ('%s') of class %s was interrupted at %s, while running on %s (pid %s). The future was reset to a vanilla, lazy future, which can be re-evaluted again", label, class(future)[1], format(when, format = "%FT%T"), sQuote(host), pid)
+      backend <- future[["backend"]]
+      if (!is.null(backend)) future <- resetFuture(backend, future)
       warning(FutureInterruptWarning(msg, future = future))
     } else if (inherits(condition, "warning")) {
       warning(condition)