First prototype of MulticoreFutureBackend

Author: HenrikBengtsson

R/backend_api-ClusterFuture-class.R

@@ -1009,7 +1009,7 @@ launchFuture.ClusterFutureBackend <- function(backend, future, ...) {
     workers = backend[["workers"]]
   )
   future <- do.call(as_ClusterFuture, args = args)
-  class(future) <- unique(c(backend$futureClasses, class(future)))
+  future <- coerceFuture(backend, future)
 
   backend0 <- future[["backend"]]
 

R/backend_api-MulticoreFuture-class.R

@@ -291,3 +291,75 @@ result.MulticoreFuture <- function(future, ...) {
 
   result
 }
+
+
+
+#' @export
+launchFuture.MulticoreFutureBackend <- function(backend, future, ...) {
+  debug <- isTRUE(getOption("future.debug"))
+  
+  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)
+
+  mcparallel <- importParallel("mcparallel")
+
+  future <- coerceFuture(backend, future)
+
+  data <- getFutureData(future, debug = debug)
+
+  t_start <- Sys.time()
+  
+  ## Get a free worker
+  reg <- sprintf("multicore-%s", session_uuid())
+  requestCore(
+    await = function() FutureRegistry(reg, action = "collect-first", earlySignal = TRUE),
+    workers = backend[["workers"]]
+  )
+
+  if (inherits(future[[".journal"]], "FutureJournal")) {
+    appendToFutureJournal(future,
+         event = "getWorker",
+      category = "other",
+        parent = "launch",
+         start = t_start,
+          stop = Sys.time()
+    )
+  }
+
+  ## Add to registry
+  FutureRegistry(reg, action = "add", future = future, earlySignal = TRUE)
+
+  job <- local({
+    oopts <- options(mc.cores = NULL)
+    on.exit(options(oopts))
+    mcparallel(evalFuture(data))
+  })
+
+  future[["job"]] <- job
+  future[["state"]] <- "running"
+
+  if (debug) mdebugf("%s started", class(future)[1])
+  
+  invisible(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]]
+  }
+  core$futureClasses <- c("MulticoreFuture", "Future")
+  core <- structure(core, class = c("MulticoreFutureBackend", "FutureBackend", class(core)))
+  core
+}

R/backend_api-UniprocessFuture-class.R

@@ -108,6 +108,16 @@ SequentialFuture <- function(expr = NULL, envir = parent.frame(), substitute = T
 
 
 
+
+coerceFuture <- function(backend, future, ...) {
+  UseMethod("coerceFuture")
+}
+
+coerceFuture.FutureBackend <- function(backend, future, ...) {
+  class(future) <- unique(c(backend$futureClasses, class(future)))
+  future
+}
+
 launchFuture <- function(backend, future, ...) {
   UseMethod("launchFuture")
 }
@@ -130,9 +140,7 @@ launchFuture.SequentialFutureBackend <- function(backend, future, ...) {
   ## 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))
+  future <- coerceFuture(backend, future)
 
   ## Launch future
   future[["state"]] <- "running"

R/zzz.R

@@ -4,6 +4,7 @@
 attr(sequential, "backend") <- SequentialFutureBackend
 attr(cluster, "backend") <- ClusterFutureBackend
 attr(multisession, "backend") <- MultisessionFutureBackend
+attr(multicore, "backend") <- MulticoreFutureBackend
 
 
 ## covr: skip=all