r-mirai

mirai 是一个极简 R 框架，基于 nanonext 构建，用于异步、并行和分布式求值。

核心原则：显式传递依赖项

mirai 会在守护进程上以干净环境求值表达式。除非显式传递，否则调用环境中的任何内容都不可用——这是错误的首要来源。

# WRONG: my_data and my_func are not available on the daemon
m <- mirai(my_func(my_data))

传递对象有两种方式，所用的名称必须与表达式中引用的名称一致。

.args（推荐）

.args 中的对象会填充表达式的本地求值环境，可在表达式中直接按名称使用。

m <- mirai(my_func(my_data), .args = list(my_func = my_func, my_data = my_data))

...（点点点）

通过 ... 传递的对象会被赋值到守护进程的全局环境。当对象需要遵循 R 的标准作用域规则来查找时，请使用此方式（例如，被其他函数调用的辅助函数）。

m <- mirai(my_func(my_data), my_func = my_func, my_data = my_data)

快捷方式：传递整个调用环境

# .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())

如何选择

场景	使用
数据和简单函数	.args
需要词法作用域、由其他函数调用的辅助函数	...
将整个本地作用域传递到本地求值环境	.args = environment()
将整个本地作用域传递到守护进程的全局环境	mirai(expr, environment())
多个任务共享的大型对象	先使用 everywhere()，再按名称引用

常见错误

未限定命名空间的包函数

守护进程启动时不会加载任何用户包。mirai_map() 回调中也是如此。

# 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))

期望立即获得结果

m$data 可访问值，但值可能尚未解析。使用 m[]（或 collect_mirai(m)）阻塞等待任务完成；使用 unresolved(m) 进行非阻塞检查。

m <- mirai(slow_computation())
result <- m[]                          # blocks until resolved
if (!unresolved(m)) result <- m$data   # non-blocking

设置守护进程

无需守护进程

无需先调用 daemons()，mirai() 也能工作——每次调用都会启动一个临时后台进程。只有在需要持久化的工作进程池时，才需要设置守护进程。

本地守护进程

# 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)

作用域内的守护进程（自动清理）

with(daemons(...), {...}) 会创建守护进程，并在代码块退出时自动清理。

with(daemons(4), {
  m <- mirai(expensive_task())
  m[]
})

作用域内切换计算配置

local_daemons() 和 with_daemons() 会将活动计算配置切换为已存在的配置——它们不会创建守护进程。

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[]
})

计算配置（多个独立池）

daemons(4, .compute = "cpu")
daemons(2, .compute = "gpu")

m1 <- mirai(cpu_work(), .compute = "cpu")
m2 <- mirai(gpu_work(), .compute = "gpu")

内存背压（memory + try_mirai()）

对于高吞吐量的生产者（Shiny、promises、摄取管道），请使用 daemons() 的 memory 参数，为分发器处排队的任务负载设定上限（MB，度量值）。搭配 try_mirai() 使用，确保主机 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 需要分发器。没有分发器（或设置 memory = NULL）时，try_mirai() 始终会返回一个 mirai。

mirai_map：并行映射

必须先设置守护进程。它会将函数逐元素应用于 .x，并将任务分配到各个守护进程。回调中使用的任何包函数都要限定命名空间（参见错误 2）。

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]

映射多个参数（数据框行）

# 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))[]

按完成顺序处理（race_mirai）

race_mirai() 会返回列表中第一个已解析 mirai 的整数索引（如果列表为空，则返回 0L）。如果想按完成顺序而非提交顺序处理结果，此函数会很有用。

remaining <- mirai_map(jobs, run)
while (length(remaining) > 0) {
  idx <- race_mirai(remaining)
  process(remaining[[idx]]$data)
  remaining <- remaining[-idx]
}

everywhere：在所有守护进程上预加载状态

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)

错误处理

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 集成

ExtendedTask 模式

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()))
}

对于高流量应用，请设置 daemons(4, memory = ...)，并使用 try_mirai() 提交任务，从而施加背压而不阻塞 Shiny 事件循环。

Promise 管道

library(promises)
mirai({Sys.sleep(1); "done"}) %...>% cat()

远程 / 分布式计算

SSH（直接连接）

daemons(
  url = host_url(tls = TRUE),
  remote = ssh_config(c("ssh://user@node1", "ssh://user@node2"))
)

SSH（隧道连接，适用于防火墙环境）

daemons(
  n = 4,
  url = local_url(tcp = TRUE),
  remote = ssh_config("ssh://user@node1", tunnel = TRUE)
)

HPC 集群（Slurm/SGE/PBS/LSF）

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 启动器（例如，Posit Workbench）

daemons(n = 2, url = host_url(), remote = http_config())

从 future 转换

future	mirai
自动检测全局变量	必须显式传入所有依赖项
future({expr})	mirai({expr}, .args = list(...))
value(f)	m[] 或 collect_mirai(m)
plan(multisession, workers = 4)	daemons(4)
plan(sequential) / 重置	daemons(0)
future_lapply(X, FUN)	mirai_map(X, FUN)[]
future_map(X, FUN)（furrr）	mirai_map(X, FUN)[]
future_promise(expr)	mirai(expr, ...)（自动转换为 promise）

关键的转换步骤：找出表达式使用的调用环境中的所有对象，并通过 .args 或 ... 显式传入。

从 parallel 转换

parallel	mirai
makeCluster(4)	daemons(4) 或 make_cluster(4)
clusterExport(cl, "x")	通过 .args / ... 传入，或使用 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)

通过 make_cluster 实现直接替换

对于已大量使用 parallel 包的代码，make_cluster() 提供了可直接替换的后端：

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")

随机数生成

# 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)

调试

# 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)

高级模式：嵌套并行

在 daemon 回调中（例如 mirai_map），使用 local_url() + launch_local()，而不要使用 daemons(n)，以避免与外层 daemon 池冲突。

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
})[]