Minimalist async evaluation framework for R.
Lightweight parallel code execution, local or distributed across the network.
Designed for simplicity, a ‘mirai’ evaluates an arbitrary expression asynchronously, resolving automatically upon completion.
Leverages ‘nanonext’ and ‘NNG’ (Nanomsg Next Gen) to provide efficient task scheduling, scalability beyond R connection limits, and transports faster than TCP/IP for inter-process communications.
mirai() returns a ‘mirai’ object immediately. ‘mirai’ (未来 みらい) is Japanese for ‘future’.
The asynchronous ‘mirai’ task runs in an ephemeral or persistent process, spawned locally or distributed across the network.
{mirai} has a tiny pure R code base, relying solely on {nanonext}, a high-performance binding for the ‘NNG’ (Nanomsg Next Gen) C library with zero package dependencies.
Installation
Install the latest release from CRAN:
install.packages("mirai")or the development version from rOpenSci R-universe:
install.packages("mirai", repos = "https://shikokuchuo.r-universe.dev")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.
library(mirai)
m <- mirai({
res <- rnorm(n) + m
res / rev(res)
}, n = 1e8, m = runif(1))
m
#> < mirai >
#> - $data for evaluated resultAbove, all named objects are passed through to the mirai.
The ‘mirai’ yields an ‘unresolved’ logical NA value whilst the async operation is ongoing.
m$data
#> 'unresolved' logi NAUpon completion, the ‘mirai’ resolves automatically to the evaluated result.
m$data |> str()
#> num [1:100000000] -10.974 -0.142 1.335 2.598 0.91 ...Alternatively, explicitly call and wait for the result using call_mirai().
call_mirai(m)$data |> str()
#> num [1:100000000] -10.974 -0.142 1.335 2.598 0.91 ...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.
A ‘mirai’ object is returned immediately.
Below, ‘.args’ accepts a list of objects already present in the calling environment to be passed to the mirai.
library(mirai)
x <- rnorm(1e6)
file <- tempfile()
m <- mirai(write.csv(x, file = file), .args = list(x, file))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")
Sys.sleep(0.5)
}
#> while unresolved
#> while unresolved
cat("Write complete:", is.null(m$data))
#> Write complete: TRUENow actions which depend on the resolution may be processed, for example the next write.
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 or 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’ process isolates this potentially-problematic code such that if it does fail, it does not bring down the entire pipeline.
library(mirai)
run_iteration <- function(i) {
if (runif(1) < 0.15) stop("random error", call. = FALSE) # simulates a stochastic error rate
sprintf("iteration %d successful", i)
}
for (i in 1:10) {
m <- mirai(run_iteration(i), .args = list(run_iteration, i))
while (is_error_value(call_mirai(m)$data)) {
cat(m$data, "\n")
m <- mirai(run_iteration(i), .args = list(run_iteration, i))
}
cat(m$data, "\n")
}
#> iteration 1 successful
#> iteration 2 successful
#> iteration 3 successful
#> iteration 4 successful
#> iteration 5 successful
#> iteration 6 successful
#> iteration 7 successful
#> iteration 8 successful
#> Error: random error
#> iteration 9 successful
#> iteration 10 successfulFurther, 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.
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.
daemons(6)
#> [1] 6To view the current status, call daemons() with no arguments. This provides the number of active connections along with a matrix of statistics for each daemon.
daemons()
#> $connections
#> [1] 1
#>
#> $daemons
#> status_online status_busy tasks_assigned tasks_complete instance #
#> abstract://n2049946055 1 0 0 0 1
#> abstract://n1227422372 1 0 0 0 1
#> abstract://n7652447120 1 0 0 0 1
#> abstract://n3038177524 1 0 0 0 1
#> abstract://n1873925846 1 0 0 0 1
#> abstract://n6017584969 1 0 0 0 1The default dispatcher = TRUE launches a dispatcher() background process that connects to individual background server() processes on the local machine. This ensures that tasks are dispatched efficiently on a first-in first-out (FIFO) basis to servers for processing. Tasks are queued at the dispatcher and sent to a server as soon as it can accept the task for immediate execution.
daemons(0)
#> [1] 0Set 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 daemon processes connect directly to the client.
daemons(6, dispatcher = FALSE)
#> [1] 6Requesting the status now shows 6 connections and 6 daemons.
daemons()
#> $connections
#> [1] 6
#>
#> $daemons
#> [1] 6This 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 remain idle.
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.
daemons(0)
#> [1] 0Set the number of daemons to zero to reset.
Distributed Computing: Remote Servers
The daemons interface may also be used to send tasks for computation to server processes on the network.
Call daemons() specifying ‘url’ as a character string the client network address and a port that is able to accept incoming connections.
The examples below use an illustrative local network IP address of ‘10.111.5.13’.
A port on the client also needs to be open and available for inbound connections from the local network, illustratively ‘5555’ in the examples below.
Connecting to Remote Servers Through Dispatcher
The default dispatcher = TRUE creates a background dispatcher() process on the local client machine, which listens to a vector of URLs that remote servers dial in to, with each server having its 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 server. In this way a dispatcher can connect to an arbitrary number of servers over a single port.
# daemons(n = 4, url = "ws://10.111.5.13:5555")
daemons(n = 4, url = "ws://:5555")
#> [1] 4Above, a single URL was 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 these URLs.
Alternatively, supplying a vector of URLs allows the use of arbitrary port numbers / paths, e.g.:
# daemons(url = c("ws://:5555/cpu", "ws://:5555/gpu", "ws://:12560", "ws://:12560/2"))Above, ‘n’ is not specified, in which case its value is inferred from the length of the ‘url’ vector supplied.
–
On the remote resource, server() may be called from an R session, or directly from a shell using Rscript. Each server instance should dial into one of the unique URLs that the dispatcher is listening to:
Rscript -e 'mirai::server("ws://10.111.5.13:5555/1")'
Rscript -e 'mirai::server("ws://10.111.5.13:5555/2")'
Rscript -e 'mirai::server("ws://10.111.5.13:5555/3")'
Rscript -e 'mirai::server("ws://10.111.5.13:5555/4")'–
Requesting status, on the client:
daemons()
#> $connections
#> [1] 1
#>
#> $daemons
#> status_online status_busy tasks_assigned tasks_complete instance #
#> ws://:5555/1 1 0 0 0 1
#> ws://:5555/2 1 0 0 0 1
#> ws://:5555/3 1 0 0 0 1
#> ws://:5555/4 1 0 0 0 1As per the local case, $connections will show the single connection to the dispatcher process, however $daemons will provide the matrix of statistics for the remote servers.
status_online shows as 1 when there is an active connection, or else 0 if a server has yet to connect or has disconnected.
status_busy will be 1 if the server is currently processing a task, or else 0.
tasks_assinged shows the cumulative number of tasks assigned to the server instance by the dispatcher.
tasks_complete shows the cumulative number of tasks completed by the server instance.
instance # will increment by 1 every time there is a new connection at a URL. When this happens, the tasks_assigned and tasks_complete statistics will also reset. This is designed to track new server instances connecting after previous ones have ended (due to time-outs etc.).
The dispatcher will automatically adjust to the number of servers actually connected. Hence it is possible to dynamically scale up or down the number of servers according to requirements (limited to the ‘n’ URLs assigned at the dispatcher).
To reset all connections and revert to default behaviour:
daemons(0)
#> [1] 0Closing the connection causes the dispatcher to exit automatically, and in turn all connected servers when their respective connections with the dispatcher are terminated.
Connecting to Remote Servers Directly
By specifying dispatcher = FALSE, remote servers connect directly to the client. The client listens at the below address, and distributes tasks to all connected server processes.
daemons(url = "tcp://10.111.5.13:45073", dispatcher = FALSE)Alternatively, simply supply a colon followed by the port number to listen on all interfaces on the local host, for example:
daemons(url = "tcp://:0", dispatcher = FALSE)
#> [1] "tcp://:45073"Note that above, the port number is specified as zero. This is a wildcard value that will automatically cause a free ephemeral port to be assigned. The actual assigned port is provided as the return value of the call, or it may be queried at any time by requesting the status using daemons().
–
On the server, server() may be called from an R session, or an Rscript invocation from a shell. This sets up a remote daemon process that connects to the client URL and receives tasks:
Rscript -e 'mirai::server("tcp://10.111.5.13:0")'–
On the client, requesting the status will return the client URL for daemons. The number of daemons connecting to this URL is not limited and network resources may be added and removed at any time, with tasks automatically distributed to all server processes.
$connections will show the actual number of connected server instances.
daemons()
#> $connections
#> [1] 0
#>
#> $daemons
#> [1] "tcp://:45073"To reset all connections and revert to default behaviour:
daemons(0)
#> [1] 0This causes all connected server instances to exit automatically.
Compute Profiles
The daemons interface allows the easy specification of compute profiles. This is for managing tasks with heterogeneous compute requirements:
- send tasks to different servers or server clusters with the appropriate specifications (in terms of CPUs / memory / GPU / accelerators etc.)
- split tasks between local and remote computation
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 launch a ‘mirai’ task using a specific compute profile, specify the ‘.compute’ argument to mirai(), which defaults to the ‘default’ compute profile.
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() can be used to test for mirai execution errors.
m1 <- mirai(stop("occurred with a custom message", call. = FALSE))
call_mirai(m1)$data
#> 'miraiError' chr Error: occurred with a custom message
m2 <- mirai(mirai::mirai())
call_mirai(m2)$data
#> 'miraiError' chr Error in mirai::mirai(): missing expression, perhaps wrap in {}?
is_mirai_error(m2$data)
#> [1] TRUE
is_error_value(m2$data)
#> [1] TRUEIf during a call_mirai() an interrupt e.g. ctrl+c is sent, 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.
is_mirai_interrupt(m2$data)
#> [1] FALSEIf 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 hang and never return.
m3 <- mirai(nanonext::msleep(1000), .timeout = 500)
call_mirai(m3)$data
#> 'errorValue' int 5 | Timed out
is_mirai_error(m3$data)
#> [1] FALSE
is_mirai_interrupt(m3$data)
#> [1] FALSE
is_error_value(m3$data)
#> [1] TRUEis_error_value() tests for all mirai execution errors, user interrupts and timeouts.
Deferred Evaluation Pipe
{mirai} implements a deferred evaluation pipe %>>% for working with potentially unresolved values.
Pipe a mirai $data value forward into a function or series of functions and it initially returns an ‘unresolvedExpr’.
The result may be queried at $data, which will return another ‘unresolvedExpr’ whilst unresolved. However when the original value resolves, the ‘unresolvedExpr’ will simultaneously resolve into a ‘resolvedExpr’, for which the evaluated result will be available at $data.
It is possible to use unresolved() around a ‘unresolvedExpr’ or its $data element to test for resolution, as in the example below.
The pipe operator semantics are similar to R’s base pipe |>:
x %>>% f is equivalent to f(x) x %>>% f() is equivalent to f(x) x %>>% f(y) is equivalent to f(x, y)
Links
{mirai} website: https://shikokuchuo.net/mirai/
{mirai} on CRAN: https://cran.r-project.org/package=mirai
Listed in CRAN Task View:
- High Performance Computing: https://cran.r-project.org/view=HighPerformanceComputing
{nanonext} website: https://shikokuchuo.net/nanonext/
{nanonext} on CRAN: https://cran.r-project.org/package=nanonext
NNG website: https://nng.nanomsg.org/
The {crew} package https://wlandau.github.io/crew/ (available on CRAN) by William Landau further extends {mirai} to different computing platforms for distributed workers.
–
Please note that this project is released with a Contributor Code of Conduct. By participating in this project you agree to abide by its terms.
