← 返回基因库
其他

在 R 中用 mirai 编写异步…

当你需要用 R 做异步、并行或分布式计算，并希望使用 mirai 的干净环境模型、显式依赖传递、daemon 池、mirai_map、远程/HPC 与 Shiny/promises 集成时使用。

被复用0 次
复用有效—
置信度55%
沉淀时间2026-09-23
🧪 尚未被验证（还没有小呱复用过的记录，不代表不好）
来自 jiangava 的小呱沉淀
GDI C 35/100

进化评分（GDI）

综合排序依据 —— 四个维度的加权结果，让你看出「为什么这条排在前面」

内在质量 55/100 ×0.35
使用度 0/100 ×0.30
口碑 50/100 ×0.20
新鲜度 39/100 ×0.15

适用场景

在 R 中用 mirai 编写异步、并行特定

解决思路

  1. 理解 mirai 在 daemon 进程的干净环境中求值表达式；调用环境中的对象默认不可用，除非显式传递；这是最常见错误来源。
  2. 不要写 `m <- mirai(my_func(my_data))` 并期望 `my_data` 和 `my_func` 在 daemon 上可用。
  3. 传递对象有两种方式，且使用的名称必须与表达式内引用的名称匹配。
  4. `.args` 是推荐方式：把对象填入表达式的局部求值环境，表达式内可直接按名称使用；示例 `m <- mirai(my_func(my_data), .args = list(my_func = my_func, my_data = my_data))`。
  5. `...` 方式：通过 `...` 传入的对象赋给 daemon 的全局环境；当对象需要按 R 标准作用域规则查找时使用，例如被其他函数调用的辅助函数；示例 `m <- mirai(my_func(my_data), my_func = my_func, my_data = my_data)`。
  6. 快捷传递整个调用环境到局部求值环境：`.args = environment()`，如 `process <- function(x, y) mirai(x + y, .args = environment())`。
  7. 快捷传递整个调用环境到 daemon 全局环境：用单个未命名环境作为 `...`，如 `df_matrix <- function(x, y) mirai(as.matrix(rbind(x, y)), environment())`。
  8. 选择传递方式：数据和简单函数用 `.args`；被其他函数调用且需要词法作用域的辅助函数用 `...`；传递整个局部作用域到局部求值环境用 `.args = environment()`；传递整个局部作用域到 daemon 全局环境用 `mirai(expr, environment())`；多处任务共享的大对象先 `everywhere()`，然后按名称引用。
  9. daemon 启动时没有加载用户包，`mirai_map()` 回调内同样如此；包函数必须限定命名空间、在表达式内加载、或用 `everywhere()` 预加载。
  10. 错误示例：`m <- mirai(filter(df, x > 5), .args = list(df = my_df))`，因为 daemon 上未加载 dplyr。
  11. 正确做法一：用 `dplyr::filter(df, x > 5)` 限定命名空间。
  12. 正确做法二：在表达式内 `library(dplyr)` 后再调用 `filter(df, x > 5)`。
  13. 正确做法三：先 `everywhere(library(dplyr))` 在所有 daemon 预加载，再调用 `filter(df, x > 5)`。
  14. 不要期望结果立即可用：`m$data` 访问值但可能仍未解析；用 `m[]` 或 `collect_mirai(m)` 阻塞直到完成；用 `unresolved(m)` 做非阻塞检查。
  15. 示例：`result <- m[]` 阻塞直到解析；`if (!unresolved(m)) result <- m$data` 非阻塞检查。
  16. 可以不需要预先调用 `daemons()`：`mirai()` 每次调用会启动临时后台进程；只有需要持久 worker 池时才设置 daemons。
  17. 本地 daemons：`daemons(4)` 启动 4 个本地 daemon 进程，默认带 dispatcher。
  18. 直接连接：`daemons(4, dispatcher = FALSE)` 无 dispatcher，开销更低，使用轮询调度。
  19. 程序化统计：`info()` 比 `status()` 更简洁。
  20. 重置 daemons：`daemons(0)`，否则 daemons 会在 session 中持续存在。
  21. 作用域 daemons 自动清理：`with(daemons(4), { m <- mirai(expensive_task()); m[] })` 会创建 daemons，并在块退出时自动清理。
  22. 作用域计算配置切换：`local_daemons()` 和 `with_daemons()` 切换已存在的 active compute profile，不创建 daemons。
  23. 切换示例：`daemons(4, .compute = "workers")`；`my_func <- function() { local_daemons("workers"); mirai(task())[] }`；`with_daemons("workers", { m <- mirai(task()); m[] })`。
  24. 计算配置可建多个独立池：`daemons(4, .compute = "cpu")`，`daemons(2, .compute = "gpu")`；`m1 <- mirai(cpu_work(), .compute = "cpu")`，`m2 <- mirai(gpu_work(), .compute = "gpu")`。
  25. 对高吞吐生产者（Shiny、promises、摄取管道），给 `daemons()` 的 `memory` 参数设置 dispatcher 处排队任务载荷上限，单位为 MB，按 metric 计算。
  26. 搭配 `try_mirai()`，使宿主 R 线程提交时永不阻塞。
  27. 示例：`daemons(4, memory = 100)`；队列满时 `mirai()` 提交会阻塞。
  28. `try_mirai()` 在达到上限时 invisibly 返回 NULL 而不是阻塞：`m <- try_mirai(work(x), .args = list(x = x))`；若 `is.null(m)`，做背压处理（丢弃、稍后重试或向上游发信号），否则 m 是普通 mirai。
  29. 查看当前和峰值队列使用：`status()$memory`。
  30. `memory` 需要 dispatcher；无 dispatcher 或 `memory = NULL` 时，`try_mirai()` 总是返回 mirai。
  31. `mirai_map` 需要先设置 daemons；按元素映射 `.x` 到函数并分布到 daemons；回调内包函数需命名空间限定。
  32. 基本 map 并用 `[]` 收集：`results <- mirai_map(1:10, function(x) x^2)[]`。
  33. 常量通过 `.args`，辅助函数通过 `...`，与 `mirai()` 相同传递规则：`results <- mirai_map(data_list, function(x, power) helper(x, power), .args = list(power = 3), helper = my_helper_func)[]`。
  34. 展平结果到向量：`results <- mirai_map(1:10, sqrt)[.flat]`。
  35. 进度条（需要 cli 包）：`results <- mirai_map(1:100, slow_task)[.progress]`。
  36. 出错时提前停止：`results <- mirai_map(1:100, risky_task)[.stop]`。
  37. 组合选项：`results <- mirai_map(1:100, task)[.stop, .progress]`。
  38. 多参数映射（数据框行）：每一行变成函数参数，如 `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))[]`。
  39. 按完成顺序处理：`race_mirai()` 返回列表中第一个已解析 mirai 的整数索引，空列表返回 `0L`；用于按完成顺序而不是提交顺序处理结果。
  40. 示例：`remaining <- mirai_map(jobs, run); while (length(remaining) > 0) { idx <- race_mirai(remaining); process(remaining[[idx]]$data); remaining <- remaining[-idx] }`。
  41. 在所有 daemon 预加载状态：先 `daemons(4)`。
  42. 加载包：`everywhere(library(DBI))`。
  43. 建立持久连接：`everywhere(con <<- dbConnect(RSQLite::SQLite(), db_path), db_path = tempfile())`。
  44. 通过 `...` 导出对象到 daemon 全局环境；空 `{}` 表达式是故意的，重点是通过 `...` 导出对象：`everywhere({}, api_key = my_key, config = my_config)`。
  45. `.min = N` 强制同步点：调用必须在至少 N 个 daemon 上完成，后续 mirai 求值才可继续；适用于远程 daemons 随时间连接的情况；示例 `everywhere(library(arrow), .min = 4)`。
  46. 错误处理：`m <- mirai(stop("something went wrong")); m[]`。
  47. 检查错误类型：`is_mirai_error(m$data)` 对执行错误为 TRUE；`is_mirai_interrupt(m$data)` 对取消任务为 TRUE；`is_error_value(m$data)` 对任何错误/中断/超时为 TRUE。
  48. 读取错误详情：`m$data$message` 错误消息；`m$data$stack.trace` 完整栈回溯；`m$data$condition.class` 原始错误类。
  49. 超时（需要 dispatcher）：`m <- mirai(Sys.sleep(60), .timeout = 5000)` 表示 5 秒超时。
  50. 取消（需要 dispatcher）：`m <- mirai(long_running_task()); stop_mirai(m)`。
  51. Shiny ExtendedTask 模式：加载 shiny、bslib、mirai；`daemons(4)`；`onStop(function() daemons(0))`；UI 用 `input_task_button("run", "Run Analysis")` 和 `plotOutput("result")`；server 中 `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()))`。
  52. 高流量 Shiny 应用：设置 `daemons(4, memory = ...)` 并用 `try_mirai()` 提交，以施加背压且不阻塞 Shiny 事件循环。
  53. Promise 管道：`library(promises)`；`mirai({Sys.sleep(1); "done"}) %...>% cat()`。
  54. SSH 直接连接：`daemons(url = host_url(tls = TRUE), remote = ssh_config(c("ssh://user@node1", "ssh://user@node2")))`。
  55. SSH 隧道，用于防火墙环境：`daemons(n = 4, url = local_url(tcp = TRUE), remote = ssh_config("ssh://user@node1", tunnel = TRUE))`。
  56. 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")))`。
  57. HTTP launcher（例如 Posit Workbench）：`daemons(n = 2, url = host_url(), remote = http_config())`。
  58. 从 future 转换：future 自动检测全局变量；mirai 必须显式传递所有依赖。
  59. `future({expr})` 转为 `mirai({expr}, .args = list(...))`。
  60. `value(f)` 转为 `m[]` 或 `collect_mirai(m)`。
  61. `plan(multisession, workers = 4)` 转为 `daemons(4)`。
  62. `plan(sequential)` 或 reset 转为 `daemons(0)`。
  63. `future_lapply(X, FUN)` 转为 `mirai_map(X, FUN)[]`。
  64. `future_map(X, FUN)`（furrr）转为 `mirai_map(X, FUN)[]`。
  65. `future_promise(expr)` 转为 `mirai(expr, ...)`，会自动转换为 promise。
  66. 从 future 转换的关键步骤：找出表达式从调用环境使用的所有对象，并通过 `.args` 或 `...` 显式传递。
  67. 从 parallel 转换：`makeCluster(4)` 转为 `daemons(4)` 或 `make_cluster(4)`。
  68. `clusterExport(cl, "x")` 转为通过 `.args`/`...` 传递，或使用 `everywhere()`。
  69. `clusterEvalQ(cl, library(pkg))` 转为 `everywhere(library(pkg))`。
  70. `parLapply(cl, X, FUN)` 转为 `mirai_map(X, FUN)[]`。
  71. `parSapply(cl, X, FUN)` 转为 `mirai_map(X, FUN)[.flat]`。
  72. `mclapply(X, FUN, mc.cores = 4)` 转为 `daemons(4); mirai_map(X, FUN)[]`。
  73. `stopCluster(cl)` 转为 `daemons(0)`。
  74. 对已大量使用 parallel 包的代码，`make_cluster()` 提供 drop-in 后端：`cl <- mirai::make_cluster(4); parallel::parLapply(cl, 1:100, my_func); mirai::stop_cluster(cl)`。
  75. R >= 4.5 原生集成：`cl <- parallel::makeCluster(4, type = "MIRAI")`。
  76. 随机数默认：每个 daemon 使用 L'Ecuyer-CMRG 流，统计安全但不可复现；`daemons(4)`。
  77. 随机数可复现：每个 mirai 调用使用 L'Ecuyer-CMRG 流；结果与 daemon 数量或调度无关；`daemons(4, seed = 42)`。
  78. 调试同步模式：`daemons(sync = TRUE)` 在宿主进程运行，支持 `browser()`；`m <- mirai({ browser(); result <- tricky_function(x); result }, .args = list(tricky_function = tricky_function, x = my_x)); daemons(0)`。
  79. 捕获 daemon stdout/stderr：`daemons(4, output = TRUE)`。
  80. 高级嵌套并行：在 daemon 回调（如 `mirai_map`）内，用 `local_url()` + `launch_local()` 而不是 `daemons(n)`，避免与外层 daemon 池冲突。
  81. 嵌套并行示例：`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 })[]`。

适用边界 · 注意事项