Address #124

build_distributed() now returns invisible()

R/Makefile.R

@@ -3,7 +3,7 @@ run_Makefile <- function( #nolint: we want Makefile capitalized.
   run = TRUE,
   debug = FALSE
 ){
-  config <- prepare_distributed(config = config)
+  prepare_distributed(config = config)
   with_output_sink(
     new = "Makefile",
     code = {
@@ -95,8 +95,8 @@ mk <- function(
   target = character(0),
   cache_path = drake::default_cache_path()
 ){
-  config <- build_distributed(target = target, cache_path = cache_path)
-  config <- inventory(config)
+  build_distributed(target = target, cache_path = cache_path)
+  config <- recover_config(cache_path)
   new_hash <- self_hash(target = target, config = config)
   if (!identical(config$old_hash, new_hash)){
     file <- time_stamp_file(target = target, config = config)

R/console.R

@@ -65,7 +65,8 @@ console_up_to_date <- function(config){
   if (config$imports_only){
     return(invisible())
   }
-  if (config$verbose && !length(config$attempted_targets)){
+  any_attempted <- length(config$cache$list(namespace = "target_attempts"))
+  if (config$verbose && !any_attempted){
     color("All targets are already up to date.\n", colors["target"]) %>%
       cat
   }

R/current.R

@@ -31,3 +31,9 @@ file_current <- function(target, hashes, config){
   }
   identical(config$cache$get(target)$value, hashes$file)
 }
+
+log_target_attempts <- Vectorize(function(targets, config){
+  config$cache$set(key = targets, value = targets,
+    namespace = "target_attempts")
+},
+"targets")

R/distributed.R

@@ -7,7 +7,7 @@ prepare_distributed <- function(config){
       file = globalenv_file(this_cache_path)
     )
   }
-  config$attempted_targets <- outdated(
+  target_attempts <- outdated(
     plan = config$plan,
     targets = config$targets,
     envir = config$envir,
@@ -18,19 +18,13 @@ prepare_distributed <- function(config){
     packages = config$packages,
     prework = config$prework
   )
+  log_target_attempts(targets = target_attempts, config = config)
   config$cache$set("config", config, namespace = "distributed")
-  invisible(config)
+  invisible()
 }
 
 build_distributed <- function(target, cache_path){
-  cache <- this_cache(cache_path)
-  config <- cache$get("config", namespace = "distributed")
-  if (identical(globalenv(), config$envir)){
-    dir <- cache_path
-    file <- globalenv_file(dir)
-    load(file = file, envir = config$envir)
-  }
-  config <- inventory(config)
+  config <- recover_config(cache_path = cache_path)
   do_prework(config = config, verbose_packages = FALSE)
   prune_envir(targets = target, config = config)
   hash_list <- hash_list(targets = target, config = config)
@@ -47,5 +41,16 @@ build_distributed <- function(target, cache_path){
       config = config
     )
   }
-  invisible(config)
+  invisible()
+}
+
+recover_config <- function(cache_path){
+  cache <- this_cache(cache_path)
+  config <- cache$get("config", namespace = "distributed")
+  if (identical(globalenv(), config$envir)){
+    dir <- cache_path
+    file <- globalenv_file(dir)
+    load(file = file, envir = config$envir)
+  }
+  inventory(config)
 }

R/make.R

@@ -247,9 +247,9 @@ make <- function(
   )
   check_config(config = config)
   store_config(config = config)
-  store_session(config = config)
+  initialize_session(config = config)
   config <- imports_only_preprocessing(config = config)
-  config <- run_parallel_backend(config = config)
+  run_parallel_backend(config = config)
   config$parallelism <- parallelism
   console_up_to_date(config = config)
   return(invisible(config))
@@ -278,7 +278,8 @@ next_targets <- function(graph_remaining_targets){
     names()
 }
 
-store_session <- function(config){
+initialize_session <- function(config){
+  config$cache$clear(namespace = "target_attempts")
   config$cache$set(
     key = "sessionInfo",
     value = sessionInfo(),

R/parLapply.R

@@ -15,11 +15,10 @@ run_parLapply <- function(config) { # nolint
   })
   clusterCall(cl = config$cluster, fun = do_prework, config = config,
     verbose_packages = FALSE)
-  config <- run_parallel(
+  run_parallel(
     config = config,
     worker = worker_parLapply # nolint
   )
-  invisible(config)
 }
 
 worker_parLapply <- function(targets, hash_list, config) { # nolint

R/parallel.R

@@ -10,7 +10,7 @@ run_parallel <- function(config, worker) {
   while (length(V(config$graph_remaining_targets))){
     config <- parallel_stage(worker = worker, config = config)
   }
-  config
+  invisible()
 }
 
 parallel_stage <- function(worker, config) {
@@ -24,10 +24,8 @@ parallel_stage <- function(worker, config) {
   build_these <- Filter(candidates,
     f = function(target)
       should_build(target = target, hash_list = hash_list, config = config))
-  config$attempted_targets <- c(
-    config$attempted_targets,
-    intersect(build_these, config$plan$target)
-  )
+  target_attempts <- intersect(build_these, config$plan$target)
+  log_target_attempts(targets = target_attempts, config = config)
   hash_list <- hash_list[build_these]
   if (length(build_these)){
     worker(targets = build_these, hash_list = hash_list,

R/timestamp.R

@@ -4,7 +4,8 @@ time_stamps <- function(config){
   dir_empty(stamp_dir)
   write_time_stamp_template(cache_path)
   targets <- intersect(V(config$graph)$name, config$plan$target)
-  stamp_these <- setdiff(targets, config$attempted_targets)
+  target_attempts <- config$cache$list(namespace = "target_attempts")
+  stamp_these <- setdiff(targets, target_attempts)
   lightly_parallelize(
     stamp_these, write_time_stamp, jobs = config$jobs, config = config)
   return(invisible())