Promises from crew

A crew controller can generate two types of promise objects for use with the promises package:

1. Single-task promises: wait until a single task finishes. The promise is fulfilled if the task succeeded and rejected if the task threw an error. In the former case, the controller asynchronously pops the completed task and returns the tibble of results and metadata. On error, task is still asynchronously popped, but the error message of the task is returned instead.
2. Multi-task promises: wait until there are no pending tasks left in the controller (or controller group). This happens when either all the tasks finish or the controller is empty. The promise is fulfilled if all tasks succeeded and rejected if at least one task threw an error. In the former case, the controller asynchronously pops all completed tasks and returns the tibble of all results and metadata (with one row per task). On error, tasks are all still asynchronously popped, but the error message of one of the tasks is returned instead.

Single-task promises

To dive into single-task promises, let’s start a local controller first.

library(crew)
library(dplyr)
library(promises)
controller <- crew_controller_local(workers = 2L)
controller$start()

Let’s push a single task.

task <- controller$push(
  name = "success",
  command = {
    Sys.sleep(2)
    "done"
  },
  save_command = TRUE
)

To create a promise specific to the task above, call as.promise() on the returned task object, then call controller$autoscale() to tell the main event loop to periodically launch any necessary crew workers.

promise <- as.promise(task)
controller$autoscale()

To create a promise that resolves when any task in the controller completes, use the promise() method of the controller. controller$promise() always calls controller$autoscale() behind the scenes, so there is no need to call it manually in this case. The following promise prints the output value asynchronously if the task succeeds.

promise <- controller$promise(mode = "one") %...>%
  mutate(result = as.character(result)) %...>%
  print()

When you run both steps above, the R interpreter runs it immediately and returns control back to you. But then the following output prints two seconds after the task was pushed.

#> # A tibble: 1 × 12
#>   name    command     result seconds  seed algorithm error trace
#>   <chr>   <chr>       <chr>    <dbl> <int> <chr>     <chr> <chr>
#> 1 success "{\n    Sy… done      2.00    NA NA        NA    NA   
#> # ℹ 4 more variables: warnings <chr>, launcher <chr>,
#> #   worker <int>, instance <chr>

The task below runs in the background for 2 seconds and then throws an error.

controller$push({
  Sys.sleep(2)
  stop("error message")
})

As before, control returns immediately when you push the task and create the promise.

promise <- then(
  controller$promise(mode = "one"),
  onRejected = function(error) {
    print(conditionMessage(error))
  }
)

But this time, an error message prints two seconds later.

#> [1] "error message"

Multi-task promises

To demonstrate multi-task promises, we push multiple tasks at once. The walk() method is like map(), except that it returns control immediately without waiting for any tasks to complete.

controller$walk(
  command = {
    Sys.sleep(2)
    argument
  },
  iterate = list(argument = c("x", "y")),
  names = "argument",
  save_command = TRUE
)

We create a promise which asynchronously resolves when all the tasks in the controller finish.

promise <- controller$promise(mode = "all") %...>%
  mutate(result = as.character(result)) %...>%
  select(any_of(c("name", "command", "result", "error", "worker"))) %...T>%
  print()

Two seconds after walk() was called, the promise resolves asynchronously and prints the results of all the tasks. Each row in the tibble below corresponds to an individual task.

#> # A tibble: 2 × 5
#>   name  command                                result error worker
#>   <chr> <chr>                                  <chr>  <chr>  <int>
#> 1 x     "{\n    Sys.sleep(2)\n    argument\n}" x      NA         1
#> 2 y     "{\n    Sys.sleep(2)\n    argument\n}" y      NA         2

A couple remarks:

1. You do not need to use walk() with multi-task promises. You can push tasks individually and still create a promise which resolves they all finish.
2. A multi-task promise is rejected if any one of the tasks fail. Due to performance concerns and limitations, the error is not discovered until all tasks resolve.

To demonstrate (1) and (2), let’s push a task that will succeed and a task that will throw an error.

controller$push(
  name = "success",
  command = {
    Sys.sleep(2)
    "done"
  },
  save_command = TRUE
)
controller$push(
  name = "error",
  command = {
    Sys.sleep(2)
    stop("one task's error message")
  },
  save_command = TRUE
)

We create a multi-task promise which prints the error message asynchronously on resolution.

promise <- then(
  controller$promise(mode = "all"),
  onRejected = function(error) {
    print(conditionMessage(error))
  }
)

Two seconds after the tasks were pushed, the error message prints.

#> [1] "one task's error message"