Package 'rrq'

Description

When you create a task with rrq and that task uses local variables these need to be copied over to the worker that will evaluate the task. So, if we had

rrq_task_create_expr(f(a, b))

that would be the objects a and b from the context where rrq_task_create_expr was called. There are a few considerations here:

• The names a and b are only useful in the immediate context of the controller at the point the task is sent and so we need to store the values referenced by a and b without reference to the names - we do this by naming the new values after their value. That is, the name becomes the hash of the object, computed by rlang::hash(), as a form of content-addressable storage.

• When doing this we note that we might end up using the value referenced by a or b many times in different tasks so we should not re-save the data more than needed, and we should not necessarily delete it when a task is deleted unless nothing else uses that value.

• The objects might tiny or could be large; if small we tend to care about how quickly they can be resolved (i.e., latency) and if large we need to be careful not to overfull Redis' database as it's a memory-based system.

To make this robust and flexible, we use a object_store object, which will allow objects to be stored either directly in Redis, or offloaded onto some "large" data store based on their size. Currently, we provide support only for offloading to disk, but in future hope to expand this.

When we create a value in the store (or reference a value that already exists) we assign a tag into the database; this means that we have for a value with hash abc123 and tag def789

• ⁠prefix:data["abc123"] => [1] f5 26 a5 b7 26 93 b3 41 b7 d0 b0...⁠ (the data stored, serialised into a redis hash by its hash, as a binary object.

• ⁠prefix:tag_hash:def789 => {abc123}⁠ (a set of hashes used by our tag)

• ⁠prefix:hash_tag:abc123 => {def789}⁠ (a set of tags that reference our hash)

If we also used the value with hash abc123 with tag fed987 this would look like

• ⁠prefix:data[abc123] => [1] f5 26 a5 b7 26 93 b3 41 b7 d0 b0...⁠ hash, as a binary object.

• ⁠prefix:tag_hash:def789 => {abc123}⁠

• ⁠prefix:tag_hash:fed987 => {abc123}⁠

• ⁠prefix:hash_tag:abc123 => {def789, fed987}⁠

As tags are dropped, then the references are dropped from the set prefix:hash_tag:abc123 and when that set becomes empty then we can delete prefix:data[abc123] as simple form of reference counting.

For rrq we will use task_ids as a tag.

For dealing with large data, we "offload" large data into a secondary store. This replaces the redis hash of ⁠hash => value⁠ with something else. Currently the only alternative we offer is object_store_offload_disk which will save the binary representation of the object at the path ⁠<path>/<hash>⁠ and will allow large values to be shared between controller and worker so long as they share a common filesystem.

Details

Create an object store. Typically this is not used by end-users, and is used internally by rrq_controller

Methods

Public methods

• object_store$new()

• object_store$list()

• object_store$tags()

• object_store$get()

• object_store$mget()

• object_store$set()

• object_store$mset()

• object_store$location()

• object_store$drop()

• object_store$destroy()

Method new()

Create a new object store (or connect to an existing one)

Usage

object_store$new(con, prefix, max_size = Inf, offload = NULL)

Arguments

Method list()

List all hashes of data known to this data store

Usage

object_store$list()

Method tags()

List all tags known to this data store

Usage

object_store$tags()

Method get()

Get a single object by its hash

Usage

object_store$get(hash)

Arguments

Method mget()

Get a number objects by their hashes. Unlike ⁠$get()⁠ this method accepts a vector of hash (length 0, 1, or more than 1) and returns a list of the same length.

Usage

object_store$mget(hash)

Arguments

Method set()

Set an object into the object store, returning the hash of that object.

Usage

object_store$set(value, tag, serialize = TRUE)

Arguments

Method mset()

Set a number of objects into the store. Unlike ⁠$set()⁠, this method sets a list of objects into the store at once, and returns a character vector of hashes the same length as the list of values.

Usage

object_store$mset(value, tag, serialize = TRUE)

Arguments

Method location()

Return the storage locations of a set of hashes. Currently the location may be redis (stored directly in the redis server), offload (stored in the offload storage) or NA (if not found, and if error = FALSE).

Usage

object_store$location(hash, error = TRUE)

Arguments

Method drop()

Delete tags from the store. This will dissociate the tags from any hashes they references and if that means that no tag points to a hash then the data at that hash will be removed. We return (invisibly) a character vector of any dropped hashes.

Usage

object_store$drop(tag)

Arguments

Method destroy()

Remove all data from the store, and all the stores metadata

Usage

object_store$destroy()

Description

Disk-based offload

Disk-based offload

Details

A disk-based offload for object_store. This is not intended at all for direct user-use.

Methods

Public methods

• object_store_offload_disk$new()

• object_store_offload_disk$mset()

• object_store_offload_disk$mget()

• object_store_offload_disk$mdel()

• object_store_offload_disk$list()

• object_store_offload_disk$destroy()

Method new()

Create the store

Usage

object_store_offload_disk$new(path)

Arguments

Method mset()

Save a number of values to disk

Usage

object_store_offload_disk$mset(hash, value)

Arguments

Method mget()

Retrieve a number of objects from the store

Usage

object_store_offload_disk$mget(hash)

Arguments

Method mdel()

Delete a number of objects from the store

Usage

object_store_offload_disk$mdel(hash)

Arguments

Method list()

List hashes stored in this offload store

Usage

object_store_offload_disk$list()

Method destroy()

Completely delete the store (by deleting the directory)

Usage

object_store_offload_disk$destroy()

Description

Configure rrq options. This function must be called before either a controller or worker connects to a queue, as the options will apply to both. The function may only be called once on a given queue as there is no facility (yet) to update options. Currently the options concern only storage, and specifically how larger objects will be saved (using object_store.

Usage

rrq_configure(
  queue_id,
  con = redux::hiredis(),
  ...,
  store_max_size = Inf,
  offload_path = NULL
)

Arguments

Value

Invisibly, a list with processed configuration information

Storage

Every time that a task is saved, or a task is completed, results are saved into the Redis database. Because Redis is an in-memory database, it's not a great idea to save very large objects into it (if you ran 100 jobs in parallel and each saved a 2GB object you'd likely take down your redis server). In addition, redux does not support directly saving objects larger than 2^31 - 1 bytes into Redis. So, for some use cases we need to consider where to store larger objects.

The strategy here is to "offload" the larger objects - bigger than some user-given size - onto some other storage system. Currently the only alternative supported is a disk store (object_store_offload_disk) but we hope to expand this later. So if your task returns a 3GB object then we will spill that to disk rather than failing to save that into Redis.

How big is an object? We serialise the object (redux::object_to_bin just wraps serialize) which creates a vector of bytes and that is saved into the database. To get an idea of how large things are you can do: length(redux::object_to_bin(your_object)). At the time this documentation was written, mtcars was 3807 bytes, and a million random numbers was ⁠8,000,031⁠ bytes. It's unlikely that a store_max_size of less than 1MB will be sensible.

Examples

tmp <- tempfile()
dir.create(tmp)
rrq_configure("rrq:offload", store_max_size = 100000, offload_path = tmp)
obj <- rrq_controller("rrq:offload")
x <- runif(100000)
t <- rrq_task_create_expr(mean(x), controller = obj)
dir(tmp)
file.size(dir(tmp, full.names = TRUE))
rrq_destroy(controller = obj)

Description

Create a new controller. This is the new interface that will replace rrq_controller soon, at which point it will rename back to rrq_controller.

Usage

rrq_controller(
  queue_id,
  con = redux::hiredis(),
  timeout_task_wait = NULL,
  follow = NULL,
  check_version = TRUE
)

Arguments

Value

An rrq_controller object, which is opaque.

Task lifecycle

• A task is queued with ⁠$enqueue()⁠, at which point it becomes PENDING

• Once a worker selects the task to run, it becomes RUNNING

• If the task completes successfully without error it becomes COMPLETE

• If the task throws an error, it becomes ERROR

• If the task was cancelled (e.g., via ⁠$task_cancel()⁠) it becomes CANCELLED

• If the task is killed by an external process, crashes or the worker dies (and is running a heartbeat) then the task becomes DIED.

• The status of an unknown task is MISSING

• Tasks in any terminal state (except IMPOSSIBLE) may be retried with task_retry at which point they become MOVED, see vignette("fault-tolerance") for details

Worker lifecycle

• A worker appears and is IDLE

• When running a task it is BUSY

• If it receives a PAUSE message it becomes PAUSED until it receives a RESUME message

• If it exits cleanly (e.g., via a STOP message or a timeout) it becomes EXITED

• If it crashes and was running a heartbeat, it becomes LOST

Messages

Most of the time workers process tasks, but you can also send them "messages". Messages take priority over tasks, so if a worker becomes idle (by coming online or by finishing a task) it will consume all available messages before starting on a new task, even if both are available.

Each message has a "command" and may have "arguments" to that command. The supported messages are:

• PING (no args): "ping" the worker, if alive it will respond with "PONG"

• ECHO (accepts an argument of a string): Print a string to the terminal and log of the worker. Will respond with OK once the message has been printed.

• EVAL (accepts a string or a quoted expression): Evaluate an arbitrary R expression on the worker. Responds with the value of this expression.

• STOP (accepts a string to print as the worker exits, defaults to "BYE"): Tells the worker to stop.

• INFO (no args): Returns information about the worker (versions of packages, hostname, pid, etc).

• PAUSE (no args): Tells the worker to stop accepting tasks (until it receives a RESUME message). Messages are processed as normal.

• RESUME (no args): Tells a paused worker to resume accepting tasks.

• REFRESH (no args): Tells the worker to rebuild their environment with the create method.

• TIMEOUT_SET (accepts a number, representing seconds): Updates the worker timeout - the length of time after which it will exit if it has not processed a task.

• TIMEOUT_GET (no args): Tells the worker to respond with its current timeout.

Examples

# Create a new controller; the id will be specific to your
# application.  Here, we use 'rrq:example'
obj <- rrq_controller("rrq:example")

# Create a task for this controller to work on:
t <- rrq_task_create_expr(runif(10), controller = obj)

# Wait for the task to complete
rrq_task_wait(t, controller = obj)

# Fetch the task's result
rrq_task_result(t, controller = obj)

Description

Set or clear a default controller for use with rrq functions. You will want to use this to avoid passing controller in as a named argument to every function.

Usage

rrq_default_controller_set(controller)

rrq_default_controller_clear()

Arguments

Value

Invisibly, the previously set default controller (or NULL if none was set)

Description

Return deferred tasks and what they are waiting on. Note this is in an arbitrary order, tasks will be added to the queue as their dependencies are satisfied.

Usage

rrq_deferred_list(controller = NULL)

Arguments

Description

Entirely destroy a queue, by deleting all keys associated with it from the Redis database. This is a very destructive action and cannot be undone.

Usage

rrq_destroy(
  delete = TRUE,
  worker_stop_type = "message",
  timeout_worker_stop = 0,
  controller = NULL
)

Arguments

Description

Helper function for creating a worker environment. This function exists to create a function suitable for passing to rrq_worker_envir_set for the common case where the worker should source some R scripts and/or load some packages on startup. This is a convenience wrapper around defining your own function, covering the most simple case. If you need more flexibility you should write your own function.

Usage

rrq_envir(packages = NULL, sources = NULL)

Arguments

Value

A function suitable for passing to rrq_worker_envir_set, which can set (or update) the environment for your workers.

Description

Create a heartbeat instance

Create a heartbeat instance

Details

Create a heartbeat instance. This can be used by running obj$start() which will reset the TTL (Time To Live) on key every period seconds (don't set this too high). If the R process dies, then the key will expire after 3 * period seconds (or set expire) and another application can tell that this R instance has died.

Methods

Public methods

• rrq_heartbeat$new()

• rrq_heartbeat$is_running()

• rrq_heartbeat$start()

• rrq_heartbeat$stop()

• rrq_heartbeat$format()

Method new()

Create a heartbeat object

Usage

rrq_heartbeat$new(
  key,
  period,
  expire = 3 * period,
  value = expire,
  config = NULL,
  start = TRUE,
  timeout = 10
)

Arguments

Method is_running()

Report if heartbeat process is running. This will be TRUE if the process has been started and has not stopped.

Usage

rrq_heartbeat$is_running()

Method start()

Start the heartbeat process. An error will be thrown if it is already running.

Usage

rrq_heartbeat$start()

Method stop()

Stop the heartbeat process

Usage

rrq_heartbeat$stop(wait = TRUE)

Arguments

Method format()

Format method, used by R6 to nicely print the object

Usage

rrq_heartbeat$format(...)

Arguments

Examples

if (redux::redis_available()) {
  rand_str <- function() {
    paste(sample(letters, 20, TRUE), collapse = "")
  }
  key <- sprintf("rrq:heartbeat:%s", rand_str())
  h <- rrq::rrq_heartbeat$new(key, 1, expire = 2)
  con <- redux::hiredis()

  # The heartbeat key exists
  con$EXISTS(key)

  # And has an expiry of less than 2000ms
  con$PTTL(key)

  # We can manually stop the heartbeat, and 2s later the key will
  # stop existing
  h$stop()

  Sys.sleep(2)
  con$EXISTS(key) # 0

  # This is required to close any processes opened by this
  # example, normally you would not need this.
  processx:::supervisor_kill()
}

Description

Send a kill signal (typically SIGTERM) to terminate a process that is running a heartbeat. This is used by rrq_controller in order to tear down workers, even if they are processing a task. When a heartbeat process is created, in its main loop it will listen for requests to kill via this function and will forward them to the worker. This is primarily useful where workers are on a different physical machine to the controller where tools::pskill() cannot be used.

Usage

rrq_heartbeat_kill(con, key, signal = tools::SIGTERM)

Arguments

Examples

if (redux::redis_available()) {
  rand_str <- function() {
    paste(sample(letters, 20, TRUE), collapse = "")
  }
  # Suppose we have a process that exposes a heartbeat running on
  # this key:
  key <- sprintf("rrq:heartbeat:%s", rand_str())

  # We can send it a SIGTERM signal over redis using:
  con <- redux::hiredis()
  rrq::rrq_heartbeat_kill(con, key, tools::SIGTERM)
}

Description

Get response to messages, waiting until the message has been responded to.

Usage

rrq_message_get_response(
  message_id,
  worker_ids = NULL,
  named = TRUE,
  delete = FALSE,
  timeout = 0,
  time_poll = 0.5,
  progress = NULL,
  controller = NULL
)

Arguments

Examples

obj <- rrq_controller("rrq:example")
id <- rrq_message_send("PING", controller = obj)
rrq_message_get_response(id, timeout = 5, controller = obj)

Description

Detect if a response is available for a message

Usage

rrq_message_has_response(
  message_id,
  worker_ids = NULL,
  named = TRUE,
  controller = NULL
)

Arguments

Value

A logical vector, possibly named (depending on the named argument)

Examples

obj <- rrq_controller("rrq:example")

id <- rrq_message_send("PING", controller = obj)
rrq_message_has_response(id, controller = obj)
rrq_message_get_response(id, timeout = 5, controller = obj)
rrq_message_has_response(id, controller = obj)

Description

Return ids for messages with responses for a particular worker.

Usage

rrq_message_response_ids(worker_id, controller = NULL)

Arguments

Value

A character vector of ids

Examples

obj <- rrq_controller("rrq:example")
w <- rrq_worker_list(controller = obj)
rrq_message_send("PING", controller = obj)

Description

Send a message to workers. Sending a message returns a message id, which can be used to poll for a response with the other ⁠rrq_message_*⁠ functions. See vignette("messages") for details for the messaging interface.

Usage

rrq_message_send(command, args = NULL, worker_ids = NULL, controller = NULL)

Arguments

Value

Invisibly, a single identifier

Examples

obj <- rrq_controller("rrq:example")

id <- rrq_message_send("PING", controller = obj)
rrq_message_get_response(id, timeout = 5, controller = obj)

Description

Send a message and wait for responses. This is a helper function around rrq_message_send() and rrq_message_get_response().

Usage

rrq_message_send_and_wait(
  command,
  args = NULL,
  worker_ids = NULL,
  named = TRUE,
  delete = TRUE,
  timeout = 600,
  time_poll = 0.05,
  progress = NULL,
  controller = NULL
)

Arguments

Value

The message response

Examples

obj <- rrq_controller("rrq:example")
rrq_message_send_and_wait("PING", controller = obj)

Description

Returns the length of the queue (the number of tasks waiting to run). This is the same as the length of the value returned by rrq_queue_list.

Usage

rrq_queue_length(queue = NULL, controller = NULL)

Arguments

Value

A number

Description

Returns the keys in the task queue.

Usage

rrq_queue_list(queue = NULL, controller = NULL)

Arguments

Description

Remove task ids from a queue.

Usage

rrq_queue_remove(task_ids, queue = NULL, controller = NULL)

Arguments

Description

Cancel a single task. If the task is PENDING it will be unqueued and the status set to CANCELED. If RUNNING then the task will be stopped if it was set to run in a separate process (i.e., queued with separate_process = TRUE). Dependent tasks will be marked as impossible.

Usage

rrq_task_cancel(task_id, wait = TRUE, timeout_wait = 10, controller = NULL)

Arguments

Value

Nothing if successfully cancelled, otherwise throws an error with task_id and status e.g. Task 123 is not running (MISSING)

Examples

obj <- rrq_controller("rrq:example")

t <- rrq_task_create_expr(Sys.sleep(4), separate_process = TRUE,
                          controller = obj)
Sys.sleep(0.5)
rrq_task_cancel(t, controller = obj)
rrq_task_status(t, controller = obj)

Description

Create a bulk set of tasks based on applying a function over a vector or data.frame. This is the bulk equivalent of rrq_task_create_call, in the same way that rrq_task_create_bulk_expr is a bulk version of rrq_task_create_expr.

Usage

rrq_task_create_bulk_call(
  fn,
  data,
  args = NULL,
  queue = NULL,
  separate_process = FALSE,
  timeout_task_run = NULL,
  depends_on = NULL,
  controller = NULL
)

Arguments

Value

A vector of task identifiers; this will have the length as data has rows if it is a data.frame, otherwise it has the same length as data

Examples

obj <- rrq_controller("rrq:example")

d <- data.frame(n = 1:10, lambda = rgamma(10, 5))
ts <- rrq_task_create_bulk_call(rpois, d, controller = obj)
rrq_task_wait(ts, controller = obj)
rrq_task_results(ts, controller = obj)

Description

Create a bulk set of tasks. Variables in data take precedence over variables in the environment in which expr was created. There is no "pronoun" support yet (see rlang docs). Use ⁠!!⁠ to pull a variable from the environment if you need to, but be careful not to inject something really large (e.g., any vector really) or you'll end up with a revolting expression and poor backtraces.

Usage

rrq_task_create_bulk_expr(
  expr,
  data,
  queue = NULL,
  separate_process = FALSE,
  timeout_task_run = NULL,
  depends_on = NULL,
  controller = NULL
)

Arguments

Value

A character vector with task identifiers; this will have a length equal to the number of row in data

Examples

obj <- rrq_controller("rrq:example")

# Create 10 tasks:
ts <- rrq_task_create_bulk_expr(sqrt(x), data.frame(x = 1:10),
                                controller = obj)
rrq_task_wait(ts, controller = obj)
rrq_task_results(ts, controller = obj)

# Note that there is no automatic simplification when fetching
# results, you might use unlist or vapply to turn this into a
# numeric vector rather than a list

# The data.frame substituted in may have multiple columns
# representing multiple variables to substitute into the
# expression
d <- expand.grid(a = 1:4, b = 1:4)
ts <- rrq_task_create_bulk_expr(a * b, d, controller = obj)
rrq_task_wait(ts, controller = obj)
rrq_task_results(ts, controller = obj)

Description

Create a task based on a function call. This is fairly similar to callr::r, and forms the basis of lapply()-like task submission. Sending a call may have slightly different semantics than you expect if you send a closure (a function that binds data), and we may change behaviour here until we find a happy set of compromises. See Details for more on this. The expression rrq_task_create_call(f, list(a, b, c)) is similar to rrq_task_create_expr(f(a, b, c)), use whichever you prefer.

Usage

rrq_task_create_call(
  fn,
  args,
  queue = NULL,
  separate_process = FALSE,
  timeout_task_run = NULL,
  depends_on = NULL,
  controller = NULL
)

Arguments

Details

Things are pretty unambiguous when you pass in a function from a package, especially when you refer to that package with its namespace (e.g. pkg::fn).

If you pass in the name without a namespace from a package that you have loaded with library() locally but you have not loaded with library within your worker environment, we may not do the right thing and you may see your task fail, or find a different function with the same name.

If you pass in an anonymous function (e.g., function(x) x + 1) we may or may not do the right thing with respect to environment capture. We never capture the global environment so if your function is a closure that tries to bind a symbol from the global environment it will not work. Like with callr::r, anonymous functions will be easiest to think about where they are fully self contained (i.e., all inputs to the functions come through args). If you have bound a local environment, we may do slightly better, but semantics here are undefined and subject to change.

R does some fancy things with function calls that we don't try to replicate. In particular you may have noticed that this works:

c <- "x"
c(c, c) # a vector of two "x"'s

You can end up in this situation locally with:

f <- function(x) x + 1
local({
  f <- 1
  f(f) # 2
})

this is because when R looks for the symbol for the call it skips over non-function objects. We don't reconstruct environment chains in exactly the same way as you would have locally so this is not possible.

Value

A task identifier (a 32 character hex string) that you can pass in to other rrq functions, notably rrq_task_status() and rrq_task_result()

Examples

obj <- rrq_controller("rrq:example")
t <- rrq_task_create_call(sqrt, list(2), controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)

Description

Create a task based on an expression. The expression passed as expr will typically be a function call (e.g., f(x)). We will analyse the expression and find all variables that you reference (in the case of f(x) this is x) and combine this with the function name to run on the worker. If x cannot be found in your calling environment we will error.

Usage

rrq_task_create_expr(
  expr,
  queue = NULL,
  separate_process = FALSE,
  timeout_task_run = NULL,
  depends_on = NULL,
  controller = NULL
)

Arguments

Details

Alternatively you may provide a multiline statement by using {} to surround multiple lines, such as:

task_create_expr({
  x <- runif(1)
  f(x)
}, ...)

in this case, we apply a simple heuristic to work out that x is locally assigned and should not be saved with the expression.

See Also

rrq_task_create_call for creating a task from a function and arguments to the function, and rrq_task_create_bulk_expr for creating many tasks from a call and a data.frame

Examples

obj <- rrq_controller("rrq:example")

# Simple use of the function to create a task based on a function call
t <- rrq_task_create_expr(sqrt(2), controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)

# The expression can contain calls to other variables, and these
# will be included in the call:
a <- 3
t <- rrq_task_create_expr(sqrt(a), controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)

# You can pass in an expression _as_ a symbol too:
expr <- quote(sqrt(4))
t <- rrq_task_create_expr(expr, controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)

# If you queue tasks into separate processes you can use a timeout
# to kill the task if it takes too long:
t <- rrq_task_create_expr(Sys.sleep(3),
                          separate_process = TRUE,
                          timeout_task_run = 1,
                          controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)

Description

Fetch internal data about a task (expert use only)

Usage

rrq_task_data(task_id, controller = NULL)

Arguments

Value

Internal data, structures subject to change

Examples

obj <- rrq_controller("rrq:example")

t <- rrq_task_create_expr(runif(1), controller = obj)
rrq_task_data(t, controller = obj)

x <- 10
y <- 20
t <- rrq_task_create_expr(x + y, controller = obj)
rrq_task_data(t, controller = obj)

Description

Delete one or more tasks

Usage

rrq_task_delete(task_ids, check = TRUE, controller = NULL)

Arguments

Value

Nothing, called for side effects only

Examples

obj <- rrq_controller("rrq:example:delete")

ts <- rrq_task_create_bulk_call(sqrt, 1:10, controller = obj)
rrq_task_exists(ts, controller = obj)

rrq_task_delete(ts[1:5], controller = obj)
rrq_task_exists(ts, controller = obj)

rrq_task_delete(ts, controller = obj)
rrq_task_exists(ts, controller = obj)

Description

Test if task ids exist (i.e., are known to this controller). Nonexistent tasks may be deleted, known to a different controller or just never have existed.

Usage

rrq_task_exists(task_ids, named = FALSE, controller = NULL)

Arguments

Value

A logical vector the same length as task_ids; TRUE where the task exists, FALSE otherwise. If named was TRUE, then this vector is named with task_ids.

Examples

obj <- rrq_controller("rrq:example")

t1 <- rrq_task_create_expr(runif(1), controller = obj)
rrq_task_exists(t1, controller = obj)

t2 <- ids::random_id()
rrq_task_exists(t2, controller = obj)

Description

Fetch information about a task. This currently includes information about where a task is (or was) running and information about any retry chain, but will expand in future. The format of the output here is subject to change (and will probably get a nice print method) but the values present in the output will be included in any future update.

Usage

rrq_task_info(task_id, controller = NULL)

Arguments

Value

A list, format currently subject to change

Examples

obj <- rrq_controller("rrq:example")

# Get information about a task
t <- rrq_task_create_expr(runif(1), controller = obj)
rrq_task_info(t, controller = obj)

# If the task has been retried, the retry chain is shown
rrq_task_wait(t, controller = obj)
rrq_task_retry(t, controller = obj)
rrq_task_info(t, controller = obj)

# If the task was queued onto a separate process, then this
# information is shown
rrq_task_create_expr(1 + 1, separate_process = TRUE, timeout_task_run = 60,
                      controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_info(t, controller = obj)

Description

List all tasks. This may be a lot of tasks, and so can be quite slow to execute.

Usage

rrq_task_list(controller = NULL)

Arguments

Value

A character vector

Examples

obj <- rrq_controller("rrq:example")

rrq_task_list(controller = obj)

Description

Fetch logs from tasks that were queued into separate processes (e.g., with rrq_task_create_expr using separate_process = TRUE). It is not knowable if a task definitely produce logs - if you have a mixture of workers that do enable task logs and some that don't, then it will depend on the worker that picks it up if logging will be enabled. Don't do this though and you should be fine.

Usage

rrq_task_log(task_id, controller = NULL)

Arguments

Value

A character vector of logs, or NULL if no log is present yet. If logging is not enabled for this task, we throw an error. Empty logs can be distinguished from "no logs yet", as they will return an empty character vector (character(0)).

Examples

obj <- rrq_controller("rrq:example")

t <- rrq_task_create_expr(message("hello!"), separate_process = TRUE,
                          controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_log(t, controller = obj)

Description

Provide a high level overview of task statuses for a set of task ids, being the count in major categories of PENDING, RUNNING, COMPLETE, ERROR, CANCELLED, DIED, TIMEOUT, IMPOSSIBLE, DEFERRED and MOVED.

Usage

rrq_task_overview(task_ids = NULL, controller = NULL)

Arguments

Value

A list with names corresponding to possible task status levels and values being the number of tasks in that state.

Examples

obj <- rrq_controller("rrq:example")

ids <- rrq_task_list(controller = obj)
t(as.data.frame(rrq_task_overview(ids, controller = obj)))

Description

Find the position of one or more tasks in the queue.

Usage

rrq_task_position(
  task_ids,
  missing = 0L,
  queue = NULL,
  follow = NULL,
  controller = NULL
)

Arguments

Value

An integer vector, the same length as task_ids

Description

List the tasks in front of task_id in the queue. If the task is missing from the queue this will return NULL. If the task is next in the queue this will return an empty character vector.

Usage

rrq_task_preceeding(task_id, queue = NULL, follow = NULL, controller = NULL)

Arguments

Description

Retrieve task progress, if set. This will be NULL if progress has never been registered, otherwise whatever value was set - can be an arbitrary R object.

Usage

rrq_task_progress(task_id, controller = NULL)

Arguments

Value

Any set progress object

Description

Post a task progress update. The progress system in rrq is agnostic about how you are going to render your progress, and so it just a convention - see Details below. Any R object can be sent as a progress value (e.g., a string, a list, etc).

Usage

rrq_task_progress_update(value, error = FALSE)

Arguments

Details

In order to report on progress, a task may, in it's code, write

rrq::rrq_task_progress_update("task is 90% done")

and this information will be fetchable by calling rrq_task_progress with the task_id.

It is also possible to register progress without acquiring a dependency on rrq. If your package/script includes code like:

progress <- function(message) {
  signalCondition(structure(list(message = message),
                            class = c("progress", "condition")))
}

(this function can be called anything - the important bit is the body function body - you must return an object with a message element and the two class attributes progress and condition).

then you can use this in the same way as rrq::rrq_task_progress_update above in your code. When run without using rrq, this function will appear to do nothing.

Examples

obj <- rrq_controller("rrq:example")

f <- function(n) {
  for (i in seq_len(n)) {
    rrq::rrq_task_progress_update(sprintf("Iteration %d / %d", i, n))
    Sys.sleep(0.1)
  }
  n
}

t <- rrq_task_create_call(f, list(5), controller = obj)
# This might be empty at first
rrq_task_progress(t, controller = obj)
# Wait for the task to complete
rrq_task_wait(t, controller = obj)
# Contains the _last_ progress message
rrq_task_progress(t, controller = obj)

Description

Get the result for a single task (see rrq_task_results for a method for efficiently getting multiple results at once). Returns the value of running the task if it is complete, and an error otherwise.

Usage

rrq_task_result(task_id, error = FALSE, follow = NULL, controller = NULL)

Arguments

Value

The result of your task. This may be an error (an object with class rrq_task_error) if your task has failed.

Examples

obj <- rrq_controller("rrq:example")

# Create a task, wait for it to finish and fetch its result
t <- rrq_task_create_expr(runif(1), controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)

# Tasks that fail do not fail on result, but instead return an
# object with the class "rrq_task_error"
t <- rrq_task_create_expr(readRDS("somefile.rds"), controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)

Get the results of a group of tasks, returning them as a list. See rrq_task_result for getting the result of a single task.

Description

Get the results of a group of tasks, returning them as a list. See rrq_task_result for getting the result of a single task.

Usage

rrq_task_results(
  task_ids,
  error = FALSE,
  named = FALSE,
  follow = NULL,
  controller = NULL
)

Arguments

Value

A list, one entry per result. This function errors if any task is not available. If named = TRUE, then this list is named with the task_ids.

Examples

obj <- rrq_controller("rrq:example")

ts <- rrq_task_create_bulk_call(sqrt, 1:10, controller = obj)
rrq_task_wait(ts, controller = obj)
rrq_task_results(ts, controller = obj)

# For a single task, rrq_task_result and rrq_task_results differ
# in the return type; rrq_task_results always returns a list:
t <- ts[[1]]
rrq_task_result(t, controller = obj)
rrq_task_results(t, controller = obj)

Description

Retry a task (or set of tasks). Typically this is after failure (e.g., ERROR, DIED or similar) but you can retry even successfully completed tasks. Once retried, functions that retrieve information about a task (e.g., rrq_task_status()⁠, [rrq_task_result()]) will behave differently depending on the value of their ⁠follow⁠argument. See⁠vignette("fault-tolerance")' for more details.

Usage

rrq_task_retry(task_ids, controller = NULL)

Arguments

Value

New task ids

Examples

obj <- rrq_controller("rrq:example")

# It's straightforward to see the effect of retrying a task with
# one that produces a different value each time, so here, we use a
# simple task that draws one normally distributed random number
t1 <- rrq_task_create_expr(rnorm(1), controller = obj)
rrq_task_wait(t1, controller = obj)
rrq_task_result(t1, controller = obj)

# If we retry the task we'll get a different value:
t2 <- rrq_task_retry(t1, controller = obj)
rrq_task_wait(t2, controller = obj)
rrq_task_result(t2, controller = obj)

# Once a task is retried, most of the time (by default) you can use
# the original id and the new one exchangeably:
rrq_task_result(t1, controller = obj)
rrq_task_result(t2, controller = obj)

# Use the 'follow' argument to modify this behaviour
rrq_task_result(t1, follow = FALSE, controller = obj)
rrq_task_result(t2, follow = FALSE, controller = obj)

# See the retry chain with rrq_task_info
rrq_task_info(t1, controller = obj)
rrq_task_info(t2, controller = obj)

Description

Return a character vector of task statuses. The name of each element corresponds to a task id, and the value will be one of the possible statuses ("PENDING", "COMPLETE", etc).

Usage

rrq_task_status(task_ids, named = FALSE, follow = NULL, controller = NULL)

Arguments

Value

A character vector the same length as task_ids

Examples

obj <- rrq_controller("rrq:example")

ts <- rrq_task_create_bulk_call(sqrt, 1:10, controller = obj)
rrq_task_status(ts, controller = obj)
rrq_task_wait(ts, controller = obj)
rrq_task_status(ts, controller = obj)

Description

Fetch times for tasks at points in their life cycle. For each task returns the time of submission, starting and completion (not necessarily successfully; this includes errors and interruptions). If a task has not reached a point yet (e.g., submitted but not run, or running but not finished) the time will be NA). Times are returned in unix timestamp format in UTC; you can use redux::redis_time_to_r to convert them to a POSIXt object.

Usage

rrq_task_times(task_ids, follow = NULL, controller = NULL)

Arguments

Value

A matrix of times, with row names corresponding to task ids. We may change this to a data.frame at some point in the future.

Examples

obj <- rrq_controller("rrq:example")

t <- rrq_task_create_expr(Sys.sleep(3), controller = obj)
rrq_task_times(t, controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_times(t, controller = obj)

Description

Wait for a task, or set of tasks, to complete. If you have used rrq prior to version 0.8.0, you might expect this function to return the result, but we now return a logical value which indicates success or not. You can fetch the task result with rrq_task_result.

Usage

rrq_task_wait(
  task_id,
  timeout = NULL,
  time_poll = 1,
  progress = NULL,
  follow = NULL,
  controller = NULL
)

Arguments

Value

A scalar logical value; TRUE if all tasks complete successfully and FALSE otherwise

Examples

obj <- rrq_controller("rrq:example")
t1 <- rrq_task_create_expr(Sys.sleep(1), controller = obj)
rrq_task_wait(t1, controller = obj)

# The return value of wait gives a summary of successfullness
# of the task
t2 <- rrq_task_create_expr(stop("Some error"), controller = obj)
rrq_task_wait(t2, controller = obj)

# If you wait on many tasks, the return value is effectively
# reduced with "all" (so the result is TRUE if all tasks were
# successful)
rrq_task_wait(c(t1, t2), controller = obj)

Description

rrq queue worker

rrq queue worker

Details

A rrq queue worker. These are not typically for interacting with but will sit and poll a queue for jobs.

Public fields

Methods

Public methods

• rrq_worker$new()

• rrq_worker$info()

• rrq_worker$log()

• rrq_worker$load_envir()

• rrq_worker$poll()

• rrq_worker$step()

• rrq_worker$loop()

• rrq_worker$format()

• rrq_worker$timer_start()

• rrq_worker$progress()

• rrq_worker$task_eval()

• rrq_worker$shutdown()

Method new()

Constructor

Usage

rrq_worker$new(
  queue_id,
  name_config = "localhost",
  worker_id = NULL,
  timeout_config = 0,
  is_child = FALSE,
  con = redux::hiredis()
)

Arguments

Method info()

Return information about this worker, a list of key-value pairs.

Usage

rrq_worker$info()

Method log()

Create a log entry. This will print a human readable format to screen and a machine-readable format to the redis database.

Usage

rrq_worker$log(label, value = NULL)

Arguments

Method load_envir()

Load the worker environment by creating a new environment object and running the create hook (if configured). See rrq_worker_envir_set() for details.

Usage

rrq_worker$load_envir()

Method poll()

Poll for work

Usage

rrq_worker$poll(immediate = FALSE)

Arguments

Method step()

Take a single "step". This consists of

1. Poll for work (⁠$poll()⁠)

2. If work found, run it (either a task or a message)

3. If work not found, check the timeout

Usage

rrq_worker$step(immediate = FALSE)

Arguments

Method loop()

The main worker loop. Use this to set up the main worker event loop, which will continue until exiting (via a timeout or message).

Usage

rrq_worker$loop(immediate = FALSE)

Arguments

Method format()

Create a nice string representation of the worker. Used automatically to print the worker by R6.

Usage

rrq_worker$format()

Method timer_start()

Start the timer

Usage

rrq_worker$timer_start()

Method progress()

Submit a progress message. See rrq_task_progress_update() for details of this mechanism.

Usage

rrq_worker$progress(value, error = TRUE)

Arguments

Method task_eval()

Evaluate a task. When running a task on a separate process, we will always set two environment variables: * RRQ_WORKER_ID this is the id field * RRQ_TASK_ID this is the task id

Usage

rrq_worker$task_eval(task_id)

Arguments

Method shutdown()

Stop the worker

Usage

rrq_worker$shutdown(status = "OK", graceful = TRUE)

Arguments

Description

Create a worker configuration, suitable to pass into rrq_worker_config_save. The results of this function should not be modified.

Usage

rrq_worker_config(
  queue = NULL,
  verbose = TRUE,
  logdir = NULL,
  poll_queue = NULL,
  timeout_idle = Inf,
  poll_process = 1,
  timeout_process_die = 2,
  heartbeat_period = NULL
)

Arguments

Value

A list of values with class rrq_worker_config; these should be considered read-only, and contain only the validated input parameters.

Examples

rrq::rrq_worker_config()

Description

Return names of worker configurations saved by rrq_worker_config_save()

Usage

rrq_worker_config_list(controller = NULL)

Arguments

Value

A character vector of names; these can be passed as the name argument to rrq_worker_config_read().

Examples

obj <- rrq_controller("rrq:example")

cfg <- rrq_worker_config("fast")
rrq_worker_config_save("use-fast", cfg, controller = obj)
rrq_worker_config_list(controller = obj)

Description

Return the value of a of worker configuration saved by rrq_worker_config_save()

Usage

rrq_worker_config_read(name, timeout = 0, controller = NULL)

Arguments

Examples

obj <- rrq_controller("rrq:example")

cfg <- rrq_worker_config("fast")
rrq_worker_config_save("use-fast", cfg, controller = obj)
rrq_worker_config_read("use-fast", controller = obj)

Description

Save a worker configuration, which can be used to start workers with a set of options with the cli. These correspond to arguments to rrq_worker. This function will be renamed soon

Usage

rrq_worker_config_save(name, config, overwrite = TRUE, controller = NULL)

Arguments

Value

Invisibly, a boolean indicating if the configuration was updated.

Examples

obj <- rrq_controller("rrq:example")

cfg <- rrq_worker_config("fast")
rrq_worker_config_save("use-fast", cfg, controller = obj)
rrq_worker_config_list(controller = obj)

Description

Cleans up workers known to have exited. See vignette("fault-tolerance") for more details.

Usage

rrq_worker_delete_exited(worker_ids = NULL, controller = NULL)

Arguments

Value

A character vector of workers that were deleted

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_delete_exited(controller = obj)

Description

Detects exited workers through a lapsed heartbeat. This differs from rrq_worker_list_exited() which lists workers that have definitely exited by checking to see if any worker that runs a heartbeat process has not reported back in time, then marks that worker as exited. See vignette("fault-tolerance") for details.

Usage

rrq_worker_detect_exited(controller = NULL)

Arguments

Value

Undefined.

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_detect_exited(controller = obj)

Description

Register a function to create an environment when creating a worker. When a worker starts, they will run this function.

Usage

rrq_worker_envir_set(create, notify = TRUE, controller = NULL)

Arguments

Examples

obj <- rrq_controller("rrq:example")

rrq_worker_envir_set(rrq_envir(packages = "ids"), controller = obj)
t <- rrq_task_create_expr(search(), controller = obj)
rrq_task_wait(t, controller = obj)
rrq_task_result(t, controller = obj)
rrq_worker_log_tail(n = 5, controller = obj)

rrq_worker_envir_set(NULL, controller = obj)

Description

Test if a worker exists

Usage

rrq_worker_exists(name, controller = NULL)

Arguments

Value

A logical value

Examples

obj <- rrq_controller("rrq:example")
w <- rrq_worker_list(controller = obj)
rrq_worker_exists(w, controller = obj)
rrq_worker_exists("bob-the-builder", controller = obj)

Description

Returns a list of information about active workers (or exited workers if worker_ids includes them).

Usage

rrq_worker_info(worker_ids = NULL, controller = NULL)

Arguments

Value

A list of worker_info objects

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_info(controller = obj)

Description

Returns the number of active workers

Usage

rrq_worker_len(controller = NULL)

Arguments

Value

An integer

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_len(controller = obj)

Description

Returns the ids of active workers. This does not include exited workers; use rrq_worker_list_exited() for that.

Usage

rrq_worker_list(controller = NULL)

Arguments

Value

A character vector of worker names

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_list(controller = obj)

Description

Returns the ids of workers known to have exited

Usage

rrq_worker_list_exited(controller = NULL)

Arguments

Value

A character vector of worker names

Examples

obj <- rrq_controller("rrq:example")

# At this point you might have an exited worker, depending on
# which examples have been run so far!
rrq_worker_list_exited(controller = obj)

# Spawn a new worker so that we can stop it:
w <- rrq_worker_spawn(1, controller = obj)$id
w$id
# Stop this worker and see that it appears in the list of exited
# workers:
rrq_worker_stop(w$id, controller = obj)
rrq_worker_list_exited(controller = obj)

# We can delete this exited worker:
rrq_worker_delete_exited(w$id, controller = obj)

# After this, it is no longer listed as exited:
rrq_worker_list_exited(controller = obj)

Description

Report on worker "load" (the number of workers being used over time). Reruns an object of class worker_load, for which a mean method exists (this function is a work in progress and the interface may change).

Usage

rrq_worker_load(worker_ids = NULL, controller = NULL)

Arguments

Value

An object of class "worker_load", which has a pretty print method.

Examples

obj <- rrq_controller("rrq:example")
mean(rrq_worker_load(controller = obj))

Description

Returns the last (few) elements in the worker log, in a programmatically useful format (see Value).

Usage

rrq_worker_log_tail(worker_ids = NULL, n = 1, controller = NULL)

Arguments

Value

A data.frame with columns:

• worker_id: the worker id

• child: the process id, an integer, where logs come from a child process from a task queued with separate_process = TRUE

• time: the time from Redis when the event happened; see redux::redis_time to convert this to an R time

• command: the command sent from or to the worker

• message: the message corresponding to that command

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_log_tail(n = 10, controller = obj)

Description

Return the contents of a worker's process log, if it is located on the same physical storage (including network storage) as the controller. This will generally behave for workers started with rrq_worker_spawn but may require significant care otherwise.

Usage

rrq_worker_process_log(worker_id, controller = NULL)

Arguments

Value

A character vector, one line per line in the log. If logging is enabled but the worker has not produced any logs, this will be an empty character vector. If logging is not enabled, then this function will throw.

Examples

obj <- rrq_controller("rrq:example")
worker_id <- rrq_worker_list(controller = obj)[[1]]
tryCatch(
  rrq_worker_process_log(worker_id, controller = obj),
  error = identity)

Description

Write a small script that can be used to launch a rrq worker. The resulting script takes the same arguments as the rrq_worker constructor, but from the command line. See Details.

Usage

rrq_worker_script(path, versioned = FALSE)

Arguments

Details

If you need to launch rrq workers from a script, it's convenient not to have to embed R code like:

Rscript -e 'rrq::rrq_worker$new("myqueue")'

as this is error-prone and unpleasant to quote and read. You can use the function rrq_worker_script to write out a small helper script which lets you write:

./path/rrq_worker myqueue

instead.

The helper script supports the same arguments as the ⁠[rrq::rrq_worker]⁠ constructor:

• queue_id as the sole positional argument

• name_config as --config

• worker_id as --worker-id

To change the redis connection settings, set the REDIS_URL environment variable (see redux::hiredis() for details).

For example to create a worker myworker with configuration myconfig on queue myqueue you might use

./rrq_worker --config=myconfig --worker-id=myworker myqueue

Value

Invisibly, the path to the script

Examples

path <- rrq::rrq_worker_script(tempfile())
readLines(path)

Description

Spawn a worker in the background

Usage

rrq_worker_spawn(
  n = 1,
  logdir = NULL,
  timeout = 600,
  name_config = "localhost",
  worker_id_base = NULL,
  time_poll = 0.2,
  progress = NULL,
  controller = NULL
)

Arguments

Details

Spawning multiple workers. If n is greater than one, multiple workers will be spawned. This happens in parallel so it does not take n times longer than spawning a single worker.

Beware that signals like Ctrl-C passed to this R instance can still propagate to the child processes and can result in them dying unexpectedly. It is probably safer to start processes in a completely separate session.

Value

An rrq_worker_manager object with fields:

• id: the ids of the spawned workers

• wait_alive: a method to wait for workers to come alive

• stop: a method to stop workers

• kill: a method to kill workers abruptly by sending a signal

• is_alive: a method that checks if a worker is currently alive

• logs: a method that returns logs for a single worker

All the methods accept a vector of worker names, or integers, except logs which requires a single worker id (as a string or integer). For all methods except logs, the default of NULL means "all managed workers".

Description

Returns a character vector of current worker statuses

Usage

rrq_worker_status(worker_ids = NULL, controller = NULL)

Arguments

Value

A character vector of statuses, named by worker

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_status(controller = obj)

Description

Stop workers, causing them to exit. Workers can be stopped in a few different ways (see Details), but after executing this function, assume that any worker targeted will no longer be available to work on tasks.

Usage

rrq_worker_stop(
  worker_ids = NULL,
  type = "message",
  timeout = 0,
  time_poll = 0.1,
  progress = NULL,
  controller = NULL
)

Arguments

Details

The type parameter indicates the strategy used to stop workers, and interacts with other parameters. The strategies used by the different values are:

• message, in which case a STOP message will be sent to the worker, which they will receive after finishing any currently running task (if RUNNING; IDLE workers will stop immediately).

• kill, in which case a kill signal will be sent via the heartbeat (if the worker is using one). This will kill the worker even if is currently working on a task, eventually leaving that task with a status of DIED.

• kill_local, in which case a kill signal is sent using operating system signals, which requires that the worker is on the same machine as the controller.

Value

The names of the stopped workers, invisibly.

Examples

obj <- rrq_controller("rrq:example")
w <- rrq_worker_spawn(controller = obj)
rrq_worker_list(controller = obj)
rrq_worker_stop(w$id, timeout = 10, controller = obj)
rrq_worker_list(controller = obj)

Description

Returns the task id that each worker is working on, if any.

Usage

rrq_worker_task_id(worker_ids = NULL, controller = NULL)

Arguments

Value

A character vector, NA where nothing is being worked on, otherwise corresponding to a task id.

Examples

obj <- rrq_controller("rrq:example")
rrq_worker_list(controller = obj)

# This example might be a bit racy: we need to run a task that
# sleeps, and then sleep a little bit for the task to be picked up
# by a worker.  Typically this happens very quickly but there are
# no guarantees.
t <- rrq_task_create_expr(Sys.sleep(1), controller = obj)
Sys.sleep(.2)
rrq_worker_task_id(controller = obj)

# You can always find out which worker did work on a task though:
rrq_task_info(t, controller = obj)$worker

Description

Wait for workers to appear.

Usage

rrq_worker_wait(
  worker_ids,
  timeout = Inf,
  time_poll = 0.2,
  progress = NULL,
  controller = NULL
)

Arguments