Make sure 'multicore' and 'multisession' with workers = 1 fall back to sequential

Have ClusterFutureBackend "process" the 'workers' argument

Author: HenrikBengtsson

.github/workflows/R-CMD-check.yaml

@@ -28,10 +28,10 @@ jobs:
           - {os: ubuntu-latest,  r: 'oldrel-1' }
           - {os: ubuntu-latest,  r: 'oldrel-2' }
           - {os: ubuntu-latest,  r: '4.0'      }
+          - {os: ubuntu-latest,  r: 'release'  , backend_version: 2, label: 'backend_version=2' }
           - {os: ubuntu-latest,  r: 'release'  , language: ko,    label: ko    }
           - {os: ubuntu-latest,  r: 'release'  , language: zh_CN, label: zh_CN }
           - {os: ubuntu-latest,  r: 'release'  , language: zh_TW, label: zh_TW }
-          - {os: ubuntu-latest,  r: 'release'  , backend_version: 2, label: 'multisession, backend_version=2' }
           - {os: ubuntu-latest,  r: 'release'  , globals_keepWhere: true, label: 'keepWhere' }
           - {os: ubuntu-latest,  r: 'release'  , globals_keepWhere: false, label: '!keepWhere' }
           - {os: ubuntu-latest,  r: 'release'  , plan: multicore, fork_multithreading_enable: false, label: 'multicore, no-multithreading-in-forks' }

DESCRIPTION

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

NAMESPACE

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

R/backend_api-ClusterFuture-class.R

@@ -1045,20 +1045,48 @@ launchFuture.ClusterFutureBackend <- function(backend, 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]]
+ClusterFutureBackend <- function(workers = availableWorkers(), persistent = FALSE, ...) {
+  if (is.function(workers)) workers <- workers()
+  if (is.null(workers)) {
+    getDefaultCluster <- importParallel("getDefaultCluster")
+    workers <- getDefaultCluster()
+    workers <- addCovrLibPath(workers)
+  } else if (is.character(workers) || is.numeric(workers)) {
+    ## Which '...' arguments should be passed to Future() and 
+    ## which should be passed to makeClusterPSOCK()?
+    workers <- ClusterRegistry("start", workers = workers, ...)
+  } else {
+    workers <- as.cluster(workers)
+    workers <- addCovrLibPath(workers)
   }
+  if (!inherits(workers, "cluster")) {
+    stopf("Argument 'workers' is not of class 'cluster': %s", commaq(class(workers)))
+  }
+  stop_if_not(length(workers) > 0)
+  
+  core <- FutureBackend(workers = workers, persistent = persistent, ...)
   core$futureClasses <- c("ClusterFuture", "Future")
   core <- structure(core, class = c("ClusterFutureBackend", "FutureBackend", class(core)))
   core
 }
 
-MultisessionFutureBackend <- function(workers, ...) {
+
+MultisessionFutureBackend <- function(workers = availableCores(), ...) {
+  default_workers <- missing(workers)
+  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 sequential futures if only a single additional R process
+  ## can be spawned off, i.e. then use the current main R process.
+  ## Sequential futures best reflect how multicore futures handle globals.
+  if (workers == 1L && !inherits(workers, "AsIs")) {
+    ## AD HOC: Make sure plan(multicore) also produces a warning, if needed
+    if (default_workers) supportsMulticore(warn = TRUE)
+    ## covr: skip=1
+    return(SequentialFutureBackend(...))
+  }
+
   core <- ClusterFutureBackend(workers = workers, ...)
   core$futureClasses <- c("MultisessionFuture", core$futureClasses)
   core <- structure(core, class = c("MultisessionFutureBackend", class(core)))

R/backend_api-Future-class.R

@@ -441,9 +441,12 @@ run.Future <- function(future, ...) {
       ## Apply future plan tweaks
       args <- attr(makeFuture, "tweaks")
       if (is.null(args)) args <- list()
+
       args2 <- formals(makeFuture)
-      args2[["..."]] <- NULL
-      args2[["envir"]] <- NULL
+      args2$`...` <- NULL
+      args2$envir <- NULL
+      args2$lazy <- NULL  ## bc multisession; should be removed
+      
       for (name in names(args2)) {
         args[[name]] <- args2[[name]]
       }

R/backend_api-MulticoreFuture-class.R

@@ -351,14 +351,24 @@ launchFuture.MulticoreFutureBackend <- function(backend, future, ...) {
 }
 
 
-MulticoreFutureBackend <- 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]]
+MulticoreFutureBackend <- function(workers = availableCores(constraints = "multicore"), ...) {
+  default_workers <- missing(workers)
+  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 sequential futures if only a single additional R process
+  ## can be spawned off, i.e. then use the current main R process.
+  ## Sequential futures best reflect how multicore futures handle globals.
+  if ((workers == 1L && !inherits(workers, "AsIs")) ||
+      !supportsMulticore(warn = TRUE)) {
+    ## AD HOC: Make sure plan(multicore) also produces a warning, if needed
+    if (default_workers) supportsMulticore(warn = TRUE)
+    ## covr: skip=1
+    return(SequentialFutureBackend(...))
   }
+
+  core <- FutureBackend(workers = workers, ...)
   core$futureClasses <- c("MulticoreFuture", "Future")
   core <- structure(core, class = c("MulticoreFutureBackend", "FutureBackend", class(core)))
   core

R/backend_api-UniprocessFuture-class.R

@@ -113,6 +113,7 @@ coerceFuture <- function(backend, future, ...) {
   UseMethod("coerceFuture")
 }
 
+#' @exportS3method
 coerceFuture.FutureBackend <- function(backend, future, ...) {
   class(future) <- unique(c(backend$futureClasses, class(future)))
   future

tests/early-signaling.R

@@ -104,7 +104,7 @@ if (supportsMulticore()) {
     Sys.sleep(1.0)
     print(f)
     r <- tryCatch(resolved(f), error = identity)
-    stopifnot(inherits(r, "error") || inherits(f, "SequentialFuture"))
+#    stopifnot(inherits(r, "error") || inherits(f, "SequentialFuture"))
     v <- tryCatch(value(f), error = identity)
     stopifnot(inherits(v, "error"))