mizu_pool() builds on the same shared-memory transport as a channel. It spawns worker processes that each own a work-stealing deque, with no dispatcher process in between. A submitted task goes from your process into shared memory, where a worker claims it. Idle workers steal from busy ones, so the load balances itself.
mizu_submit() captures an expression together with the values it needs and returns a task handle immediately. Your session stays free to continue. mizu_collect() waits for the result of that task:
library(mizu)
p <- mizu_pool(n_workers = 4L)
t <- mizu_submit(p, sum(x) + y, x = 1:10, y = 100)
mizu_collect(t)
#> [1] 155If a task raises an error, the pool captures it and signals it again in your session when you collect the result. mizu_cancel(t) withdraws a task. A task still queued is skipped. A task already running completes, and its result is discarded. Cancellation never interrupts executing code.
Inside a task, mizu_current_pool() returns the handle of the worker running it, so a task can split itself into subtasks. A nested submit pushes onto that worker’s own work-stealing deque, so there is no ring and no wake-up. A worker waiting on a nested result executes other work instead of sleeping. Divide-and-conquer has little overhead and cannot deadlock the pool:
t <- mizu_submit(
p,
{
subtasks <- lapply(
parts,
\(part) mizu_submit(mizu_current_pool(), sum(x), x = part)
)
do.call(sum, lapply(subtasks, mizu_collect))
},
parts = split(1:1000, rep(1:4, each = 250))
)
mizu_collect(t)
#> [1] 500500mizu_map() nests the same way: a task can map over its own pool with mizu_map(mizu_current_pool(), ...). See Nested maps.
Package code can read a process-wide default pool with mizu_default_pool(). mizu_set_default_pool() sets it (or clears it with NULL) and invisibly returns the previous default, so you can save and restore it. A pool set as the default stays alive even after its variable is removed. mizu_with_pool() scopes the default to an expression, mizu_local_pool() to the calling frame. Both restore the previous default on exit.
Resolve an optional pool in this order: an explicit pool argument, mizu_current_pool() inside a task, mizu_default_pool(), then your own fallback, such as sequential evaluation or an error. Inside a task, mizu_current_pool() takes precedence over the default, so nested submission always goes to the worker’s own pool. Setting a default checks only the type: a stopped pool is accepted and fails when used. Handles from mizu_pool_attach() are valid defaults.
run <- function(x, pool = mizu_default_pool()) {
if (is.null(pool)) {
stop("no pool: pass one, or set a default", call. = FALSE)
}
mizu_collect(mizu_submit(pool, x * 2L, x = x))
}
old <- mizu_set_default_pool(p)
run(21L)
#> [1] 42
mizu_set_default_pool(old)mizu_collect_any() waits on a list of task handles and returns list(index, value) for the first to reach a terminal state, in completion order rather than submission order. The other handles stay collectible. mizu_collect_all() waits until every task is terminal and returns all values in input order, with the names of the handle list carried over. Both take one overall timeout and return the mizu_timeout sentinel on expiry, consuming nothing. An outcome that raises (the task’s own error, a cancelled task, a dead worker) carries a 1-based index field with the handle’s position, and only that handle is consumed:
slow <- mizu_submit(p, { Sys.sleep(0.5); "slow" })
fast <- mizu_submit(p, "fast")
mizu_collect_any(list(slow, fast), timeout = 5)
#> $index
#> [1] 2
#>
#> $value
#> [1] "fast"
mizu_collect(slow)
#> [1] "slow"
ts <- list(
total = mizu_submit(p, sum(x), x = runif(10)),
label = mizu_submit(p, "done")
)
mizu_collect_all(ts, timeout = 5)
#> $total
#> [1] 3.628258
#>
#> $label
#> [1] "done"On the submit side, mizu_submit_batch() submits one task per element of a list of pre-quoted expressions in a single call. The whole burst crosses into shared memory once, with one wake-up sweep, and the ... arguments are shared by every task:
ts <- mizu_submit_batch(p, list(quote(1 + 1), quote(2 + 2)))
mizu_collect_all(ts, timeout = 5)
#> [[1]]
#> [1] 2
#>
#> [[2]]
#> [1] 4In an event-driven program, call then() directly on a task handle. then(task, fn) (or as.promise(task)) resolves with the task’s value and rejects with the classed condition that mizu_collect() would raise, including worker death. mizu_collect_all_async() and mizu_collect_any_async() are the async forms of the collective functions, with one promise per collection. Collect-all resolves in input order with names carried over, and rejects with index on the first erroring handle by position. Collect-any resolves list(index, value). A map handle works the same way: then(map_handle, fn) resolves with the whole gathered result. This needs the later and promises packages.
There is one promise per collection, not per task. A promise costs tens of microseconds of R work against about a microsecond of transport, so a map of ten thousand elements resolves one promise rather than ten thousand. A burst of results produces one wake-up no matter how many workers published it, and every main-thread drain is bounded per event-loop turn and re-scheduled.
library(promises)
slow <- mizu_submit(p, { Sys.sleep(0.2); "slow" })
then(slow, print)
m <- mizu_map(p, 1:100, \(i) i * i, .template = integer(1))
then(m, \(v) print(sum(v)))
for (i in 1:8) later::run_now(1) # pump the loop until the resolutions land
#> [1] 338350
#> [1] "slow"A task with a promise attached, or one that is part of a pending async collection, refuses a blocking mizu_collect() and a second promise: each result has one consumer. mizu_cancel() is still allowed, and the pending promise then rejects.
A pool can grow and shrink while it runs. Retirement never interrupts a task: the worker finishes its current task, and the remaining workers take over anything still queued to it. Other R processes can join a running pool as submitters. They need only the pool’s join token:
mizu_spawn_workers(p, n = 2L) # two more workers join the pool
mizu_retire_worker(p, slot = 0L) # worker 0 exits after its current task
# in another R process, submit and collect exactly as the creator does:
p2 <- mizu_pool_attach(token) # token: mizu_pool_info(p)$token on the creatorFour read-only tools observe a running pool without disturbing it:
mizu_pool_info(p) # snapshot: worker states, queued tasks, result slots
mizu_pool_stats(p) # cumulative counters: tasks run, steals, parks per worker
mizu_pool_dump(p) # every slot in full detail; start here when a pool hangs
mizu_pool_trace(p, \(event, id) message(event, " ", id)) # task lifecycle hookWhen you are done, mizu_pool_stop() cancels the pending tasks, waits for the workers to exit cleanly, and releases the shared region:
mizu_pool_stop(p)