r-mirai

mirai は、nanonext を基盤とする、非同期・並列・分散評価のためのミニマルな R フレームワークです。

中核となる原則：依存関係を明示的に渡す

mirai は、デーモンプロセス上のクリーンな環境で式を評価します。明示的に渡さない限り、呼び出し元の環境にあるものは何も利用できません。これが間違いの最大の原因です。

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

オブジェクトを渡す方法は 2 つあり、使用する名前は、式内で参照する名前と一致している必要があります。

.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) / 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, ...)（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)

高度なパターン：ネストされた並列処理

デーモンのコールバック内（例：mirai_map）では、外側のデーモンプールとの競合を避けるため、daemons(n) の代わりに local_url() と launch_local() を使用します。

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