redis: Redis-based futures

In bwlewis/future.redis: An Elastic Distributed Computing Backend for the Future Package

Redis-based futures

Description

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

Usage

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

Arguments

expr	An R expression.
envir	The environment from where global objects should be identified.
substitute	If TRUE, argument expr is substitute():ed, otherwise not.
globals	(optional) a logical, a character vector, a named list, or a [globals::Globals] object. If 'TRUE', globals are identified by code inspection based on 'expr' and 'tweak' searching from environment 'envir'. If 'FALSE', no globals are used. If a character vector, then globals are identified by lookup based their names 'globals' searching from environment 'envir'. If a named list or a Globals object, the globals are used as is.
queue	Redis key name of the task queue (Redis list).
config	Redis config.
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.
...	Additional named elements of the future.

Value

An object of class [RedisFuture].

See Also

redis_config, worker, removeQ

Examples

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

# Register the redis plan on a specified task queue:
plan(redis, queue = "R jobs")

# Start some local R worker processes:
startLocalWorkers(n=2, queue="R jobs", linger=1)

# Alternatively, use the following to run the workers quietly without
# showing messages as they run:
# startLocalWorkers(n=2, queue="R jobs", linger=1, log="/dev/null")

# 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
Reduce(sum, Map(value, replicate(M, f()))) / M

# Clean up
removeQ("R jobs")
}

bwlewis/future.redis documentation built on May 12, 2023, 1:52 a.m.