Record R options future.wait.{timeout,interval,alpha} when setting up the Cluster- and MulicoreFutureBackend

Author: HenrikBengtsson

R/backend_api-ClusterFutureBackend-class.R

@@ -50,7 +50,17 @@ ClusterFutureBackend <- function(workers = availableWorkers(), persistent = FALS
   ## Name of the FutureRegistry
   reg <- sprintf("workers-%s", name)
 
-  core <- FutureBackend(workers = workers, persistent = persistent, reg = reg, earlySignal = earlySignal, gc = gc, ...)
+  core <- FutureBackend(
+    workers = workers,
+    persistent = persistent,
+    reg = reg,
+    earlySignal = earlySignal,
+    gc = gc,
+    future.wait.timeout = getOption("future.wait.timeout", 30 * 24 * 60 * 60),
+    future.wait.interval = getOption("future.wait.interval", 0.01),
+    future.wait.alpha = getOption("future.wait.alpha", 1.01),
+    ...
+  )
   core[["futureClasses"]] <- c("ClusterFuture", core[["futureClasses"]])
   core <- structure(core, class = c("ClusterFutureBackend", "FutureBackend", class(core)))
   core
@@ -86,9 +96,13 @@ launchFuture.ClusterFutureBackend <- function(backend, future, ...) {
   ## (1) Get a free worker. This will block until one is available
   if (debug) mdebug("requestWorker() ...")
 
+  timeout <- backend[["future.wait.timeout"]]
+  delta <- backend[["future.wait.interval"]]
+  alpha <- backend[["future.wait.alpha"]]
+
   node_idx <- requestNode(await = function() {
     FutureRegistry(reg, action = "collect-first", earlySignal = TRUE)
-  }, workers = workers)
+  }, workers = workers, timeout = timeout, delta = delta, alpha = alpha)
   future[["node"]] <- node_idx
 
   if (inherits(future[[".journal"]], "FutureJournal")) {

R/backend_api-MulticoreFuture-class.R

@@ -228,13 +228,18 @@ launchFuture.MulticoreFutureBackend <- function(backend, future, ...) {
   data <- getFutureData(future, debug = debug)
 
   t_start <- Sys.time()
-  
+
+  workers <- backend[["workers"]]
+  reg <- backend[["reg"]]
+
+  timeout <- backend[["future.wait.timeout"]]
+  delta <- backend[["future.wait.interval"]]
+  alpha <- backend[["future.wait.alpha"]]
+
   ## Get a free worker
-  reg <- sprintf("multicore-%s", session_uuid())
-  requestCore(
-    await = function() FutureRegistry(reg, action = "collect-first", earlySignal = TRUE),
-    workers = backend[["workers"]]
-  )
+  requestCore(await = function() {
+    FutureRegistry(reg, action = "collect-first", earlySignal = TRUE)
+  }, workers = workers, timeout = timeout, delta = delta, alpha = alpha)
 
   if (inherits(future[[".journal"]], "FutureJournal")) {
     appendToFutureJournal(future,

R/backend_api-MulticoreFutureBackend-class.R

@@ -27,7 +27,16 @@ MulticoreFutureBackend <- function(workers = availableCores(constraints = "multi
     return(SequentialFutureBackend(...))
   }
 
-  core <- FutureBackend(workers = workers, ...)
+  reg <- sprintf("multicore-%s", session_uuid())
+
+  core <- FutureBackend(
+    workers = workers,
+    reg = reg,
+    future.wait.timeout = getOption("future.wait.timeout", 30 * 24 * 60 * 60),
+    future.wait.interval = getOption("future.wait.interval", 0.01),
+    future.wait.alpha = getOption("future.wait.alpha", 1.01),
+    ...
+  )
   core[["futureClasses"]] <- c("MulticoreFuture", "MultiprocessFuture", core[["futureClasses"]])
   core <- structure(core, class = c("MulticoreFutureBackend", "FutureBackend", class(core)))
   core