In gfkse/bettermc: Enhanced Fork-Based Parallelization

knitr::opts_chunk$set(cache = TRUE)

# knitr hook function to allow an output.lines option
# e.g., 
#   output.lines=12 prints lines 1:12 ...
#   output.lines=1:12 does the same
#   output.lines=3:15 prints lines ... 3:15 ...
#   output.lines=-(1:8) removes lines 1:8 and prints ... 9:n ...
#   No allowance for anything but a consecutive range of lines
#   
# adopted from https://stackoverflow.com/a/23205752

create_output_hook <- function(type) {
  hook_output <- knitr::knit_hooks$get(type)
  function(x, options) {
    lines <- options$output.lines
    if (is.null(lines)) {
      return(hook_output(x, options))  # pass to default hook
    }
    x <- unlist(strsplit(x, "\n"))
    more <- "..."
    if (length(lines) == 1) {  # first n lines
      if (length(x) > lines) {
        # truncate the output, but add ...
        x <- c(head(x, lines), more)
      }
    } else {
      x <- c(if (abs(lines[1]) > 1) more else NULL, 
             x[lines], 
             if (length(x) > lines[abs(length(lines))]) more else NULL
      )
    }
    # paste these lines together
    x <- paste(c(x, ""), collapse = "\n")
    hook_output(x, options)
  }
}

knitr::knit_hooks$set(output = create_output_hook("output"))
knitr::knit_hooks$set(error = create_output_hook("error"))
knitr::knit_hooks$set(warning = create_output_hook("warning"))
knitr::knit_hooks$set(message = create_output_hook("message"))

bettermc

The bettermc package provides a wrapper around the parallel::mclapply function for better performance, error handling, seeding and UX.

Installation of the Development Version

# install.packages("devtools")
devtools::install_github("gfkse/bettermc")

Supported Platforms

bettermc was originally developed for 64-bit Linux. By now it should also compile and run on 32-bit systems, and on macOS and Solaris. However, as stated in the respective help pages, not all features are supported on macOS. Porting to other POSIX-compliant Unix flavors should be fairly straightforward. The Windows support is very limited and mainly provided for compatibility reasons only, i.e. to allow the serial execution of code using bettermc::mclapply, which was originally developed for Linux or macOS.

Features

Here is a short overview on its main features ...

Progress Bar

progress bar

Error Handling, Tracebacks and Crashdumps

By default, crashdumps and full tracebacks are generated on errors in child processes:

g <- function(x) x + 1
f <- function(x) g(as.character(x))
bettermc::mclapply(1:2, f)

# in a non-interactive session a file "last.dump.rds" is created
last.dump <- readRDS("last.dump.rds")

# in an interactive session use debugger() instead of print() for actual debugging
print(attr(last.dump[[1L]], "dump.frames"))

As shown in the example above, bettermc by default fails if there are errors in child processes. This behavior can be changed to merely warn about both fatal and non-fatal error:

ret <- bettermc::mclapply(1:4, function(i) {
  if (i == 1L)
    stop(i)
  else if (i == 4L)
    system(paste0("kill ", Sys.getpid()))
  NULL
}, mc.allow.fatal = TRUE, mc.allow.error = TRUE, mc.preschedule = FALSE)

Also in this case, full tracebacks and crash dumps are available:

stopifnot(inherits(ret[[1]], "try-error"))
names(attributes(ret[[1L]]))

Additionally, results affected by fatal errors are clearly indicated and can be differentiated from legitimate NULL values:

lapply(ret, class)

You can use mc.allow.fatal = NULL to instead return NULL on fatal errors as does parallel::mclapply.

Output, Messages and Warnings

In contrast to parallel::mclapply, neither output nor messages nor warnings from the child processes are lost. All of these can be forwarded to the parent process and are prefixed with the index of the function invocation from which they originate:

f <- function(i) {
  if (i == 1) message(letters[i])
  else if (i == 2) warning(letters[i])
  else print(letters[i])

  i
}
ret <- bettermc::mclapply(1:4, f)

Similarly, other conditions can also be re-signaled in the parent process.

Reproducible Seeding

By default, bettermc reproducibly seeds all function calls:

set.seed(538)
a <- bettermc::mclapply(1:4, function(i) runif(1), mc.cores = 3)
set.seed(538)
b <- bettermc::mclapply(1:4, function(i) runif(1), mc.cores = 1)
a
stopifnot(identical(a, b))

Compare with parallel:

set.seed(594)
a <- parallel::mclapply(1:4, function(i) runif(1), mc.cores = 3)
set.seed(594)
b <- parallel::mclapply(1:4, function(i) runif(1), mc.cores = 3)
stopifnot(identical(a, b))

POSIX Shared Memory

Many types of objects can be returned from the child processes using POSIX shared memory. This includes e.g. logical, numeric, complex and raw vectors and arrays as well as factors. In doing so, the overhead of getting larger results back into the parent R process is reduced:

X <- data.frame(
  x = runif(3e7),
  y = sample(c(TRUE, FALSE), 3e7, TRUE),
  z = 1:3e7
)
f <- function(i) X

microbenchmark::microbenchmark(
  bettermc1 = bettermc::mclapply(1:2, f, mc.share.copy = FALSE),
  bettermc2 = bettermc::mclapply(1:2, f),
  bettermc3 = bettermc::mclapply(1:2, f, mc.share.vectors = FALSE),
  bettermc4 = bettermc::mclapply(1:2, f, mc.share.vectors = FALSE, mc.shm.ipc = FALSE),
  parallel = parallel::mclapply(1:2, f),
  times = 10, setup = gc()
)

In examples bettermc1 and bettermc2, the child processes place the columns of the return value X in shared memory. The object which needs to be serialized for transfer from child to parent processes hence becomes:

X_shm <- bettermc:::vectors2shm(X, name_prefix = "/bettermc_README_")
str(X_shm)

Column z is an ALTREP and, because it can be serialized efficiently, is left alone by default. The parent process can recover the original object from X_shm:

Y <- bettermc:::shm2vectors(X_shm)
stopifnot(identical(X, Y))

The internal functions vectors2shm() and shm2vectors() recursively walk the return value and apply the exported functions copy2shm() and allocate_from_shm(), respectively.

In bettermc1, the shared memory objects are used directly by the parent process. In bettermc2, which is the default, new vectors are allocated in the parent process and the data is merely copied from the shared memory objects, which are freed afterwards. See ?copy2shm for more details on this topic and why the slower mc.share.copy = TRUE might be a sensible default.

In bettermc3, the original X is serialized and the resulting raw vector is placed in shared memory, from where it is deserialized in the parent process.

bettermc4 does not involve any POSIX shared memory and hence is equivalent to parallel, i.e. the original X is serialized and transferred to the parent process using pipes.

Character Compression

In practice, character vectors often contain a substantial amount of duplicated values. This is exploited by bettermc to speed up the returning of larger character vectors from child processes:

X <- rep(as.character(runif(1e6)), 30)
f <- function(i) X
microbenchmark::microbenchmark(
  bettermc1 = bettermc::mclapply(1:2, f),
  parallel =  parallel::mclapply(1:2, f),
  times = 1, setup = gc()
)

By default, bettermc replaces character vectors with objects of type char_map before returning them to the parent process:

X_comp <- bettermc::compress_chars(X)
str(X_comp)

The important detail here is the length of the chars vector, which just contains the unique elements of X and hence is significantly faster to (de)serialize than the original vector. The parent process can recover the original character vectors:

Y <- bettermc::uncompress_chars(X_comp)
stopifnot(identical(X, Y))

The functions compress_chars() and uncompress_chars() recursively walk the return value and apply the functions char_map() and map2char(), respectively.

char_map() is implemented using a radix sort, which makes it very efficient:

microbenchmark::microbenchmark(
  char_map = bettermc::char_map(X),
  match = {chars <- unique(X); idx <- match(X, chars)},
  times = 3, setup = gc()
)

Retries

bettermc supports automatic retries on both fatal and non-fatal errors. mc.force.fork ensures that FUN is called in a child process, even if X is of length 1. This is useful if FUN might encounter a fatal error and we want to protect the parent process against it. With retires, length(X) might drop to 1 if all other values could already be processed. This is also why we need mc.force.fork in the following example:

set.seed(456)
res <-
  bettermc::mclapply(1:20, function(i) {
    r <- runif(1)
    if (r < 0.25)
      system(paste0("kill ", Sys.getpid()))
    else if (r < 0.5)
      stop(i)
    else
      i
  }, mc.retry = 50, mc.cores = 10, mc.force.fork = TRUE)
stopifnot(identical(res, as.list(1:20)))

Additionally, it is possible to automatically decrease the number of cores with every retry by specifying a negative value for mc.retry. This is useful if we expect failures to be caused simply by too many concurrent processes, e.g. if system load or the size of input data is unpredictable and might lead to the Linux Out Of Memory Killer stepping in. In such a case it makes sense to retry using fewer concurrent processes:

ppid <- Sys.getpid()
res <-
  bettermc::mclapply(1:20, function(i) {
    Sys.sleep(0.25)  # wait for the other child processes
    number_of_child_processes <- length(system(paste0("pgrep -P ", ppid), intern = TRUE))
    if (number_of_child_processes >= 5) system(paste0("kill ", Sys.getpid()))
    i
  }, mc.retry = -3, mc.cores = 10, mc.force.fork = TRUE)

stopifnot(identical(res, as.list(1:20)))

If there are still errors after the retries, we regularly fail:

set.seed(123)
res <-
  bettermc::mclapply(1:20, function(i) {
    r <- runif(1)
    if (r < 0.25)
      system(paste0("kill ", Sys.getpid()))
    else if (r < 0.5)
      stop(i)
    else
      i
  }, mc.retry = 1, mc.cores = 10, mc.force.fork = TRUE)