Fix #333

NEWS.md

@@ -1,5 +1,9 @@
 # targets 0.2.0.9000
 
+## New features
+
+* Implement an exponential backoff algorithm for polling the priority queue in `tar_make_clustermq()` and `tar_make_future()`. Add a `backoff` option in `tar_option_set()` to set the maximum upper bound (seconds) for the polling interval (#333).
+
 ## Enhancements
 
 * Use `rlang::check_installed()` inside `assert_package()` (#331, @malcolmbarrett).

R/class_backoff.R

@@ -1,6 +1,6 @@
 # Exponential backoff algorithm
 # similar to https://en.wikipedia.org/wiki/Exponential_backoff
-backoff_init <- function(min = 0.01, max = 10, rate = 1.5) {
+backoff_init <- function(min = 0.01, max = 5, rate = 1.5) {
   backoff_new(
     min = max(sqrt(.Machine$double.eps), min),
     max = max,

R/class_scheduler.R

@@ -13,7 +13,7 @@ scheduler_init <- function(
   queued <- counter_init(names)
   progress <- progress_init(queued = queued)
   reporter <- reporter_init(reporter)
-  backoff <- backoff_init()
+  backoff <- backoff_init(max = tar_option_get("backoff"))
   scheduler_new(graph, queue, progress, reporter, backoff)
 }
 

R/tar_option_get.R

@@ -49,6 +49,7 @@ tar_option_default <- function(option) {
     garbage_collection = FALSE,
     deployment = "worker",
     priority = 0,
+    backoff = 5,
     resources = list(),
     storage = "main",
     retrieval = "main",

R/tar_option_set.R

@@ -20,6 +20,20 @@
 #'   everything in `tar_option_get("imports")`.
 #' @param envir Environment containing functions and global objects
 #'   used in the R commands to run targets.
+#' @param backoff Numeric of length 1, must be greater than or equal to 0.01.
+#'   Maximum upper bound of the random polling interval
+#'   for the priority queue (seconds).
+#'   In high-performance computing (e.g. [tar_make_clustermq()]
+#'   and [tar_make_future()]) it can be expensive to repeatedly poll the
+#'   priority queue if no targets are ready to process. The number of seconds
+#'   between polls is `runif(1, 0.01, max(backoff, 0.01 * 1.5 ^ index))`,
+#'   where `index` is the number of consecutive polls so far that found
+#'   no targets ready to skip or run.
+#'   (If no target is ready, `index` goes up by 1. If a target is ready,
+#'   `index` resets to 0. For more information on exponential,
+#'   backoff, visit <https://en.wikipedia.org/wiki/Exponential_backoff>).
+#'   Raising `backoff` is kinder to the CPU etc. but may incur delays
+#'   in some instances.
 #' @param debug Character vector of names of targets to run in debug mode.
 #'   To use effectively, you must set `callr_function = NULL` and
 #'   restart your R session just before running. You should also
@@ -67,6 +81,7 @@ tar_option_set <- function(
   garbage_collection = NULL,
   deployment = NULL,
   priority = NULL,
+  backoff = NULL,
   resources = NULL,
   storage = NULL,
   retrieval = NULL,
@@ -87,6 +102,7 @@ tar_option_set <- function(
   tar_option_set_garbage_collection(garbage_collection)
   tar_option_set_deployment(deployment)
   tar_option_set_priority(priority)
+  tar_option_set_backoff(backoff)
   tar_option_set_resources(resources)
   tar_option_set_storage(storage)
   tar_option_set_retrieval(retrieval)
@@ -178,6 +194,15 @@ tar_option_set_priority <- function(priority) {
   assign("priority", priority, envir = tar_envir_options)
 }
 
+tar_option_set_backoff <- function(backoff) {
+  backoff <- backoff %||% tar_option_get("backoff")
+  assert_dbl(backoff, msg = "backoff must be numeric")
+  assert_scalar(backoff, msg = "backoff must have length 1")
+  assert_ge(backoff, 0.01, msg = "backoff cannot be less than 0.01")
+  assert_le(backoff, 1e9, msg = "backoff cannot be greater than 1e9")
+  assign("backoff", backoff, envir = tar_envir_options)
+}
+
 tar_option_set_resources <- function(resources) {
   resources <- resources %||% tar_option_get("resources")
   assert_list(resources, "resources in tar_option_set() must be a named list.")

tests/testthat/test-tar_options.R

@@ -6,3 +6,7 @@ tar_test("tar_option_set() works", {
 tar_test("tar_option_get() must take a valid option name", {
   expect_error(tar_option_get("nope"))
 })
+
+tar_test("tar_option_get() must take a valid option name", {
+  expect_error(tar_option_get("nope"))
+})