mirai - Minimalist Async Evaluation Framework for R

Table of Contents

mirai logo

  1. Example 1: Compute-intensive Operations
  2. Example 2: I/O-bound Operations
  3. Example 3: Resilient Pipelines
  4. Daemons: Local Persistent Processes
  5. Distributed Computing: Remote Daemons
  6. Distributed Computing: Launching Daemons
  7. Distributed Computing: TLS Secure Connections
  8. Compute Profiles
  9. Errors, Interrupts and Timeouts

Example 1: Compute-intensive Operations

Use case: minimise execution times by performing long-running tasks concurrently in separate processes.

Multiple long computes (model fits etc.) can be performed in parallel on available computing cores.

Use mirai() to evaluate an expression asynchronously in a separate, clean R process.

A ‘mirai’ object is returned immediately.


m <- mirai(
    res <- rnorm(n) + m
    res / rev(res)
  m = runif(1),
  n = 1e8

#> < mirai | $data >

Above, all specified name = value pairs are passed through to the ‘mirai’.

The ‘mirai’ yields an ‘unresolved’ logical NA whilst the async operation is ongoing.

#> 'unresolved' logi NA

Upon completion, the ‘mirai’ resolves automatically to the evaluated result.

m$data |> str()
#>  num [1:100000000] 0.403 1.447 1.004 0.364 1.525 ...

Alternatively, explicitly call and wait for the result using call_mirai().

call_mirai(m)$data |> str()
#>  num [1:100000000] 0.403 1.447 1.004 0.364 1.525 ...

For easy programmatic use of mirai(), ‘.expr’ accepts a pre-constructed language object, and also a list of named arguments passed via ‘.args’. So, the following would be equivalent to the above:

expr <- quote({
  res <- rnorm(n) + m
  res / rev(res)

args <- list(m = runif(1), n = 1e8)

m <- mirai(.expr = expr, .args = args)

call_mirai(m)$data |> str()
#>  num [1:100000000] 3.105 -0.931 0.943 3.262 0.151 ...

« Back to ToC

Example 2: I/O-bound Operations

Use case: ensure execution flow of the main process is not blocked.

High-frequency real-time data cannot be written to file/database synchronously without disrupting the execution flow.

Cache data in memory and use mirai() to perform periodic write operations concurrently in a separate process.

Below, ‘.args’ is used to pass a list of objects already present in the calling environment to the mirai by name. This is an alternative use of ‘.args’, and may be combined with ... to also pass in name = value pairs.


x <- rnorm(1e6)
file <- tempfile()

m <- mirai(write.csv(x, file = file), .args = list(x, file))

A ‘mirai’ object is returned immediately.

unresolved() may be used in control flow statements to perform actions which depend on resolution of the ‘mirai’, both before and after.

This means there is no need to actually wait (block) for a ‘mirai’ to resolve, as the example below demonstrates.

# unresolved() queries for resolution itself so no need to use it again within the while loop

while (unresolved(m)) {
  cat("while unresolved\n")
#> while unresolved
#> while unresolved

cat("Write complete:", is.null(m$data))
#> Write complete: TRUE

Now actions which depend on the resolution may be processed, for example the next write.

« Back to ToC

Example 3: Resilient Pipelines

Use case: isolating code that can potentially fail in a separate process to ensure continued uptime.

As part of a data science / machine learning pipeline, iterations of model training may periodically fail for stochastic and uncontrollable reasons (e.g. buggy memory management on graphics cards).

Running each iteration in a ‘mirai’ isolates this potentially-problematic code such that even if it does fail, it does not bring down the entire pipeline.


run_iteration <- function(i) {

  if (runif(1) < 0.1) stop("random error\n", call. = FALSE) # simulates a stochastic error rate
  sprintf("iteration %d successful\n", i)


for (i in 1:10) {

  m <- mirai(run_iteration(i), .args = list(run_iteration, i))
  while (is_error_value(call_mirai(m)$data)) {
    m <- mirai(run_iteration(i), .args = list(run_iteration, i))

#> iteration 1 successful
#> iteration 2 successful
#> iteration 3 successful
#> iteration 4 successful
#> iteration 5 successful
#> iteration 6 successful
#> Error: random error
#> iteration 7 successful
#> iteration 8 successful
#> iteration 9 successful
#> iteration 10 successful

Further, by testing the return value of each ‘mirai’ for errors, error-handling code is then able to automate recovery and re-attempts, as in the above example. Further details on error handling can be found in the section below.

The end result is a resilient and fault-tolerant pipeline that minimises downtime by eliminating interruptions of long computes.

« Back to ToC

Daemons: Local Persistent Processes

Daemons, or persistent background processes, may be set to receive ‘mirai’ requests.

This is potentially more efficient as new processes no longer need to be created on an ad hoc basis.

With Dispatcher (default)

Call daemons() specifying the number of daemons to launch.

#> [1] 6

To view the current status, status() provides the number of active connections along with a matrix of statistics for each daemon.

#> $connections
#> [1] 1
#> $daemons
#>                                     i online instance assigned complete
#> abstract://08b4b97238bde0689bc9514b 1      1        1        0        0
#> abstract://63a8dca236eb13ad5303cdae 2      1        1        0        0
#> abstract://64a11588f7f48743a45c7bcc 3      1        1        0        0
#> abstract://f6e2b2f97f506bb617820921 4      1        1        0        0
#> abstract://73bfd5425ff5b62788bad26f 5      1        1        0        0
#> abstract://688211a2852bbecf432754e9 6      1        1        0        0

The default dispatcher = TRUE creates a dispatcher() background process that connects to individual daemon processes on the local machine. This ensures that tasks are dispatched efficiently on a first-in first-out (FIFO) basis to daemons for processing. Tasks are queued at the dispatcher and sent to a daemon as soon as it can accept the task for immediate execution.

Dispatcher uses synchronisation primitives from nanonext, waiting upon rather than polling for tasks, which is efficient both in terms of consuming no resources while waiting, and also being fully synchronised with events (having no latency).

#> [1] 0

Set the number of daemons to zero to reset. This reverts to the default of creating a new background process for each ‘mirai’ request.

Without Dispatcher

Alternatively, specifying dispatcher = FALSE, the background daemons connect directly to the host process.

daemons(6, dispatcher = FALSE)
#> [1] 6

Requesting the status now shows 6 connections, along with the host URL at $daemons.

#> $connections
#> [1] 6
#> $daemons
#> [1] "abstract://8a07989c110b62dab5b65a4a"

This implementation sends tasks immediately, and ensures that tasks are evenly-distributed amongst daemons. This means that optimal scheduling is not guaranteed as the duration of tasks cannot be known a priori. As an example, tasks could be queued at a daemon behind a long-running task, whilst other daemons are idle having already completed their tasks.

The advantage of this approach is that it is low-level and does not require an additional dispatcher process. It is well-suited to working with similar-length tasks, or where the number of concurrent tasks typically does not exceed available daemons.

everywhere() may be used to evaluate an expression on all connected daemons and persist the resultant state, regardless of a daemon’s ‘cleanup’ setting.


The above keeps the parallel package loaded for all evaluations. Other types of setup task may also be performed such as making a common resource available, etc.

#> [1] 0

Set the number of daemons to zero to reset.


everywhere() may be used to evaluate an expression on all connected daemons and persist the resultant state, regardless of a daemon’s ‘cleanup’ setting. This may be used for performing setup of the evaluation environment, with particular packages loaded, or common resources made available, etc.

« Back to ToC

Distributed Computing: Remote Daemons

The daemons interface may also be used to send tasks for computation to remote daemon processes on the network.

Call daemons() specifying ‘url’ as a character string such as: ‘tcp://’ at which daemon processes should connect to. Alternatively, use host_url() to automatically construct a valid URL.

IPv6 addresses are also supported and must be enclosed in square brackets [] to avoid confusion with the final colon separating the port. For example, port 5555 on the IPv6 address ::ffff:a6f:50d would be specified as tcp://[::ffff:a6f:50d]:5555.

For options on actually launching the daemons, please see the next section.

Connecting to Remote Daemons Through Dispatcher

The default dispatcher = TRUE creates a background dispatcher() process on the local machine, which listens to a vector of URLs that remote daemon() processes dial in to, with each daemon having its own unique URL.

It is recommended to use a websocket URL starting ws:// instead of TCP in this scenario (used interchangeably with tcp://). A websocket URL supports a path after the port number, which can be made unique for each daemon. In this way a dispatcher can connect to an arbitrary number of daemons over a single port.

Supplying a vector of URLs allows the use of arbitrary port numbers / paths. ‘n’ does not need to be specified if it can be inferred from the length of the ‘url’ vector, for example:

daemons(url = c("ws://", "ws://", "ws://"))

Alternatively, below a single URL is supplied, along with n = 4 to specify that the dispatcher should listen at 4 URLs. In such a case, an integer sequence is automatically appended to the path /1 through /4 to produce the URLs.

daemons(n = 4, url = host_url(port = 5555))
#> [1] 4

Requesting status on the host machine:

#> $connections
#> [1] 1
#> $daemons
#>                     i online instance assigned complete
#> tcp://hostname:5555 1      0        0        0        0
#> tcp://hostname:5556 2      0        0        0        0
#> tcp://hostname:5557 3      0        0        0        0
#> tcp://hostname:5558 4      0        0        0        0

As per the local case, $connections shows the single connection to dispatcher, however $daemons now provides a matrix of statistics for the remote daemons.

Dispatcher automatically adjusts to the number of daemons actually connected. Hence it is possible to dynamically scale up or down the number of daemons according to requirements (limited to the ‘n’ URLs assigned).

To reset all connections and revert to default behaviour:

#> [1] 0

Closing the connection causes the dispatcher to exit automatically, and in turn all connected daemons when their respective connections with the dispatcher are terminated.

Connecting to Remote Daemons Directly

By specifying dispatcher = FALSE, remote daemons connect directly to the host process. The host listens at a single URL, and distributes tasks to all connected daemons.

daemons(url = host_url(), dispatcher = FALSE)
#> [1] "tcp://hostname:36117"

Note that above, calling host_url() without a port value uses the default of ‘0’. This is a wildcard value that will automatically cause a free ephemeral port to be assigned. The actual assigned port is provided in the return value of the call, or it may be queried at any time via status().

The number of daemons connecting to the host URL is not limited and network resources may be added or removed at any time, with tasks automatically distributed to all connected daemons.

$connections will show the actual number of connected daemons.

#> $connections
#> [1] 0
#> $daemons
#> [1] "tcp://hostname:36117"

To reset all connections and revert to default behaviour:

#> [1] 0

This causes all connected daemons to exit automatically.

« Back to ToC

Distributed Computing: Launching Daemons

To launch remote daemons, supply a remote launch configuration to the ‘remote’ argument of daemons() when setting up daemons, or launch_remote() at any time afterwards.

ssh_config() may be used to generate a remote launch configuration if there is SSH access to the remote machine, or else remote_config() provides a flexible method for generating a configuration involving a custom resource manager / application.

SSH Direct Connection

The first example below launches 4 daemons on the machine (using the default SSH port of 22 as this was not specified), connecting back to the dispatcher URLs:

  n = 4,
  url = host_url(ws = TRUE, port = 5555),
  remote = ssh_config(remotes = "ssh://")

The second example below launches one daemon on each of and using the custom SSH port of 222:

  n = 2,
  url = host_url(ws = TRUE, port = 5555),
  remote = ssh_config(c("ssh://", "ssh://"))

In the above examples, as the remote daemons connect back directly, port 5555 on the local machine must be open to incoming connections from the remote addresses.

SSH Tunnelling

Use of SSH tunnelling provides a convenient way to launch remote daemons without requiring the remote machine to be able to access the host. Often firewall configurations or security policies may prevent opening a port to accept outside connections.

In these cases SSH tunnelling offers a solution by creating a tunnel once the initial SSH connection is made. For simplicity, this SSH tunnelling implementation uses the same port on both the side of the host and that of the corresponding node. SSH key-based authentication must also already be in place.

Tunnelling requires the hostname for ‘url’ specified when setting up daemons to be either ‘’ or ‘localhost’. This is as the tunnel is created between or equivalently localhost:port on each machine. The host listens to its localhost:port and the remotes each dial into localhost:port on their own respective machines.

The below example launches 2 nodes on the remote machine using SSH tunnelling over port 5555 (‘url’ hostname is specified as ‘localhost’):

  url = "tcp://localhost:5555",
  remote = ssh_config(
    remotes = c("ssh://", "ssh://"),
    tunnel = TRUE

Manual Deployment

As an alternative to automated launches, calling launch_remote() without specifying ‘remote’ may be used to return the shell commands for deploying daemons manually. The printed return values may be copy / pasted directly to a remote machine.

daemons(n = 2, url = host_url())
#> [1] 2

#> [1]
#> Rscript -e "mirai::daemon('tcp://hostname:43115',rs=c(10407,-512668947,-1821112118,19352515,924029480,-1421864439,1967253046))"
#> [2]
#> Rscript -e "mirai::daemon('tcp://hostname:38413',rs=c(10407,1394391231,-727687268,-591259728,-971773911,72728696,-575275953))"

#> [1] 0

Note that daemons() should be set up on the host machine before launching daemon() on remote resources, otherwise the daemon instances will exit if a connection is not immediately available. Alternatively, specifying the argument autoexit = FALSE will allow daemons to wait (indefinitely) for a connection to become available.

« Back to ToC

Distributed Computing: TLS Secure Connections

TLS is available as an option to secure communications from the local machine to remote daemons.


An automatic zero-configuration default is implemented. Simply specify a secure URL of the form wss:// or tls+tcp:// when setting daemons, or use host_url(tls = TRUE), for example:

daemons(n = 4, url = host_url(ws = TRUE, tls = TRUE))
#> [1] 4

Single-use keys and certificates are automatically generated and configured, without requiring any further intervention. The private key is always retained on the host machine and never transmitted.

The generated self-signed certificate is available via launch_remote(). This function conveniently constructs the full shell command to launch a daemon, including the correctly specified ‘tls’ argument to daemon().

#> [1]
#> Rscript -e "mirai::daemon('wss://hostname:38377/1',tls=c('-----BEGIN CERTIFICATE-----
#> ADCCAgoCggIBALGoWelheZ9mzpBV73aaH8a6ugnsYLXANCNpRBQCz5oDjNP2D6W4
#> t8aUXh6hSwC/VSB67DfTOeFHrni+1HlBlTdg8nJy/pKFMYD1MgAtIc/KKYt8oiWU
#> eqoQ+XLZlAFPentM0yaWZCGqLDhqJ8+zY59XSsWr3VIOsgTY9lrcdYnE/HxlfTIp
#> +bpkykANP981fuPCY4GMddo119BzY4b4IultA8o44C/uK/jUFN/Yq7CUYXb9cgt3
#> 80qNfAhYHDCB6ATZP+i5zCzClOAhQ0NXLdlF10yXqC2NuE84bP2wCDbxP5tIKo4q
#> g+6tUUWSM9s9yvEnnmH8CWLGaRj8oKqzWWcPJTXFux+2zXHiR48tVSuOBzZ03au0
#> ee3xY8RiXmvWnpDC9vPH7hwc9NZuCdaFwggx9MGQNRbhGsYGQCqtXwjd/3/Ebt2+
#> ZsvuOQOtoSPS54Ji9/z4WMI4De6CgcYOdqiT/izAW8TfhA+jijAX2UXQgInsoi4h
#> hMPhcF+mv/AzDtVPYn16ZcuDtEX5hZR532fmuc8gmArkkQ/ZWduoRNZ0rw6ESwhP
#> lP7QbPDhkrNjRIdt3bGst+J/BzZNveJQmgBarRanN2RCgIF8nGn6NT7X3urITn2g
#> 27IFWHh2m0bs1OVPCdtx6WmaxPhOhNCKo7c6q+UockZkP7CtzD66/SmBAgMBAAGj
#> tN+m7wjHMB8GA1UdIwQYMBaAFLpumbCa69wKwNnAfVuhtN+m7wjHMA0GCSqGSIb3
#> K2cJUiFa99RGNH8QO0gzKb5jB/wNm6eLK2sw2wytb0KXy13fQ/IlEv02lP5PtyQ9
#> acQv9DeUE9Q1GOkYd1j/lNqJmd3Ka3dK555rfJiF+s8czMpZBCHueZHpqJ3EQFI2
#> gdtvkpSDWAPe6AjkKpJtO3IJ89VCWZAoj2k6dpkJUokytW6vI+5IpELJZn4voNQz
#> QtK3Z+Z9XOHg3nYE9B8n2GwYOE+mtU4utIA7+bn3NY5SKE58DY/Ub46EQYhuakYS
#> sls7DTJsmu4/Ff8rvWiS9cLA94CIap3/EmI51rkzZCjwCBIpLaFMwJkcCr7yfFlI
#> afvurXOdtOrzpwKTWJ+DHVvJUr/4QfjQWM36xRTqewuJo09PMMe8BUBEj9grxx5B
#> WId5pRs//nSQ1//FQ+D0x6kd4yIGqdSa7UYF2Q7OVPqugoEmPO8d0GeYBtbEsfdw
#> IhKTHS1lRE6P/TJLCMX9CZaCipeR1PTWsvfFxyQi+tWfF6pUOS1eZKHqXwNzfchy
#> 7IXahUfi52tUHb0/3Zvs6bAVO9VnkkhiYy4FeA57uo1kpxyHV0DMPcdcRwT8n78o
#> qZAcVptJiXXtMReinMYxhIHncdCrgXuTirmgqV4McBhEPpbnOFyWBMIiqA==
#> -----END CERTIFICATE-----
#> ',''),rs=c(10407,-1949455173,305385664,98036161,-1903821682,-341019721,-290458484))"

The printed value may be deployed directly on a remote machine.

« Back to ToC

CA Signed Certificates

As an alternative to the zero-configuration default, a certificate may also be generated via a Certificate Signing Request (CSR) to a Certificate Authority (CA), which may be a public CA or a CA internal to an organisation.

  1. Generate a private key and CSR. The following resources describe how to do so:
  1. Provide the generated CSR to the CA for it to sign a new TLS certificate.
  1. When setting daemons, the TLS certificate and private key should be provided to the ‘tls’ argument of daemons().
  1. When launching daemons, the certificate chain to the CA should be supplied to the ‘tls’ argument of daemon() or launch_remote().

« Back to ToC

Compute Profiles

The daemons() interface also allows the specification of compute profiles for managing tasks with heterogeneous compute requirements:

Simply specify the argument .compute when calling daemons() with a profile name (which is ‘default’ for the default profile). The daemons settings are saved under the named profile.

To create a ‘mirai’ task using a specific compute profile, specify the ‘.compute’ argument to mirai(), which defaults to the ‘default’ compute profile.

Similarly, functions such as status(), launch_local() or launch_remote() should be specified with the desired ‘.compute’ argument.

« Back to ToC

Errors, Interrupts and Timeouts

If execution in a mirai fails, the error message is returned as a character string of class ‘miraiError’ and ‘errorValue’ to facilitate debugging. is_mirai_error() may be used to test for mirai execution errors.

m1 <- mirai(stop("occurred with a custom message", call. = FALSE))
#> 'miraiError' chr Error: occurred with a custom message

m2 <- mirai(mirai::mirai())
#> 'miraiError' chr Error in mirai::mirai(): missing expression, perhaps wrap in {}?

#> [1] TRUE
#> [1] TRUE

If a daemon instance is sent a user interrupt, the mirai will resolve to an empty character string of class ‘miraiInterrupt’ and ‘errorValue’. is_mirai_interrupt() may be used to test for such interrupts.

#> [1] FALSE

If execution of a mirai surpasses the timeout set via the ‘.timeout’ argument, the mirai will resolve to an ‘errorValue’. This can, amongst other things, guard against mirai processes that have the potential to hang and never return.

m3 <- mirai(nanonext::msleep(1000), .timeout = 500)
#> 'errorValue' int 5 | Timed out

#> [1] FALSE
#> [1] FALSE
#> [1] TRUE

is_error_value() tests for all mirai execution errors, user interrupts and timeouts.

« Back to ToC