Add future_lapply_staged parallelism

R/distributed.R

@@ -43,11 +43,16 @@ finish_distributed <- function(config){
   unlink(file, force = TRUE)
 }
 
-build_distributed <- function(target, cache_path){
+build_distributed <- function(target, cache_path, check = TRUE){
   config <- recover_drake_config(cache_path = cache_path)
   eval(parse(text = "base::require(drake, quietly = TRUE)"))
   do_prework(config = config, verbose_packages = FALSE)
-  build_check_store(target = target, config = config)
+  if (check){
+    build_check_store(target = target, config = config)
+  } else {
+    prune_envir(targets = target, config = config)
+    build_and_store(target = target, config = config)
+  }
   invisible()
 }
 

R/future.R

@@ -71,24 +71,7 @@ new_worker <- function(id, target, config, protect){
   # Avoid potential name conflicts with other globals.
   # When we solve #296, the need for such a clumsy workaround
   # should go away.
-  globals <- list(
-    DRAKE_GLOBALS__ = list(
-      target = target,
-      meta = meta,
-      config = config
-    )
-  )
-  if (identical(config$envir, globalenv())){
-    # Unit tests should not modify global env # nocov
-    if (exists("DRAKE_GLOBALS__", config$envir)){ # nocov # nolint
-      warning( # nocov
-        "Do not define an object named `DRAKE_GLOBALS__` ", # nocov
-        "in the global environment", # nocov
-        call. = FALSE # nocov
-      ) # nocov
-    } # nocov
-    globals <- c(globals, as.list(config$envir, all.names = TRUE)) # nocov
-  }
+  globals <- future_globals(target = target, meta = meta, config = config)
   evaluator <- drake_plan_override(
     target = target,
     field = "evaluator",
@@ -111,6 +94,28 @@ new_worker <- function(id, target, config, protect){
   )
 }
 
+future_globals <- function(target, meta, config){
+  globals <- list(
+    DRAKE_GLOBALS__ = list(
+      target = target,
+      meta = meta,
+      config = config
+    )
+  )
+  if (identical(config$envir, globalenv())){
+    # Unit tests should not modify global env # nocov
+    if (exists("DRAKE_GLOBALS__", config$envir)){ # nocov # nolint
+      warning( # nocov
+        "Do not define an object named `DRAKE_GLOBALS__` ", # nocov
+        "in the global environment", # nocov
+        call. = FALSE # nocov
+      ) # nocov
+    } # nocov
+    globals <- c(globals, as.list(config$envir, all.names = TRUE)) # nocov
+  }
+  globals
+}
+
 empty_worker <- function(target){
   structure(NA, target = target)
 }

R/parallel_ui.R

@@ -107,6 +107,7 @@ parallelism_choices <- function(distributed_only = FALSE) {
   distributed <- c(
     "future",
     "future_lapply",
+    "future_lapply_staged",
     "Makefile"
   )
   if (distributed_only){

R/staged.R

@@ -14,7 +14,7 @@ next_stage <- function(config, schedule) {
     new_meta <- lightly_parallelize(
       X = new_leaves,
       FUN = drake_meta,
-      jobs = config$jobs,
+      jobs = config$jobs_imports,
       config = config
     )
     names(new_meta) <- new_leaves
@@ -27,7 +27,7 @@ next_stage <- function(config, schedule) {
           config = config
         )
       },
-      jobs = config$jobs
+      jobs = config$jobs_imports
     ) %>%
       unlist
     targets <- c(targets, new_leaves[do_build])
@@ -139,3 +139,24 @@ run_parLapply_staged <- function(config) { # nolint
   }
   invisible()
 }
+
+run_future_lapply_staged <- function(config){
+  prepare_distributed(config = config)
+  schedule <- config$schedule
+  while (length(V(schedule)$name)){
+    stage <- next_stage(config = config, schedule = schedule)
+    schedule <- stage$schedule
+    if (!length(stage$targets)){
+      break
+    } else if (any(stage$targets %in% config$plan$target)){
+      set_attempt_flag(key = "_attempt", config = config)
+    }
+    tmp <- future.apply::future_lapply(
+      X = stage$targets,
+      FUN = build_distributed,
+      cache_path = config$cache_path,
+      check = FALSE
+    )
+  }
+  invisible()
+}

tests/testthat/test-future.R

@@ -6,9 +6,9 @@ test_with_dir("future package functionality", {
   scenario <- get_testing_scenario()
   e <- eval(parse(text = scenario$envir))
   load_mtcars_example(envir = e)
-  backends <- c("future_lapply", rep("future", 2))
-  caching <- c(rep("worker", 2), "master")
-  for (i in 1:3){
+  backends <- c("future_lapply", rep("future", 2), "future_lapply_staged")
+  caching <- c(rep("worker", 2), rep("master", 2))
+  for (i in 1:4){
     clean(destroy = TRUE)
     config <- make(
       e$my_plan,