--- name: mirai description: Help users write correct R code for async, parallel, and distributed computing using mirai. Use when users need to run R code asynchronously or in parallel, write mirai code with correct dependency passing, set up parallel workers, convert from future or parallel, use mirai_map, integrate with Shiny or promises, or configure cluster/HPC computing. metadata: author: Charlie Gao (@shikokuchuo) version: "1.2" license: MIT --- mirai is a minimalist R framework for async, parallel, and distributed evaluation, built on nanonext. ## Core Principle: Explicit Dependency Passing mirai evaluates expressions in a **clean environment** on a daemon process. Nothing from the calling environment is available unless passed explicitly — this is the #1 source of mistakes. ```r # WRONG: my_data and my_func are not available on the daemon m <- mirai(my_func(my_data)) ``` There are two ways to pass objects, and the names used **must match** the names referenced in the expression. ### `.args` (recommended) Objects in `.args` populate the expression's **local evaluation environment** — available directly by name inside the expression. ```r m <- mirai(my_func(my_data), .args = list(my_func = my_func, my_data = my_data)) ``` ### `...` (dot-dot-dot) Objects passed via `...` are assigned to the **daemon's global environment**. Use this when objects need to be found by R's standard scoping rules (e.g., helper functions called by other functions). ```r m <- mirai(my_func(my_data), my_func = my_func, my_data = my_data) ``` ### Shortcut: pass the whole calling environment ```r # .args form — populates local eval env process <- function(x, y) mirai(x + y, .args = environment()) # ... form — single unnamed environment, populates daemon global env df_matrix <- function(x, y) mirai(as.matrix(rbind(x, y)), environment()) ``` ### When to use which | Scenario | Use | |----------|-----| | Data and simple functions | `.args` | | Helper functions called by other functions that need lexical scoping | `...` | | Pass entire local scope to local eval env | `.args = environment()` | | Pass entire local scope to daemon global env | `mirai(expr, environment())` | | Large objects shared across many tasks | `everywhere()` first, then reference by name | ## Common Mistakes ### Unqualified package functions Daemons start with no user packages loaded. Same applies inside `mirai_map()` callbacks. ```r # WRONG: dplyr is not loaded on the daemon m <- mirai(filter(df, x > 5), .args = list(df = my_df)) # CORRECT: namespace-qualify m <- mirai(dplyr::filter(df, x > 5), .args = list(df = my_df)) # CORRECT: load inside the expression m <- mirai({ library(dplyr) filter(df, x > 5) }, .args = list(df = my_df)) # CORRECT: pre-load on all daemons everywhere(library(dplyr)) m <- mirai(filter(df, x > 5), .args = list(df = my_df)) ``` ### Expecting results immediately `m$data` accesses the value but may still be unresolved. Use `m[]` (or `collect_mirai(m)`) to block until done; use `unresolved(m)` for a non-blocking check. ```r m <- mirai(slow_computation()) result <- m[] # blocks until resolved if (!unresolved(m)) result <- m$data # non-blocking ``` ## Setting Up Daemons ### No daemons required `mirai()` works without calling `daemons()` first — it launches a transient background process per call. Setting up daemons is only needed for persistent pools of workers. ### Local daemons ```r # Start 4 local daemon processes (with dispatcher, the default) daemons(4) # Direct connection (no dispatcher) — lower overhead, round-robin scheduling daemons(4, dispatcher = FALSE) # Concise programmatic statistics (vs. the richer status()) info() # Reset (daemons otherwise persist for the session) daemons(0) ``` ### Scoped daemons (auto-cleanup) `with(daemons(...), {...})` **creates** daemons and automatically cleans them up when the block exits. ```r with(daemons(4), { m <- mirai(expensive_task()) m[] }) ``` ### Scoped compute profile switching `local_daemons()` and `with_daemons()` **switch** the active compute profile to one that already exists — they do not create daemons. ```r daemons(4, .compute = "workers") # Switch active profile for the duration of the calling function my_func <- function() { local_daemons("workers") mirai(task())[] # uses "workers" profile } # Switch active profile for a block with_daemons("workers", { m <- mirai(task()) m[] }) ``` ### Compute profiles (multiple independent pools) ```r daemons(4, .compute = "cpu") daemons(2, .compute = "gpu") m1 <- mirai(cpu_work(), .compute = "cpu") m2 <- mirai(gpu_work(), .compute = "gpu") ``` ## Memory Backpressure (`memory` + `try_mirai()`) For high-throughput producers (Shiny, promises, ingest pipelines), use the `memory` argument to `daemons()` to cap the queued task payload at dispatcher (MB, metric). Pair it with `try_mirai()` so the host R thread never blocks on submission. ```r # 100 MB queue cap. mirai() blocks on submission once the queue is full. daemons(4, memory = 100) # try_mirai() returns NULL (invisibly) instead of blocking when the cap is hit. m <- try_mirai(work(x), .args = list(x = x)) if (is.null(m)) { # backpressure: drop, retry later, or signal upstream } else { # m is a regular mirai } # Inspect current and peak queue usage status()$memory ``` `memory` requires dispatcher. Without dispatcher (or with `memory = NULL`), `try_mirai()` always returns a mirai. ## mirai_map: Parallel Map Requires daemons to be set. Maps `.x` element-wise over a function, distributing across daemons. Namespace-qualify any package functions used inside the callback (see Mistake 2). ```r daemons(4) # Basic map — collect with [] results <- mirai_map(1:10, function(x) x^2)[] # Constants via .args, helpers via ... (same passing rules as mirai()) results <- mirai_map( data_list, function(x, power) helper(x, power), .args = list(power = 3), helper = my_helper_func )[] # Flatten results to a vector results <- mirai_map(1:10, sqrt)[.flat] # Progress bar (requires cli package) results <- mirai_map(1:100, slow_task)[.progress] # Early stopping on error results <- mirai_map(1:100, risky_task)[.stop] # Combine options results <- mirai_map(1:100, task)[.stop, .progress] ``` ### Mapping over multiple arguments (data frame rows) ```r # Each row becomes arguments to the function params <- data.frame(mean = 1:5, sd = c(0.1, 0.5, 1, 2, 5)) results <- mirai_map(params, function(mean, sd) rnorm(100, mean, sd))[] ``` ### Process as completed (`race_mirai`) `race_mirai()` returns the integer index of the first resolved mirai in a list (or `0L` if empty). Useful when you want to handle results in completion order rather than submission order. ```r remaining <- mirai_map(jobs, run) while (length(remaining) > 0) { idx <- race_mirai(remaining) process(remaining[[idx]]$data) remaining <- remaining[-idx] } ``` ## everywhere: Pre-load State on All Daemons ```r daemons(4) # Load packages on all daemons everywhere(library(DBI)) # Set up persistent connections everywhere(con <<- dbConnect(RSQLite::SQLite(), db_path), db_path = tempfile()) # Export objects to daemon global environment via ... # The empty {} expression is intentional — the point is to export objects via ... everywhere({}, api_key = my_key, config = my_config) # .min = N forces a synchronization point: the call must complete on at least # N daemons before subsequent mirai evaluations proceed. Useful when launching # remote daemons that connect over time. everywhere(library(arrow), .min = 4) ``` ## Error Handling ```r m <- mirai(stop("something went wrong")) m[] is_mirai_error(m$data) # TRUE for execution errors is_mirai_interrupt(m$data) # TRUE for cancelled tasks is_error_value(m$data) # TRUE for any error/interrupt/timeout m$data$message # Error message m$data$stack.trace # Full stack trace m$data$condition.class # Original error classes # Timeouts (requires dispatcher) m <- mirai(Sys.sleep(60), .timeout = 5000) # 5-second timeout # Cancellation (requires dispatcher) m <- mirai(long_running_task()) stop_mirai(m) ``` ## Shiny / Promises Integration ### ExtendedTask pattern ```r library(shiny) library(bslib) library(mirai) daemons(4) onStop(function() daemons(0)) ui <- page_fluid( input_task_button("run", "Run Analysis"), plotOutput("result") ) server <- function(input, output, session) { task <- ExtendedTask$new( function(n) mirai(rnorm(n), .args = list(n = n)) ) |> bind_task_button("run") observeEvent(input$run, task$invoke(input$n)) output$result <- renderPlot(hist(task$result())) } ``` For high-traffic apps, set `daemons(4, memory = ...)` and submit with `try_mirai()` to apply backpressure without stalling the Shiny event loop. ### Promise piping ```r library(promises) mirai({Sys.sleep(1); "done"}) %...>% cat() ``` ## Remote / Distributed Computing ### SSH (direct connection) ```r daemons( url = host_url(tls = TRUE), remote = ssh_config(c("ssh://user@node1", "ssh://user@node2")) ) ``` ### SSH (tunnelled, for firewalled environments) ```r daemons( n = 4, url = local_url(tcp = TRUE), remote = ssh_config("ssh://user@node1", tunnel = TRUE) ) ``` ### HPC cluster (Slurm/SGE/PBS/LSF) ```r daemons( n = 1, url = host_url(), remote = cluster_config( command = "sbatch", options = "#SBATCH --job-name=mirai\n#SBATCH --mem=8G\n#SBATCH --array=1-50", rscript = file.path(R.home("bin"), "Rscript") ) ) ``` ### HTTP launcher (e.g., Posit Workbench) ```r daemons(n = 2, url = host_url(), remote = http_config()) ``` ## Converting from future | future | mirai | |--------|-------| | Auto-detects globals | Must pass all dependencies explicitly | | `future({expr})` | `mirai({expr}, .args = list(...))` | | `value(f)` | `m[]` or `collect_mirai(m)` | | `plan(multisession, workers = 4)` | `daemons(4)` | | `plan(sequential)` / reset | `daemons(0)` | | `future_lapply(X, FUN)` | `mirai_map(X, FUN)[]` | | `future_map(X, FUN)` (furrr) | `mirai_map(X, FUN)[]` | | `future_promise(expr)` | `mirai(expr, ...)` (auto-converts to promise) | The key conversion step: identify all objects the expression uses from the calling environment and pass them explicitly via `.args` or `...`. ## Converting from parallel | parallel | mirai | |----------|-------| | `makeCluster(4)` | `daemons(4)` or `make_cluster(4)` | | `clusterExport(cl, "x")` | Pass via `.args` / `...`, or use `everywhere()` | | `clusterEvalQ(cl, library(pkg))` | `everywhere(library(pkg))` | | `parLapply(cl, X, FUN)` | `mirai_map(X, FUN)[]` | | `parSapply(cl, X, FUN)` | `mirai_map(X, FUN)[.flat]` | | `mclapply(X, FUN, mc.cores = 4)` | `daemons(4); mirai_map(X, FUN)[]` | | `stopCluster(cl)` | `daemons(0)` | ### Drop-in replacement via make_cluster For code that already uses the parallel package extensively, `make_cluster()` provides a drop-in backend: ```r cl <- mirai::make_cluster(4) parallel::parLapply(cl, 1:100, my_func) mirai::stop_cluster(cl) # R >= 4.5: native integration cl <- parallel::makeCluster(4, type = "MIRAI") ``` ## Random Number Generation ```r # Default: L'Ecuyer-CMRG stream per daemon (statistically safe, non-reproducible) daemons(4) # Reproducible: L'Ecuyer-CMRG stream per mirai call. # Results are the same regardless of daemon count or scheduling. daemons(4, seed = 42) ``` ## Debugging ```r # Synchronous mode — runs in the host process, supports browser() daemons(sync = TRUE) m <- mirai({ browser() result <- tricky_function(x) result }, .args = list(tricky_function = tricky_function, x = my_x)) daemons(0) # Capture daemon stdout/stderr daemons(4, output = TRUE) ``` ## Advanced Pattern: Nested Parallelism Inside daemon callbacks (e.g., `mirai_map`), use `local_url()` + `launch_local()` instead of `daemons(n)` to avoid conflicting with the outer daemon pool. ```r mirai_map(1:10, function(x) { daemons(url = local_url()) launch_local(2) result <- mirai_map(1:5, function(y, x) x * y, .args = list(x = x))[] daemons(0) result })[] ```