First prototype of a SequentialFutureBackend class

Author: HenrikBengtsson

DESCRIPTION

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

NAMESPACE

@@ -21,6 +21,7 @@ S3method(journal,Future)
 S3method(journal,FutureJournal)
 S3method(journal,FutureJournalCondition)
 S3method(journal,list)
+S3method(launchFuture,SequentialFutureBackend)
 S3method(mandelbrot,matrix)
 S3method(mandelbrot,numeric)
 S3method(nbrOfFreeWorkers,"NULL")

R/backend_api-Future-class.R

@@ -427,6 +427,34 @@ run.Future <- function(future, ...) {
   makeFuture <- plan("next")
   if (debug) mdebug("- Future backend: ", commaq(class(makeFuture)))
 
+  ## Use new FutureBackend approach?
+  if (getOption("future.backend.version", 1L) == 2L) {
+    ## Implements a FutureBackend?
+    backend <- attr(makeFuture, "backend")
+    if (is.function(backend)) {
+      if (debug) mdebug("Using FutureBackend ...")
+      mdebug("- state: ", sQuote(future[["state"]]))
+      on.exit(mdebug("run() for ", sQuote(class(future)[1]), " ... done"), add = TRUE)
+
+      ## Apply future plan tweaks
+      args <- attr(makeFuture, "tweaks")
+      if (is.null(args)) args <- list()
+      backend <- do.call(backend, args = args)
+      stop_if_not(inherits(backend, "FutureBackend"))
+
+
+      if (debug) mdebug(" - Launching futures ...")
+      future2 <- launchFuture(backend, future = future)
+      if (debug) mdebug(" - Launching futures ... done")
+      if (debug) mdebug(" - Future launched: ", commaq(class(future2)))
+      stop_if_not(inherits(future2, "Future"))
+      if (debug) mdebug("Using FutureBackend ... DONE")
+      
+      return(future2)
+    }
+  }
+
+
   ## AD HOC/WORKAROUND: /HB 2020-12-21
   args <- list(
     quote(future[["expr"]]),

R/backend_api-UniprocessFuture-class.R

@@ -22,7 +22,7 @@ UniprocessFuture <- function(expr = NULL, substitute = TRUE, envir = parent.fram
 #' @export
 run.UniprocessFuture <- function(future, ...) {
   debug <- isTRUE(getOption("future.debug"))
-  
+   
   if (future[["state"]] != 'created') {
     label <- future[["label"]]
     if (is.null(label)) label <- "<none>"
@@ -105,3 +105,74 @@ SequentialFuture <- function(expr = NULL, envir = parent.frame(), substitute = T
   f <- UniprocessFuture(expr = expr, envir = envir, substitute = FALSE, lazy = lazy, globals = globals, ...)
   structure(f, class = c("SequentialFuture", class(f)))
 }
+
+
+
+launchFuture <- function(backend, future, ...) {
+  UseMethod("launchFuture")
+}
+
+#' @export
+launchFuture.SequentialFutureBackend <- function(backend, future, ...) {
+  debug <- isTRUE(getOption("future.debug"))
+  if (debug) {
+    mdebugf("launchFuture() for %s ...", commaq(class(backend)))
+    on.exit(mdebugf("launchFuture() for %s ... DONE", commaq(class(backend))))
+  }
+
+  if (future[["state"]] != 'created') {
+    label <- future[["label"]]
+    if (is.null(label)) label <- "<none>"
+    stop(FutureError(sprintf("A future ('%s') can only be launched once", label), future = future))
+  }
+
+  ## Assert that the process that created the future is
+  ## also the one that evaluates/resolves/queries it.
+  assertOwner(future)
+
+  ## Coerce to a SequentialFuture
+  ## NOTE: Has to be done before getFutureData() is called
+  class(future) <- c("SequentialFuture", "UniprocessFuture", class(future))
+
+  ## Launch future
+  future[["state"]] <- "running"
+
+  ## Get future
+  data <- getFutureData(future, debug = debug)
+
+  ## Apply backend tweaks
+  split <- backend[["split"]]
+  if (!is.null(split)) data$capture$split <- split
+  earlySignal <- backend[["earlySignal"]]
+  if (!is.null(earlySignal)) future$earlySignal <- earlySignal
+
+  future[["result"]] <- evalFuture(data)
+  
+  future[["state"]] <- "finished"
+
+  if (debug) mdebugf("%s started (and completed)", class(future)[1])
+
+  ## Always signal immediateCondition:s and as soon as possible.
+  ## They will always be signaled if they exist.
+  signalImmediateConditions(future)
+
+  ## Signal conditions early, iff specified for the given future
+  signalEarly(future, collect = FALSE)
+
+  future
+}
+
+SequentialFutureBackend <- function(...) {
+  core <- new.env(parent = emptyenv())
+
+  ## Record future plan tweaks, if any
+  args <- list(...)
+  for (name in names(args)) {
+    core[[name]] <- args[[name]]
+  }
+  
+  core <- structure(core, class = c("SequentialFutureBackend", "FutureBackend", class(core)))
+    
+  core
+}
+

R/utils-options.R

@@ -362,4 +362,7 @@ update_package_options <- function(debug = FALSE) {
   ## future 1.34.0:
   update_package_option("future.globals.objectSize.method", mode = "character", debug = debug)
   update_package_option("future.plan.cleanup.legacy", mode = "logical", debug = debug)
+
+  ## future (>= 1.34.0-9000):
+  update_package_option("future.backend.version", mode = "integer", default = 1L, debug = debug)
 }

R/utils_api-tweak.R

@@ -134,6 +134,9 @@ tweak.future <- function(strategy, ..., penvir = parent.frame()) {
   ## Restore attributes including class
   attributes(strategy2) <- attrs
 
+  ## Record what tweaks were made
+  attr(strategy2, "tweaks") <- args
+  
   ## Flag that it is tweaked
   class(strategy2) <- c("tweaked", class)
 

R/zzz.R

@@ -1,5 +1,8 @@
 .package <- new.env()
 
+## Update FutureBackend:s
+attr(sequential, "backend") <- SequentialFutureBackend
+
 
 ## covr: skip=all
 #' @importFrom utils packageVersion