Console messages when a target is pending

ropensci · Dec 4, 2023 · b7f2788 · b7f2788
1 parent 384c6b0
commit b7f2788
Show file tree

Hide file tree

Showing 8 changed files with 61 additions and 35 deletions.
diff --git a/R/class_builder.R b/R/class_builder.R
@@ -49,10 +49,20 @@ target_read_value.tar_builder <- function(target, pipeline = NULL) {
 }
 
 #' @export
-target_prepare.tar_builder <- function(target, pipeline, scheduler, meta) {
+target_prepare.tar_builder <- function(
+  target,
+  pipeline,
+  scheduler,
+  meta,
+  pending = FALSE
+) {
   target_patternview_dispatched(target, pipeline, scheduler)
   scheduler$progress$register_dispatched(target)
-  scheduler$reporter$report_dispatched(target, scheduler$progress)
+  scheduler$reporter$report_dispatched(
+    target = target,
+    progress = scheduler$progress,
+    pending = pending
+  )
   builder_ensure_deps(target, pipeline, "main")
   builder_update_subpipeline(target, pipeline)
   builder_marshal_subpipeline(target)

diff --git a/R/class_crew.R b/R/class_crew.R
@@ -157,29 +157,25 @@ crew_class <- R6::R6Class(
       globals <- self$exports$globals
       resources <- target$settings$resources$crew
       name <- target_get_name(target)
-      saturated <- self$controller$saturated(controller = resources$controller)
-      if (saturated) {
-        # Requires a slow test. Covered in the saturation tests in
-        # tests/hpc/test-crew_local.R # nolint
-        # nocov start
-        self$scheduler$queue$append0(name = name)
-        self$backoff_requeue$wait()
-        # nocov end
-      } else {
-        target_prepare(target, self$pipeline, self$scheduler, self$meta)
-        self$sync_meta_time()
-        self$controller$push(
-          command = command,
-          data = data,
-          globals = globals,
-          substitute = FALSE,
-          name = name,
-          controller = resources$controller,
-          scale = TRUE,
-          seconds_timeout = resources$seconds_timeout
-        )
-        self$backoff_requeue$reset()
-      }
+      target_prepare(
+        target = target,
+        pipeline = self$pipeline, 
+        scheduler = self$scheduler,
+        meta = self$meta,
+        pending = self$controller$saturated(controller = resources$controller)
+      )
+      self$sync_meta_time()
+      self$controller$push(
+        command = command,
+        data = data,
+        globals = globals,
+        substitute = FALSE,
+        name = name,
+        controller = resources$controller,
+        scale = TRUE,
+        seconds_timeout = resources$seconds_timeout
+      )
+      self$backoff_requeue$reset()
     },
     run_main = function(target) {
       target_prepare(target, self$pipeline, self$scheduler, self$meta)

diff --git a/R/class_reporter.R b/R/class_reporter.R
@@ -53,7 +53,11 @@ reporter_class <- R6::R6Class(
         cli_warned(progress$warned$count)
       }
     },
-    report_dispatched = function(target = NULL, progress = NULL) {
+    report_dispatched = function(
+      target = NULL,
+      progress = NULL,
+      pending = FALSE
+    ) {
     },
     report_completed = function(target = NULL, progress = NULL) {
     },

diff --git a/R/class_summary.R b/R/class_summary.R
@@ -24,7 +24,7 @@ summary_class <- R6::R6Class(
     },
     report_error = function(error) {
     },
-    report_dispatched = function(target = NULL, progress) {
+    report_dispatched = function(target = NULL, progress, pending = FALSE) {
       self$report_progress(progress)
     },
     report_completed = function(target = NULL, progress) {

diff --git a/R/class_target.R b/R/class_target.R
@@ -209,12 +209,24 @@ target_skip <- function(target, pipeline, scheduler, meta, active) {
   UseMethod("target_skip")
 }
 
-target_prepare <- function(target, pipeline, scheduler, meta) {
+target_prepare <- function(
+  target,
+  pipeline,
+  scheduler,
+  meta,
+  pending = FALSE
+) {
   UseMethod("target_prepare")
 }
 
 #' @export
-target_prepare.default <- function(target, pipeline, scheduler, meta) {
+target_prepare.default <- function(
+  target,
+  pipeline,
+  scheduler,
+  meta,
+  pending = FALSE
+) {
 }
 
 target_should_run <- function(target, meta) {

diff --git a/R/class_timestamp.R b/R/class_timestamp.R
@@ -19,13 +19,14 @@ timestamp_class <- R6::R6Class(
       self$buffer[length(self$buffer) + 1L] <- msg
       self$poll()
     },
-    report_dispatched = function(target, progress = NULL) {
+    report_dispatched = function(target, progress = NULL, pending = FALSE) {
       self$buffer_message(
         cli_dispatched(
           target_get_name(target),
           target_get_type_cli(target),
           time_stamp = TRUE,
-          print = FALSE
+          print = FALSE,
+          pending = pending
         )
       )
     },

diff --git a/R/class_verbose.R b/R/class_verbose.R
@@ -19,12 +19,13 @@ verbose_class <- R6::R6Class(
       self$buffer[length(self$buffer) + 1L] <- msg
       self$poll()
     },
-    report_dispatched = function(target, progress = NULL) {
+    report_dispatched = function(target, progress = NULL, pending = FALSE) {
       self$buffer_message(
         cli_dispatched(
           target_get_name(target),
           target_get_type_cli(target),
-          print = FALSE
+          print = FALSE,
+          pending = pending
         )
       )
     },

diff --git a/R/utils_cli.R b/R/utils_cli.R
@@ -2,10 +2,12 @@ cli_dispatched <- function(
   name,
   prefix = NULL,
   time_stamp = FALSE,
-  print = TRUE
+  print = TRUE,
+  pending = FALSE
 ) {
   time <- if_any(time_stamp, time_stamp(), NULL)
-  msg <- paste(c(time, "dispatched", prefix, name), collapse = " ")
+  action <- if_any(pending, "dispatched (pending)", "dispatched")
+  msg <- paste(c(time, action, prefix, name), collapse = " ")
   cli_blue_play(msg, print = print)
 }