Start one or more background R worker processes on the local system.

Use startLocalWorkers() to start one or more future.redis R worker processes in the background. The worker processes are started on the local system using the worker() function. Additional workers can be launched by calling startLocalWorkers() multiple times.

startLocalWorkers(
  n,
  queue = getOption("future.redis.queue", "{{session}}"),
  config = redis_config(),
  iter = Inf,
  linger = 10,
  log = nullfile(),
  Rbin = paste(R.home(component = "bin"), "R", sep = "/")
)

stopLocalWorkers(
  queue = getOption("future.redis.queue", "{{session}}"),
  config = redis_config()
)

Arguments

n

number of workers to start.

queue

Redis task queue name.

config

A redux::redis_config Redis configuration object.

iter

Maximum number of tasks to acquire before exiting.

linger

in seconds, max time before system checks (including termination).

log

divert stdout and messages to log file.

Rbin

full path to the command-line R program.

Value

startLocalWorkers() returns, invisibly, a RedisWorkerConfiguration

object, which comprise of the arguments passed to each of the background workers on startup.

stopLocalWorkers() returns nothing.

Details

stopLocalWorkers() can remove the task queue for these workers. All workers that listen to the task queue will self-terminate after a linger interval (seconds) if the task queue is no longer available, or if network communication with the Redis server encounters an error.

When passing an RedisWorkerConfiguration object to startLocalWorkers() and stopLocalWorkers(), the queue and config values are extracted from that object.

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.142062