Become an expert in R — Interactive courses, Cheat Sheets, certificates and more!
Get Started for Free

makeClusterPSOCK

Create a PSOCK Cluster of R Workers for Parallel Processing


Description

The makeClusterPSOCK() function creates a cluster of R workers for parallel processing. These R workers may be background R sessions on the current machine, R sessions on external machines (local or remote), or a mix of such. For external workers, the default is to use SSH to connect to those external machines. This function works similarly to makePSOCKcluster() of the parallel package, but provides additional and more flexibility options for controlling the setup of the system calls that launch the background R workers, and how to connect to external machines.

Usage

makeClusterPSOCK(
  workers,
  makeNode = makeNodePSOCK,
  port = c("auto", "random"),
  ...,
  autoStop = FALSE,
  tries = getOptionOrEnvVar("parallelly.makeNodePSOCK.tries", 3L),
  delay = getOptionOrEnvVar("parallelly.makeNodePSOCK.tries.delay", 15),
  validate = getOptionOrEnvVar("parallelly.makeNodePSOCK.validate", TRUE),
  verbose = getOptionOrEnvVar("parallelly.debug", FALSE)
)

makeNodePSOCK(
  worker = "localhost",
  master = NULL,
  port,
  connectTimeout = getOptionOrEnvVar("parallelly.makeNodePSOCK.connectTimeout", 2 * 60),
  timeout = getOptionOrEnvVar("parallelly.makeNodePSOCK.timeout", 30 * 24 * 60 * 60),
  rscript = NULL,
  homogeneous = NULL,
  rscript_args = NULL,
  rscript_envs = NULL,
  rscript_libs = NULL,
  rscript_startup = NULL,
  methods = TRUE,
  useXDR = getOptionOrEnvVar("parallelly.makeNodePSOCK.useXDR", FALSE),
  outfile = "/dev/null",
  renice = NA_integer_,
  rshcmd = getOptionOrEnvVar("parallelly.makeNodePSOCK.rshcmd", NULL),
  user = NULL,
  revtunnel = TRUE,
  rshlogfile = NULL,
  rshopts = getOptionOrEnvVar("parallelly.makeNodePSOCK.rshopts", NULL),
  rank = 1L,
  manual = FALSE,
  dryrun = FALSE,
  quiet = FALSE,
  verbose = FALSE
)

Arguments

workers

The hostnames of workers (as a character vector) or the number of localhost workers (as a positive integer).

makeNode

A function that creates a "SOCKnode" or "SOCK0node" object, which represents a connection to a worker.

port

The port number of the master used for communicating with all the workers (via socket connections). If an integer vector of ports, then a random one among those is chosen. If "random", then a random port in is chosen from 11000:11999, or from the range specified by environment variable R_PARALLELLY_RANDOM_PORTS. If "auto" (default), then the default (single) port is taken from environment variable R_PARALLEL_PORT, otherwise "random" is used. Note, do not use this argument to specify the port number used by rshcmd, which typically is an SSH client. Instead, if the SSH daemon runs on a different port than the default 22, specify the SSH port by appending it to the hostname, e.g. "remote.server.org:2200" or via SSH options -p, e.g. rshopts = c("-p", "2200").

...

Optional arguments passed to makeNode(workers[i], ..., rank = i) where i = seq_along(workers).

autoStop

If TRUE, the cluster will be automatically stopped using stopCluster() when it is garbage collected, unless already stopped. See also autoStopCluster().

tries, delay

Maximum number of attempts done to launch each node with makeNode() and the delay (in seconds) in-between attempts. If argument port specifies more than one port, e.g. port = "random" then a random port will be drawn and validated at most tries times.

validate

If TRUE, after the nodes have been created, they are all validated that they work by inquiring about their session information, which is saved in attribute session_info of each node.

verbose

If TRUE, informative messages are outputted.

worker

The hostname or IP number of the machine where the worker should run.

master

The hostname or IP number of the master / calling machine, as known to the workers. If NULL (default), then the default is Sys.info()[["nodename"]] unless worker is localhost or revtunnel = TRUE in case it is "localhost".

connectTimeout

The maximum time (in seconds) allowed for each socket connection between the master and a worker to be established (defaults to 2 minutes). See note below on current lack of support on Linux and macOS systems.

timeout

The maximum time (in seconds) allowed to pass without the master and a worker communicate with each other (defaults to 30 days).

rscript, homogeneous

The system command for launching Rscript on the worker and whether it is installed in the same path as the calling machine or not. For more details, see below.

rscript_args

Additional arguments to Rscript (as a character vector). This argument can be used to customize the R environment of the workers before they launches. For instance, use rscript_args = c("-e", shQuote('setwd("/path/to")')) to set the working directory to ‘/path/to’ on all workers.

rscript_envs

A named character vector environment variables to set on worker at startup, e.g. rscript_envs = c(FOO = "3.14", "HOME", "UNKNOWN"). If an element is not named, then the value of that variable will be used as the name and the value will be the value of Sys.getenv() for that variable. Non-existing environment variables will be dropped. These variables are set using Sys.setenv().

rscript_libs

A character vector of R library paths that will be used for the library search path of the R workers. An asterisk ("*") will be resolved to the default .libPaths() on the worker. That is, to prepend a folder, instead of replacing the existing ones, use rscript_libs = c("new_folder", "*"). To pass down a non-default library path currently set on the main R session to the workers, use rscript_libs = .libPaths().

rscript_startup

An R expression or a character vector of R code, or a list with a mix of these, that will be evaluated on the R worker prior to launching the worker's event loop. For instance, use rscript_startup = 'setwd("/path/to")' to set the working directory to ‘/path/to’ on all workers.

methods

If TRUE, then the methods package is also loaded.

useXDR

If TRUE, the communication between master and workers, which is binary, will use big-endian (XDR).

outfile

Where to direct the stdout and stderr connection output from the workers. If NULL, then no redirection of output is done, which means that the output is relayed in the terminal on the local computer. On Windows, the output is only relayed when running R from a terminal but not from a GUI.

renice

A numerical 'niceness' (priority) to set for the worker processes.

rshcmd, rshopts

The command (character vector) to be run on the master to launch a process on another host and any additional arguments (character vector). These arguments are only applied if machine is not localhost. For more details, see below.

user

(optional) The user name to be used when communicating with another host.

revtunnel

If TRUE, a reverse SSH tunnel is set up for each worker such#' that the worker R process sets up a socket connection to its local port (port - rank + 1) which then reaches the master on port port. If FALSE, then the worker will try to connect directly to port port on master. For more details, see below.

rshlogfile

(optional) If a filename, the output produced by the rshcmd call is logged to this file, of if TRUE, then it is logged to a temporary file. The log file name is available as an attribute as part of the return node object. Warning: This only works with SSH clients that support option -E out.log. For example, PuTTY's plink does not support this option, and any attempts to specify rshlogfile will cause the SSH connection to fail.

rank

A unique one-based index for each worker (automatically set).

manual

If TRUE the workers will need to be run manually. The command to run will be displayed.

dryrun

If TRUE, nothing is set up, but a message suggesting how to launch the worker from the terminal is outputted. This is useful for troubleshooting.

quiet

If TRUE, then no output will be produced other than that from using verbose = TRUE.

Value

An object of class c("RichSOCKcluster", "SOCKcluster", "cluster") consisting of a list of "SOCKnode" or "SOCK0node" workers (that also inherit from RichSOCKnode).

makeNodePSOCK() returns a "SOCKnode" or "SOCK0node" object representing an established connection to a worker.

Definition of localhost

A hostname is considered to be localhost if it equals:

  • "localhost",

  • "127.0.0.1", or

  • Sys.info()[["nodename"]].

It is also considered localhost if it appears on the same line as the value of Sys.info()[["nodename"]] in file ‘/etc/hosts’.

Default SSH client and options (arguments rshcmd and rshopts)

Arguments rshcmd and rshopts are only used when connecting to an external host.

The default method for connecting to an external host is via SSH and the system executable for this is given by argument rshcmd. The default is given by option parallelly.makeNodePSOCK.rshcmd. If that is not set, then the default is to use ssh on Unix-like systems, including macOS as well as Windows 10. On older MS Windows versions, which does not have a built-in ssh client, the default is to use (i) plink from the PuTTY project, and then (ii) the ssh client that is distributed with RStudio.

PuTTY puts itself on Windows' system PATH when installed, meaning this function will find PuTTY automatically if installed. If not, to manually set specify PuTTY as the SSH client, specify the absolute pathname of ‘plink.exe’ in the first element and option -ssh in the second as in rshcmd = c("C:/Path/PuTTY/plink.exe", "-ssh"). This is because all elements of rshcmd are individually "shell" quoted and element rshcmd[1] must be on the system PATH.

Furthermore, when running R from RStudio on Windows, the ssh client that is distributed with RStudio will also be considered. This client, which is from MinGW MSYS, is searched for in the folder given by the RSTUDIO_MSYS_SSH environment variable - a variable that is (only) set when running RStudio. To use this SSH client outside of RStudio, set RSTUDIO_MSYS_SSH accordingly.

You can override the default set of SSH clients that are searched for by specifying them in argument rshcmd or via option parallelly.makeNodePSOCK.rshcmd using the format <...>, e.g. rshcmd = c("<rstudio-ssh>", "<putty-plink>", "<ssh>"). See below for examples.

If no SSH-client is found, an informative error message is produced.

Additional SSH options may be specified via argument rshopts, which defaults to option parallelly.makeNodePSOCK.rshopts. For instance, a private SSH key can be provided as rshopts = c("-i", "~/.ssh/my_private_key"). PuTTY users should specify a PuTTY PPK file, e.g. rshopts = c("-i", "C:/Users/joe/.ssh/my_keys.ppk"). Contrary to rshcmd, elements of rshopts are not quoted.

Accessing external machines that prompts for a password

IMPORTANT: With one exception, it is not possible to for these functions to log in and launch R workers on external machines that requires a password to be entered manually for authentication. The only known exception is the PuTTY client on Windows for which one can pass the password via command-line option -pw, e.g. rshopts = c("-pw", "MySecretPassword").

Note, depending on whether you run R in a terminal or via a GUI, you might not even see the password prompt. It is also likely that you cannot enter a password, because the connection is set up via a background system call.

The poor man's workaround for setup that requires a password is to manually log into the each of the external machines and launch the R workers by hand. For this approach, use manual = TRUE and follow the instructions which include cut'n'pasteable commands on how to launch the worker from the external machine.

However, a much more convenient and less tedious method is to set up key-based SSH authentication between your local machine and the external machine(s), as explain below.

Accessing external machines via key-based SSH authentication

The best approach to automatically launch R workers on external machines over SSH is to set up key-based SSH authentication. This will allow you to log into the external machine without have to enter a password.

Key-based SSH authentication is taken care of by the SSH client and not R. To configure this, see the manuals of your SSH client or search the web for "ssh key authentication".

Reverse SSH tunneling

The default is to use reverse SSH tunneling (revtunnel = TRUE) for workers running on other machines. This avoids the complication of otherwise having to configure port forwarding in firewalls, which often requires static IP address as well as privileges to edit the firewall on your outgoing router, something most users don't have. It also has the advantage of not having to know the internal and / or the public IP address / hostname of the master. Yet another advantage is that there will be no need for a DNS lookup by the worker machines to the master, which may not be configured or is disabled on some systems, e.g. compute clusters.

Argument rscript

If homogeneous is FALSE, the rscript defaults to "Rscript", i.e. it is assumed that the Rscript executable is available on the PATH of the worker. If homogeneous is TRUE, the rscript defaults to file.path(R.home("bin"), "Rscript"), i.e. it is basically assumed that the worker and the caller share the same file system and R installation.

If specified, argument rscript should be a character vector with one more more elements. all elements are automatically shell quoted using base::shQuote(), except those that are of format <ENVVAR>=<VALUE>, that is, the ones matching the regular expression '^[[:alpha:]_][[:alnum:]_]*=.*'. Another exception is when rscript inherits from 'AsIs'.

Default value of argument homogeneous

The default value of homogeneous is TRUE if and only if either of the following is fulfilled:

  • worker is localhost

  • revtunnel is FALSE and master is localhost

  • worker is neither an IP number nor a fully qualified domain name (FQDN). A hostname is considered to be a FQDN if it contains one or more periods

In all other cases, homogeneous defaults to FALSE.

Connection time out

Argument connectTimeout does not work properly on Unix and macOS due to limitation in R itself. For more details on this, please see R-devel thread 'BUG?: On Linux setTimeLimit() fails to propagate timeout error when it occurs (works on Windows)' on 2016-10-26 (https://stat.ethz.ch/pipermail/r-devel/2016-October/073309.html). When used, the timeout will eventually trigger an error, but it won't happen until the socket connection timeout timeout itself happens.

Communication time out

If there is no communication between the master and a worker within the timeout limit, then the corresponding socket connection will be closed automatically. This will eventually result in an error in code trying to access the connection.

Failing to set up local workers

When setting up a cluster of localhost workers, that is, workers running on the same machine as the master R process, occasionally a connection to a worker ("cluster node") may fail to be set up. When this occurs, an informative error message with troubleshooting suggestions will be produced. The most common reason for such localhost failures is due to port clashes. Retrying will often resolve the problem.

Failing to set up remote workers

A cluster of remote workers runs R processes on external machines. These external R processes are launched over, typically, SSH to the remote machine. For this to work, each of the remote machines needs to have R installed, which preferably is of the same version as what is on the main machine. For this to work, it is required that one can SSH to the remote machines. Ideally, the SSH connections use authentication based on public-private SSH keys such that the set up of the remote workers can be fully automated (see above). If makeClusterPSOCK() fails to set up one or more remote R workers, then an informative error message is produced. There are a few reasons for failing to set up remote workers. If this happens, start by asserting that you can SSH to the remote machine and launch ‘Rscript’ by calling something like:

{local}$ ssh -l alice remote.server.org
{remote}$ Rscript --version
R scripting front-end version 3.6.1 (2019-07-05)
{remote}$ logout
{local}$

When you have confirmed the above to work, then confirm that you can achieve the same in a single command-line call;

{local}$ ssh -l alice remote.server.org Rscript --version
R scripting front-end version 3.6.1 (2019-07-05)
{local}$

The latter will assert that you have proper startup configuration also for non-interactive shell sessions on the remote machine.

Another reason for failing to setup remote workers could be that they are running an R version that is not compatible with the version that your main R session is running. For instance, if we run R (>= 3.6.0) locally and the workers run R (< 3.5.0), we will get: Error in unserialize(node$con) : error reading from connection. This is because R (>= 3.6.0) uses serialization format version 3 whereas R (< 3.5.0) only supports version 2. We can see the version of the R workers by adding rscript_args = c("-e", shQuote("getRversion()")) when calling makeClusterPSOCK().

Examples

options(parallelly.debug = TRUE)

## NOTE: Drop 'dryrun = TRUE' below in order to actually connect.  Add
## 'verbose = TRUE' if you run into problems and need to troubleshoot.

## EXAMPLE: Two workers on the local machine
workers <- c("localhost", "localhost")
cl <- makeClusterPSOCK(workers, dryrun = TRUE)

## EXAMPLE: Three remote workers
## Setup of three R workers on two remote machines are set up
workers <- c("n1.remote.org", "n2.remote.org", "n1.remote.org")
cl <- makeClusterPSOCK(workers, dryrun = TRUE)


## EXAMPLE: Local and remote workers
## Same setup when the two machines are on the local network and
## have identical software setups
cl <- makeClusterPSOCK(
  workers,
  revtunnel = FALSE, homogeneous = TRUE,
  dryrun = TRUE
)

## EXAMPLE: Remote workers with specific setup
## Setup of remote worker with more detailed control on
## authentication and reverse SSH tunnelling
cl <- makeClusterPSOCK(
  "remote.server.org", user = "johnny",
  ## Manual configuration of reverse SSH tunnelling
  revtunnel = FALSE,
  rshopts = c("-v", "-R 11000:gateway:11942"),
  master = "gateway", port = 11942,
  ## Run Rscript nicely and skip any startup scripts
  rscript = c("nice", "/path/to/Rscript"),
  rscript_args = c("--vanilla"),
  dryrun = TRUE
)

## EXAMPLE: Two workers running in Docker on the local machine
## Setup of 2 Docker workers running rocker/r-parallel
cl <- makeClusterPSOCK(
  rep("localhost", times = 2L),
  ## Launch Rscript inside Docker container
  rscript = c(
    "docker", "run", "--net=host", "rocker/r-parallel",
    "Rscript"
  ),
  ## IMPORTANT: Because Docker runs inside a virtual machine (VM) on macOS
  ## and Windows (not Linux), when the R worker tries to connect back to
  ## the default 'localhost' it will fail, because the main R session is
  ## not running in the VM, but outside on the host.  To reach the host on
  ## macOS and Windows, make sure to use master = "host.docker.internal"
  # master = "host.docker.internal",  # <= macOS & Windows
  dryrun = TRUE
)


## EXAMPLE: Two workers running in Singularity on the local machine
## Setup of 2 Singularity workers running rocker/r-parallel
cl <- makeClusterPSOCK(
  rep("localhost", times = 2L),
  ## Launch Rscript inside Linux container
  rscript = c(
    "singularity", "exec", "docker://rocker/r-parallel",
    "Rscript"
  ),
  dryrun = TRUE
)


## EXAMPLE: One worker running in udocker on the local machine
## Setup of a single udocker.py worker running rocker/r-parallel
cl <- makeClusterPSOCK(
  "localhost",
  ## Launch Rscript inside Docker container (using udocker)
  rscript = c(
    "udocker.py", "run", "rocker/r-parallel",
    "Rscript"
  ), 
  ## Manually launch parallel workers
  ## (need double shQuote():s because udocker.py drops one level)
  rscript_args = c(
    "-e", shQuote(shQuote("parallel:::.slaveRSOCK()"))
  ),
  dryrun = TRUE
)


## EXAMPLE: Remote worker running on AWS
## Launching worker on Amazon AWS EC2 running one of the
## Amazon Machine Images (AMI) provided by RStudio
## (http://www.louisaslett.com/RStudio_AMI/)
public_ip <- "1.2.3.4"
ssh_private_key_file <- "~/.ssh/my-private-aws-key.pem"
cl <- makeClusterPSOCK(
  ## Public IP number of EC2 instance
  public_ip,
  ## User name (always 'ubuntu')
  user = "ubuntu",
  ## Use private SSH key registered with AWS
  rshopts = c(
    "-o", "StrictHostKeyChecking=no",
    "-o", "IdentitiesOnly=yes",
    "-i", ssh_private_key_file
  ),
  ## Set up .libPaths() for the 'ubuntu' user
  ## and then install the future package
  rscript_startup = quote(local({
    p <- Sys.getenv("R_LIBS_USER")
    dir.create(p, recursive = TRUE, showWarnings = FALSE)
    .libPaths(p)
    install.packages("future")
  })),
  dryrun = TRUE
)


## EXAMPLE: Remote worker running on GCE
## Launching worker on Google Cloud Engine (GCE) running a
## container based VM (with a #cloud-config specification)
public_ip <- "1.2.3.4"
user <- "johnny"
ssh_private_key_file <- "~/.ssh/google_compute_engine"
cl <- makeClusterPSOCK(
  ## Public IP number of GCE instance
  public_ip,
  ## User name (== SSH key label (sic!))
  user = user,
  ## Use private SSH key registered with GCE
  rshopts = c(
    "-o", "StrictHostKeyChecking=no",
    "-o", "IdentitiesOnly=yes",
    "-i", ssh_private_key_file
  ),
  ## Launch Rscript inside Docker container
  rscript = c(
    "docker", "run", "--net=host", "rocker/r-parallel",
    "Rscript"
  ),
  dryrun = TRUE
)


## EXAMPLE: Remote worker running on Linux from Windows machine
## Connect to remote Unix machine 'remote.server.org' on port 2200
## as user 'bob' from a Windows machine with PuTTY installed.
## Using the explicit special rshcmd = "<putty-plink>", will force
## makeClusterPSOCK() to search for and use the PuTTY plink software,
## preventing it from using other SSH clients on the system search PATH.
cl <- makeClusterPSOCK(
  "remote.server.org", user = "bob",
  rshcmd = "<putty-plink>",
  rshopts = c("-P", 2200, "-i", "C:/Users/bobby/.ssh/putty.ppk"),
  dryrun = TRUE
)


## EXAMPLE: Remote worker running on Linux from RStudio on Windows
## Connect to remote Unix machine 'remote.server.org' on port 2200
## as user 'bob' from a Windows machine via RStudio's SSH client.
## Using the explicit special rshcmd = "<rstudio-ssh>", will force
## makeClusterPSOCK() to use the SSH client that comes with RStudio,
## preventing it from using other SSH clients on the system search PATH.
cl <- makeClusterPSOCK(
  "remote.server.org", user = "bob", rshcmd = "<rstudio-ssh>",
  dryrun = TRUE
)

parallelly

Enhancing the 'parallel' Package

v1.25.0
LGPL (>= 2.1)
Authors
Henrik Bengtsson [aut, cre, cph]
Initial release

We don't support your browser anymore

Please choose more modern alternatives, such as Google Chrome or Mozilla Firefox.