Package 'crew'

Description

In computationally demanding analysis projects, statisticians and data scientists asynchronously deploy long-running tasks to distributed systems, ranging from traditional clusters to cloud services. The NNG-powered mirai R package is a sleek and sophisticated scheduler that efficiently processes these intense workloads. The crew package extends mirai with a unifying interface for third-party worker launchers. Inspiration also comes from packages future, rrq, clustermq, and batchtools.

Description

Assert that a condition is true.

Usage

crew_assert(value = NULL, ..., message = NULL, envir = parent.frame())

Arguments

Value

NULL (invisibly). Throws an error if the condition is not true.

See Also

Other utility: crew_clean(), crew_deprecate(), crew_eval(), crew_random_name(), crew_retry(), crew_terminate_process(), crew_terminate_signal(), crew_worker()

Examples

crew_assert(1 < 2)
crew_assert("object", !anyNA(.), nzchar(.))
tryCatch(
  crew_assert(2 < 1),
  crew_error = function(condition) message("false")
)

Description

Create an R6 object to manage local asynchronous quick tasks with error detection.

Usage

crew_async(workers = NULL)

Arguments

Details

crew_async() objects are created inside launchers to allow launcher plugins to run local tasks asynchronously, such as calls to cloud APIs to launch serious remote workers.

Value

An R6 async client object.

See Also

Other async: crew_class_async

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
x <- crew_async()
x$start()
out <- x$eval(1 + 1)
mirai::call_mirai_(out)
out$data # 2
x$terminate()
}

Description

R6 class for async configuration.

Details

See crew_async().

Active bindings

Methods

Public methods

• crew_class_async$new()

• crew_class_async$validate()

• crew_class_async$start()

• crew_class_async$terminate()

• crew_class_async$started()

• crew_class_async$asynchronous()

• crew_class_async$eval()

Method new()

TLS configuration constructor.

Usage

crew_class_async$new(workers = NULL)

Arguments

Returns

An R6 object with TLS configuration.

Method validate()

Validate the object.

Usage

crew_class_async$validate()

Returns

NULL (invisibly).

Method start()

Start the local workers and error handling socket.

Usage

crew_class_async$start()

Details

Does not create workers or an error handling socket if workers is NULL or the object is already started.

Returns

NULL (invisibly).

Method terminate()

Start the local workers and error handling socket.

Usage

crew_class_async$terminate()

Details

Waits for existing tasks to complete first.

Returns

NULL (invisibly).

Method started()

Show whether the object is started.

Usage

crew_class_async$started()

Returns

Logical of length 1, whether the object is started.

Method asynchronous()

Show whether the object is asynchronous (has real workers).

Usage

crew_class_async$asynchronous()

Returns

Logical of length 1, whether the object is asynchronous.

Method eval()

Run a local asynchronous task using a local compute profile.

Usage

crew_class_async$eval(
  command,
  substitute = TRUE,
  data = list(),
  packages = character(0L),
  library = NULL
)

Arguments

Details

Used for launcher plugins with asynchronous launches and terminations. If processes is NULL, the task will run locally. Otherwise, the task will run on a local process in the local mirai compute profile.

Returns

If the processes field is NULL, a list with an object named data containing the result of evaluating expr synchronously. Otherwise, the task is evaluated asynchronously, and the result is a mirai task object. Either way, the data element of the return value will contain the result of the task.

See Also

Other async: crew_async()

Description

R6 class for mirai clients.

Details

See crew_client().

Active bindings

Methods

Public methods

• crew_class_client$new()

• crew_class_client$validate()

• crew_class_client$start()

• crew_class_client$terminate()

• crew_class_client$condition()

• crew_class_client$resolved()

• crew_class_client$summary()

• crew_class_client$pids()

Method new()

mirai client constructor.

Usage

crew_class_client$new(
  name = NULL,
  workers = NULL,
  host = NULL,
  port = NULL,
  tls = NULL,
  seconds_interval = NULL,
  seconds_timeout = NULL,
  retry_tasks = NULL,
  relay = NULL
)

Arguments

Returns

An R6 object with the client.

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
client$log()
client$terminate()
}

Method validate()

Validate the client.

Usage

crew_class_client$validate()

Returns

NULL (invisibly).

Method start()

Start listening for workers on the available sockets.

Usage

crew_class_client$start()

Returns

NULL (invisibly).

Method terminate()

Stop the mirai client and disconnect from the worker websockets.

Usage

crew_class_client$terminate()

Returns

NULL (invisibly).

Method condition()

Get the nanonext condition variable which tasks signal on resolution.

Usage

crew_class_client$condition()

Returns

The nanonext condition variable which tasks signal on resolution. The return value is NULL if the client is not running.

Method resolved()

Get the true value of the nanonext condition variable.

Usage

crew_class_client$resolved()

Returns

The value of the nanonext condition variable.

Method summary()

Show an informative worker log.

Usage

crew_class_client$summary()

Returns

A tibble with information on the workers, or NULL if the client is not started. The tibble has 1 row per worker and the following columns:

• worker: integer index of the worker.

• online: TRUE if the worker is online and connected to the websocket URL, FALSE otherwise.

• instances: integer, number of instances of mirai daemons (crew workers) that have connected to the websocket URL during the life cycle of the listener.

• assigned: number of tasks assigned to the current websocket URL.

• complete: number of tasks completed at the current websocket URL.

• socket: websocket URL. crew changes the token at the end of the URL path periodically as a safeguard while managing workers.

Method pids()

Get the process IDs of the local process and the mirai dispatcher (if started).

Usage

crew_class_client$pids()

Returns

An integer vector of process IDs of the local process and the mirai dispatcher (if started).

See Also

Other client: crew_client()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
client$log()
client$terminate()
}

## ------------------------------------------------
## Method `crew_class_client$new`
## ------------------------------------------------

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
client$log()
client$terminate()
}

Description

R6 class for controllers.

Details

See crew_controller().

Active bindings

Methods

Public methods

• crew_class_controller$new()

• crew_class_controller$validate()

• crew_class_controller$empty()

• crew_class_controller$nonempty()

• crew_class_controller$resolved()

• crew_class_controller$unresolved()

• crew_class_controller$unpopped()

• crew_class_controller$saturated()

• crew_class_controller$start()

• crew_class_controller$started()

• crew_class_controller$launch()

• crew_class_controller$scale()

• crew_class_controller$autoscale()

• crew_class_controller$descale()

• crew_class_controller$push()

• crew_class_controller$walk()

• crew_class_controller$map()

• crew_class_controller$pop()

• crew_class_controller$collect()

• crew_class_controller$promise()

• crew_class_controller$wait()

• crew_class_controller$push_backlog()

• crew_class_controller$pop_backlog()

• crew_class_controller$summary()

• crew_class_controller$cancel()

• crew_class_controller$pids()

• crew_class_controller$terminate()

Method new()

mirai controller constructor.

Usage

crew_class_controller$new(client = NULL, launcher = NULL)

Arguments

Returns

An R6 controller object.

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
launcher <- crew_launcher_local()
controller <- crew_controller(client = client, launcher = launcher)
controller$start()
controller$push(name = "task", command = sqrt(4))
controller$wait()
controller$pop()
controller$terminate()
}

Method validate()

Validate the client.

Usage

crew_class_controller$validate()

Returns

NULL (invisibly).

Method empty()

Check if the controller is empty.

Usage

crew_class_controller$empty(controllers = NULL)

Arguments

Details

A controller is empty if it has no running tasks or completed tasks waiting to be retrieved with push().

Returns

TRUE if the controller is empty, FALSE otherwise.

Method nonempty()

Check if the controller is nonempty.

Usage

crew_class_controller$nonempty(controllers = NULL)

Arguments

Details

A controller is empty if it has no running tasks or completed tasks waiting to be retrieved with push().

Returns

TRUE if the controller is empty, FALSE otherwise.

Method resolved()

Number of resolved mirai() tasks.

Usage

crew_class_controller$resolved(controllers = NULL)

Arguments

Details

resolved() is cumulative: it counts all the resolved tasks over the entire lifetime of the controller session.

Returns

Non-negative integer of length 1, number of resolved mirai() tasks. The return value is 0 if the condition variable does not exist (i.e. if the client is not running).

Method unresolved()

Number of unresolved mirai() tasks.

Usage

crew_class_controller$unresolved(controllers = NULL)

Arguments

Returns

Non-negative integer of length 1, number of unresolved mirai() tasks.

Method unpopped()

Number of resolved mirai() tasks available via pop().

Usage

crew_class_controller$unpopped(controllers = NULL)

Arguments

Returns

Non-negative integer of length 1, number of resolved mirai() tasks available via pop().

Method saturated()

Check if the controller is saturated.

Usage

crew_class_controller$saturated(
  collect = NULL,
  throttle = NULL,
  controller = NULL
)

Arguments

Details

A controller is saturated if the number of unresolved tasks is greater than or equal to the maximum number of workers. In other words, in a saturated controller, every available worker has a task. You can still push tasks to a saturated controller, but tools that use crew such as targets may choose not to.

Returns

TRUE if the controller is saturated, FALSE otherwise.

Method start()

Start the controller if it is not already started.

Usage

crew_class_controller$start(controllers = NULL)

Arguments

Details

Register the mirai client and register worker websockets with the launcher.

Returns

NULL (invisibly).

Method started()

Check whether the controller is started.

Usage

crew_class_controller$started(controllers = NULL)

Arguments

Details

Actually checks whether the client is started.

Returns

TRUE if the controller is started, FALSE otherwise.

Method launch()

Launch one or more workers.

Usage

crew_class_controller$launch(n = 1L, controllers = NULL)

Arguments

Returns

NULL (invisibly).

Method scale()

Auto-scale workers out to meet the demand of tasks.

Usage

crew_class_controller$scale(throttle = TRUE, controllers = NULL)

Arguments

Details

The scale() method re-launches all inactive backlogged workers, then any additional inactive workers needed to accommodate the demand of unresolved tasks. A worker is "backlogged" if it was assigned more tasks than it has completed so far.

Methods push(), pop(), and wait() already invoke scale() if the scale argument is TRUE. For finer control of the number of workers launched, call launch() on the controller with the exact desired number of workers.

Returns

NULL (invisibly).

Method autoscale()

Run worker auto-scaling in a private later loop every controller$client$seconds_interval seconds.

Usage

crew_class_controller$autoscale(controllers = NULL)

Arguments

Details

Call controller$descale() to terminate the auto-scaling loop.

Returns

NULL (invisibly).

Method descale()

Terminate the auto-scaling loop started by controller$autoscale().

Usage

crew_class_controller$descale(controllers = NULL)

Arguments

Returns

NULL (invisibly).

Method push()

Push a task to the head of the task list.

Usage

crew_class_controller$push(
  command,
  data = list(),
  globals = list(),
  substitute = TRUE,
  seed = NULL,
  algorithm = NULL,
  packages = character(0),
  library = NULL,
  seconds_timeout = NULL,
  scale = TRUE,
  throttle = TRUE,
  name = NA_character_,
  save_command = FALSE,
  controller = NULL
)

Arguments

Returns

Invisibly return the mirai object of the pushed task. This allows you to interact with the task directly, e.g. to create a promise object with promises::as.promise().

Method walk()

Apply a single command to multiple inputs, and return control to the user without waiting for any task to complete.

Usage

crew_class_controller$walk(
  command,
  iterate,
  data = list(),
  globals = list(),
  substitute = TRUE,
  seed = NULL,
  algorithm = NULL,
  packages = character(0),
  library = NULL,
  seconds_timeout = NULL,
  names = NULL,
  save_command = FALSE,
  scale = TRUE,
  throttle = TRUE,
  controller = NULL
)

Arguments

Details

In contrast to walk(), map() blocks the local R session and waits for all tasks to complete.

Returns

Invisibly returns a list of mirai task objects for the newly created tasks. The order of tasks in the list matches the order of data in the iterate argument.

Method map()

Apply a single command to multiple inputs, wait for all tasks to complete, and return the results of all tasks.

Usage

crew_class_controller$map(
  command,
  iterate,
  data = list(),
  globals = list(),
  substitute = TRUE,
  seed = NULL,
  algorithm = NULL,
  packages = character(0),
  library = NULL,
  seconds_interval = 0.5,
  seconds_timeout = NULL,
  names = NULL,
  save_command = FALSE,
  error = "stop",
  warnings = TRUE,
  verbose = interactive(),
  scale = TRUE,
  throttle = TRUE,
  controller = NULL
)

Arguments

Details

map() cannot be used unless all prior tasks are completed and popped. You may need to wait and then pop them manually. Alternatively, you can start over: either call terminate() on the current controller object to reset it, or create a new controller object entirely.

Returns

A tibble of results and metadata: one row per task and columns corresponding to the output of pop().

Method pop()

Pop a completed task from the results data frame.

Usage

crew_class_controller$pop(
  scale = TRUE,
  collect = NULL,
  throttle = TRUE,
  error = NULL,
  controllers = NULL
)

Arguments

Details

If not task is currently completed, pop() will attempt to auto-scale workers as needed.

Returns

If there is no task to collect, return NULL. Otherwise, return a one-row tibble with the following columns.

• name: the task name if given.

• command: a character string with the R command if save_command was set to TRUE in push().

• result: a list containing the return value of the R command.

• seconds: number of seconds that the task ran.

• seed: the single integer originally supplied to push(), NA otherwise. The pseudo-random number generator state just prior to the task can be restored using set.seed(seed = seed, kind = algorithm), where seed and algorithm are part of this output.

• algorithm: name of the pseudo-random number generator algorithm originally supplied to push(), NA otherwise. The pseudo-random number generator state just prior to the task can be restored using set.seed(seed = seed, kind = algorithm), where seed and algorithm are part of this output.

• status: a character string. "success" if the task did not throw an error, "cancel" if the task was canceled with the cancel() controller method, or "error" if the task threw an error.

• code: an integer code denoting the specific exit status: 0 for successful tasks, 1 for tasks with an error in the R command of the task, and another positive integer with an NNG status code if there is an error at the NNG/nanonext level.

• error: the first 2048 characters of the error message if the task threw an error, NA otherwise.

• trace: the first 2048 characters of the text of the traceback if the task threw an error, NA otherwise.

• warnings: the first 2048 characters. of the text of warning messages that the task may have generated, NA otherwise.

• launcher: name of the crew launcher where the task ran.

Method collect()

Pop all available task results and return them in a tidy tibble.

Usage

crew_class_controller$collect(
  scale = TRUE,
  throttle = TRUE,
  error = NULL,
  controllers = NULL
)

Arguments

Returns

A tibble of results and metadata of all resolved tasks, with one row per task. Returns NULL if there are no tasks to collect.

Method promise()

Create a promises::promise() object to asynchronously pop or collect one or more tasks.

Usage

crew_class_controller$promise(
  mode = "one",
  seconds_interval = 0.1,
  scale = NULL,
  throttle = NULL,
  controllers = NULL
)

Arguments

Details

Please be aware that pop() or collect() will happen asynchronously at a some unpredictable time after the promise object is created, even if your local R process appears to be doing something completely different. This behavior is highly desirable in a Shiny reactive context, but please be careful as it may be surprising in other situations.

Returns

A promises::promise() object whose eventual value will be a tibble with results from one or more popped tasks. If mode = "one", only one task is popped and returned (one row). If mode = "all", then all the tasks are returned in a tibble with one row per task (or NULL is returned if there are no tasks to pop).

Method wait()

Wait for tasks.

Usage

crew_class_controller$wait(
  mode = "all",
  seconds_interval = 0.5,
  seconds_timeout = Inf,
  scale = TRUE,
  throttle = TRUE,
  controllers = NULL
)

Arguments

Details

The wait() method blocks the calling R session and repeatedly auto-scales workers for tasks that need them. The function runs until it either times out or the condition in mode is met.

Returns

A logical of length 1, invisibly. TRUE if the condition in mode was met, FALSE otherwise.

Method push_backlog()

Push the name of a task to the backlog.

Usage

crew_class_controller$push_backlog(name, controller = NULL)

Arguments

Details

pop_backlog() pops the tasks that can be pushed without saturating the controller.

Returns

NULL (invisibly).

Method pop_backlog()

Pop the task names from the head of the backlog which can be pushed without saturating the controller.

Usage

crew_class_controller$pop_backlog(controllers = NULL)

Arguments

Returns

Character vector of task names which can be pushed to the controller without saturating it. If the controller is saturated, character(0L) is returned.

Method summary()

Summarize the workers and tasks of the controller.

Usage

crew_class_controller$summary(controllers = NULL)

Arguments

Returns

A data frame of summary statistics on the workers and tasks. It has one row per worker websocket and the following columns:

• controller: name of the controller. . * worker: integer index of the worker.

• tasks: number of tasks which were completed by a worker at the websocket and then returned by calling pop() on the controller object.

• seconds: total number of runtime and seconds of all the tasks that ran on a worker connected to this websocket and then were retrieved by calling pop() on the controller object.

• errors: total number of tasks which ran on a worker at the website, encountered an error in R, and then retrieved with pop().

• warnings: total number of tasks which ran on a worker at the website, encountered one or more warnings in R, and then retrieved with pop(). Note: warnings is actually the number of tasks, not the number of warnings. (A task could throw more than one warning.

Method cancel()

Cancel one or more tasks.

Usage

crew_class_controller$cancel(names = character(0L), all = FALSE)

Arguments

Method pids()

Get the process IDs of the local process and the mirai dispatcher (if started).

Usage

crew_class_controller$pids(controllers = NULL)

Arguments

Returns

An integer vector of process IDs of the local process and the mirai dispatcher (if started).

Method terminate()

Terminate the workers and the mirai client.

Usage

crew_class_controller$terminate(controllers = NULL)

Arguments

Returns

NULL (invisibly).

See Also

Other controller: crew_controller()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
launcher <- crew_launcher_local()
controller <- crew_controller(client = client, launcher = launcher)
controller$start()
controller$push(name = "task", command = sqrt(4))
controller$wait()
controller$pop()
controller$terminate()
}

## ------------------------------------------------
## Method `crew_class_controller$new`
## ------------------------------------------------

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
launcher <- crew_launcher_local()
controller <- crew_controller(client = client, launcher = launcher)
controller$start()
controller$push(name = "task", command = sqrt(4))
controller$wait()
controller$pop()
controller$terminate()
}

Description

R6 class for controller groups.

Details

See crew_controller_group().

Active bindings

Methods

Public methods

• crew_class_controller_group$new()

• crew_class_controller_group$validate()

• crew_class_controller_group$empty()

• crew_class_controller_group$nonempty()

• crew_class_controller_group$resolved()

• crew_class_controller_group$unresolved()

• crew_class_controller_group$unpopped()

• crew_class_controller_group$saturated()

• crew_class_controller_group$start()

• crew_class_controller_group$started()

• crew_class_controller_group$launch()

• crew_class_controller_group$scale()

• crew_class_controller_group$autoscale()

• crew_class_controller_group$descale()

• crew_class_controller_group$push()

• crew_class_controller_group$walk()

• crew_class_controller_group$map()

• crew_class_controller_group$pop()

• crew_class_controller_group$collect()

• crew_class_controller_group$promise()

• crew_class_controller_group$wait()

• crew_class_controller_group$push_backlog()

• crew_class_controller_group$pop_backlog()

• crew_class_controller_group$summary()

• crew_class_controller_group$pids()

• crew_class_controller_group$terminate()

Method new()

Multi-controller constructor.

Usage

crew_class_controller_group$new(controllers = NULL, relay = NULL)

Arguments

Returns

An R6 object with the controller group object.

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
persistent <- crew_controller_local(name = "persistent")
transient <- crew_controller_local(
  name = "transient",
  tasks_max = 1L
)
group <- crew_controller_group(persistent, transient)
group$start()
group$push(name = "task", command = sqrt(4), controller = "transient")
group$wait()
group$pop()
group$terminate()
}

Method validate()

Validate the client.

Usage

crew_class_controller_group$validate()

Returns

NULL (invisibly).

Method empty()

See if the controllers are empty.

Usage

crew_class_controller_group$empty(controllers = NULL)

Arguments

Details

A controller is empty if it has no running tasks or completed tasks waiting to be retrieved with push().

Returns

TRUE if all the selected controllers are empty, FALSE otherwise.

Method nonempty()

Check if the controller group is nonempty.

Usage

crew_class_controller_group$nonempty(controllers = NULL)

Arguments

Details

A controller is empty if it has no running tasks or completed tasks waiting to be retrieved with push().

Returns

TRUE if the controller is empty, FALSE otherwise.

Method resolved()

Number of resolved mirai() tasks.

Usage

crew_class_controller_group$resolved(controllers = NULL)

Arguments

Details

resolved() is cumulative: it counts all the resolved tasks over the entire lifetime of the controller session.

Returns

Non-negative integer of length 1, number of resolved mirai() tasks. The return value is 0 if the condition variable does not exist (i.e. if the client is not running).

Method unresolved()

Number of unresolved mirai() tasks.

Usage

crew_class_controller_group$unresolved(controllers = NULL)

Arguments

Returns

Non-negative integer of length 1, number of unresolved mirai() tasks.

Method unpopped()

Number of resolved mirai() tasks available via pop().

Usage

crew_class_controller_group$unpopped(controllers = NULL)

Arguments

Returns

Non-negative integer of length 1, number of resolved mirai() tasks available via pop().

Method saturated()

Check if a controller is saturated.

Usage

crew_class_controller_group$saturated(
  collect = NULL,
  throttle = NULL,
  controller = NULL
)

Arguments

Details

A controller is saturated if the number of unresolved tasks is greater than or equal to the maximum number of workers. In other words, in a saturated controller, every available worker has a task. You can still push tasks to a saturated controller, but tools that use crew such as targets may choose not to.

Returns

TRUE if all the selected controllers are saturated, FALSE otherwise.

Method start()

Start one or more controllers.

Usage

crew_class_controller_group$start(controllers = NULL)

Arguments

Returns

NULL (invisibly).

Method started()

Check whether all the given controllers are started.

Usage

crew_class_controller_group$started(controllers = NULL)

Arguments

Details

Actually checks whether all the given clients are started.

Returns

TRUE if the controllers are started, FALSE if any are not.

Method launch()

Launch one or more workers on one or more controllers.

Usage

crew_class_controller_group$launch(n = 1L, controllers = NULL)

Arguments

Returns

NULL (invisibly).

Method scale()

Automatically scale up the number of workers if needed in one or more controller objects.

Usage

crew_class_controller_group$scale(throttle = TRUE, controllers = NULL)

Arguments

Details

See the scale() method in individual controller classes.

Returns

NULL (invisibly).

Method autoscale()

Run worker auto-scaling in a private later loop every controller$client$seconds_interval seconds.

Usage

crew_class_controller_group$autoscale(controllers = NULL)

Arguments

Returns

NULL (invisibly).

Method descale()

Terminate the auto-scaling loop started by controller$autoscale().

Usage

crew_class_controller_group$descale(controllers = NULL)

Arguments

Returns

NULL (invisibly).

Method push()

Push a task to the head of the task list.

Usage

crew_class_controller_group$push(
  command,
  data = list(),
  globals = list(),
  substitute = TRUE,
  seed = NULL,
  algorithm = NULL,
  packages = character(0),
  library = NULL,
  seconds_timeout = NULL,
  scale = TRUE,
  throttle = TRUE,
  name = NA_character_,
  save_command = FALSE,
  controller = NULL
)

Arguments

Returns

Invisibly return the mirai object of the pushed task. This allows you to interact with the task directly, e.g. to create a promise object with promises::as.promise().

Method walk()

Apply a single command to multiple inputs, and return control to the user without waiting for any task to complete.

Usage

crew_class_controller_group$walk(
  command,
  iterate,
  data = list(),
  globals = list(),
  substitute = TRUE,
  seed = NULL,
  algorithm = NULL,
  packages = character(0),
  library = NULL,
  seconds_timeout = NULL,
  names = NULL,
  save_command = FALSE,
  scale = TRUE,
  throttle = TRUE,
  controller = NULL
)

Arguments

Details

In contrast to walk(), map() blocks the local R session and waits for all tasks to complete.

Returns

Invisibly returns a list of mirai task objects for the newly created tasks. The order of tasks in the list matches the order of data in the iterate argument.

Method map()

Apply a single command to multiple inputs.

Usage

crew_class_controller_group$map(
  command,
  iterate,
  data = list(),
  globals = list(),
  substitute = TRUE,
  seed = NULL,
  algorithm = NULL,
  packages = character(0),
  library = NULL,
  seconds_interval = 0.5,
  seconds_timeout = NULL,
  names = NULL,
  save_command = FALSE,
  error = "stop",
  warnings = TRUE,
  verbose = interactive(),
  scale = TRUE,
  throttle = TRUE,
  controller = NULL
)

Arguments

Details

The idea comes from functional programming: for example, the map() function from the purrr package.

Returns

A tibble of results and metadata: one row per task and columns corresponding to the output of pop().

Method pop()

Pop a completed task from the results data frame.

Usage

crew_class_controller_group$pop(
  scale = TRUE,
  collect = NULL,
  throttle = TRUE,
  error = NULL,
  controllers = NULL
)

Arguments

Returns

If there is no task to collect, return NULL. Otherwise, return a one-row tibble with the same columns as pop() for ordinary controllers.

Method collect()

Pop all available task results and return them in a tidy tibble.

Usage

crew_class_controller_group$collect(
  scale = TRUE,
  throttle = TRUE,
  error = NULL,
  controllers = NULL
)

Arguments

Returns

A tibble of results and metadata of all resolved tasks, with one row per task. Returns NULL if there are no available results.

Method promise()

Create a promises::promise() object to asynchronously pop or collect one or more tasks.

Usage

crew_class_controller_group$promise(
  mode = "one",
  seconds_interval = 0.1,
  scale = NULL,
  throttle = NULL,
  controllers = NULL
)

Arguments

Details

Please be aware that pop() or collect() will happen asynchronously at a some unpredictable time after the promise object is created, even if your local R process appears to be doing something completely different. This behavior is highly desirable in a Shiny reactive context, but please be careful as it may be surprising in other situations.

Returns

A promises::promise() object whose eventual value will be a tibble with results from one or more popped tasks. If mode = "one", only one task is popped and returned (one row). If mode = "all", then all the tasks are returned in a tibble with one row per task (or NULL is returned if there are no tasks to pop).

Method wait()

Wait for tasks.

Usage

crew_class_controller_group$wait(
  mode = "all",
  seconds_interval = 0.5,
  seconds_timeout = Inf,
  scale = TRUE,
  throttle = TRUE,
  controllers = NULL
)

Arguments

Details

The wait() method blocks the calling R session and repeatedly auto-scales workers for tasks that need them. The function runs until it either times out or the condition in mode is met.

Returns

A logical of length 1, invisibly. TRUE if the condition in mode was met, FALSE otherwise.

Method push_backlog()

Push the name of a task to the backlog.

Usage

crew_class_controller_group$push_backlog(name, controller = NULL)

Arguments

Details

pop_backlog() pops the tasks that can be pushed without saturating the controller.

Returns

NULL (invisibly).

Method pop_backlog()

Pop the task names from the head of the backlog which can be pushed without saturating the controller.

Usage

crew_class_controller_group$pop_backlog(controllers = NULL)

Arguments

Returns

Character vector of task names which can be pushed to the controller without saturating it. If the controller is saturated, character(0L) is returned.

Method summary()

Summarize the workers of one or more controllers.

Usage

crew_class_controller_group$summary(controllers = NULL)

Arguments

Returns

A data frame of aggregated worker summary statistics of all the selected controllers. It has one row per worker, and the rows are grouped by controller. See the documentation of the summary() method of the controller class for specific information about the columns in the output.

Method pids()

Get the process IDs of the local process and the mirai dispatchers (if started).

Usage

crew_class_controller_group$pids(controllers = NULL)

Arguments

Returns

An integer vector of process IDs of the local process and the mirai dispatcher (if started).

Method terminate()

Terminate the workers and disconnect the client for one or more controllers.

Usage

crew_class_controller_group$terminate(controllers = NULL)

Arguments

Returns

NULL (invisibly).

See Also

Other controller_group: crew_controller_group()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
persistent <- crew_controller_local(name = "persistent")
transient <- crew_controller_local(
  name = "transient",
  tasks_max = 1L
)
group <- crew_controller_group(persistent, transient)
group$start()
group$push(name = "task", command = sqrt(4), controller = "transient")
group$wait()
group$pop()
group$terminate()
}

## ------------------------------------------------
## Method `crew_class_controller_group$new`
## ------------------------------------------------

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
persistent <- crew_controller_local(name = "persistent")
transient <- crew_controller_local(
  name = "transient",
  tasks_max = 1L
)
group <- crew_controller_group(persistent, transient)
group$start()
group$push(name = "task", command = sqrt(4), controller = "transient")
group$wait()
group$pop()
group$terminate()
}

Description

R6 abstract class to build other subclasses which launch and manage workers.

Active bindings

Methods

Public methods

• crew_class_launcher$new()

• crew_class_launcher$validate()

• crew_class_launcher$set_name()

• crew_class_launcher$settings()

• crew_class_launcher$call()

• crew_class_launcher$start()

• crew_class_launcher$terminate()

• crew_class_launcher$summary()

• crew_class_launcher$tally()

• crew_class_launcher$unlaunched()

• crew_class_launcher$booting()

• crew_class_launcher$active()

• crew_class_launcher$done()

• crew_class_launcher$rotate()

• crew_class_launcher$launch()

• crew_class_launcher$forward()

• crew_class_launcher$errors()

• crew_class_launcher$wait()

• crew_class_launcher$scale()

• crew_class_launcher$launch_worker()

• crew_class_launcher$crashes()

• crew_class_launcher$terminate_worker()

• crew_class_launcher$terminate_workers()

Method new()

Launcher constructor.

Usage

crew_class_launcher$new(
  name = NULL,
  seconds_interval = NULL,
  seconds_timeout = NULL,
  seconds_launch = NULL,
  seconds_idle = NULL,
  seconds_wall = NULL,
  seconds_exit = NULL,
  tasks_max = NULL,
  tasks_timers = NULL,
  reset_globals = NULL,
  reset_packages = NULL,
  reset_options = NULL,
  garbage_collection = NULL,
  crashes_error = NULL,
  launch_max = NULL,
  tls = NULL,
  processes = NULL,
  r_arguments = NULL,
  options_metrics = NULL
)

Arguments

Returns

An R6 object with the launcher.

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(workers = client$workers)
launcher$launch(index = 1L)
m <- mirai::mirai("result", .compute = client$name)
Sys.sleep(0.25)
m$data
client$terminate()
}

Method validate()

Validate the launcher.

Usage

crew_class_launcher$validate()

Returns

NULL (invisibly).

Method set_name()

Set the name of the launcher.

Usage

crew_class_launcher$set_name(name)

Arguments

Returns

NULL (invisibly).

Method settings()

List of arguments for mirai::daemon().

Usage

crew_class_launcher$settings(socket)

Arguments

Returns

List of arguments for mirai::daemon().

Method call()

Create a call to crew_worker() to help create custom launchers.

Usage

crew_class_launcher$call(socket, launcher, worker, instance)

Arguments

Returns

Character of length 1 with a call to crew_worker().

Examples

launcher <- crew_launcher_local()
launcher$call(
  socket = "ws://127.0.0.1:5000/3/cba033e58",
  launcher = "launcher_a",
  worker = 3L,
  instance = "cba033e58"
)

Method start()

Start the launcher.

Usage

crew_class_launcher$start(sockets = NULL)

Arguments

Details

Creates the workers data frame. Meant to be called once at the beginning of the launcher life cycle, after the client has started.

Returns

NULL (invisibly).

Method terminate()

Terminate the whole launcher, including all workers.

Usage

crew_class_launcher$terminate()

Returns

NULL (invisibly).

Method summary()

Summarize the workers.

Usage

crew_class_launcher$summary()

Returns

NULL if the launcher is not started. Otherwise, a tibble with one row per crew worker and the following columns:

• worker: integer index of the worker.

• launches: number of times the worker was launched. Each launch occurs at a different websocket because the token at the end of the URL is rotated before each new launch.

• online: logical vector, whether the current instance of each worker was actively connected to its NNG socket during the time of the last call to tally().

• discovered: logical vector, whether the current instance of each worker had connected to its NNG socket at some point (and then possibly disconnected) during the time of the last call to tally().

• assigned: cumulative number of tasks assigned, reported by mirai::daemons() and summed over all completed instances of the worker. Does not reflect the activity of the currently running instance of the worker.

• complete: cumulative number of tasks completed, reported by mirai::daemons() and summed over all completed instances of the worker. Does not reflect the activity of the currently running instance of the worker.

• crashes: number of consecutive times a worker launched without completing all its assigned tasks.

Method tally()

Update the daemons-related columns of the internal workers data frame.

Usage

crew_class_launcher$tally(daemons = NULL)

Arguments

Returns

NULL (invisibly).

Method unlaunched()

Get indexes of unlaunched workers.

Usage

crew_class_launcher$unlaunched(n = Inf)

Arguments

Details

A worker is "unlaunched" if it has never connected to the current instance of its websocket. Once a worker launches with the launch() method, it is considered "launched" until it disconnects and its websocket is rotated with rotate().

Returns

Integer index of workers available for launch. The backlogged workers are listed first. A worker is backlogged if it is assigned more tasks than it completed.

Method booting()

Get workers that may still be booting up.

Usage

crew_class_launcher$booting()

Details

A worker is "booting" if its launch time is within the last seconds_launch seconds. seconds_launch is a configurable grace period when crew allows a worker to start up and connect to the mirai dispatcher. The booting() function does not know about the actual worker connection status, it just knows about launch times, so it may return TRUE for workers that have already connected and started doing tasks.

Method active()

Get active workers.

Usage

crew_class_launcher$active()

Details

A worker is "active" if its current instance is online and connected, or if it is within its booting time window and has never connected. In other words, "active" means online | (!discovered & booting).

Returns

Logical vector with TRUE for active workers and FALSE for inactive ones.

Method done()

Get done workers.

Usage

crew_class_launcher$done()

Details

A worker is "done" if it is launched and inactive. A worker is "launched" if launch() was called and the worker websocket has not been rotated since.

Returns

Integer index of inactive workers.

Method rotate()

Usage

crew_class_launcher$rotate()

Details

Rotate websockets at all unlaunched workers and throw an error if a worker launched at least crashes_error times in a row without completing all its assigned tasks.

Returns

NULL (invisibly).

Method launch()

Launch a worker.

Usage

crew_class_launcher$launch(index)

Arguments

Returns

NULL (invisibly).

Method forward()

Forward an asynchronous launch/termination error condition of a worker.

Usage

crew_class_launcher$forward(index, condition = "error")

Arguments

Returns

Throw an error, throw a warning, or return a character string, depending on the condition argument.

Method errors()

Collect and return the most recent error messages from asynchronous worker launching and termination.

Usage

crew_class_launcher$errors()

Returns

Character vector of all the most recent error messages from asynchronous worker launching and termination. NULL if there are no errors.

Method wait()

Wait for any local asynchronous launch or termination tasks to complete.

Usage

crew_class_launcher$wait()

Details

Only relevant if processes is a positive integer.

Returns

NULL (invisibly).

Method scale()

Auto-scale workers out to meet the demand of tasks.

Usage

crew_class_launcher$scale(demand, throttle = TRUE)

Arguments

Returns

NULL (invisibly)

Method launch_worker()

Abstract worker launch method.

Usage

crew_class_launcher$launch_worker(call, name, launcher, worker, instance)

Arguments

Details

Launcher plugins will overwrite this method.

Returns

A handle to mock the worker launch.

Method crashes()

Return the number of consecutive times a worker launched without completing all its assigned tasks.

Usage

crew_class_launcher$crashes(index)

Arguments

Returns

Non-negative integer, number of consecutive times a worker launched without completing all its assigned tasks.

Method terminate_worker()

Abstract worker termination method.

Usage

crew_class_launcher$terminate_worker(handle)

Arguments

Details

Launcher plugins will overwrite this method.

Returns

A handle to mock worker termination.

Method terminate_workers()

Terminate one or more workers.

Usage

crew_class_launcher$terminate_workers(index = NULL)

Arguments

Returns

NULL (invisibly).

See Also

Other launcher: crew_launcher()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(workers = client$workers)
launcher$launch(index = 1L)
m <- mirai::mirai("result", .compute = client$name)
Sys.sleep(0.25)
m$data
client$terminate()
}

## ------------------------------------------------
## Method `crew_class_launcher$new`
## ------------------------------------------------

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(workers = client$workers)
launcher$launch(index = 1L)
m <- mirai::mirai("result", .compute = client$name)
Sys.sleep(0.25)
m$data
client$terminate()
}

## ------------------------------------------------
## Method `crew_class_launcher$call`
## ------------------------------------------------

launcher <- crew_launcher_local()
launcher$call(
  socket = "ws://127.0.0.1:5000/3/cba033e58",
  launcher = "launcher_a",
  worker = 3L,
  instance = "cba033e58"
)

Description

R6 class to launch and manage local process workers.

Details

See crew_launcher_local().

Super class

crew::crew_class_launcher -> crew_class_launcher_local

Active bindings

Methods

Public methods

• crew_class_launcher_local$new()

• crew_class_launcher_local$validate()

• crew_class_launcher_local$launch_worker()

• crew_class_launcher_local$terminate_worker()

Method new()

Local launcher constructor.

Usage

crew_class_launcher_local$new(
  name = NULL,
  seconds_interval = NULL,
  seconds_timeout = NULL,
  seconds_launch = NULL,
  seconds_idle = NULL,
  seconds_wall = NULL,
  seconds_exit = NULL,
  tasks_max = NULL,
  tasks_timers = NULL,
  reset_globals = NULL,
  reset_packages = NULL,
  reset_options = NULL,
  garbage_collection = NULL,
  crashes_error = NULL,
  tls = NULL,
  processes = NULL,
  r_arguments = NULL,
  options_metrics = NULL,
  options_local = NULL
)

Arguments

Returns

An R6 object with the local launcher.

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(sockets = client$summary()$socket)
launcher$launch(index = 1L)
task <- mirai::mirai("result", .compute = client$name)
mirai::call_mirai_(task)
task$data
client$terminate()
}

Method validate()

Validate the local launcher.

Usage

crew_class_launcher_local$validate()

Returns

NULL (invisibly).

Method launch_worker()

Launch a local process worker which will dial into a socket.

Usage

crew_class_launcher_local$launch_worker(call, name, launcher, worker, instance)

Arguments

Details

The call argument is R code that will run to initiate the worker. Together, the launcher, worker, and instance arguments are useful for constructing informative job names.

Returns

A handle object to allow the termination of the worker later on.

Method terminate_worker()

Terminate a local process worker.

Usage

crew_class_launcher_local$terminate_worker(handle)

Arguments

Returns

A list with the process ID of the worker.

See Also

Other plugin_local: crew_controller_local(), crew_launcher_local()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(sockets = client$summary()$socket)
launcher$launch(index = 1L)
task <- mirai::mirai("result", .compute = client$name)
mirai::call_mirai_(task)
task$data
client$terminate()
}

## ------------------------------------------------
## Method `crew_class_launcher_local$new`
## ------------------------------------------------

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(sockets = client$summary()$socket)
launcher$launch(index = 1L)
task <- mirai::mirai("result", .compute = client$name)
mirai::call_mirai_(task)
task$data
client$terminate()
}

Description

Local monitor R6 class

Details

See crew_monitor_local().

Methods

Public methods

• crew_class_monitor_local$dispatchers()

• crew_class_monitor_local$daemons()

• crew_class_monitor_local$workers()

• crew_class_monitor_local$terminate()

Method dispatchers()

List the process IDs of the running mirai dispatcher processes.

Usage

crew_class_monitor_local$dispatchers(user = ps::ps_username())

Arguments

Returns

Integer vector of process IDs of the running mirai dispatcher processes.

Method daemons()

List the process IDs of the locally running mirai daemon processes which are not crew workers. The crew_async() object can launch such processes: for example, when a positive integer is supplied to the processes argument of e.g. crew.aws.batch::crew_controller_aws_batch().

Usage

crew_class_monitor_local$daemons(user = ps::ps_username())

Arguments

Returns

Integer vector of process IDs of the locally running mirai daemon processes which are not crew workers.

Method workers()

List the process IDs of locally running crew workers launched by the local controller (crew_controller_local()).

Usage

crew_class_monitor_local$workers(user = ps::ps_username())

Arguments

Details

Only the workers running on your local computer are listed. Workers that are not listed include jobs on job schedulers like SLURM or jobs on cloud services like AWS Batch. To monitor those worker processes, please consult the monitor objects in the relevant third-party launcher plugins such as crew.cluster and crew.aws.batch.

Returns

Integer vector of process IDs of locally running crew workers launched by the local controller (crew_controller_local()).

Method terminate()

Terminate the given process IDs.

Usage

crew_class_monitor_local$terminate(pids)

Arguments

Details

Termination happens with the operating system signal given by crew_terminate_signal().

Returns

NULL (invisibly).

See Also

Other monitor: crew_monitor_local()

Description

R6 class for relay configuration.

Details

See crew_relay().

Active bindings

Methods

Public methods

• crew_class_relay$validate()

• crew_class_relay$start()

• crew_class_relay$terminate()

• crew_class_relay$set_from()

• crew_class_relay$set_to()

• crew_class_relay$wait()

Method validate()

Validate the object.

Usage

crew_class_relay$validate()

Returns

NULL (invisibly).

Method start()

Start the relay object.

Usage

crew_class_relay$start()

Returns

NULL (invisibly).

Method terminate()

Terminate the relay object.

Usage

crew_class_relay$terminate()

Returns

NULL (invisibly).

Method set_from()

Set the condition variable to relay from.

Usage

crew_class_relay$set_from(from)

Arguments

Returns

NULL (invisibly).

Method set_to()

Set the condition variable to relay to.

Usage

crew_class_relay$set_to(to)

Arguments

Returns

NULL (invisibly).

Method wait()

Wait until an unobserved task resolves or the timeout is reached.

Usage

crew_class_relay$wait(seconds_timeout = 1000)

Arguments

Returns

NULL (invisibly).

See Also

Other relay: crew_relay()

Examples

crew_relay()

Description

R6 class for throttle configuration.

Details

See crew_throttle().

Active bindings

Methods

Public methods

• crew_class_throttle$new()

• crew_class_throttle$validate()

• crew_class_throttle$poll()

• crew_class_throttle$reset()

Method new()

Throttle constructor.

Usage

crew_class_throttle$new(seconds_interval = NULL)

Arguments

Returns

An R6 object with throttle configuration.

Examples

throttle <- crew_throttle(seconds_interval = 0.5)
throttle$poll()
throttle$poll()

Method validate()

Validate the object.

Usage

crew_class_throttle$validate()

Returns

NULL (invisibly).

Method poll()

Poll the throttler.

Usage

crew_class_throttle$poll()

Returns

TRUE if poll() did not return TRUE in the last seconds_interval seconds, FALSE otherwise.

Method reset()

Reset the throttle object so the next poll() returns TRUE.

Usage

crew_class_throttle$reset()

Returns

NULL (invisibly).

See Also

Other throttle: crew_throttle()

Examples

throttle <- crew_throttle(seconds_interval = 0.5)
throttle$poll()
throttle$poll()

## ------------------------------------------------
## Method `crew_class_throttle$new`
## ------------------------------------------------

throttle <- crew_throttle(seconds_interval = 0.5)
throttle$poll()
throttle$poll()

Description

R6 class for TLS configuration.

Details

See crew_tls().

Active bindings

Methods

Public methods

• crew_class_tls$new()

• crew_class_tls$validate()

• crew_class_tls$client()

• crew_class_tls$worker()

Method new()

TLS configuration constructor.

Usage

crew_class_tls$new(
  mode = NULL,
  key = NULL,
  password = NULL,
  certificates = NULL
)

Arguments

Returns

An R6 object with TLS configuration.

Examples

crew_tls(mode = "automatic")

Method validate()

Validate the object.

Usage

crew_class_tls$validate(test = TRUE)

Arguments

Returns

NULL (invisibly).

Method client()

TLS credentials for the crew client.

Usage

crew_class_tls$client()

Returns

NULL or character vector, depending on the mode.

Method worker()

TLS credentials for crew workers.

Usage

crew_class_tls$worker(name)

Arguments

Returns

NULL or character vector, depending on the mode.

See Also

Other tls: crew_tls()

Examples

crew_tls(mode = "automatic")

## ------------------------------------------------
## Method `crew_class_tls$new`
## ------------------------------------------------

crew_tls(mode = "automatic")

Description

Terminate mirai dispatchers and/or crew workers which may be lingering from previous workloads.

Usage

crew_clean(
  dispatchers = TRUE,
  workers = TRUE,
  user = ps::ps_username(),
  seconds_interval = 0.5,
  seconds_timeout = 60,
  verbose = TRUE
)

Arguments

Details

Behind the scenes, mirai uses an external R process called a "dispatcher" to send tasks to crew workers. This dispatcher usually shuts down when you terminate the controller or quit your R session, but sometimes it lingers. Likewise, sometimes crew workers do not shut down on their own. The crew_clean() function searches the process table on your local machine and manually terminates any mirai dispatchers and crew workers associated with your user name (or the user name you select in the user argument. Unfortunately, it cannot reach remote workers such as those launched by a crew.cluster controller.

Value

NULL (invisibly). If verbose is TRUE, it does print out a message for every terminated process.

See Also

Other utility: crew_assert(), crew_deprecate(), crew_eval(), crew_random_name(), crew_retry(), crew_terminate_process(), crew_terminate_signal(), crew_worker()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
crew_clean()
}

Description

Create an R6 wrapper object to manage the mirai client.

Usage

crew_client(
  name = NULL,
  workers = 1L,
  host = NULL,
  port = NULL,
  tls = crew::crew_tls(),
  tls_enable = NULL,
  tls_config = NULL,
  seconds_interval = 0.5,
  seconds_timeout = 5,
  retry_tasks = TRUE
)

Arguments

See Also

Other client: crew_class_client

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
client$summary()
client$terminate()
}

Description

This function is for developers of crew launcher plugins. Users should use a specific controller helper such as crew_controller_local().

Usage

crew_controller(client, launcher, auto_scale = NULL)

Arguments

See Also

Other controller: crew_class_controller

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
launcher <- crew_launcher_local()
controller <- crew_controller(client = client, launcher = launcher)
controller$start()
controller$push(name = "task", command = sqrt(4))
controller$wait()
controller$pop()
controller$terminate()
}

Description

Create an R6 object to submit tasks and launch workers through multiple crew controllers.

Usage

crew_controller_group(...)

Arguments

See Also

Other controller_group: crew_class_controller_group

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
persistent <- crew_controller_local(name = "persistent")
transient <- crew_controller_local(
  name = "transient",
  tasks_max = 1L
)
group <- crew_controller_group(persistent, transient)
group$start()
group$push(name = "task", command = sqrt(4), controller = "transient")
group$wait()
group$pop()
group$terminate()
}

Description

Create an R6 object to submit tasks and launch workers on local processes.

Usage

crew_controller_local(
  name = NULL,
  workers = 1L,
  host = "127.0.0.1",
  port = NULL,
  tls = crew::crew_tls(),
  tls_enable = NULL,
  tls_config = NULL,
  seconds_interval = 0.5,
  seconds_timeout = 60,
  seconds_launch = 30,
  seconds_idle = 300,
  seconds_wall = Inf,
  seconds_exit = NULL,
  retry_tasks = TRUE,
  tasks_max = Inf,
  tasks_timers = 0L,
  reset_globals = TRUE,
  reset_packages = FALSE,
  reset_options = FALSE,
  garbage_collection = FALSE,
  crashes_error = 5L,
  launch_max = NULL,
  r_arguments = c("--no-save", "--no-restore"),
  options_metrics = crew::crew_options_metrics(),
  options_local = crew::crew_options_local(),
  local_log_directory = NULL,
  local_log_join = NULL
)

Arguments

See Also

Other plugin_local: crew_class_launcher_local, crew_launcher_local()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
controller <- crew_controller_local()
controller$start()
controller$push(name = "task", command = sqrt(4))
controller$wait()
controller$pop()
controller$terminate()
}

Description

Show an informative warning when a crew feature is deprecated.

Usage

crew_deprecate(
  name,
  date,
  version,
  alternative,
  condition = "warning",
  value = "x",
  skip_cran = FALSE,
  frequency = "always"
)

Arguments

Value

NULL (invisibly). Throws a warning if a feature is deprecated.

See Also

Other utility: crew_assert(), crew_clean(), crew_eval(), crew_random_name(), crew_retry(), crew_terminate_process(), crew_terminate_signal(), crew_worker()

Examples

suppressWarnings(
  crew_deprecate(
    name = "auto_scale",
    date = "2023-05-18",
    version = "0.2.0",
    alternative = "use the scale argument of push(), pop(), and wait()."
  )
)

Description

Not a user-side function. Do not call directly.

Usage

crew_eval(
  command,
  name = NA_character_,
  string = NA_character_,
  data = list(),
  globals = list(),
  seed = NULL,
  algorithm = NULL,
  packages = character(0),
  library = NULL
)

Arguments

Details

The crew_eval() function evaluates an R expression in an encapsulated environment and returns a monad with the results, including warnings and error messages if applicable. The random number generator seed, globals, and global options are restored to their original values on exit.

Value

A monad object with results and metadata.

See Also

Other utility: crew_assert(), crew_clean(), crew_deprecate(), crew_random_name(), crew_retry(), crew_terminate_process(), crew_terminate_signal(), crew_worker()

Examples

crew_eval(quote(1 + 1))

Description

This function is useful for inheriting argument documentation in functions that create custom third-party launchers. See ⁠@inheritParams crew::crew_launcher⁠ in the source code file of crew_launcher_local().

Usage

crew_launcher(
  name = NULL,
  seconds_interval = 0.5,
  seconds_timeout = 60,
  seconds_launch = 30,
  seconds_idle = 300,
  seconds_wall = Inf,
  seconds_exit = NULL,
  tasks_max = Inf,
  tasks_timers = 0L,
  reset_globals = TRUE,
  reset_packages = FALSE,
  reset_options = FALSE,
  garbage_collection = FALSE,
  crashes_error = 5L,
  launch_max = NULL,
  tls = crew::crew_tls(),
  processes = NULL,
  r_arguments = c("--no-save", "--no-restore"),
  options_metrics = crew::crew_options_metrics()
)

Arguments

See Also

Other launcher: crew_class_launcher

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(sockets = client$summary()$socket)
launcher$launch(index = 1L)
task <- mirai::mirai("result", .compute = client$name)
mirai::call_mirai_(task)
task$data
client$terminate()
}

Description

Create an R6 object to launch and maintain local process workers.

Usage

crew_launcher_local(
  name = NULL,
  seconds_interval = 0.5,
  seconds_timeout = 60,
  seconds_launch = 30,
  seconds_idle = Inf,
  seconds_wall = Inf,
  seconds_exit = NULL,
  tasks_max = Inf,
  tasks_timers = 0L,
  reset_globals = TRUE,
  reset_packages = FALSE,
  reset_options = FALSE,
  garbage_collection = FALSE,
  crashes_error = 5L,
  launch_max = NULL,
  tls = crew::crew_tls(),
  r_arguments = c("--no-save", "--no-restore"),
  options_metrics = crew::crew_options_metrics(),
  options_local = crew::crew_options_local(),
  local_log_directory = NULL,
  local_log_join = NULL
)

Arguments

See Also

Other plugin_local: crew_class_launcher_local, crew_controller_local()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
client <- crew_client()
client$start()
launcher <- crew_launcher_local(name = client$name)
launcher$start(sockets = client$summary()$socket)
launcher$launch(index = 1L)
task <- mirai::mirai("result", .compute = client$name)
mirai::call_mirai_(task)
task$data
client$terminate()
}

Description

Create an R6 object to monitor local processes created by crew and mirai.

Usage

crew_monitor_local()

See Also

Other monitor: crew_class_monitor_local

Description

Options for the local crew launcher.

Usage

crew_options_local(log_directory = NULL, log_join = TRUE)

Arguments

Value

A classed list of options for the local launcher.

See Also

Other options: crew_options_metrics()

Examples

crew_options_local()

Description

crew_options_metrics() configures the crew to record resource usage metrics (such as CPU and memory usage) for each running worker. To be activate resource usage logging, the autometric R package version 0.1.0 or higher must be installed.

Logging happens in the background (through a detached POSIX) so as not to disrupt the R session. On Unix-like systems, crew_options_metrics() can specify ⁠/dev/stdout⁠ or ⁠/dev/stderr⁠ as the log files, which will redirect output to existing logs you are already using. autometric::log_read() and autometric::log_plot() can read and visualize resource usage data from multiple log files, even if those files are mixed with other messages.

Usage

crew_options_metrics(path = NULL, seconds_interval = 5)

Arguments

Value

A classed list of options for logging resource usage metrics.

See Also

Other options: crew_options_local()

Examples

crew_options_metrics()

Description

Generate a random string that can be used as a name for a worker or task.

Usage

crew_random_name(n = 12L)

Arguments

Details

The randomness is not reproducible and cannot be set with e.g. set.seed() in R.

Value

A random character string.

See Also

Other utility: crew_assert(), crew_clean(), crew_deprecate(), crew_eval(), crew_retry(), crew_terminate_process(), crew_terminate_signal(), crew_worker()

Examples

crew_random_name()

Description

Create an R6 crew relay object.

Usage

crew_relay()

Details

A crew relay object keeps the signaling relationships among condition variables.

Value

An R6 crew relay object.

See Also

Other relay: crew_class_relay

Examples

crew_relay()

Description

Repeatedly retry a function while it keeps returning FALSE and exit the loop when it returns TRUE

Usage

crew_retry(
  fun,
  args = list(),
  seconds_interval = 1,
  seconds_timeout = 60,
  max_tries = Inf,
  error = TRUE,
  message = character(0),
  envir = parent.frame()
)

Arguments

Value

NULL (invisibly).

See Also

Other utility: crew_assert(), crew_clean(), crew_deprecate(), crew_eval(), crew_random_name(), crew_terminate_process(), crew_terminate_signal(), crew_worker()

Examples

crew_retry(fun = function() TRUE)

Description

Manually terminate a local process.

Usage

crew_terminate_process(pid)

Arguments

Value

NULL (invisibly).

See Also

Other utility: crew_assert(), crew_clean(), crew_deprecate(), crew_eval(), crew_random_name(), crew_retry(), crew_terminate_signal(), crew_worker()

Examples

if (identical(Sys.getenv("CREW_EXAMPLES"), "true")) {
process <- processx::process$new("sleep", "60")
process$is_alive()
crew_terminate_process(pid = process$get_pid())
process$is_alive()
}

Description

Get a supported operating system signal for terminating a local process.

Usage

crew_terminate_signal()

Value

An integer of length 1: tools::SIGTERM if your platform supports SIGTERM. If not, then crew_crew_terminate_signal()() checks SIGQUIT, then SIGINT, then SIGKILL, and then returns the first signal it finds that your operating system can use.

See Also

Other utility: crew_assert(), crew_clean(), crew_deprecate(), crew_eval(), crew_random_name(), crew_retry(), crew_terminate_process(), crew_worker()

Examples

crew_terminate_signal()

Description

Create an R6 object for throttling.

Usage

crew_throttle(seconds_interval = 0.5)

Arguments

Details

Throttling is a technique that limits how often a function is called in a given period of time. crew_throttle() objects support the throttle argument of controller methods, which ensures auto-scaling only happen every seconds_interval seconds. This helps avoid overburdening the mirai dispatcher and other resources.

Value

An R6 object with throttle configuration settings and methods.

See Also

Other throttle: crew_class_throttle

Examples

throttle <- crew_throttle(seconds_interval = 0.5)
throttle$poll()
throttle$poll()

Description

Create an R6 object with transport layer security (TLS) configuration for crew.

Usage

crew_tls(
  mode = "none",
  key = NULL,
  password = NULL,
  certificates = NULL,
  validate = TRUE
)

Arguments

Details

crew_tls() objects are input to the tls argument of crew_client(), crew_controller_local(), etc. See https://wlandau.github.io/crew/articles/risks.html for details.

Value

An R6 object with TLS configuration settings and methods.

See Also

Other tls: crew_class_tls

Examples

crew_tls(mode = "automatic")

Description

Launches a crew worker which runs a mirai daemon. Not a user-side function. Users should not call crew_worker() directly. See launcher plugins like crew_launcher_local() for examples.

Usage

crew_worker(
  settings,
  launcher,
  worker,
  instance,
  options_metrics = crew::crew_options_metrics()
)

Arguments

Value

NULL (invisibly)

See Also

Other utility: crew_assert(), crew_clean(), crew_deprecate(), crew_eval(), crew_random_name(), crew_retry(), crew_terminate_process(), crew_terminate_signal()