Redis-based futures

Use the Redis key/value database to define partially fault-tolerant, asynchronous task queues for elastic distributed computing.

redis(
  expr,
  substitute = TRUE,
  envir = parent.frame(),
  ...,
  queue = getOption("future.redis.queue", "{{session}}"),
  config = redis_config(),
  output_queue = NA_character_,
  max_retries = 3L
)

Arguments

expr

An R expression.

substitute

If TRUE, argument expr is substitute():ed, otherwise not.

envir

The environment from where global objects should be identified.

...

Additional named elements of the future.

queue

A Redis key name of the task queue, or a RedisWorkerConfiguration object as returned by startLocalWorkers().

config

A redux::redis_config Redis configuration object.

output_queue

(optional) Redis key name of the work output queue (note: reserved for future use).

max_retries

Maximum number of times the future can be re-submitted to the task queue in the event of failure.

Value

An object of class RedisFuture.

See also

redux::redis_config(), worker(), removeQ()

Examples

## The example assumes that a Redis server is running
## on the local host and the standard Redis port (6379)
if (redux::redis_available()) {

# Start two local R worker processes running in the background
workers <- startLocalWorkers(2L, linger = 1.0)

plan(redis)

# A function that returns a future, note that N uses lexical scoping...
f <- \() future({4 * sum((runif(N) ^ 2 + runif(N) ^ 2) < 1) / N}, seed = TRUE)

# Run a simple sampling approximation of pi in parallel using  M * N points:
N <- 1e6  # samples per worker
M <- 10   # iterations
pi_est <- Reduce(sum, Map(value, replicate(M, f()))) / M
print(pi_est)

# Make sure to stop the workers
stopLocalWorkers(workers)

}
#> [1] 3.141237