---
title: "Task pools"
format:
html:
toc: true
vignette: >
%\VignetteIndexEntry{Task pools}
%\VignetteEngine{quarto::html}
%\VignetteEncoding{UTF-8}
---
`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:
``` r
library(mizu)
p <- mizu_pool(n_workers = 4L)
t <- mizu_submit(p, sum(x) + y, x = 1:10, y = 100)
mizu_collect(t)
#> [1] 155
```
If 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.
## Nested tasks
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:
``` r
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] 500500
```
`mizu_map()` nests the same way: a task can map over its own pool with `mizu_map(mizu_current_pool(), ...)`.
See [Nested maps](map.html#nested-maps).
## A default pool for package code
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.
``` r
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)
```
## Waiting on several tasks
`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:
``` r
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:
``` r
ts <- mizu_submit_batch(p, list(quote(1 + 1), quote(2 + 2)))
mizu_collect_all(ts, timeout = 5)
#> [[1]]
#> [1] 2
#>
#> [[2]]
#> [1] 4
```
## Promises and handlers
In 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.
``` r
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.
## Sizing, sharing and watching a pool
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:
``` r
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 creator
```
Four read-only tools observe a running pool without disturbing it:
``` r
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 hook
```
When you are done, `mizu_pool_stop()` cancels the pending tasks, waits for the workers to exit cleanly, and releases the shared region:
``` r
mizu_pool_stop(p)
```