Merge pull request #145 from wlandau/backlog

push_backlog() and pop_backlog()

DESCRIPTION

@@ -16,7 +16,7 @@ Description: In computationally demanding analysis projects,
   'clustermq' by Schubert (2019) <doi:10.1093/bioinformatics/btz284>),
   and 'batchtools' by Lang, Bischel, and Surmann (2017)
   <doi:10.21105/joss.00135>.
-Version: 0.8.0.9002
+Version: 0.8.0.9003
 License: MIT + file LICENSE
 URL: https://wlandau.github.io/crew/, https://github.com/wlandau/crew
 BugReports: https://github.com/wlandau/crew/issues

NEWS.md

@@ -1,7 +1,8 @@
-# crew 0.8.0.9002 (development)
+# crew 0.8.0.9003 (development)
 
 * Require `nanonext` >= 0.12.0 and `mirai` >= 0.12.0.
 * Return to always re-launching backlogged inactive workers (#79, https://github.com/shikokuchuo/mirai/discussions/95).
+* Implement `push_backlog()` and `pop_backlog()` to manage cases when it is not desirable to push to saturated controllers (https://github.com/ropensci/targets/issues/1220).
 
 # crew 0.8.0
 

R/crew_controller.R

@@ -66,6 +66,7 @@ crew_class_controller <- R6::R6Class(
     .popped = NULL,
     .log = NULL,
     .error = NULL,
+    .backlog = NULL,
     .shove = function(
       command,
       data = list(),
@@ -141,6 +142,10 @@ crew_class_controller <- R6::R6Class(
     #'   from the last call to `map(error = "stop)`.
     error = function() {
       .subset2(private, ".error")
+    },
+    #' @field backlog Character vector of explicitly backlogged tasks.
+    backlog = function() {
+      .subset2(private, ".backlog")
     }
   ),
   public = list(
@@ -180,6 +185,7 @@ crew_class_controller <- R6::R6Class(
         message = "client and launcher must have the same name"
       )
       crew_assert(private$.log, is.null(.) || is.list(.))
+      crew_assert(private$.backlog, is.null(.) || is.character(.))
       invisible()
     },
     #' @description Check if the controller is empty.
@@ -280,6 +286,7 @@ crew_class_controller <- R6::R6Class(
           errors = rep(0L, workers),
           warnings = rep(0L, workers)
         )
+        private$.backlog <- character(0L)
       }
       invisible()
     },
@@ -998,6 +1005,45 @@ crew_class_controller <- R6::R6Class(
       )
       invisible(envir$result)
     },
+    #' @description Push the name of a task to the backlog.
+    #' @details `pop_backlog()` pops the tasks that can be pushed
+    #'   without saturating the controller.
+    #' @param name Character of length 1 with the task name to push to
+    #'   the backlog.
+    #' @return `NULL` (invisibly).
+    #' @param controller Not used. Included to ensure the signature is
+    #'   compatible with the analogous method of controller groups.
+    push_backlog = function(name, controller = NULL) {
+      crew_assert(
+        name,
+        is.character(.),
+        length(.) == 1L,
+        !anyNA(.),
+        nzchar(.),
+        message = "'name' in push_backlog() must be a valid character string"
+      )
+      n <- length(.subset2(private, ".backlog")) + 1L
+      private$.backlog[[n]] <- name
+      invisible()
+    },
+    #' @description Pop the task names from the head of the backlog which
+    #'   can be pushed without saturating the controller.
+    #' @return Character vector of task names which can be pushed to the
+    #'   controller without saturating it. If the controller is saturated,
+    #'   `character(0L)` is returned.
+    #' @param controllers Not used. Included to ensure the signature is
+    #'   compatible with the analogous method of controller groups.
+    pop_backlog = function(controllers = NULL) {
+      n <- .subset2(.subset2(self, "client"), "workers") -
+        .subset2(self, "unresolved")()
+      if (n < 1L) {
+        return(character(0L))
+      }
+      backlog <- .subset2(private, ".backlog")
+      out <- utils::head(x = backlog, n = n)
+      private$.backlog <- backlog[-seq_len(n)]
+      out
+    },
     #' @description Summarize the workers and tasks of the controller.
     #' @return A data frame of summary statistics on the workers and tasks.
     #'   It has one row per worker websocket and the following columns:

R/crew_controller_group.R

@@ -604,6 +604,34 @@ crew_class_controller_group <- R6::R6Class(
       )
       invisible(out)
     },
+    #' @description Push the name of a task to the backlog.
+    #' @details `pop_backlog()` pops the tasks that can be pushed
+    #'   without saturating the controller.
+    #' @param name Character of length 1 with the task name to push to
+    #'   the backlog.
+    #' @return `NULL` (invisibly).
+    #' @param controller Character vector of length 1 with the controller name.
+    #'   Set to `NULL` to select the default controller that `push_backlog()`
+    #'   would choose.
+    push_backlog = function(name, controller = NULL) {
+      control <- .subset2(
+        private,
+        ".select_single_controller"
+      )(name = controller)
+      control$push_backlog(name = name)
+    },
+    #' @description Pop the task names from the head of the backlog which
+    #'   can be pushed without saturating the controller.
+    #' @return Character vector of task names which can be pushed to the
+    #'   controller without saturating it. If the controller is saturated,
+    #'   `character(0L)` is returned.
+    #' @param controllers Character vector of controller names.
+    #'   Set to `NULL` to select all controllers.
+    pop_backlog = function(controllers = NULL) {
+      control <- private$.select_controllers(controllers)
+      out <- map(control, ~.subset2(.x, "pop_backlog")())
+      unlist(out, use.names = FALSE)
+    },
     #' @description Summarize the workers of one or more controllers.
     #' @return A data frame of aggregated worker summary statistics
     #'   of all the selected controllers. It has one row per worker,