First prototype of ClusterFutureBackend and MultisessionFutureBackend classes

Author: HenrikBengtsson

NAMESPACE

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

R/backend_api-ClusterFuture-class.R

@@ -72,7 +72,7 @@ as_ClusterFuture <- function(future, workers = NULL, ...) {
   ## futures' values.
   ##  workers <- add_cluster_session_info(workers)
 
-  backend <- ClusterFutureBackend(workers, persistent =isTRUE(future[["persistent"]]))
+  backend <- ClusterFutureBackend0(workers, persistent = isTRUE(future[["persistent"]]))
 
   future[["backend"]] <- backend
 
@@ -87,11 +87,11 @@ as_ClusterFuture <- function(future, workers = NULL, ...) {
 
 #' @export
 run.ClusterFuture <- function(future, ...) {
-  debug <- isTRUE(getOption("future.debug"))
-  if (debug) {
-    mdebug("run.ClusterFuture() ...")
-    on.exit(mdebug("run.ClusterFuture() ... done"))
+  if (getOption("future.backend.version", 1L) == 2L) {
+    return(NextMethod())
   }
+
+  debug <- getOption("future.debug", FALSE)
 
   if (future[["state"]] != 'created') {
     label <- future[["label"]]
@@ -103,31 +103,84 @@ run.ClusterFuture <- function(future, ...) {
   ## also the one that evaluates/resolves/queries it.
   assertOwner(future)
 
-  backend <- future[["backend"]]
+  workers <- future[["workers"]]
+  data <- getFutureData(future)
+  persistent <- isTRUE(future[["persistent"]])
+
+  ## FutureRegistry to use
+  reg <- sprintf("workers-%s", attr(workers, "name", exact = TRUE))
 
   ## Next available cluster node
   t_start <- Sys.time()
+  node_idx <- requestNode(await = function() {
+    FutureRegistry(reg, action = "collect-first", earlySignal = TRUE)
+  }, workers = workers)
+  future[["node"]] <- node_idx
 
-  ## (1) Get a free worker. This will block until one is available
-  backend$requestWorker(future)
-
-  ## (2) Attach packages that needs to be attached
-  ##     NOTE: Already take care of by evalFuture().
-  ##     However, if we need to get an early error about missing packages,
-  ##     we can get the error here before launching the future.
-  if (future[["earlySignal"]]) {
-    backend$requirePackages(future = future)
+  ## Cluster node to use
+  cl <- workers[node_idx]
+  
+  if (inherits(future[[".journal"]], "FutureJournal")) {
+    appendToFutureJournal(future,
+         event = "getWorker",
+      category = "overhead",
+        parent = "launch",
+         start = t_start,
+          stop = Sys.time()
+    )
   }
 
-  ## (2) Reset global environment of cluster node such that
+
+  ## (i) Reset global environment of cluster node such that
   ##     previous futures are not affecting this one, which
   ##     may happen even if the future is evaluated inside a
   ##     local, e.g. local({ a <<- 1 }).
-  ##     If the persistent = TRUE, this will be skipped.
-  backend$eraseGlobalEnvironment(future = future)
+  if (!persistent) {
+    t_start <- Sys.time()
+    cluster_call_blocking(cl, fun = grmall, future = future, when = "call grmall() on")
+    if (inherits(future[[".journal"]], "FutureJournal")) {
+      appendToFutureJournal(future,
+           event = "eraseWorker",
+        category = "overhead",
+          parent = "launch",
+           start = t_start,
+            stop = Sys.time()
+      )
+    }
+  }
 
-  ## (3) Launch future
-  backend$launchFuture(future)
+
+  ## (ii) Attach packages that needs to be attached
+  ##      NOTE: Already take care of by evalFuture().
+  ##      However, if we need to get an early error about missing packages,
+  ##      we can get the error here before launching the future.
+  t_start <- Sys.time()
+  packages <- future[["packages"]]
+  if (future[["earlySignal"]] && length(packages) > 0) {
+    if (debug) mdebugf("Attaching %d packages (%s) on cluster node #%d ...",
+                      length(packages), hpaste(sQuote(packages)), node_idx)
+
+    cluster_call_blocking(cl, fun = requirePackages, packages, future = future, when = "call requirePackages() on")
+
+    if (debug) mdebugf("Attaching %d packages (%s) on cluster node #%d ... DONE",
+                      length(packages), hpaste(sQuote(packages)), node_idx)
+  }
+  
+  if (inherits(future[[".journal"]], "FutureJournal")) {
+    appendToFutureJournal(future,
+         event = "attachPackages",
+      category = "overhead",
+        parent = "launch",
+         start = t_start,
+          stop = Sys.time()
+    )
+  }
+
+  ## Add to registry
+  FutureRegistry(reg, action = "add", future = future, earlySignal = FALSE)
+
+  ## (iv) Launch future
+  node_call_nonblocking(cl[[1L]], fun = evalFuture, args = list(data), when = "launch future on")
 
   future[["state"]] <- 'running'
 
@@ -318,7 +371,7 @@ receiveMessageFromWorker <- function(future, ...) {
     if (future[["gc"]]) {
       if (debug) mdebug("- Garbage collecting worker ...")
       ## Cleanup global environment while at it
-      if (!future[["persistent"]]) {
+      if (!isTRUE(future[["persistent"]])) {
         ## Blocking cluster-node call
         cluster_call_blocking(cl[1], fun = grmall, future = future, when = "call grmall() on")
       }
@@ -517,6 +570,12 @@ post_mortem_cluster_failure <- function(ex, when, node, future) {
 
   ## (4) POST-MORTEM ANALYSIS:
   postmortem <- list()
+
+  ## (a) Inspect the 'reason' for known clues
+  if (grepl("ignoring SIGPIPE signal", reason)) {
+    postmortem$sigpipe <- "The SIGPIPE error suggests that the R socket connection to the parallel worker broke, which can happen for different reasons, e.g. the parallel worker crashed"
+  }
+
   ## (a) Did the worker process terminate?
   if (!is.null(host) && is.numeric(pid)) {
     if (localhost) {
@@ -646,7 +705,7 @@ assertValidConnection <- function(future) {
 
 
 
-ClusterFutureBackend <- local({
+ClusterFutureBackend0 <- local({
   indexOf <- function(futures, future) {
     for (ii in seq_along(futures)) {
       if (identical(future, futures[[ii]])) return(ii)
@@ -915,6 +974,77 @@ ClusterFutureBackend <- local({
       }
     )
   }
-}) ## ClusterFutureBackend()
+}) ## ClusterFutureBackend0()
 
 
+
+
+#' @export
+launchFuture.ClusterFutureBackend <- function(backend, future, ...) {
+  debug <- isTRUE(getOption("future.debug"))
+  if (debug) {
+    mdebug("launchFuture() for ClusterFutureBackend ...")
+    on.exit(mdebug("launchFuture() for ClusterFutureBackend ... done"))
+  }
+
+  ## Coerce Future to ClusterFuture
+  args <- list(
+    future,
+    workers = backend[["workers"]]
+  )
+  future <- do.call(as_ClusterFuture, args = args)
+  class(future) <- unique(c(backend$futureClasses, class(future)))
+  
+  backend0 <- future[["backend"]]
+  
+  ## Next available cluster node
+  t_start <- Sys.time()
+
+  ## (1) Get a free worker. This will block until one is available
+  backend0$requestWorker(future)
+
+  ## (2) Attach packages that needs to be attached
+  ##     NOTE: Already take care of by evalFuture().
+  ##     However, if we need to get an early error about missing packages,
+  ##     we can get the error here before launching the future.
+  if (future[["earlySignal"]]) {
+    backend0$requirePackages(future = future)
+  }
+
+  ## (2) Reset global environment of cluster node such that
+  ##     previous futures are not affecting this one, which
+  ##     may happen even if the future is evaluated inside a
+  ##     local, e.g. local({ a <<- 1 }).
+  ##     If the persistent = TRUE, this will be skipped.
+  backend0$eraseGlobalEnvironment(future = future)
+
+  ## (3) Launch future
+  backend0$launchFuture(future)
+
+  future[["state"]] <- "running"
+
+  if (debug) mdebugf("%s started", class(future)[1])
+  
+  invisible(future)
+}
+
+
+ClusterFutureBackend <- function(workers, persistent = FALSE, ...) {
+  core <- new.env(parent = emptyenv())
+
+  ## Record future plan tweaks, if any
+  args <- list(workers = workers, persistent = persistent, ...)
+  for (name in names(args)) {
+    core[[name]] <- args[[name]]
+  }
+  core$futureClasses <- c("ClusterFuture", "Future")
+  core <- structure(core, class = c("ClusterFutureBackend", "FutureBackend", class(core)))
+  core
+}
+
+MultisessionFutureBackend <- function(workers, ...) {
+  core <- ClusterFutureBackend(workers = workers, ...)
+  core$futureClasses <- c("MultisessionFuture", core$futureClasses)
+  core <- structure(core, class = c("MultisessionFutureBackend", class(core)))
+  core
+}

R/backend_api-Future-class.R

@@ -395,9 +395,9 @@ assertOwner <- local({
 run.Future <- function(future, ...) {
   debug <- isTRUE(getOption("future.debug"))
   if (debug) {
-    mdebug("run() for ", sQuote(class(future)[1]), " ...")
+    mdebugf("run() for Future (%s) ...", sQuote(class(future)[1]))
     mdebug("- state: ", sQuote(future[["state"]]))
-    on.exit(mdebug("run() for ", sQuote(class(future)[1]), " ... done"), add = TRUE)
+    on.exit(mdebugf("run() for Future (%s) ... done", sQuote(class(future)[1])), add = TRUE)
   }
 
   if (future[["state"]] != "created") {
@@ -408,7 +408,7 @@ run.Future <- function(future, ...) {
   }
 
   ## Sanity check: This method should only called for lazy futures
-  stop_if_not(future[["lazy"]])
+#  stop_if_not(future[["lazy"]])
 
   if (is.null(future[["owner"]])) {
     future[["owner"]] <- session_uuid()
@@ -436,10 +436,19 @@ run.Future <- function(future, ...) {
       mdebug("- state: ", sQuote(future[["state"]]))
       on.exit(mdebug("run() for ", sQuote(class(future)[1]), " ... done"), add = TRUE)
 
+      if (debug) mprint(backend)
+
       ## Apply future plan tweaks
       args <- attr(makeFuture, "tweaks")
       if (is.null(args)) args <- list()
+      args2 <- formals(makeFuture)
+      args2[["..."]] <- NULL
+      args2[["envir"]] <- NULL
+      for (name in names(args2)) {
+        args[[name]] <- args2[[name]]
+      }
       backend <- do.call(backend, args = args)
+      if (debug) mdebug(" - FutureBackend: ", commaq(class(backend)))
       stop_if_not(inherits(backend, "FutureBackend"))
 
 

R/backend_api-UniprocessFuture-class.R

@@ -162,7 +162,7 @@ launchFuture.SequentialFutureBackend <- function(backend, future, ...) {
   future
 }
 
-SequentialFutureBackend <- function(...) {
+FutureBackend <- function(...) {
   core <- new.env(parent = emptyenv())
 
   ## Record future plan tweaks, if any
@@ -171,8 +171,15 @@ SequentialFutureBackend <- function(...) {
     core[[name]] <- args[[name]]
   }
 
-  core <- structure(core, class = c("SequentialFutureBackend", "FutureBackend", class(core)))
-    
+  core$futureClasses <- c("FutureBackend")
+  core <- structure(core, class = c("FutureBackend", class(core)))
+}
+
+
+SequentialFutureBackend <- function(...) {
+  core <- FutureBackend(...)
+  core$futureClasses <- c("SequentialFuture", "UniprocessFuture", core$futureClasses)
+  core <- structure(core, class = c("SequentialFutureBackend", class(core)))
   core
 }
 

R/utils-prune_pkg_code.R

@@ -26,7 +26,9 @@ prune_fcn <- function(name, envir) {
     body0 <- body(fcn)
     body <- walkAST(body0, call = prune_fcns)
     if (!identical(body, body0)) {
+      attrs <- attributes(fcn)
       body(fcn) <- body
+      attributes(fcn) <- attrs ## attributes are lost if body is changed
       assign(name, fcn, envir = envir, inherits = FALSE)
       return(TRUE)
     }

R/utils_api-tweak.R

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

R/zzz.R

@@ -2,6 +2,8 @@
 
 ## Update FutureBackend:s
 attr(sequential, "backend") <- SequentialFutureBackend
+attr(cluster, "backend") <- ClusterFutureBackend
+attr(multisession, "backend") <- MultisessionFutureBackend
 
 
 ## covr: skip=all
@@ -13,7 +15,7 @@ attr(sequential, "backend") <- SequentialFutureBackend
   if (isTRUE(as.logical(Sys.getenv("R_FUTURE_PRUNE_PKG_CODE", "FALSE")))) {
     prune_pkg_code()
   }
-  
+
   update_package_option("future.debug", mode = "logical")
   debug <- isTRUE(getOption("future.debug"))
 
@@ -31,7 +33,6 @@ attr(sequential, "backend") <- SequentialFutureBackend
   ## .GlobalEnv$.Random.seed.
   session_uuid(attributes = FALSE)
 
-
   ## Report on future plan, if set
   strategy <- getOption("future.plan")
   if (!is.null(strategy)) {
@@ -44,7 +45,6 @@ attr(sequential, "backend") <- SequentialFutureBackend
     }
   }
 
-
   args <- parseCmdArgs()
   p <- args$p
   if (!is.null(p)) {