Parallel and Asynchronous Computing

source("scenario.R")

# The unit of work: the share of one member's loans that are overdue.
# It takes everything it needs as arguments and uses no globals.
overdue_rate <- function(rows, today, days) {
  end <- rows$return_date
  end[is.na(end)] <- today
  mean(as.integer(end - rows$loan_date) > days, na.rm = TRUE)
}

# Clean the member names, drop the empty one, and split loans by member.
members <- tolower(trimws(loans$member))
keep <- nzchar(members)
by_member <- split(loans[keep, ], members[keep])
length(by_member)
#> [1] 8

serial <- lapply(by_member, overdue_rate, today = shelf_today, days = loan_days)
# Percent of each member's loans that are overdue.
round(100 * unlist(serial))
#>      ana ruiz    ben okafor  chloe martin     dev patel eva lindqvist
#>            62            60            67            67            33
#>  farid karimi     grace liu    hugo silva
#>             0            57            50

source("scenario.R")
library(parallel)

overdue_rate <- function(rows, today, days) {
  end <- rows$return_date
  end[is.na(end)] <- today
  mean(as.integer(end - rows$loan_date) > days, na.rm = TRUE)
}
members <- tolower(trimws(loans$member))
keep <- nzchar(members)
by_member <- split(loans[keep, ], members[keep])
serial <- lapply(by_member, overdue_rate, today = shelf_today, days = loan_days)

# Forking only exists on Unix-alikes (Linux, macOS), not on Windows.
.Platform$OS.type
#> [1] "unix"
formals(mclapply)$mc.cores
#> getOption("mc.cores", 2L)
getOption("mc.cores")
#> NULL

forked <- mclapply(
  by_member, overdue_rate,
  today = shelf_today, days = loan_days, mc.cores = 2
)
identical(forked, serial)
#> [1] TRUE

# The work ran in child processes, not in this one.
pids <- unlist(mclapply(1:4, \(i) Sys.getpid(), mc.cores = 2))
length(unique(pids))
#> [1] 2
Sys.getpid() %in% pids
#> [1] FALSE

# mc.cores = 1 does not fork at all: it calls lapply().
one <- mclapply(
  by_member, overdue_rate,
  today = shelf_today, days = loan_days, mc.cores = 1
)
identical(one, serial)
#> [1] TRUE

# Does it pay off? Make each unit take about a quarter of a second.
slow_rate <- function(rows, today, days) {
  Sys.sleep(0.25)
  overdue_rate(rows, today, days)
}
t_serial <- system.time(
  lapply(by_member, slow_rate, today = shelf_today, days = loan_days)
)[["elapsed"]]
t_forked <- system.time(
  mclapply(by_member, slow_rate, today = shelf_today, days = loan_days,
           mc.cores = 2)
)[["elapsed"]]
t_serial / t_forked > 1.5
#> [1] TRUE

library(parallel)

# mc.preschedule = TRUE (the default): the values are dealt out to the
# cores in turn, one forked process per core.
pids <- unlist(mclapply(1:6, \(i) Sys.getpid(), mc.cores = 2))
match(pids, unique(pids))
#> [1] 1 2 1 2 1 2

# mc.preschedule = FALSE: one forked process per value.
pids <- unlist(
  mclapply(1:6, \(i) Sys.getpid(), mc.cores = 2, mc.preschedule = FALSE)
)
length(unique(pids))
#> [1] 6

# An error is caught per job, not per value: values 1 and 3 share a core,
# so both come back as "try-error" although only 3 failed.
warned <- character()
res <- withCallingHandlers(
  mclapply(1:4, \(i) if (i == 3) stop("bad item ", i) else i, mc.cores = 2),
  warning = \(w) {
    warned <<- conditionMessage(w)
    invokeRestart("muffleWarning")
  }
)
warned
#> [1] "scheduled core 1 encountered error in user code, all values of the job will be affected"
vapply(res, inherits, logical(1), what = "try-error")
#> [1]  TRUE FALSE  TRUE FALSE
cat(res[[3]])
#> Error in FUN(X[[i]], ...) : bad item 3
conditionMessage(attr(res[[3]], "condition"))
#> [1] "bad item 3"

# With mc.preschedule = FALSE only the value that failed is affected.
res <- suppressWarnings(
  mclapply(1:4, \(i) if (i == 3) stop("bad item ", i) else i,
           mc.cores = 2, mc.preschedule = FALSE)
)
vapply(res, inherits, logical(1), what = "try-error")
#> [1] FALSE FALSE  TRUE FALSE

# A child that dies delivers nothing: its value is NULL.
res <- withCallingHandlers(
  mclapply(1:2, \(i) if (i == 2) tools::pskill(Sys.getpid(), 9) else i,
           mc.cores = 2, mc.preschedule = FALSE),
  warning = \(w) {
    warned <<- conditionMessage(w)
    invokeRestart("muffleWarning")
  }
)
warned
#> [1] "1 parallel job did not deliver a result"
vapply(res, is.null, logical(1))
#> [1] FALSE  TRUE

# Changes made in a child never reach the parent.
counter <- 0
invisible(mclapply(1:2, \(i) {
  counter <<- counter + 1
  counter
}, mc.cores = 2))
counter
#> [1] 0

# mcaffinity() is where the CPU numbers are (NULL where unsupported).
length(mcaffinity()) >= 1L
#> [1] TRUE

source("scenario.R")
library(parallel)

# pvec() splits a vector, runs the function on each piece, and joins the
# pieces. The function must be a pure vectorised map.
valid <- pvec(loans$isbn, is_valid_isbn13, mc.cores = 2)
identical(valid, is_valid_isbn13(loans$isbn))
#> [1] TRUE
sum(valid)
#> [1] 57

overdue_rate <- function(rows, today, days) {
  end <- rows$return_date
  end[is.na(end)] <- today
  mean(as.integer(end - rows$loan_date) > days, na.rm = TRUE)
}
members <- tolower(trimws(loans$member))
keep <- nzchar(members)
by_member <- split(loans[keep, ], members[keep])

# mcparallel() starts one job and returns at once; mccollect() waits.
job_a <- mcparallel(
  overdue_rate(by_member[["ana ruiz"]], shelf_today, loan_days),
  name = "ana"
)
job_b <- mcparallel(
  overdue_rate(by_member[["ben okafor"]], shelf_today, loan_days),
  name = "ben"
)
class(job_a)
#> [1] "parallelJob"  "childProcess" "process"
res <- mccollect(list(job_a, job_b))
names(res)
#> [1] "ana" "ben"
round(unlist(res), 2)
#>  ana  ben
#> 0.62 0.60

# wait = FALSE only looks: NULL means "nothing is ready yet".
slow <- mcparallel({
  Sys.sleep(0.5)
  "done"
})
is.null(mccollect(slow, wait = FALSE))
#> [1] TRUE
mccollect(slow)[[1]]
#> [1] "done"

# mcmapply() and mcMap() are the forking versions of mapply() and Map().
mcmapply(
  \(rows, label) paste(label, nrow(rows)),
  by_member[1:3], names(by_member)[1:3],
  mc.cores = 2
)
#>         ana ruiz       ben okafor     chloe martin
#>    "ana ruiz 14"   "ben okafor 5" "chloe martin 9"

source("scenario.R")
library(parallel)

overdue_rate <- function(rows, today, days) {
  end <- rows$return_date
  end[is.na(end)] <- today
  mean(as.integer(end - rows$loan_date) > days, na.rm = TRUE)
}
members <- tolower(trimws(loans$member))
keep <- nzchar(members)
by_member <- split(loans[keep, ], members[keep])
serial <- lapply(by_member, overdue_rate, today = shelf_today, days = loan_days)

# makeCluster(2) starts two fresh copies of R (type "PSOCK") on this machine.
cl <- makeCluster(2)
cl
#> socket cluster with 2 nodes on host ‘localhost’
class(cl)
#> [1] "SOCKcluster" "cluster"
length(cl)
#> [1] 2

# Each worker is a separate process with its own workspace.
pids <- unlist(clusterCall(cl, Sys.getpid))
length(unique(pids))
#> [1] 2
Sys.getpid() %in% pids
#> [1] FALSE

# parLapply() splits X into one chunk per worker and sends each chunk over
# the socket together with the function.
par <- parLapply(
  cl, by_member, overdue_rate, today = shelf_today, days = loan_days
)
identical(par, serial)
#> [1] TRUE

# The workers stay alive, and keep their workspaces, until you stop them.
invisible(clusterEvalQ(cl, counter <- 0))
invisible(clusterEvalQ(cl, counter <- counter + 1))
unlist(clusterEvalQ(cl, counter))
#> [1] 1 1
stopCluster(cl)

# A function that always cleans up after itself.
with_cluster <- function(n, code) {
  cl <- makeCluster(n)
  on.exit(stopCluster(cl))
  code(cl)
}
with_cluster(2, \(cl) unlist(parLapply(cl, 1:4, \(i) i^2)))
#> [1]  1  4  9 16

source("scenario.R")
library(parallel)

cl <- makeCluster(2)

# A function defined at top level is sent to the workers, but its free
# variables are looked up in each worker's own global environment.
days_left <- function(i) loan_days - i # nolint: object_usage_linter.
tryCatch(
  parLapply(cl, 1:2, days_left),
  error = conditionMessage
)
#> [1] "2 nodes produced errors; first error: object 'loan_days' not found"

# Fix 1: pass what the function needs as an argument.
days_left_arg <- function(i, days) days - i
unlist(parLapply(cl, 1:2, days_left_arg, days = loan_days))
#> [1] 20 19

# Fix 2: copy the variable into every worker's global environment.
clusterExport(cl, "loan_days")
unlist(parLapply(cl, 1:2, days_left))
#> [1] 20 19

# Fix 3: run code on every worker, for example to load a package.
invisible(clusterEvalQ(cl, library(stats)))
unlist(clusterEvalQ(cl, "package:stats" %in% search()))
#> [1] TRUE TRUE

# clusterExport() looks in the global environment by default. Inside a
# function the variable is local, so say where it is with envir.
export_local <- function(cl) {
  secret <- 42 # nolint: object_usage_linter.
  clusterExport(cl, "secret")
}
tryCatch(export_local(cl), error = conditionMessage)
#> [1] "object 'secret' not found"

export_local_ok <- function(cl) {
  secret <- 42 # nolint: object_usage_linter.
  clusterExport(cl, "secret", envir = environment())
}
export_local_ok(cl)
unlist(clusterEvalQ(cl, secret))
#> [1] 42 42

# A function made inside another function drags its environment along.
big <- numeric(1e6)
make_task <- function(values) {
  force(values)
  \(i) i + 1
}
task_closure <- make_task(big)
task_global <- \(i) i + 1
length(serialize(task_closure, NULL)) > 8e6
#> [1] TRUE
length(serialize(task_global, NULL)) < 1e4
#> [1] TRUE
stopCluster(cl)

source("scenario.R")
library(parallel)

overdue_rate <- function(rows, today, days) {
  end <- rows$return_date
  end[is.na(end)] <- today
  mean(as.integer(end - rows$loan_date) > days, na.rm = TRUE)
}
members <- tolower(trimws(loans$member))
keep <- nzchar(members)
by_member <- split(loans[keep, ], members[keep])
serial <- lapply(by_member, overdue_rate, today = shelf_today, days = loan_days)

cl <- makeCluster(2)

# clusterApply(): element i goes to node i, recycling the nodes.
nodes <- unlist(clusterApply(cl, 1:6, \(i) Sys.getpid()))
match(nodes, unique(nodes))
#> [1] 1 2 1 2 1 2

# clusterSplit() shows how parLapply() cuts a vector into one chunk per node.
clusterSplit(cl, 1:7)
#> [[1]]
#> [1] 1 2 3 4
#>
#> [[2]]
#> [1] 5 6 7
#>

# parLapply() is clusterApply() on those chunks, then lapply() in each.
par <- parLapply(
  cl, by_member, overdue_rate, today = shelf_today, days = loan_days
)
identical(par, serial)
#> [1] TRUE

# The load-balancing versions hand out work as nodes become free. Which node
# runs which value then varies from run to run, but the result does not.
lb <- parLapplyLB(
  cl, by_member, overdue_rate, today = shelf_today, days = loan_days
)
identical(lb, serial)
#> [1] TRUE
lb2 <- clusterApplyLB(
  cl, by_member, overdue_rate, today = shelf_today, days = loan_days
)
identical(lb2, unname(serial))
#> [1] TRUE

# chunk.size = 1 makes every value its own unit of scheduling.
par1 <- parLapply(
  cl, by_member, overdue_rate, today = shelf_today, days = loan_days,
  chunk.size = 1
)
identical(par1, serial)
#> [1] TRUE

# clusterMap() is mapply() on the cluster.
clusterMap(cl, \(a, b) a * b, 1:3, 4:6)
#> [[1]]
#> [1] 4
#>
#> [[2]]
#> [1] 10
#>
#> [[3]]
#> [1] 18
#>
rates <- clusterMap(
  cl, overdue_rate, by_member[1:3],
  days = c(7, 14, 21), MoreArgs = list(today = shelf_today)
)
# Percent overdue, with 7, 14 and 21 days allowed.
round(100 * unlist(rates))
#>     ana ruiz   ben okafor chloe martin
#>           92           80           67

# parSapply() simplifies like sapply(): fine for a number per element, but
# the type of the result depends on the data. Prefer vapply() on parLapply().
checked <- vapply(par, identity, numeric(1))
all.equal(checked, unlist(serial))
#> [1] TRUE

# Extra arguments are matched by name to the formals of parLapply() first.
tryCatch(
  parSapply(cl, 1:3, \(i, x) i + x, x = 10),
  error = conditionMessage
)
#> [1] "formal argument \"x\" matched by multiple actual arguments"
unlist(parLapply(cl, 1:3, \(i, offset) i + offset, offset = 10))
#> [1] 11 12 13
stopCluster(cl)

library(parallel)

# The cluster types this R knows about.
registerClusterType()
#> [1] "PSOCK"  "FORK"   "MIRAI"  "MPI"    "RPSOCK" "SOCK"

# makeCluster(2, type = "FORK") is makeForkCluster(2): the workers are
# forks of this session, so they inherit its workspace as it is right now.
x_before <- 42
cl <- makeForkCluster(2)
x_after <- 1
unlist(clusterEvalQ(cl, exists("x_before")))
#> [1] TRUE TRUE
unlist(clusterEvalQ(cl, exists("x_after")))
#> [1] FALSE FALSE
stopCluster(cl)

# Register a default cluster so that the cl argument can be left out.
getDefaultCluster()
#> NULL
cl <- makeCluster(2)
setDefaultCluster(cl)
identical(getDefaultCluster(), cl)
#> [1] TRUE
unlist(parLapply(NULL, 1:3, \(i) i * 2))
#> [1] 2 4 6
stopCluster(cl)
getDefaultCluster()
#> NULL
tryCatch(parLapply(NULL, 1:3, \(i) i * 2), error = conditionMessage)
#> [1] "no cluster 'cl' supplied and none is registered"

# Worker output is discarded unless outfile says where to put it.
log_file <- tempfile(fileext = ".log")
cl <- makeCluster(2, outfile = log_file)
invisible(clusterEvalQ(cl, {
  cat("hello from a worker\n")
  message("a message from a worker")
}))
stopCluster(cl)
log_lines <- readLines(log_file)
sort(log_lines[!startsWith(log_lines, "starting worker")])
#> [1] "a message from a worker" "a message from a worker"
#> [3] "hello from a worker"     "hello from a worker"
unlink(log_file)

# Worker warnings are lost too: only errors travel back.
cl <- makeCluster(2)
res <- withCallingHandlers(
  parLapply(cl, 1:2, \(i) {
    warning("careful")
    i
  }),
  warning = \(w) cat("the master saw a warning\n")
)
unlist(res)
#> [1] 1 2
stopCluster(cl)

# Options of the PSOCK launcher: how long to wait while the workers start
# (setup_timeout), the longest silence on a connection (timeout), the port
# (port, or the environment variable R_PARALLEL_PORT).
cl <- makeCluster(2, setup_timeout = 60, timeout = 3600)
length(cl)
#> [1] 2
stopCluster(cl)

library(parallel)

# An error on a worker comes back to the master as one error.
cl <- makeCluster(2)
e <- tryCatch(parLapply(cl, 1:2, \(i) stop("no good ", i)), error = identity)
conditionMessage(e)
#> [1] "2 nodes produced errors; first error: no good 1"
conditionCall(e)
#> checkForRemoteErrors(val)

# A call to something the worker does not have.
tryCatch(
  parLapply(cl, 1:2, \(i) not_a_function(i)),
  error = conditionMessage
)
#> [1] "2 nodes produced errors; first error: could not find function \"not_a_function\""
tryCatch(clusterEvalQ(cl, library(notAPackage)), error = conditionMessage)
#> [1] "2 nodes produced errors; first error: there is no package called ‘notAPackage’"

# A worker that dies.
tryCatch(
  clusterEvalQ(cl, quit(save = "no", status = 1, runLast = FALSE)),
  error = conditionMessage
)
#> [1] "error reading from connection"
try(stopCluster(cl), silent = TRUE)

# A cluster that was stopped cannot be used again.
cl <- makeCluster(2)
stopCluster(cl)
tryCatch(parLapply(cl, 1:2, \(i) i), error = conditionMessage)
#> [1] "invalid connection"

library(parallel)

# detectCores() counts the CPUs of the machine; it may be NA, and it does not
# know what this user or job is allowed to use.
n_machine <- detectCores()
is.na(n_machine)
#> [1] FALSE
n_machine >= detectCores(logical = FALSE)
#> [1] TRUE

# parallelly::availableCores() starts from the same number and then applies
# every limit it can find: cgroups, the mc.cores option, batch schedulers.
n_allowed <- parallelly::availableCores()
unname(n_allowed <= n_machine)
#> [1] TRUE
options(mc.cores = 2)
unname(parallelly::availableCores(methods = "mc.cores"))
#> [1] 2
unname(parallelly::availableCores() <= 2L)
#> [1] TRUE
options(mc.cores = NULL)

# A sensible default: leave one core for the main session, and never ask for
# more workers than there are tasks.
choose_workers <- function(tasks, cores = parallelly::availableCores()) {
  max(1L, min(tasks, cores - 1L))
}
choose_workers(8L, cores = 4L)
#> [1] 3
choose_workers(2L, cores = 16L)
#> [1] 2
choose_workers(8L, cores = 1L)
#> [1] 1

source("scenario.R")
library(parallel)

# The loan durations, in days, of the loans that were returned.
durations <- as.integer(loans$return_date - loans$loan_date)
durations <- durations[!is.na(durations) & durations >= 0L]

# One bootstrap replicate: the mean of a resample.
boot_mean <- function(i, values) mean(sample(values, replace = TRUE))

# Without seeding, every run draws different numbers.
cl <- makeCluster(2)
run_1 <- parSapply(cl, 1:6, boot_mean, values = durations)
run_2 <- parSapply(cl, 1:6, boot_mean, values = durations)
identical(run_1, run_2)
#> [1] FALSE

# clusterSetRNGStream() gives each worker its own L'Ecuyer-CMRG stream.
clusterSetRNGStream(cl, 282)
seeded_1 <- parSapply(cl, 1:6, boot_mean, values = durations)
clusterSetRNGStream(cl, 282)
seeded_2 <- parSapply(cl, 1:6, boot_mean, values = durations)
identical(seeded_1, seeded_2)
#> [1] TRUE
round(seeded_1, 2)
#> [1] 20.67 19.51 20.08 18.71 20.98 19.47

# A different seed gives different streams.
clusterSetRNGStream(cl, 283)
other <- parSapply(cl, 1:6, boot_mean, values = durations)
identical(other, seeded_1)
#> [1] FALSE

# It also switched the generator of the master process.
RNGkind()
#> [1] "L'Ecuyer-CMRG" "Inversion"     "Rejection"
stopCluster(cl)

# The streams belong to the workers, so three workers draw differently.
cl <- makeCluster(3)
clusterSetRNGStream(cl, 282)
three <- parSapply(cl, 1:6, boot_mean, values = durations)
stopCluster(cl)
identical(three, seeded_1)
#> [1] FALSE

source("scenario.R")
library(parallel)

durations <- as.integer(loans$return_date - loans$loan_date)
durations <- durations[!is.na(durations) & durations >= 0L]
boot_mean <- function(i, values) mean(sample(values, replace = TRUE))

# mclapply() gives each child the next L'Ecuyer-CMRG stream, but only when
# that generator is selected before the seed is set.
RNGkind("L'Ecuyer-CMRG")
set.seed(282)
fork_1 <- unlist(mclapply(1:6, boot_mean, values = durations, mc.cores = 2))
set.seed(282)
fork_2 <- unlist(mclapply(1:6, boot_mean, values = durations, mc.cores = 2))
identical(fork_1, fork_2)
#> [1] TRUE
round(fork_1, 2)
#> [1] 18.71 19.76 20.98 19.80 19.47 21.41

# The result depends on how many children there are.
set.seed(282)
fork_3 <- unlist(mclapply(1:6, boot_mean, values = durations, mc.cores = 3))
identical(fork_1, fork_3)
#> [1] FALSE

# With the default generator, children are seeded from the clock and the
# process id: different from each other and not reproducible.
RNGkind("Mersenne-Twister")
set.seed(282)
mt_1 <- mclapply(1:2, \(i) runif(1), mc.cores = 2)
set.seed(282)
mt_2 <- mclapply(1:2, \(i) runif(1), mc.cores = 2)
identical(mt_1[[1]], mt_1[[2]])
#> [1] FALSE
identical(mt_1, mt_2)
#> [1] FALSE

# mc.set.seed = FALSE: the children inherit the state of the parent, so once
# the parent has drawn numbers they all draw the same ones.
set.seed(282)
same <- mclapply(1:2, \(i) runif(1), mc.cores = 2, mc.set.seed = FALSE)
identical(same[[1]], same[[2]])
#> [1] TRUE

# mc.reset.stream() restarts the stream sequence for mcparallel().
RNGkind("L'Ecuyer-CMRG")
set.seed(282)
mc.reset.stream()
job_1 <- mccollect(mcparallel(runif(1)))[[1]]
set.seed(282)
mc.reset.stream()
job_2 <- mccollect(mcparallel(runif(1)))[[1]]
identical(job_1, job_2)
#> [1] TRUE

# nextRNGStream() and nextRNGSubStream() move a seed along the streams.
seed <- .Random.seed
length(seed)
#> [1] 7
identical(nextRNGStream(seed), seed)
#> [1] FALSE
identical(nextRNGStream(seed), nextRNGSubStream(seed))
#> [1] FALSE

source("scenario.R")
library(parallel)

durations <- as.integer(loans$return_date - loans$loan_date)
durations <- durations[!is.na(durations) & durations >= 0L]

# One seed per task, made in the master from a single set.seed().
make_streams <- function(n, seed) {
  RNGkind("L'Ecuyer-CMRG")
  set.seed(seed)
  stream <- get(".Random.seed", envir = globalenv())
  out <- vector("list", n)
  for (i in seq_len(n)) {
    out[[i]] <- stream
    stream <- nextRNGStream(stream)
  }
  out
}

# The task installs its own stream, whatever process runs it.
boot_task <- function(stream, values) {
  RNGkind("L'Ecuyer-CMRG")
  assign(".Random.seed", # nolint: object_name_linter.
         stream, envir = globalenv())
  mean(sample(values, replace = TRUE))
}

streams <- make_streams(6L, 282)
in_loop <- vapply(streams, boot_task, numeric(1), values = durations)
forked <- unlist(mclapply(streams, boot_task, values = durations, mc.cores = 2))

cl2 <- makeCluster(2)
two <- parSapply(cl2, streams, boot_task, values = durations)
stopCluster(cl2)

cl3 <- makeCluster(3)
three <- unlist(clusterApply(cl3, streams, boot_task, values = durations))
stopCluster(cl3)

# The same numbers, whatever ran the tasks and however many workers there were.
same <- list(forked, two, three)
vapply(same, identical, logical(1), y = in_loop)
#> [1] TRUE TRUE TRUE
round(in_loop, 2)
#> [1] 20.67 18.71 19.76 21.90 21.10 22.31

library(parallel)

# A new worker is a new R session: it starts with R's defaults.
cl <- makeCluster(2)
unlist(clusterEvalQ(cl, RNGkind())[[1]])
#> [1] "Mersenne-Twister" "Inversion"        "Rejection"

# R 3.6.0 changed the default of sample(); the old method needs a warning
# to be accepted.
withCallingHandlers(
  RNGkind(sample.kind = "Rounding"),
  warning = \(w) {
    cat("warning:", conditionMessage(w), "\n")
    invokeRestart("muffleWarning")
  }
)
#> warning: non-uniform 'Rounding' sampler used
RNGkind()[3]
#> [1] "Rounding"
unlist(clusterEvalQ(cl, RNGkind())[[1]])[3]
#> [1] "Rejection"

# clusterSetRNGStream() builds the streams from the master's settings, so
# the workers now sample the old way as well.
clusterSetRNGStream(cl, 282)
unlist(clusterEvalQ(cl, RNGkind())[[1]])
#> [1] "L'Ecuyer-CMRG" "Inversion"     "Rounding"
stopCluster(cl)
RNGkind(sample.kind = "Rejection")

source("scenario.R")
library(future)

# A plan says where futures are evaluated. The default is sequential: in
# this very R process.
inherits(plan(), "sequential")
#> [1] TRUE
nbrOfWorkers()
#> [1] 1

# multisession: background R sessions on this machine (PSOCK workers).
plan(multisession, workers = 2)
nbrOfWorkers()
#> [1] 2

# future() starts the work and returns at once; value() waits for it.
f <- future({
  Sys.sleep(0.5)
  Sys.getpid()
})
class(f)[1]
#> [1] "MultisessionFuture"
resolved(f)
#> [1] FALSE
value(f) == Sys.getpid()
#> [1] FALSE
resolved(f)
#> [1] TRUE

# A future assignment: x is a promise that is filled in when first used.
x %<-% {
  Sys.sleep(0.2)
  sum(seq_len(10))
}
x
#> [1] 55

# Globals are found by inspecting the code, then copied to the worker.
overdue_days <- function(loan_date, return_date) {
  as.integer(return_date - loan_date) - loan_days # nolint: object_usage_linter.
}
g <- future(overdue_days(loans$loan_date[1], loans$return_date[1]))
names(g$globals)
#> [1] "overdue_days" "loans"        "loan_days"
value(g)
#> [1] 0

# Output and conditions are relayed to the master when the value is read.
h <- future({
  cat("printed on the worker\n")
  message("a message from the worker")
  1
})
res <- withCallingHandlers(
  value(h),
  message = \(m) {
    cat("relayed:", conditionMessage(m))
    invokeRestart("muffleMessage")
  }
)
#> printed on the worker
#> relayed: a message from the worker

# An error is re-signalled by value().
e <- future(stop("failure on the worker"))
tryCatch(value(e), error = conditionMessage)
#> [1] "failure on the worker"

# Always put the plan back.
plan(sequential)
nbrOfWorkers()
#> [1] 1

source("scenario.R")
library(future)

# Change the plan for one block of code only (future 1.40.0 and later).
with(plan(multisession, workers = 2), {
  cat("inside:", nbrOfWorkers(), "workers\n")
})
#> inside: 2 workers
cat("after:", nbrOfWorkers(), "worker\n")
#> after: 1 worker

# Inside a function, local = TRUE restores the plan when the function exits.
count_workers <- function() {
  with(plan(multisession, workers = 2), local = TRUE)
  nbrOfWorkers()
}
count_workers()
#> [1] 2
nbrOfWorkers()
#> [1] 1

# The older idiom: plan() returns the previous plan, to put back later.
old_plan <- plan(multisession, workers = 2)
nbrOfWorkers()
#> [1] 2
plan(old_plan)
nbrOfWorkers()
#> [1] 1

# Forked workers: only where forking is available and safe.
can_fork <- parallelly::supportsMulticore()
can_fork
#> [1] TRUE
if (can_fork) {
  plan(multicore, workers = 2)
  f <- future(Sys.getpid())
  cat("forked worker:", value(f) != Sys.getpid(), "\n")
  plan(sequential)
}
#> forked worker: TRUE

# The cluster backend accepts a cluster you made yourself.
cl <- parallel::makeCluster(2)
plan(cluster, workers = cl)
nbrOfWorkers()
#> [1] 2
f <- future(Sys.getpid())
value(f) %in% unlist(parallel::clusterCall(cl, Sys.getpid))
#> [1] TRUE
plan(sequential)
parallel::stopCluster(cl)

# A sequential plan evaluates in this process, which makes debugging easy.
plan(sequential)
f <- future(Sys.getpid())
value(f) == Sys.getpid()
#> [1] TRUE

# multiprocess was removed in future 1.33.0 (2023); its replacement is
# multisession or multicore.
exists("multiprocess", envir = asNamespace("future"))
#> [1] FALSE

library(future)
plan(multisession, workers = 2)

# A future that draws random numbers without seed = TRUE is flagged.
f <- future(rnorm(1))
w <- tryCatch(value(f), warning = identity)
class(w)[1:2]
#> [1] "RngFutureWarning" "FutureWarning"
msg <- conditionMessage(w)
sub("\\(<unnamed-[0-9]+>\\)", "(<unnamed-n>)", sub(" \\[future .*$", "", msg))
#> [1] "UNRELIABLE VALUE: Future (<unnamed-n>) unexpectedly generated random numbers without specifying argument 'seed'. There is a risk that those random numbers are not statistically sound and the overall results might be invalid. To fix this, specify 'seed=TRUE'. This ensures that proper, parallel-safe random numbers are produced. To disable this check, use 'seed=NULL', or set option 'future.rng.onMisuse' to \"ignore\"."

# seed = TRUE gives each future a proper L'Ecuyer-CMRG stream. The numbers
# are random, so the stream is fixed by a seed given as an integer.
a <- value(future(rnorm(2), seed = 282))
b <- value(future(rnorm(2), seed = 282))
identical(a, b)
#> [1] TRUE
d <- value(future(rnorm(2), seed = 283))
identical(a, d)
#> [1] FALSE

# The warning can be turned off for good by declaring the use of RNG...
old <- options(future.rng.onMisuse = "ignore")
length(value(future(rnorm(1))))
#> [1] 1
options(old)

# ... or the check can be raised to an error.
options(future.rng.onMisuse = "error")
r <- tryCatch(value(future(rnorm(1))), error = \(e) class(e)[1])
r
#> [1] "RngFutureError"
options(future.rng.onMisuse = "warning")
plan(sequential)

source("scenario.R")
library(future)
library(future.apply)

overdue_rate <- function(rows, today, days) {
  end <- rows$return_date
  end[is.na(end)] <- today
  mean(as.integer(end - rows$loan_date) > days, na.rm = TRUE)
}
members <- tolower(trimws(loans$member))
keep <- nzchar(members)
by_member <- split(loans[keep, ], members[keep])
serial <- lapply(by_member, overdue_rate, today = shelf_today, days = loan_days)

plan(multisession, workers = 2)

# future_lapply() keeps the names and the shape of lapply(); the globals
# overdue_rate, shelf_today and loan_days are found and exported for us.
par <- future_lapply(by_member, \(rows) {
  overdue_rate(rows, shelf_today, loan_days)
})
identical(par, serial)
#> [1] TRUE

# future_vapply() checks the type of every result, like vapply().
rates <- future_vapply(
  by_member, overdue_rate, numeric(1), today = shelf_today, days = loan_days
)
all.equal(rates, unlist(serial))
#> [1] TRUE

# future_mapply() walks several vectors together.
future_mapply(\(a, b) a + b, 1:3, 4:6)
#> [1] 5 7 9

# Chunking: one chunk per worker by default (future.scheduling = 1).
# future.chunk.size sets the number of elements per chunk.
par_chunks <- future_lapply(
  by_member, overdue_rate, today = shelf_today, days = loan_days,
  future.chunk.size = 2
)
identical(par_chunks, serial)
#> [1] TRUE

# Random numbers need future.seed. With it the result does not depend on
# the plan, nor on the number of workers.
durations <- as.integer(loans$return_date - loans$loan_date)
durations <- durations[!is.na(durations) & durations >= 0L]
boot_mean <- function(i) mean(sample(durations, replace = TRUE))
on_two <- future_sapply(1:6, boot_mean, future.seed = 282)
plan(multisession, workers = 3)
on_three <- future_sapply(1:6, boot_mean, future.seed = 282)
plan(sequential)
in_process <- future_sapply(1:6, boot_mean, future.seed = 282)
c(identical(on_two, on_three), identical(on_two, in_process))
#> [1] TRUE TRUE
round(on_two, 2)
#> [1] 18.98 20.73 20.82 19.31 18.08 21.63

# Without future.seed the same call warns.
plan(multisession, workers = 2)
w <- tryCatch(future_sapply(1:2, boot_mean), warning = identity)
class(w)[1]
#> [1] "RngFutureWarning"
msg <- sub(" \\[future .*$", "", conditionMessage(w))
sub("\\(‘future_sapply-[0-9]+’\\)", "(‘future_sapply-n’)", msg)
#> [1] "UNRELIABLE VALUE: One of the ‘future.apply’ iterations (‘future_sapply-n’) unexpectedly generated random numbers without declaring so. There is a risk that those random numbers are not statistically sound and the overall results might be invalid. To fix this, specify 'future.seed=TRUE'. This ensures that proper, parallel-safe random numbers are produced via a parallel RNG method. To disable this check, use 'future.seed = NULL', or set option 'future.rng.onMisuse' to \"ignore\"."
plan(sequential)

source("scenario.R")
library(mirai)

overdue_rate <- function(rows, today, days) {
  end <- rows$return_date
  end[is.na(end)] <- today
  mean(as.integer(end - rows$loan_date) > days, na.rm = TRUE)
}
members <- tolower(trimws(loans$member))
keep <- nzchar(members)
by_member <- split(loans[keep, ], members[keep])
serial <- lapply(by_member, overdue_rate, today = shelf_today, days = loan_days)

# daemons(2) starts two persistent background R processes.
daemons(2)
status()$connections
#> [1] 2

# mirai() returns at once; the code runs in a clean environment, so
# everything it needs is passed by name.
m <- mirai(
  {
    Sys.sleep(0.5)
    overdue_rate(rows, today, days)
  },
  overdue_rate = overdue_rate, rows = by_member[[1]],
  today = shelf_today, days = loan_days
)
class(m)
#> [1] "mirai"   "recvAio"
unresolved(m)
#> [1] TRUE
m[]
#> [1] 0.6153846
round(m$data, 2)
#> [1] 0.62
unresolved(m)
#> [1] FALSE

# The value is read from $data; before the result arrives it is an
# "unresolved" NA, and call_mirai() waits for it.
m2 <- mirai({
  Sys.sleep(0.3)
  "late"
})
is.na(m2$data)
#> [1] TRUE
call_mirai(m2)$data
#> [1] "late"

# What is not passed is not there.
m3 <- mirai(overdue_rate(rows, today, days), rows = by_member[[1]],
            today = shelf_today, days = loan_days)
m3[]
#> 'miraiError' chr Error in overdue_rate(rows, today, days): could not find function "overdue_rate"

# An error is a value, not an exception.
m4 <- mirai(stop("boom"))
call_mirai(m4)
is_mirai_error(m4$data)
#> [1] TRUE
is_error_value(m4$data)
#> [1] TRUE
class(m4$data)
#> [1] "miraiError" "errorValue" "try-error"

# A timeout is a value as well: the integer 5.
m5 <- mirai(Sys.sleep(5), .timeout = 200)
call_mirai(m5)$data
#> 'errorValue' int 5 | Timed out
is_error_value(m5$data)
#> [1] TRUE

# mirai_map() maps over a list, one mirai per element, and collects with [].
par <- mirai_map(
  by_member, overdue_rate,
  .args = list(today = shelf_today, days = loan_days)
)[]
identical(par, serial)
#> [1] TRUE

# Extra arguments in ... are objects for the daemons, not arguments of .f.
bad <- mirai_map(1:2, \(x, y) x + y, y = 10)[]
bad[[1]]
#> 'miraiError' chr Error in (function (x, y) : argument "y" is missing, with no default
ok <- mirai_map(1:2, \(x, y) x + y, .args = list(y = 10))[.flat]
ok
#> [1] 11 12

# Errors in a map are returned element by element, so only they can be re-run.
mixed <- mirai_map(list(1, "a", 3), \(x) x + 1)[]
vapply(mixed, is_mirai_error, logical(1))
#> [1] FALSE  TRUE FALSE

# Run code on every daemon, for example to load a package.
invisible(everywhere(library(stats)))

# seed makes the random numbers of mirai_map() reproducible, whatever the
# number of daemons.
daemons(0)
daemons(2, seed = 282)
first <- mirai_map(1:4, \(i) rnorm(1))[.flat]
daemons(0)
daemons(3, seed = 282)
second <- mirai_map(1:4, \(i) rnorm(1))[.flat]
daemons(0)
identical(first, second)
#> [1] TRUE

# Without daemons, mirai() starts a fresh background process per request.
m6 <- mirai(Sys.getpid())
call_mirai(m6)$data == Sys.getpid()
#> [1] FALSE
tryCatch(mirai_map(1:2, identity), error = conditionMessage)
#> [1] "No daemons set.\nUse e.g. `mirai::daemons(6)` to set 6 local daemons."

library(purrr)
mirai::daemons(2)

offset <- 10
map_dbl(1:3, in_parallel(\(x) x + offset, offset = offset))

mirai::daemons(0)

library(parallel)
library(mirai)

# R 4.6.1 knows a "MIRAI" cluster type: the parallel API on mirai daemons.
cl <- makeCluster(2, type = "MIRAI")
class(cl)
#> [1] "miraiCluster" "cluster"
unlist(parLapply(cl, 1:4, \(i) i^2))
#> [1]  1  4  9 16
stopCluster(cl)

# purrr 1.1.0 added in_parallel(); it needs the carrier package.
exists("in_parallel", asNamespace("purrr"))
#> [1] TRUE
tryCatch(purrr::in_parallel(\(x) x + 1), error = \(e) conditionMessage(e))
#> [1] "The package \"carrier\" (>= 0.3.0) is required for parallel map."
isTRUE(requireNamespace("carrier", quietly = TRUE))
#> [1] FALSE

# The adapter picks up the daemons you set; without them it runs serially.
exists("ExtendedTask", asNamespace("shiny"))
#> [1] TRUE
daemons(0)

library(later)
library(promises)

# later() schedules a function; nothing runs until the event loop gets
# control. In a script that is run_now(); at the console it is the prompt.
log <- "before"
later(\() log <<- c(log, "ran later"), delay = 0.1)
log
#> [1] "before"
run_now(timeoutSecs = 1)
log
#> [1] "before"    "ran later"

# The value returned by later() cancels the callback.
stop_it <- later(\() cat("never printed\n"), delay = 5)
stop_it()
loop_empty()
#> [1] TRUE

# A promise stands for a value that is not there yet.
slow_value <- promise(\(resolve, reject) {
  later(\() resolve(42), delay = 0.1)
})
class(slow_value)
#> [1] "promise"
slow_value |> then(\(v) cat("got", v, "\n"))
cat("this line comes first\n")
#> this line comes first
while (!loop_empty()) run_now(timeoutSecs = 0.5)
#> got 42

# then() chains, catch() handles a rejection.
failing <- promise(\(resolve, reject) {
  later(\() reject("failed upstream"), delay = 0.1)
})
failing |>
  then(\(v) cat("not reached\n")) |>
  catch(\(e) cat("caught:", conditionMessage(e), "\n"))
while (!loop_empty()) run_now(timeoutSecs = 0.5)
#> caught: failed upstream

# The same with the older promise pipe, which is superseded by then().
slow_value %...>% (\(v) cat("piped", v, "\n"))
while (!loop_empty()) run_now(timeoutSecs = 0.5)
#> piped 42

# A mirai is a promise: as.promise() lets the event loop deliver it.
library(mirai)
daemons(2)
m <- mirai({
  Sys.sleep(0.3)
  "mirai finished"
})
as.promise(m) |> then(\(v) cat(v, "\n"))
while (unresolved(m)) run_now(timeoutSecs = 0.1)
run_now()
daemons(0)

# So is a future, through future_promise() (a promise-friendly future()).
library(future)
plan(multisession, workers = 2)
fp <- future_promise({
  Sys.sleep(0.3)
  "future finished"
})
fp |> then(\(v) cat(v, "\n"))
while (!loop_empty()) run_now(timeoutSecs = 0.5)
#> mirai finished
#> future finished
plan(sequential)

library(callr)

# r_bg() runs a function in a fresh background R process and returns.
bg <- r_bg(function(n) {
  Sys.sleep(0.3)
  sum(seq_len(n))
}, args = list(10L))
class(bg)[1]
#> [1] "r_process"
bg$is_alive()
#> [1] TRUE
invisible(bg$wait())
bg$is_alive()
#> [1] FALSE
bg$get_result()
#> [1] 55
bg$get_exit_status()
#> [1] 0

# The function runs in a clean session: pass what it needs in args.
x <- 1
bad <- r_bg(function() x + 1)
invisible(bad$wait())
tryCatch(bad$get_result(), error = conditionMessage)
#> [1] "! in callr subprocess.\nCaused by error in `(function () ...`:\n! object 'x' not found"
good <- r_bg(function(x) x + 1, args = list(x = x))
invisible(good$wait())
good$get_result()
#> [1] 2

# Errors in the child are signalled by get_result().
failing <- r_bg(function() stop("child failed"))
invisible(failing$wait())
err <- tryCatch(failing$get_result(), error = identity)
class(err)[1]
#> [1] "callr_status_error"
conditionMessage(err)
#> [1] "! in callr subprocess.\nCaused by error in `(function () ...`:\n! child failed"

# Output can be piped back, and a long job can be killed.
talker <- r_bg(function() {
  cat("hello from the child\n")
  1
}, stdout = "|")
invisible(talker$wait())
talker$read_all_output()
#> [1] "hello from the child\n"

sleeper <- r_bg(function() Sys.sleep(30))
sleeper$is_alive()
#> [1] TRUE
invisible(sleeper$kill())
invisible(sleeper$wait())
sleeper$is_alive()
#> [1] FALSE

# r() is the blocking version.
callr::r(function() Sys.getpid()) == Sys.getpid()
#> [1] FALSE