Add support for state 'submitted' as well; this will be useful for HPC schedulers, but also other types of backends with a queue

Author: HenrikBengtsson

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: future
-Version: 1.67.0-9011
+Version: 1.67.0-9012
 Title: Unified Parallel and Distributed Processing in R for Everyone
 Depends:
     R (>= 3.2.0)

R/backend_api-Future-class.R

@@ -1024,7 +1024,7 @@ getExpression.Future <- local({
 #' @export
 `$<-.Future` <- function(x, name, value) {
   if (name == "state") {
-    if (!is.element(value, c("created", "running", "finished", "failed", "canceled", "interrupted"))) {
+    if (!is.element(value, c("created", "submitted", "running", "finished", "failed", "canceled", "interrupted"))) {
       action <- getOption("future.state.onInvalid", "warning")
 
       if (action != "ignore") {

R/backend_api-UniprocessFuture-class.R

@@ -23,7 +23,7 @@ UniprocessFuture <- function(expr = NULL, substitute = TRUE, envir = parent.fram
 run.UniprocessFuture <- function(future, ...) {
   debug <- isTRUE(getOption("future.debug"))
 
-  if (future[["state"]] != 'created') {
+  if (future[["state"]] != "created") {
     label <- sQuoteLabel(future)
     stop(FutureError(sprintf("A future (%s) can only be launched once", label), future = future))
   }
@@ -33,10 +33,10 @@ run.UniprocessFuture <- function(future, ...) {
   assertOwner(future)
 
   ## Run future
-  future[["state"]] <- 'running'
+  future[["state"]] <- "running"
   data <- getFutureData(future, debug = debug)
   future[["result"]] <- evalFuture(data)
-  future[["state"]] <- 'finished'
+  future[["state"]] <- "finished"
 
   if (debug) mdebugf("%s started (and completed)", class(future)[1])
 

R/core_api-cancel.R

@@ -85,9 +85,9 @@ cancel.Future <- function(x, interrupt = TRUE, ...) {
     on.exit(mdebug_pop())
   }
 
-  ## Only running futures can be canceled, ignore everything else
-  if (future[["state"]] != "running") {
-    if (debug) mdebug("Skipping, because a non-running future")
+  ## Only submitted or running futures can be canceled, ignore everything else
+  if (!future[["state"]] %in% c("submitted", "running")) {
+    if (debug) mdebug("Skipping, because not a submitted or a running future")
     return(invisible(future))
   }
 

R/core_api-reset.R

@@ -65,13 +65,14 @@ reset.environment <- function(x, ...) {
 reset.Future <- function(x, ...) {
   future <- x
 
-  if (future[["state"]] == "running") {
+  if (future[["state"]] %in% c("submitted", "running")) {
+    msg <- sprintf("Cannot reset a %s future", future[["state"]])
     backend <- future[["backend"]]
     if (!inherits(backend, "FutureBackend")) {
-      warning(FutureWarning("Cannot reset a running future", future = future))
+      warning(FutureWarning(msg, future = future))
       return(future)
     }
-    stop(FutureError("Cannot reset a running future", future = future))
+    stop(FutureError(msg, future = future))
   }
 
   core_fields <- c(

R/core_api-value.R

@@ -659,7 +659,7 @@ value.list <- function(x, idxs = NULL, recursive = 0, reduce = NULL, stdout = TR
         ## so that future can be resolved in the asynchronously
         if (inherits(obj, "Future")) {
           ## Lazy future that is not yet launched?
-          if (obj[["state"]] == 'created') obj <- run(obj)
+          if (obj[["state"]] == "created") obj <- run(obj)
 
           if (!resolved(obj)) {
             next

R/protected_api-resolve.R

@@ -92,7 +92,7 @@ resolve.Future <- function(x, idxs = NULL, recursive = 0, result = FALSE, stdout
   result <- result || relay
 
   ## Lazy future that is not yet launched?
-  if (future[["state"]] == 'created') future <- run(future)
+  if (future[["state"]] == "created") future <- run(future)
 
   ## Poll for the Future to finish
   while (!resolved(future)) {
@@ -251,7 +251,7 @@ resolve.list <- function(x, idxs = NULL, recursive = 0, result = FALSE, stdout =
         ## so that future can be resolved in the asynchronously
         if (inherits(obj, "Future")) {
           ## Lazy future that is not yet launched?
-          if (obj[["state"]] == 'created') obj <- run(obj)
+          if (obj[["state"]] == "created") obj <- run(obj)
           if (!resolved(obj)) next
           if (debug) mdebugf("Future #%d", ii)
           if (result) {
@@ -397,7 +397,7 @@ resolve.environment <- function(x, idxs = NULL, recursive = 0, result = FALSE, s
         ## so that future can be resolved in the asynchronously
         if (inherits(obj, "Future")) {
           ## Lazy future that is not yet launched?
-          if (obj[["state"]] == 'created') obj <- run(obj)
+          if (obj[["state"]] == "created") obj <- run(obj)
           if (!resolved(obj)) next
           if (debug) mdebugf("Future #%d", ii)
           if (result) value(obj, stdout = FALSE, signal = FALSE)
@@ -543,7 +543,7 @@ resolve.listenv <- function(x, idxs = NULL, recursive = 0, result = FALSE, stdou
         ## so that future can be resolved in the asynchronously
         if (inherits(obj, "Future")) {
           ## Lazy future that is not yet launched?
-          if (obj[["state"]] == 'created') obj <- run(obj)
+          if (obj[["state"]] == "created") obj <- run(obj)
           if (!resolved(obj)) next
           if (debug) mdebugf("Future #%d", ii)
           if (result) value(obj, stdout = FALSE, signal = FALSE)