FutureBackend cleanups: Move cluster() and multisession() to use Cluster- and MultisessionFutureBackend

Author: HenrikBengtsson

NAMESPACE

@@ -28,13 +28,17 @@ S3method(launchFuture,SequentialFutureBackend)
 S3method(mandelbrot,matrix)
 S3method(mandelbrot,numeric)
 S3method(nbrOfFreeWorkers,"NULL")
+S3method(nbrOfFreeWorkers,ClusterFutureBackend)
+S3method(nbrOfFreeWorkers,SequentialFutureBackend)
 S3method(nbrOfFreeWorkers,cluster)
 S3method(nbrOfFreeWorkers,future)
 S3method(nbrOfFreeWorkers,logical)
 S3method(nbrOfFreeWorkers,multicore)
 S3method(nbrOfFreeWorkers,multiprocess)
 S3method(nbrOfFreeWorkers,uniprocess)
 S3method(nbrOfWorkers,"NULL")
+S3method(nbrOfWorkers,ClusterFutureBackend)
+S3method(nbrOfWorkers,SequentialFutureBackend)
 S3method(nbrOfWorkers,cluster)
 S3method(nbrOfWorkers,future)
 S3method(nbrOfWorkers,multiprocess)

R/backend_api-ClusterFuture-class.R

@@ -247,9 +247,10 @@ resolved.ClusterFuture <- function(x, run = TRUE, timeout = NULL, ...) {
 
   future <- x
   backend <- future[["backend"]]
+  stop_if_not(inherits(backend, "FutureBackend"))
   workers <- backend$workers
   reg <- backend$reg
-
+  
   ## A lazy future not even launched?
   if (future[["state"]] == "created") {
     if (run) {

R/backend_api-ClusterFutureBackend-class.R

@@ -13,7 +13,7 @@
 #' @keywords internal
 #' @rdname FutureBackend
 #' @export
-ClusterFutureBackend <- function(workers = availableWorkers(), persistent = FALSE, ...) {
+ClusterFutureBackend <- function(workers = availableWorkers(), persistent = FALSE, earlySignal = TRUE, ...) {
   if (is.function(workers)) workers <- workers()
   if (is.null(workers)) {
     getDefaultCluster <- importParallel("getDefaultCluster")

R/backend_api-FutureBackend-class.R

@@ -61,7 +61,7 @@ makeFutureBackend <- function(evaluator, ...) {
   args2 <- formals(evaluator)
   args2$`...` <- NULL
   args2$envir <- NULL
-  args2$lazy <- NULL  ## bc multisession; should be removed
+  args2$lazy <- NULL         ## bc multisession; should be removed
   for (name in names(args2)) {
     args[[name]] <- args2[[name]]
   }

R/backend_api-MulticoreFutureBackend-class.R

@@ -27,7 +27,7 @@ MulticoreFutureBackend <- function(workers = availableCores(constraints = "multi
   }
 
   core <- FutureBackend(workers = workers, ...)
-  core[["futureClasses"]] <- c("MulticoreFuture", core[["futureClasses"]])
+  core[["futureClasses"]] <- c("MulticoreFuture", "MultiprocessFuture", core[["futureClasses"]])
   core <- structure(core, class = c("MulticoreFutureBackend", "FutureBackend", class(core)))
   core
 }

R/backend_api-cluster.R

@@ -40,9 +40,11 @@
 #'
 #' @export
 cluster <- function(..., persistent = FALSE, workers = availableWorkers(), envir = parent.frame()) {
-  future <- ClusterFuture(..., persistent = persistent, workers = workers, envir = envir)
-  if (!future[["lazy"]]) future <- run(future)
-  invisible(future)
+  f <- Future(..., envir = envir)
+  f[["workers"]] <- workers
+  f[["persistent"]] <- persistent
+  class(f) <- c("ClusterFuture", "MultiprocessFuture", "Future")
+  f
 }
 class(cluster) <- c("cluster", "multiprocess", "future", "function")
 attr(cluster, "init") <- TRUE

R/backend_api-multisession.R

@@ -65,19 +65,10 @@
 #'
 #' @export
 multisession <- function(..., workers = availableCores(), lazy = FALSE, rscript_libs = .libPaths(), envir = parent.frame()) {
-  if (is.function(workers)) workers <- workers()
-  workers <- structure(as.integer(workers), class = class(workers))
-  stop_if_not(length(workers) == 1, is.finite(workers), workers >= 1)
-
-  ## Fall back to lazy sequential futures if only a single R session can be used,
-  ## that is, then use the current main R process.
-  if (workers == 1L && !inherits(workers, "AsIs")) {
-    return(sequential(..., lazy = TRUE, envir = envir))
-  }
-
-  future <- MultisessionFuture(..., workers = workers, lazy = lazy, rscript_libs = rscript_libs, envir = envir)
-  if (!future[["lazy"]]) future <- run(future)
-  invisible(future)
+  f <- Future(..., lazy = lazy, envir = envir)
+  f[["workers"]] <- workers
+  class(f) <- c("MultisessionFuture", "MultiprocessFuture", "Future")
+  f
 }
 class(multisession) <- c("multisession", "cluster", "multiprocess", "future", "function")
 attr(multisession, "init") <- TRUE

R/utils_api-nbrOfWorkers.R

@@ -17,26 +17,27 @@ nbrOfWorkers <- function(evaluator = NULL) {
 
 
 #' @export
-nbrOfWorkers.cluster <- function(evaluator) {
-  assert_no_positional_args_but_first()
-  
-  expr <- formals(evaluator)$workers
-  workers <- eval(expr, enclos = baseenv())
-  if (is.function(workers)) workers <- workers()
-  if (is.character(workers)) {
-    stop_if_not(!anyNA(workers))
-    workers <- length(workers)
-  } else if (is.numeric(workers)) {
-  } else if (inherits(workers, "cluster")) {
-    workers <- length(workers)
-  } else {
-    stopf("Unsupported type of 'workers' for evaluator of class %s: %s", commaq(class(evaluator)), class(workers)[1])
-  }
+nbrOfWorkers.ClusterFutureBackend <- function(evaluator) {
+  backend <- evaluator
+  workers <- backend[["workers"]]
+  stop_if_not(length(workers) > 0L, inherits(workers, "cluster"))
+  workers <- length(workers)
   stop_if_not(length(workers) == 1L, !is.na(workers), workers >= 1L, is.finite(workers))
-
   workers
 }
 
+#' @export
+nbrOfWorkers.SequentialFutureBackend <- function(evaluator) {
+  1L
+}
+
+#' @export
+nbrOfWorkers.cluster <- function(evaluator) {
+  assert_no_positional_args_but_first()
+  backend <- makeFutureBackend(evaluator)
+  nbrOfWorkers(backend)
+}
+
 #' @export
 nbrOfWorkers.uniprocess <- function(evaluator) {
   assert_no_positional_args_but_first()
@@ -106,22 +107,14 @@ nbrOfFreeWorkers <- function(evaluator = NULL, background = FALSE, ...) {
 
 
 #' @export
-nbrOfFreeWorkers.cluster <- function(evaluator, background = FALSE, ...) {
-  assert_no_positional_args_but_first()
-  
-  workers <- nbrOfWorkers(evaluator)
-  
-  ## Create a dummy, lazy future based on the future strategy ("evaluator")
-  f <- evaluator(NULL, lazy = TRUE)
+nbrOfFreeWorkers.ClusterFutureBackend <- function(evaluator, ...) {
+  backend <- evaluator
+  workers <- backend[["workers"]]
+  stop_if_not(length(workers) > 0L, inherits(workers, "cluster"))
+  workers <- length(workers)
+  reg <- backend$reg
+  stop_if_not(length(reg) == 1L, is.character(reg), nzchar(reg))
 
-  ## Special case
-  if (inherits(f, "SequentialFuture")) {
-    return(if (isTRUE(background)) 0L else 1L)
-  }
-  
-  name <- attr(f$workers, "name", exact = TRUE)
-  stop_if_not(is.character(name), length(name) == 1L)
-  reg <- sprintf("workers-%s", name)
   ## Number of unresolved cluster futures
   usedNodes <- length(FutureRegistry(reg, action = "list", earlySignal = FALSE))
 
@@ -131,6 +124,20 @@ nbrOfFreeWorkers.cluster <- function(evaluator, background = FALSE, ...) {
   workers
 }
 
+#' @export
+nbrOfFreeWorkers.SequentialFutureBackend <- function(evaluator, background = FALSE, ...) {
+  assert_no_positional_args_but_first()
+  if (isTRUE(background)) 0L else 1L
+}
+
+
+#' @export
+nbrOfFreeWorkers.cluster <- function(evaluator, background = FALSE, ...) {
+  assert_no_positional_args_but_first()
+  backend <- makeFutureBackend(evaluator)
+  nbrOfFreeWorkers(backend, background = background, ...)
+}
+
 
 #' @export
 nbrOfFreeWorkers.uniprocess <- function(evaluator, background = FALSE, ...) {