----------------------------------------------------------------------
This is the API documentation for the pymizu library.
----------------------------------------------------------------------


## Channels

A two-way message link between a Python process and a helper process that it spawns.


Channel()

A shared-memory SPSC channel handle (process-private).

Create the host side with [`Channel.create()`](`pymizu.Channel.create`)
(which spawns the peer); the peer side attaches with
[`Channel.attach()`](`pymizu.Channel.attach`) —
``python -m pymizu.child`` does this. Handles do not survive
``fork()``.


## Channel Methods

Methods for the Channel class


create(peer: 'str', *, capacity: 'int' = 16384, slot_size: 'int' = 256, arena_size: 'int' = 4194304, spin: 'bool' = False, startup_timeout: 'float' = 30.0, launcher: '_Callable[[str], _Any] | None' = None) -> 'Channel'

Create a channel and spawn its peer.

Parameters
----------
peer
    A Python source string, evaluated in the peer process with
    ``ch`` bound to the peer-side handle.
capacity
    Ring capacity in slots; a power of two.
slot_size
    Bytes per ring slot; a power of two. The inline payload
    budget is ``slot_size - 16``.
arena_size
    Bytes of shared staging arena for out-of-line payloads.
spin
    Spin rather than park while waiting.
startup_timeout
    Seconds to wait for the peer to attach.
launcher
    A ``callable(token)`` arranging for a Python process to run
    ``python -m pymizu.child <token>``; the default spawns
    ``sys.executable`` directly.
    [`r_launcher()`](`pymizu.r_launcher`) returns one spawning
    an R peer (the R package ``mizu``).

Returns
-------
    The host-side channel handle.

Examples
--------
>>> import pymizu
>>> with pymizu.Channel.create(
...     """
... import pymizu
... while True:
...     x = ch.recv()
...     if x is pymizu.CLOSED:
...         break
...     ch.send(x)
... """
... ) as ch:
...     ch.send([1, "a", None])
...     received = ch.recv(timeout=5)
>>> received
[1, 'a', None]

attach(token: 'str') -> 'Channel'

Attach to the channel named by a join token (the peer side).

Parameters
----------
token
    The join token from the host side's ``ch.token``.

Returns
-------
    The peer-side channel handle.

Details
-------
Low-level: the caller consumes ``ch.drop`` before signalling
``ch.ready_set()``. ``python -m pymizu.child`` is the reference
peer entry.

token

The join token for the peer's attach.

send(self, x: '_Any') -> 'Sentinel | None'

Send one payload.

Parameters
----------
x
    The payload. ``None`` itself is a valid payload; bytes and
    numpy arrays travel raw, everything else rides pickle
    protocol 4.

Returns
-------
    None, or the `FULL` / `CLOSED` / `PEER_GONE` sentinel
    (identity-tested).

send_batch(self, xs: '_Iterable[_Any]') -> 'int'

Send several payloads in one crossing; return the number
accepted (short on ring-full or a terminal state).

recv(self, timeout: 'float | None' = None) -> '_Any'

Receive one payload.

Parameters
----------
timeout
    Seconds to wait; None waits indefinitely.

Returns
-------
    The payload, or the `TIMEOUT` / `CLOSED` / `PEER_GONE`
    sentinel on the non-payload outcomes.

recv_async(self) -> '_Any'

The event-driven receive: an `asyncio.Future` for the next
payload, driven by the channel's notify endpoint — never blocks,
never polls.

Returns
-------
    An `asyncio.Future` resolving with the identical object a
    blocking [`recv()`](`pymizu.Channel.recv`) would have
    returned (the recv runs at drain time, on the loop thread),
    zero-copy views included. Terminal states stay values: the
    Future resolves with the `CLOSED` / `PEER_GONE` sentinel.

Details
-------
Several outstanding calls on one channel resolve in FIFO call
order, one message each, from the per-handle pending queue.
Arming is loss-free: data already in the ring resolves from the
arm's synchronous drain; only genuinely pending entries ride the
watch. Cancelling the Future detaches the waiter — the next
message goes to the next waiter. There is no `timeout` argument:
`asyncio.timeout` / `asyncio.wait_for` are the idiom. A spin
channel cannot arm (its publish path skips the wake checkpoint
by design) and raises.

__aiter__(self) -> '_Any'

Iterate over received payloads until CLOSED or PEER_GONE —
the async form of the sync iterator.

Yields payloads as they arrive, awaiting each through
[`recv_async()`](`pymizu.Channel.recv_async`). Between
iterations nothing is armed; the next iteration re-arms and the
arm's synchronous drain picks up whatever arrived meanwhile
(loss-free).

recv_batch(self, n: 'int' = 256, timeout: 'float | None' = None) -> 'list[_Any] | Sentinel'

Receive up to ``n`` payloads in one crossing; a list
(possibly short or empty), or a terminal sentinel.

close(self, timeout: 'float' = 5.0) -> 'bool'

Orderly close: signal the peer and wait up to ``timeout``
seconds for it to observe the close. True on a clean handshake;
False (with a warning) on timeout, in which case resources
release when the handle is garbage collected.

close_signal(self) -> 'None'

Signal close without waiting (the peer's recv side sees
CLOSED once the ring drains).

destroy(self) -> 'None'

Tear down the handle immediately, without the close
handshake: the peer observes the channel closed (PEER_GONE is
the hard-death outcome). Safe from any thread: a verb in
flight on another thread pins the handle, and teardown
completes when the last one returns — an indefinite wait
(recv, recv_batch, ready_wait, close) breaks out within a
bounded slice, so the teardown (and the peer's observation of
it) follows promptly. Pending recv futures settle with
PEER_GONE.

alive(self) -> 'bool'

True while the peer process is alive.

info(self) -> 'dict'

A read-only wire-state snapshot of the channel (dict).

__enter__(self) -> 'Channel'

__exit__(self, *exc: '_Any') -> 'bool'

__iter__(self) -> '_Iterator[_Any]'

Iterate over received payloads until CLOSED or PEER_GONE.

Blocks indefinitely between payloads (``recv()`` with no
timeout); use [`Channel.recv()`](`pymizu.Channel.recv`) directly
when a bound is needed.


## Task pools

A set of worker processes that divide submitted tasks among themselves.


Pool()

A shared-memory work-stealing task pool handle (process-private).

Create the controller side with [`Pool.create()`](`pymizu.Pool.create`)
(which spawns the workers, ``python -m pymizu.worker``); other
processes join as submitters with
[`Pool.attach()`](`pymizu.Pool.attach`). The pool's lifetime is
bound to the creating process: dropping the handle shuts the pool
down as [`Pool.stop()`](`pymizu.Pool.stop`) does, but without the
wait. Handles do not survive ``fork()``.

Task callables ride pickle: under stock pickle a submitted callable
must be an importable reference (the multiprocessing constraint);
installing cloudpickle lifts that transparently.

_Task()

A task handle from _Pool.submit(). Collected exactly once; an uncollected handle's finalizer releases its slot.

PreparedMap(pool: 'pymizu.Pool', fn: '_Callable[..., _Any]', x: '_Any', *, args: '_Any' = (), kwargs: 'dict | None' = None, n_chunks: 'int | None' = None, seed: 'int | bytes | bytearray | tuple[int | bytes | bytearray, int] | None' = None, template: '_Any' = None, collect: 'str | None' = None, stream: 'bool' = False) -> 'None'

A map staged once into a persistent region, run many times.

Create with [`Pool.map_prepare()`](`pymizu.Pool.map_prepare`); run
with `run()`, which returns the run's map handle (the collect
happens on it, as for [`Pool.map()`](`pymizu.Pool.map`)). Re-arming
is O(1) — a generation bump, a cursor reset, a cancel-word clear —
and workers reuse their cached contexts, so a re-run pays neither
the descriptor pickle nor the region create nor the worker-side
re-attach. A view-collected run hands its region to the view; the
next run restages into a fresh one (the mirror of mizu). Close with
`close()` (or a with block) to unlink the region.

Parameters
----------
pool
    The [`Pool`](`pymizu.Pool`) to run on.
fn
    The function to map, or a [`call`](`pymizu.call`) spec — as for
    [`Pool.map()`](`pymizu.Pool.map`).
x
    The data to map over.
args
    Constant positional arguments appended to every call.
kwargs
    Constant keyword arguments for every call.
n_chunks
    Overrides the morsel count (the scheduling granularity).
seed
    Deterministic per-element seeding — as for
    [`Pool.map()`](`pymizu.Pool.map`).
template
    An exemplar buffer declaring each result's dtype and length —
    as for [`Pool.map()`](`pymizu.Pool.map`).
collect
    'copy' or 'view' with `template`; 'list' without.
stream
    Stream slices of `x` instead of staging it whole — as for
    [`Pool.map()`](`pymizu.Pool.map`).

Examples
--------
Stage the map once, then re-run it over replacement data at memcpy
cost (same dtype and length swap in place):

```{python}
import numpy as np
import pymizu

with pymizu.Pool.create(2) as pool:
    with pool.map_prepare(np.square, np.arange(4.0)) as pm:
        first = pm.run().collect(timeout=10)
        second = pm.run(np.array([4.0, -5.0, 6.0, 7.0])).collect(
            timeout=10
        )
second
```

PoolExecutor(pool: 'pymizu.Pool', *, stop_pool: 'bool' = False) -> 'None'

A `concurrent.futures.Executor` over a pymizu pool.

`submit()` returns a real `concurrent.futures.Future`: `wait` /
`as_completed` / `result(timeout)` / `asyncio.wrap_future` and the
stdlib base `Executor.map` all work unchanged, so code written
against `ProcessPoolExecutor` drops in. `create()` spawns an owned
pool instead.

Parameters
----------
pool
    The pool to wrap; its lifetime stays with its owner unless
    `stop_pool` is set — then `shutdown()` stops it.
stop_pool
    When True, `shutdown()` stops the wrapped pool.

Notes
-----
Deltas from stdlib semantics: a task's failure surfaces as
`pymizu.TaskError` (the remote error envelope);
`Future.running()` is always False (futures resolve at completion);
cancellation is advisory, mirroring [`Task.cancel()`](`pymizu.Task`)
— and `shutdown(cancel_futures=True)` cancels every outstanding
future, since the pool exposes no queued-versus-running signal (a
running task completes, but its result is discarded). An
infrastructure failure (e.g. the wrapped pool is stopped from under
the executor) fails every outstanding future; later submissions
raise instead of queueing onto a broken executor.
Resolution — including any `add_done_callback` callables — runs on
the core's listener thread, the same discipline as the stdlib
executors' management thread: keep callbacks short, and never block
one on the pool itself.
`map` is the stdlib base implementation over `submit` — for bulk
maps, [`Pool.map`](`pymizu.Pool.map`) is the faster path.

Examples
--------
Use as a drop-in `concurrent.futures.Executor`:

```{python}
import pymizu

with pymizu.PoolExecutor.create(2) as ex:
    futures = [ex.submit(pow, 2, i) for i in range(4)]
    results = [f.result() for f in futures]
results
```


## Pool Methods

Methods for the Pool class


create(workers: 'int' = 1, *, max_workers: 'int | None' = None, max_submitters: 'int' = 8, injection_cap: 'int' = 1024, per_worker_cap: 'int' = 1024, result_slots: 'int' = 4096, slot_size: 'int' = 512, launcher: '_Callable[[str, int], _Any] | None' = None, startup_timeout: 'float' = 30.0) -> 'Pool'

Create a pool and spawn its worker processes.

Parameters
----------
workers
    Number of worker processes to spawn; they join the pool's
    registry (capacity ``max_workers``).
max_workers
    Registry capacity; defaults to ``workers``.
max_submitters
    Maximum number of submitter processes.
injection_cap
    Capacity of the injection ring.
per_worker_cap
    Capacity of each worker's work-stealing deque.
result_slots
    Bounds each submitter's outstanding (uncollected) tasks.
slot_size
    Bytes per queue entry and result slot — a payload past the
    inline budget travels in a fresh region per payload.
launcher
    A ``callable(token, slot)`` arranging for a Python process
    to run ``python -m pymizu.worker <token> <slot>``; the
    default spawns ``sys.executable`` directly.
startup_timeout
    Seconds to wait for the workers to attach.

Returns
-------
    The controller-side pool handle.

Examples
--------
>>> import pymizu
>>> with pymizu.Pool.create(2) as pool:
...     task = pool.submit(pow, 2, 16)
...     results = pool.collect_all([task], timeout=10)
>>> results
[65536]

attach(token: 'str') -> 'Pool'

Attach to a live pool as a submitter, by its join token.

Parameters
----------
token
    The join token; it travels out of band and is
    ``pool.token`` on the creator.

Returns
-------
    The submitter-side pool handle.

token

The join token for worker/submitter attach.

submit(self, fn: '_Callable[..., _Any]', /, *args: '_Any', timeout: 'float | None' = None, **kwargs: '_Any') -> 'Task'

Submit ``fn(*args, **kwargs)`` as a task; return a Task handle.

Blocks only for injection-ring space, up to ``timeout`` seconds
(None waits indefinitely).

Parameters
----------
fn
    The callable to run, or a [`call`](`pymizu.call`) spec.
args
    Positional arguments for ``fn``.
timeout
    Seconds to wait for injection-ring space; None waits
    indefinitely.
kwargs
    Keyword arguments for ``fn``.

Returns
-------
    A [`Task`](`pymizu.Task`) handle.

Raises
------
SubmitTimeoutError
    On ``timeout`` expiry.
SlotsExhaustedError
    When no result slot is free.
StoppedError
    When the pool is stopped.

Details
-------
``timeout`` belongs to the submission, not to ``fn``: a callable
taking its own ``timeout=`` keyword argument cannot receive it
through ``**kwargs`` here — bind it first with
`functools.partial` as ``functools.partial(fn, timeout=...)``.

A buffer-protocol argument (e.g. a numpy array) past the
zero-copy floor crosses as a read-only view over shared pages,
not a writable copy.

With a [`call`](`pymizu.call`) spec as ``fn`` (no ``*args`` /
``**kwargs`` — the spec carries them), the task stream crosses
in the neutral interchange format: this is how a pool of another
language's workers is driven (spawn them with
[`r_pool_launcher()`](`pymizu.r_pool_launcher`)). On a foreign
pool a plain callable errors locally, naming the spec verb.

Examples
--------
See [`Pool.create()`](`pymizu.Pool.create`) for a submit and
collect example.

submit_batch(self, fns: '_Iterable[_Callable[[], _Any]]', *, timeout: 'float | None' = None) -> 'list[Task]'

Submit one task per zero-arg callable in ``fns`` in one
crossing.

Parameters
----------
fns
    Zero-arg callables, one task each. Use `functools.partial`
    to bind arguments.
timeout
    Seconds to wait for injection-ring space; None waits
    indefinitely.

Returns
-------
    A list of [`Task`](`pymizu.Task`) handles. Ring-full past
    ``timeout`` ends the batch short — the returned handles
    stay valid and collectible. A mid-batch failure raises with
    an ``accepted`` attribute holding the handles already
    submitted, which stay valid and collectible.

collect_any(self, tasks: '_Iterable[Task]', timeout: 'float | None' = None) -> 'tuple[int, _Any] | Sentinel'

Wait on several tasks.

Parameters
----------
tasks
    The [`Task`](`pymizu.Task`) handles to wait on.
timeout
    Seconds to wait; None waits indefinitely.

Returns
-------
    ``(index, value)`` of the first terminal task, or the
    `TIMEOUT` sentinel. A non-OK outcome raises with an
    ``index`` attribute (0-based).

collect_all(self, tasks: '_Iterable[Task]', timeout: 'float | None' = None) -> 'list[_Any] | Sentinel'

Wait until every task is terminal.

Parameters
----------
tasks
    The [`Task`](`pymizu.Task`) handles to wait on.
timeout
    Seconds to wait; None waits indefinitely.

Returns
-------
    All values in input order. On the first non-OK outcome by
    position, raises with an ``index`` attribute (0-based) —
    only the reported handle is consumed; every other, the
    results ahead of it included, stays collectible. A result
    that fails to read raises with ``index`` and a ``values``
    attribute carrying the consumed prefix. The `TIMEOUT`
    sentinel consumes nothing.

collect_any_async(self, tasks: '_Iterable[Task]') -> '_Any'

The event-driven [`collect_any()`](`pymizu.Pool.collect_any`):
exactly one `asyncio.Future` per collection — never one per task
— driven by the enrolled tasks' notify announces.

Parameters
----------
tasks
    The [`Task`](`pymizu.Task`) handles to wait on.

Returns
-------
    An `asyncio.Future` resolving with the first terminal task's
    ``(index, value)`` (the sync verb's tie-breaks); a non-OK
    outcome rejects with the exception the sync verb raises
    (``index`` attribute carried). There is no `timeout`
    argument: `asyncio.timeout` / `asyncio.wait_for` are the
    idiom. Cancelling the Future un-enrolls every task — each
    returns to blocking-collectible — and never cancels them.

collect_all_async(self, tasks: '_Iterable[Task]') -> '_Any'

The event-driven [`collect_all()`](`pymizu.Pool.collect_all`):
exactly one `asyncio.Future` per collection.

Parameters
----------
tasks
    The [`Task`](`pymizu.Task`) handles to wait on.

Returns
-------
    An `asyncio.Future` resolving with all values in input
    order; the first non-OK outcome by position rejects with the
    exception the sync verb raises (``index`` attribute carried
    — only that handle is consumed, the rest stay collectible).
    Cancelling the Future un-enrolls every task and never
    cancels them.

map(self, fn: '_Callable[..., _Any]', x: '_Iterable[_Any]', *, args: '_Iterable[_Any]' = (), kwargs: 'dict[str, _Any] | None' = None, n_chunks: 'int | None' = None, seed: 'int | bytes | bytearray | tuple[int | bytes | bytearray, int] | None' = None, timeout: 'float | None' = None, template: '_Any' = None, collect: 'str | None' = None, stream: 'bool' = False) -> '_Any'

Map ``fn`` over the elements of ``x`` on the pool; return the
run's map handle.

Parameters
----------
fn
    The callable to apply, or a [`call`](`pymizu.call`) spec.
x
    The iterable of elements.
args
    Constant positional arguments appended to every call.
kwargs
    Constant keyword arguments of every call.
n_chunks
    Overrides the morsel count (the scheduling granularity).
seed
    An int or bytes deriving deterministic per-element streams
    of the stdlib `random` module; see Details. Pass
    ``seed=(seed, offset)`` to shift every element's stream by
    ``offset`` positions, for maps split across runs or
    processes.
timeout
    Seconds to wait for completion; None waits indefinitely.
template
    An exemplar buffer (e.g. ``numpy.empty(m, dtype=...)``)
    declaring that every ``fn`` result is ``m`` values of that
    dtype; see Details.
collect
    ``"copy"`` (the default) or ``"view"``; governs how a
    template-backed output area is returned.
stream
    Stream slices of ``x`` to workers instead of staging the
    whole of ``x`` in shared memory; see Details.

Returns
-------
    The run's map handle. ``handle.collect()`` blocks for the
    results — a list in input order, or with ``template`` the
    gathered output area (a view with ``collect="view"``) —
    raising the errors below; ``handle.collect_async()`` /
    ``await handle`` resolves one `asyncio.Future` with the same
    gather (not on a streaming map: its sliding window needs the
    blocking turn loop, so it collects blocking only);
    ``handle.cancel()`` stops the run. On ``timeout`` expiry
    mid-submission, `map()` itself returns the `TIMEOUT`
    sentinel (no handle); a length-0 ``x`` returns a handle
    whose collect yields the empty result.

Raises
------
TaskError
    A task error re-raises from the collect as TaskError
    carrying the failing element's 0-based ``index``; failure is
    fail-fast (peers stop within about one batch).
WorkerDiedError
    Worker death carries the lost element ranges as ``lost``
    (0-based half-open ``(lo, hi)`` pairs, conservative).
TypeError
    A [`call`](`pymizu.call`) spec as ``fn`` with
    ``stream=True``.
DeclinedError
    With a spec, a non-portable constant or element at stage
    time.

Details
-------
One call stages ``fn``, the constant ``args``/``kwargs``, and
``x`` exactly once (a shared region, or inline in chunk tasks
when small), then submits one *runner* task per live worker;
runners self-schedule adaptively sized element batches off a
shared cursor. A C-contiguous buffer of a supported dtype
(float64/int32/int64/complex128/uint8) travels as bare bytes —
workers wrap it once and index per element.

``seed`` derives deterministic per-element streams of the stdlib
`random` module: element ``i`` runs under
``random.seed(SHA-256(seed_bytes + i.to_bytes(8, "little")))``,
identical for any chunking, worker count, or steal order.
``seed`` covers the stdlib `random` module only: a task drawing
from numpy calls [`current_rng()`](`pymizu.current_rng`) inside
the task (the element's own memoized
``numpy.random.Generator``, derived from the same seed
material); other RNG universes are out of scope. The legacy
``np.random.*`` module functions draw from the worker's shared
global RandomState and stay order-dependent. A seeded map
element must not nested-submit and collect: worker helping can
run another seeded map's batches mid-element, wiping this
element's [`current_rng()`](`pymizu.current_rng`) stash — a
later call rebuilds from the digest, restarting the stream
instead of continuing it (the stdlib streams survive: the helped
batch's save/restore nests inside this batch's own).

``fn`` may be a [`call`](`pymizu.call`) specification instead of
a callable — the way to map over a foreign pool (one spawned
with [`r_pool_launcher()`](`pymizu.r_pool_launcher`)). A spec
always stages a shared region: the descriptor crosses in the
interchange format and each runner task carries a region
reference any worker language reads. The element fills the
spec's first positional slot (name kind) or binds as ``x``
(source kind), and the spec's own constant arguments ride with
it — so ``args`` and ``kwargs`` must be empty with a spec.
Constants and elements must be portable values; a non-portable
one raises [`DeclinedError`](`pymizu.DeclinedError`) at stage
time. ``seed=`` carries as a language-neutral pair and each
worker language derives its own streams, so a spec map takes int
seeds only (on R workers, the seed 32-bit-ranged and the offset
at most 2^53); invariance holds within a worker language, never
identical draws across languages.

``template`` declares that every ``fn`` result is ``m`` values
of the exemplar's dtype: results are written in place into a
shared ``n x m`` output area and never serialized. Each result
must be a matching buffer — or, for ``m == 1``, a plain Python
scalar. With a template, ``collect="copy"`` (the default)
returns the area as one gathered numpy array (a memoryview
without numpy) of shape ``(n, m)`` — ``(n,)`` for ``m == 1``;
``collect="view"`` returns it zero-copy, with the map region's
teardown deferred to the view's.

With ``stream=True``, the map never stages the whole of ``x``
into shared memory: it streams slices of ``x`` to workers as
they take work; the return value is unchanged. Fixed x-slices
ride ordinary chunk tasks under a sliding submit/collect window
of at most ``min(n_chunks, 2 * live workers, free result
slots)`` outstanding tasks, so shared-memory residency is
bounded by ``window x slice`` instead of ``sizeof(x)`` — with
the default chunk count (``min(len(x), 32 * live workers)``)
that is roughly ``(2 * workers) / n_chunks`` of the serialized
``x``. ``n_chunks=`` overrides the chunk count outright
(``n_chunks=len(x)`` is the mirai-style extreme of one element
per task). Everything else — result order, ``template`` and
``collect``, ``seed`` invariance, the error taxonomy — is
exactly the non-streaming map's. Fail-fast latency coarsens from
about one adaptive morsel batch to about one chunk (the bound
moves with ``n_chunks``), the adaptive batch sizing of the
morsel machinery is lost (skew mitigation is to raise
``n_chunks``), and slices cross via the serialized tiers, so
per-chunk staging costs an ordinary submit's serialization
rather than the raw section's zero-copy slicing. A streaming map
with Python workers shares one description of the map with every
worker, each read at most once. A [`call`](`pymizu.call`) spec
as ``fn`` streams too, over workers of any language: each chunk
in the window stages a small shared region of its own (the slice
plus the task that runs on it, one runner task per chunk),
removed as soon as the chunk completes. Shared-memory use stays
bounded by the window plus a small per-worker cache rather than
by ``sizeof(x)``. Each chunk also writes its results separately,
so they cannot combine into one view: ``collect="view"`` raises
``TypeError`` on a streaming spec map, as does preparing one
with [`map_prepare()`](`pymizu.Pool.map_prepare`).

``timeout`` bounds submission only (the ring-space wait staging
the tasks); the collection deadline is the handle's
``collect(timeout)``, whose expiry cancels the outstanding work
and returns the `TIMEOUT` sentinel, never raised.

[`starmap()`](`pymizu.Pool.starmap`) is the unpacking variant —
``fn(*element, *args, **kwargs)``, the
`multiprocessing.Pool.starmap` convention.

Examples
--------
>>> import pymizu
>>> with pymizu.Pool.create(2) as pool:
...     results = pool.map(abs, range(-5, 5)).collect()
>>> results
[5, 4, 3, 2, 1, 0, 1, 2, 3, 4]

starmap(self, fn: '_Callable[..., _Any]', x: '_Iterable[_Any]', *, args: '_Iterable[_Any]' = (), kwargs: 'dict[str, _Any] | None' = None, n_chunks: 'int | None' = None, seed: 'int | bytes | bytearray | tuple[int | bytes | bytearray, int] | None' = None, timeout: 'float | None' = None, template: '_Any' = None, collect: 'str | None' = None, stream: 'bool' = False) -> '_Any'

Map ``fn`` over ``x``, unpacking each element as the call's
positional arguments: ``fn(*element, *args, **kwargs)``;
return the run's map handle, as [`map()`](`pymizu.Pool.map`).

The `multiprocessing.Pool.starmap` convention; for the
multi-iterable shape of `concurrent.futures.Executor.map`,
zip first: ``pool.starmap(fn, zip(xs, ys))``. Elements must be
iterables (a 2-D buffer's rows qualify; a 1-D buffer's scalars
do not). A [`call`](`pymizu.call`) spec cannot starmap: element
unpacking is Python-only. Everything else — seeding, templates,
streaming, the error taxonomy — is exactly
[`map()`](`pymizu.Pool.map`)'s.

map_prepare(self, fn: '_Callable[..., _Any]', x: '_Iterable[_Any]', *, args: '_Iterable[_Any]' = (), kwargs: 'dict[str, _Any] | None' = None, n_chunks: 'int | None' = None, seed: 'int | bytes | bytearray | tuple[int | bytes | bytearray, int] | None' = None, template: '_Any' = None, collect: 'str | None' = None, stream: 'bool' = False) -> 'PreparedMap'

Stage a map once for repeated runs; return a
[`PreparedMap`](`pymizu.PreparedMap`).

Takes the same arguments as [`map()`](`pymizu.Pool.map`) (minus
``timeout``, which is per-run and moves to
``PreparedMap.run``). The descriptor
pickle, the region create, and the worker-side attach are paid
once here; each ``pm.run()`` re-arms in O(1), reuses the
workers' cached contexts, and returns the run's map handle (the
collect happens on it). A run collected with ``collect="view"``
hands its region to the view, so the next run restages into a
fresh one. With ``stream=True`` the staged ``x`` stays
submitter-side, so a replacement ``x`` of any shape simply
re-slices — only a length change under ``template`` restages (the
output area is sized for the staged length). Close the handle (or
use it as a context manager) to unlink the region.

retire(self, slot: 'int') -> 'None'

Ask the worker in ``slot`` to exit cleanly (non-blocking,
controller only).

spawn_workers(self, n: 'int' = 1, *, launcher: '_Callable[[str, int], _Any] | None' = None, startup_timeout: 'float' = 30.0) -> 'list[int]'

Spawn ``n`` additional workers into free registry slots and
wait for them to join (controller only).

Parameters
----------
n
    Number of workers to spawn.
launcher
    As for [`Pool.create()`](`pymizu.Pool.create`); the default
    spawns ``sys.executable`` directly.
startup_timeout
    Seconds to wait for the workers to attach.

Returns
-------
    The slot indices spawned into.

stop(self, timeout: 'float' = 5.0) -> 'bool'

Orderly shutdown (controller only).

Broadcast shutdown, cancel pending tasks, wait up to
``timeout`` seconds for clean worker exits, and unlink.
Idempotent.

Returns
-------
    True on clean worker exits within ``timeout``.

shutdown(self, timeout: 'float' = 5.0) -> 'bool'

Alias for [`stop()`](`pymizu.Pool.stop`) — the
`concurrent.futures.Executor` spelling.

destroy(self) -> 'None'

Tear down the pool handle immediately, without the shutdown
broadcast or the wait. Workers exit once they observe the
controller gone. Safe from any thread: a verb in flight on
another thread pins the handle, and teardown completes when the
last one returns — an indefinite wait (submit, collect,
collect_any, collect_all, ready_wait) breaks out within a
bounded slice; stop() instead waits out its own deadline, and
a worker's idle run() can delay the teardown (and the shutdown
observation) until it returns. Pending async consumers fail
with the closed-handle error.

status(self) -> 'dict'

A read-only wire-state snapshot of the pool (dict).

dump(self) -> 'dict'

A read-only debugging snapshot of the whole pool region (dict).

stats(self) -> 'dict'

Cumulative per-worker and per-submitter counters since each
participant joined (dict with ``workers`` and ``submitters``
lists).

trace(self, fn: '_Callable[[str, str], None] | None') -> 'None'

Register a hook called as ``fn(event, id)`` at each task
lifecycle event this process observes: ``"submit"`` on the
submitting thread; ``"start"``, ``"done"``, ``"error"``,
``"drop"``, ``"rehome"`` on worker handles. ``id`` is
``"<submitter slot>:<counter>"``, stable across processes.
Registration is per-handle and per-process; ``None`` removes the
hook. A hook exception is written as unraisable, never propagated
into the pool.

__enter__(self) -> 'Pool'

__exit__(self, *exc: '_Any') -> 'bool'


## Task Methods

Methods for the Task class


state

The task's state ("pending" / "ok" / "err" / "cancel" / "died" / "collected" / "dropped"); informational, racy against slot reuse.

pool

The pool handle this task was submitted on.

collect

collect(timeout=None) -> value | sentinel

Wait up to `timeout` seconds (None indefinitely, 0 polls) for the task's
terminal state and return its result. A task error re-raises as
pymizu.TaskError (carrying remote_type / remote_traceback), a cancellation
as pymizu.CancelledError, a dead worker as pymizu.WorkerDiedError, a lost
payload region as pymizu.TaskLostError.
pymizu.TIMEOUT on expiry; a task is collected exactly once.

result

result(timeout=None) -> value | sentinel

Alias for collect() — the concurrent.futures.Future spelling.

cancel(self, /)

cancel() -> bool

Advisory and discard-only, never preemptive: a still-queued task is
skipped; an executing one runs to completion and its result is dropped.
True when this call cancelled the task; every other edge folds to False.

collect_async(self, /)

collect_async() -> asyncio.Future

The event-driven collect: an asyncio.Future resolving with the task's
result (or raising the flattened remote error collect() raises), driven
by the submitter's notify endpoint — never blocks, never polls. Cancelling
the Future detaches the waiter (a later blocking collect() works again);
it never cancels the task — cancel() stays the explicit verb.

__await__(self, /)

Return an iterator to be used in await expression.


## Cross-language interop

R peers and workers (the mizu R package), neutral task specifications, and the data.frame's Python home.


call(name: 'str | None' = None, /, *args: '_Any', source: 'str | None' = None, **kwargs: '_Any')

A task specification for a pool of another language's workers.

Describes a call for [`Pool.submit()`](`pymizu.Pool.submit`): a
qualified name (``"mod.fn"`` for Python workers, ``"pkg::fn"`` for
R workers) or a ``source=`` string in the workers' language, plus
the constant arguments. The spec describes a call; it is not a
value. The language never appears at the call site — `Pool.submit`
resolves the workers' language from the pool itself — and a bare
(unqualified) name errors at submit, not here.

Parameters
----------
name
    A qualified name: ``"mod.fn"`` for Python workers,
    ``"pkg::fn"`` for R workers.
args
    Constant positional arguments of the call.
source
    A source string in the workers' language. Exactly one of
    ``name`` or ``source`` must be given.
kwargs
    Constant keyword arguments of the call.

Details
-------
Unnamed arguments map to the positional list and keyword arguments
to the named dict, matching R's mixed-call convention. Arguments
must be portable values (the interchange subset documented under
[`Channel.send()`](`pymizu.Channel.send`)): a non-portable argument
raises [`DeclinedError`](`pymizu.DeclinedError`) at submit, never a
fallback.

A ``source=`` task evaluates in a fresh namespace with the keyword
arguments bound as names and positional arguments bound as ``_1``,
``_2``, ... The result is the trailing expression's value, or None
when the source ends with a statement.

Large arguments cross to foreign workers by reference rather than by
copy: one fresh buffer argument past the zero-copy floor stages a
single layout write into a shared region (the worker reads a view
over it), and an argument that is already a shared view crosses as
its identifier alone — zero payload bytes. On a pool whose workers
predate the ref reader, such a task declines locally at submit
naming the remedy.

Examples
--------
A spec task also runs on a normal (Python-worker) pool:

>>> import pymizu
>>> with pymizu.Pool.create(2) as pool:
...     task = pool.submit(pymizu.call("math.sqrt", 2.0))
...     results = pool.collect_all([task], timeout=10)
>>> results
[1.4142135623730951]

r_launcher(*, rscript: 'str | None' = None, stdout: '_Any' = None, stderr: '_Any' = None) -> '_Callable[[str], _subprocess.Popen]'

Return a [`Channel.create()`](`pymizu.Channel.create`) launcher
spawning an R peer.

The peer runs the R package `mizu`: the returned `callable(token)`
spawns `rscript` on the package's static child runner with
`mizu:::peer_main(<token>)` as the entry expression and the probed
library paths propagated in argv.

Parameters
----------
rscript
    Path to Rscript; the default searches the PATH.
stdout
    Forwarded to `subprocess.Popen`; the default inherits the
    console, where the peer's error epilogue lands.
stderr
    Forwarded to `subprocess.Popen`.

Returns
-------
    A `callable(token)` launcher for
    [`Channel.create()`](`pymizu.Channel.create`).

Raises
------
MizuError
    Raised here, before the channel is created, when Rscript is
    not found or no installed `mizu` with source string support is
    available.

r_pool_launcher(*, rscript: 'str | None' = None, stdout: '_Any' = None, stderr: '_Any' = None) -> '_Callable[[str, int], _subprocess.Popen]'

Return a [`Pool.create()`](`pymizu.Pool.create`) launcher
spawning R workers.

Each worker runs the R package `mizu`: the returned
`callable(token, slot)` spawns `rscript` on the package's static
child runner with `mizu:::worker_main(<token>, <slot>)` as the
entry expression and the probed library paths propagated in argv
(without them the workers cannot `library(mizu)` from the host's
libraries). The mirror of the R package's `mizu_py_pool_launcher()`.

Parameters
----------
rscript
    Path to Rscript; the default searches the PATH.
stdout
    Forwarded to `subprocess.Popen`.
stderr
    Forwarded to `subprocess.Popen`.

Returns
-------
    A `callable(token, slot)` launcher for
    [`Pool.create()`](`pymizu.Pool.create`).

Raises
------
MizuError
    Raised here, before any pool exists, when Rscript is not found
    or no installed `mizu` with source string support is available.

Notes
-----
The first worker's join records the workers' language in the pool,
so the launcher carries no language attribute: a pool of R workers
takes [`pymizu.call()`](`pymizu.call`) specifications through
[`Pool.submit()`](`pymizu.Pool.submit`), and a plain callable
errors locally naming the spec verb. A launcher that spawns the
wrong language fails at join, not at the first task.

Examples
--------
```python
import pymizu

with pymizu.Pool.create(4, launcher=pymizu.r_pool_launcher()) as pool:
    spec = pymizu.call(source="summary(cars$speed)")
    print(pool.submit(spec).collect())
```

Frame()

A data.frame's Python home: named columns with a row count.
to_dict() gives the column dict without an Arrow library;
__arrow_c_stream__ exports to polars / pyarrow / pandas.
from_arrow() constructs one from an Arrow producer (the
opt-in zero-copy path for Python-to-Python channels).
R frame metadata rides the _r_attrs stash: it attaches at
read and validates at send, but it rides the pristine
received Frame only — the Frame immutability invariant:
any future mutator must clear it first (a rebuild arrives
in R as a plain data.frame, never a mislabeled one).

DimArray(shape, dtype=None, buffer=None, offset=0, strides=None, order=None)

An `ndarray` carrying R labels (`names` / `dimnames` /
`axis_names` / `r_class`) for faithful R ↔ Python round trips.

A received labeled vector or array homes as a DimArray; an
unmodified echo restores every label in R (`identical()`).
Attach labels explicitly to send a labeled value: `names` for a
1-D array, `dimnames` (plus optional `axis_names`, R's
`names(dimnames(x))`) for a matrix or n-D array, and
`r_class = "table"` for R's table class. The two label slots are
mutually exclusive at send; `axis_names` requires `dimnames` and
has one entry per axis.

Labels never propagate implicitly — any numpy operation that
builds a new array (a transpose, arithmetic, a slice, a copy)
drops them, so a reshaped or permuted array never inherits labels
its data no longer matches.


## Frame Methods

Methods for the Frame class


names

The column names (a tuple of str).

row_names

The row names metadata: None for automatic, else a list of str or
an int32 array. Not carried by to_dict() or the Arrow export.

to_dict(self, /)

to_dict() -> dict

The frame as a dict of columns: numpy arrays for numeric columns (or
memoryviews without numpy), datetime64 for Date / POSIXct, and
list[str | None] for string and factor columns. Complex columns carry
complex128 here (the Arrow export has no complex type).

from_arrow

from_arrow(obj) -> Frame

Construct a frame from an Arrow producer, adopting its buffers
wherever a column's frame layout is the Arrow layout (a single-batch
producer on offset-0 boundaries), so construction copies nothing it
does not have to — nulls included.

This is the opt-in zero-copy frame path for Python-to-Python
channels: past the zero-copy floor the frame crosses as one
shared-memory region and arrives region-backed, where a plain Arrow
object pickles (container-exact, but a full copy). The frame reads
back value-exact but type-normalized: a chunked or sliced producer's
columns concatenate, uints widen, nanosecond timestamps rescale to
microseconds, and the received value is a Frame, not the producer's
type.

Parameters
----------
obj
    Any ``__arrow_c_stream__`` producer: a pyarrow ``Table`` or
    ``RecordBatchReader``, a polars ``DataFrame``, a duckdb relation.
    ``Table.combine_chunks()`` is the route to the adopt path for a
    chunked table.

Returns
-------
    The frame. An adopted column keeps the producer's buffers alive
    for the frame's lifetime (numpy-view semantics: a frame from a
    4 GB table pins it).

Raises
------
MizuError
    The Arrow type has no portable home (the column is named), or
    the producer's export failed.

Examples
--------
>>> import pyarrow as pa
>>> import pymizu
>>> with pymizu.Channel.create(
...     """
... import pymizu
... while True:
...     x = ch.recv()
...     if x is pymizu.CLOSED:
...         break
...     ch.send(x)
... """
... ) as ch:
...     f = pymizu.Frame.from_arrow(
...         pa.table({"x": pa.array([1.5, None] * 100000)})
...     )
...     ch.send(f)
...     echoed = pa.table(ch.recv(timeout=5))
>>> echoed.column("x").null_count
100000

__arrow_c_stream__

__arrow_c_stream__(requested_schema=None) -> capsule

The Arrow C Stream Interface export: the frame as one struct batch.
Consumers take it in one line: pl.from_arrow(f), pa.table(f), or
pd.DataFrame.from_arrow(f). A complex column raises (no Arrow type).

__len__(self, /)

Return len(self).

__reduce__(self, /)


## Outcomes and errors

Terminal transport states return as the sentinel singletons FULL / TIMEOUT / CLOSED / PEER_GONE (identity-tested, never raised); real failures raise from the exception hierarchy.


Sentinel()

Terminal-state sentinel returned by pymizu verbs. Identity-tested against pymizu.FULL / TIMEOUT / CLOSED / PEER_GONE.

is_sentinel(object, /)

is_sentinel(x) -> bool

Provenance, not shape: True only for the exact sentinel singletons this
process's verbs return, so a look-alike payload never passes.

is_remote_error(x: '_Any') -> 'bool'

Test whether a received channel value is a remote error.

An uncaught error in a channel peer crosses as a value, not a raised
exception (transport states are values, payloads are values — user
code decides to raise). The value is a
[`TaskError`](`pymizu.TaskError`) carrying the original exception's
class name as ``remote_type`` and its traceback text as
``remote_traceback``; raise it to propagate.

Returns
-------
    True for a received remote error, False otherwise.

MizuError

Base class for all pymizu errors.

ShmError

A shared-memory region operation failed. Also an OSError: the failure is at the OS layer.

StartupError

A channel peer or pool worker failed to attach within the startup timeout.

SubmitTimeoutError

Pool.submit() timed out waiting for injection-ring space. Also a TimeoutError.

SlotsExhaustedError

Pool.submit() found no free result slot: too many outstanding (uncollected) tasks.

StoppedError

The pool is stopped; no further submission is possible.

CancelledError

The task was cancelled before it ran.

TaskError

The task callable raised. Carries 'remote_type' and 'remote_traceback' attributes describing the worker-side exception.

WorkerDiedError

The executing worker died mid-task. Carries 'slot' and 'pid' attributes identifying the worker.

DeclinedError

A send on a foreign-language channel of a value outside the portable interchange subset. Carries 'path' and 'reason' attributes.


## Housekeeping

The process-wide default pool, the evaluating worker's own pool handle, and orphan reaping.


default_pool() -> 'Pool | None'

The process-wide default pool, or None when none is set.

Set with [`set_default_pool()`](`pymizu.set_default_pool`); scoped
use with [`using_pool()`](`pymizu.using_pool`). Package code taking
an optional pool resolves it in this order: an explicit ``pool``
argument, then [`current_pool()`](`pymizu.current_pool`) inside a
task (the evaluating worker's own pool, for nested submission), then
``default_pool()``, then the caller's own fallback — sequential
execution or an error.

Setting a default checks the type only: a stopped pool is accepted
(liveness is transient; a probe would prove nothing about use time)
and fails at use time with the usual stopped-pool errors. Handles
from [`Pool.attach()`](`pymizu.Pool.attach`) are valid defaults;
ownership and teardown stay with the pool's creator.

The default is process-global — every thread sees the same pool, and
[`current_pool()`](`pymizu.current_pool`) remains the thread-local
mechanism. After a ``fork()``, a child process reads it as unset:
handles are process-private.

set_default_pool(pool: 'Pool | None') -> 'Pool | None'

Set the process-wide default pool; None clears it.

Returns
-------
    The previous default (a Pool or None), so callers can save and
    restore.

Details
-------
The registry anchors the handle: a pool set as the default stays
alive even after its variable is deleted, until the default is
cleared or replaced. See [`default_pool()`](`pymizu.default_pool`)
for the full semantics.

using_pool(pool: 'Pool | None') -> '_Generator[Pool | None, None, None]'

Use ``pool`` as the default for the with block, then restore.

Parameters
----------
pool
    The pool to make the default; ``None`` scopes a cleared
    default.

Returns
-------
    A context manager yielding ``pool``. The previous default
    returns on exit, including on exception.

current_pool() -> 'Pool | None'

The evaluating worker's own pool handle, inside a task.

Returns
-------
    The worker's own pool handle inside a task; None outside one.

Details
-------
A task uses it for nested submission: a nested submit pushes onto the
worker's own work-stealing deque (no ring, no wait), and a nested
collect helps — executes work — instead of parking, so nested fan-outs
run at fork/join cost and never deadlock the pool. The user-set
process-wide default is
[`default_pool()`](`pymizu.default_pool`), which never overrides
this runtime-owned binding.

current_rng() -> '_np.random.Generator | None'

The running element's own numpy Generator, inside a seeded map.

Returns
-------
    The element's memoized ``numpy.random.Generator`` inside a
    seeded map element; None outside one.

Raises
------
TypeError
    When numpy is not installed.

Details
-------
[`Pool.map()`](`pymizu.Pool.map`) ``seed=`` seeds the stdlib
`random` module per element; a task drawing from numpy calls this
instead: the element's memoized ``numpy.random.Generator``, derived
from the same seed material on first call in the element
(domain-separated from the stdlib derivation, so those streams are
unchanged). Two calls in one element continue one stream; distinct
elements get distinct streams — results identical for any chunking,
worker count, or steal order.

Outside a seeded map element (an unseeded map, an ordinary task, the
submitter process) the return is None. A seeded map element must
not nested-submit and collect:
worker helping can run another map's batches mid-element, wiping
this element's stash — a later call rebuilds from the digest,
restarting the stream instead of continuing it.

prune() -> 'list[str]'

Remove orphaned shared memory regions.

Removes the pymizu shared memory regions that dead processes leave
behind. Needed only in exceptional circumstances. Cleanup after a
crash is automatic as long as any participant of the pool or channel
survives it. The survivor detects the death through the liveness lock
and reclaims the regions of the dead process itself. Orphans arise
only when every attached process dies at once (for example, the whole
process group is killed), and no survivor remains to run that cleanup.
They persist until pruned or until the machine reboots.

Regions of running processes are never touched. If another pymizu
session on the machine is itself recovering from a crash, do not run
this. Payloads sent by a now-dead peer stay deliverable to its
survivor until drained, and this function reaps them.

A crashed process cannot clean up after itself, and a new process
that happens to reuse its PID cannot reap its orphans either (it
reads its own PID as alive). Run ``prune()`` while the PID is free,
before reuse.

Returns
-------
    The region names removed, or an empty list if none were. On
    platforms whose shared memory namespace cannot be enumerated
    (Windows, where orphans cannot exist), always an empty list.


----------------------------------------------------------------------
This is the User Guide documentation for the package.
----------------------------------------------------------------------


## Guide

## Requirements

- Python 3.10 or later, on a 64-bit platform.
- Linux: kernel 5.3 or later.
- Windows: the build requires clang-cl.

## Install

Install the development version from GitHub:

```sh
pip install git+https://github.com/shikokuchuo/pymizu
```

The extension compiles the vendored C core, so no system library is necessary.

Optional extras:

- `pymizu[numpy]`: zero-copy array views and raw-tier array staging.
- `pymizu[cloudpickle]`: lambdas, closures, and local functions as pool tasks.

To request an extra with the GitHub install:

```sh
pip install "pymizu[numpy] @ git+https://github.com/shikokuchuo/pymizu"
```

## Deploying on Linux

On Linux, POSIX shared memory is `/dev/shm`, a bounded tmpfs mount.
Everything pymizu creates must fit in it: pool and channel regions, plus the per-payload regions that carry anything past the in-slot budget.
Container runtimes set it to 64 MB by default.
Use `docker run --shm-size=4g`, or the equivalent in your orchestrator.
Size it to your peak in-flight payload traffic.

## Linux memory allocator

This section applies only to Linux with glibc.

When pymizu starts a channel peer or a pool worker, that process changes two settings of the C memory allocator.
It raises the mmap threshold to 32 MB and the trim threshold to 128 MB.
This keeps large payloads in fast memory.
Without this change, glibc asks the kernel to map and unmap each large payload, and that work is slow.

Your own Python process does not change.
If you want the same settings there, set them before Python starts:

```sh
export GLIBC_TUNABLES=glibc.malloc.mmap_threshold=33554432:glibc.malloc.trim_threshold=134217728
```

If you have set `GLIBC_TUNABLES`, pymizu respects your values.


### Benchmarks

Communication overhead against the stdlib `concurrent.futures` pools (Apple M4 Pro, from `benchmarks/mizu-stdlib-bench.py`).

Against `ProcessPoolExecutor` (tasks run in separate processes, with pickled payloads):

| Benchmark | pymizu | ProcessPoolExecutor | Speedup |
|----|----|----|----|
| Trivial task round trip | 0.7 µs | 94.2 µs | 135x |
| Pipelined throughput, 1 worker | 2,390,000 tasks/s | 18,500 tasks/s | 129x |
| Parallel map overhead, trivial function, 4 workers | 0.4 µs/elt* | 70.3 µs/elt | 176x |
| Parallel map of 2,000 ~5 µs tasks, 4 workers | 4.3 ms | 129 ms | 30x |

Against `ThreadPoolExecutor` (tasks share one process, so the GIL caps CPU-bound work at a single core):

| Benchmark | pymizu | ThreadPoolExecutor | Speedup |
|----|----|----|----|
| Trivial task round trip | 0.7 µs | 9.5 µs | 14x |
| Pipelined throughput, 1 worker | 2,390,000 tasks/s | 405,000 tasks/s | 5.9x |
| Parallel map overhead, trivial function, 4 workers | 0.4 µs/elt* | 2.6 µs/elt | 6.5x |
| Parallel map of 2,000 ~5 µs tasks, 4 workers | 4.3 ms | 50 ms | 12x |

Against `InterpreterPoolExecutor` (Python 3.14+, subinterpreters with per-interpreter GILs), the trivial-task rows land between the thread and process pools: 21.3 µs per round trip and 59,600 tasks/s pipelined on this machine (measured 2026-10-07).
It cannot run the map rows at all.
Task payloads must be shareable: builtin scalars, plain containers, or buffer-protocol objects.
numpy arrays and numpy scalars raise `NotShareableError`.

\* elt = element, in microseconds of wall time per map element.

## How pymizu compares

pymizu occupies a specific point in the design space: parallelism across processes on a single machine, with data moving through shared memory instead of serialized copies.
The tools in this table overlap at the edges.
They differ in what has to run, how data moves, and what survives a crash.

| Tool | What runs | How data moves | Best for |
|----|----|----|----|
| `multiprocessing`<br>`concurrent.futures` | Nothing beyond the stdlib | Pickle copies per task | Coarse tasks, maximum portability |
| `joblib` | A reusable worker pool (loky) | Pickle; large arrays memmapped to disk | Coarse parallel loops, anywhere Python runs |
| `ipyparallel` | A controller process plus engines | Pickle over ZeroMQ sockets | Interactive parallelism from Jupyter |
| Ray | Scheduler, object store, dashboard | Object store (zero-copy on one node) | Cluster scaling, actors, GPU placement |
| Dask | A scheduler (+ `distributed` workers) | Process copies; threads share a heap | Out-of-core dataframes and arrays |
| Free-threaded CPython | Threads | Nothing; threads share one heap | CPU work needing no isolation |
| `concurrent.interpreters` | Nothing beyond the stdlib | Builtins copy, buffers share; no numpy | Process-free isolation (3.14+) |
| pymizu | A library | Shared-memory views, µs handoff | Fine-grained tasks, large arrays, R interop |

### Process pools: `multiprocessing`, `concurrent.futures`, joblib, ipyparallel

The stdlib pools are the baseline that the benchmark tables measure.

joblib's `Parallel`/`delayed` is the long-standing answer for an embarrassingly parallel loop over numpy data.
Its `loky` backend is the closest in spirit to pymizu: a reusable process pool that memmaps arrays past a size threshold to a temp folder instead of pickling them.
That removes the pickle copy.
The array still round-trips through the filesystem, and a task round trip stays in the millisecond range.

ipyparallel answers the same question for Jupyter.
A controller process brokers pickled messages over ZeroMQ sockets to engine processes.
A controller in the loop and socket copies are the price of its generality.

`multiprocessing.shared_memory` exposes raw shared buffers and leaves synchronization, lifetime tracking, and crash cleanup to you.
pymizu is that idea finished: the same pages, with the waiting, framing, and peer-death detection built in.

### Distributed frameworks: Ray and Dask

Ray and Dask are designed to spread work across machines.
On one machine you pay for that design.
A local Ray instance still starts a scheduler, an object store, and a dashboard.
A task round trip costs on the order of a millisecond.

Dask's `distributed` scheduler is similar.
Its process-free threaded scheduler shares memory, but the GIL caps it for Python-level work.
Ray's object store does read numpy arrays without copying on the same node.
The difference is overhead and footprint.
If the work fits one machine and the tasks are measured in microseconds, the framework overhead dominates the work.

### Modern CPython: free-threading and subinterpreters

Python 3.14 officially supports a build with the GIL removed, in which threads share one heap and exchange data with no copies at all.
This is the right answer for single-language CPU parallelism when isolation is not needed.
A crashing thread takes the process down, C extensions must be built and validated for it, and it does nothing for crossing into R.
pymizu targets the standard GIL build and does not yet support the free-threaded one.
The two approaches solve different problems: threads trade isolation for sharing, and pymizu keeps the isolation and makes the sharing cheap.

Python 3.14 also added `concurrent.interpreters` to the stdlib ([PEP 734](https://peps.python.org/pep-0734/)).
It runs multiple interpreters in one process, each with its own GIL.
CPU work parallelizes without new processes on the standard build.
Objects that pass between interpreters must be shareable.
Builtin scalars and plain containers of them cross as copies.
Buffer-protocol objects can share their underlying memory, with synchronization left to you.
Anything else, numpy arrays included, is rejected outright.
The isolation is real but not OS-level: a fatal error in one interpreter ends the process.
The surrounding machinery (queues, lifetimes, crash handling) is yours to build.
`concurrent.interpreters` is also Python 3.14 and later only.
pymizu runs on Python 3.10 and later.

### The one thing only pymizu does

A pool whose workers are R processes, exchanging numpy arrays and data frames as shared-memory views ([R interop](interop.qmd)).
None of the other tools in this guide cross the language boundary.

The short version: choose pymizu when the work fits one machine, when tasks are fine-grained or arrays large, or when R needs to be in the loop.
Choose Ray or Dask when you outgrow one machine.
Choose `concurrent.interpreters` when you want stdlib-only isolation and can require Python 3.14.
Choose free-threading when one language and shared-everything is acceptable.

`benchmarks/mizu-bench.py` runs the pymizu rows standalone, `benchmarks/mizu-stdlib-bench.py` the comparison, and `benchmarks/crosslang-map-bench.py` the cross-language map comparison.
The scripts are report-only and assert nothing about the numbers: CI runner timing is too variable.
Dated records live in `benchmarks/notes.md`.


### Channels

A channel is a two-way message link between a Python process and a helper process that it spawns: one lock-free ring per direction, over shared memory.
One process writes data, and the other reads it in place, never copied through a socket, pipe, or file.

## Creating a channel

`Channel.create()` spawns the peer process and connects both ends.
The peer program is a Python source string, evaluated in the peer with `ch` bound to its side of the channel:

```{python}
import pymizu

ch = pymizu.Channel.create("""
import pymizu
while True:
    x = ch.recv()
    if x is pymizu.CLOSED:
        break
    ch.send(x)
""")
```

The peer is a full Python process: it imports anything importable in the spawning interpreter's environment.
`Channel.create()` returns after the peer connects.
It waits up to `startup_timeout` seconds and raises `StartupError` on expiry.

A `launcher` changes how pymizu spawns the peer.
It is one callable that takes the channel's token and arranges for the peer process to start.
`pymizu.r_launcher()` is the built-in example: it spawns an R peer instead (see [R interop](interop.qmd)).
User code spawns peers through a launcher.
`Channel.attach(token)` is the low-level entry for a peer that attaches itself.

## Sending and receiving

`ch.send()` publishes a value, visible to the peer the moment the call returns.
Sends never block for ring space.
`ch.recv()` returns the next value, waiting up to `timeout` seconds (indefinitely when omitted):

```{python}
ch.send([1, "a", None])
ch.recv(timeout=5)
```

The outcomes that end a conversation come back as sentinel singletons, never raised:

| Sentinel | Returned when |
|---|---|
| `pymizu.FULL` | a send finds the ring full |
| `pymizu.TIMEOUT` | a receive's timeout expires |
| `pymizu.CLOSED` | the peer has closed and the ring is drained |
| `pymizu.PEER_GONE` | the peer process has died and the ring is drained |

Test them by identity (`x is pymizu.TIMEOUT`) or with `pymizu.is_sentinel(x)`.
No call needs an error handler.
If the peer dies, receives first drain what it already sent, then report `PEER_GONE`.

`send_batch()` and `recv_batch()` move several values in one call.
Use them at rates where the per-call overhead starts to matter:

```{python}
ch.send_batch(range(4))
ch.recv_batch(4, timeout=5)
```

`ch.alive()` reports whether the peer still runs, without touching the rings.
`ch.info()` returns a read-only snapshot of the channel's state.

## asyncio

`ch.recv_async()` is the event-driven receive: an `asyncio.Future` for the next value, resolved off the channel's notify endpoint — no blocking, no polling, and no timeout argument (`asyncio.timeout` / `asyncio.wait_for` are the idiom).
Terminal states stay values: the Future resolves with `CLOSED` / `PEER_GONE`.
Several outstanding calls resolve in FIFO order, one message each.
Cancelling the Future detaches the waiter, passing the next message to the one after.

```{python}
ch.send(1)
assert await ch.recv_async() == 1
ch.send(2)
await ch.recv_async()
```

`async for x in ch` is the stream form: it awaits each value through `recv_async()` and ends on `CLOSED` or `PEER_GONE`, exactly as the sync iterator ends.

## Closing

`ch.close()` is the orderly shutdown: it signals the peer and waits for it to finish draining before releasing the shared resources:

```{python}
ch.close()
```

A channel is also a context manager: the `with` block closes it on exit.
`close_signal()` signals without waiting.
`destroy()` tears the handle down immediately, without the close handshake.
The peer observes the channel as closed (`PEER_GONE` is the hard-death outcome).
A destroy is safe from any thread.
A verb in flight on another thread pins the handle, and teardown completes when the last one returns.
An indefinite wait (`recv()`, `recv_batch()`, `ready_wait()`, `close()`) breaks out within a bounded slice, so the teardown and the peer's observation of it follow promptly.

## What crosses a channel

pymizu moves each value the cheapest way it can:

1. `None` crosses as a marker: no data moves.
2. Buffer-protocol objects (`bytes`, 1-D C-contiguous numpy arrays) cross without serialization.
   float64, int32, int64, complex128, and uint8 keep their exact values and arrive as arrays (`bytes` arrives as a uint8 array).
   numpy scalars and 0-d arrays keep their exact type and shape.
   A large buffer arrives as a read-only zero-copy view over the shared pages: no copy, no parse.
   A contiguous buffer below 256 KB instead crosses as one copy into the channel's spill arena — cheaper than region machinery at those sizes; from 256 KB the receive is a view.
3. A `str` within the inline budget crosses as raw UTF-8.
4. Booleans, numbers, and flat lists or dicts of them cross in a compact binary form.
5. Everything else crosses as a pickle (protocol 4).

Each end of a channel knows the peer's language.
Between two Python processes that is the whole story: arrays of the dtypes in this list cross unchanged, and everything else pickles.
With an R peer, values must fit a portable subset that both languages share.
A value outside it raises `pymizu.DeclinedError` at send time, naming the value and the reason.
The channel is unharmed.
See [R interop](interop.qmd).

An uncaught error in the peer crosses as a value, not a raised exception: a `pymizu.TaskError` carrying the original exception's class name and traceback text.
`pymizu.is_remote_error(x)` tests a received value.
Raise it to propagate the error.

## Frames without pickling

Between two Python processes, a pyarrow `Table`, polars `DataFrame`, or duckdb relation pickles by default: container-exact, but a full serialize/parse round trip.
`pymizu.Frame.from_arrow()` opts an Arrow producer into the frame path instead: one shared-memory region on the wire, read in place at receive:

```{python}
import pyarrow as pa

with pymizu.Channel.create("""
import pymizu
while True:
    x = ch.recv()
    if x is pymizu.CLOSED:
        break
    ch.send(x)
""") as ch2:
    f = pymizu.Frame.from_arrow(
        pa.table({"x": pa.array([1.5, None] * 100_000)})
    )
    ch2.send(f)
    echoed = pa.table(ch2.recv(timeout=5))
echoed.column("x").null_count
```

Construction adopts the producer's own buffers wherever the frame's column layout *is* the Arrow layout: a single-batch producer on offset-0 boundaries, nulls included (the validity bitmap adopts with the values).
It copies nothing it does not have to:

| Stage | Copies |
|---|---|
| producer → `Frame.from_arrow()` | 0 per adopted column; a per-column copy only where its type forces it |
| `ch.send(f)` | 1 — the layout write into the shared region (the region is the wire) |
| `ch.recv()` | 0 — the frame reads the region in place |
| `pa.table(f)` | 0 — the frame's `__arrow_c_stream__` exports borrowed buffers |

Values cross exactly, but types normalize.
A column also copies, never declines, where its type forces it:

| In | Out (the received frame's Arrow export) |
|---|---|
| `string` | `large_string` |
| `string_view` | `large_string` (copies) |
| `bool` | `bool` (copies: bit-packed) |
| int8 / int16 / uint16 | int32 (copies) |
| uint32 / half_float / float32 | float64 (copies) |
| uint64 | uint64 (adopts) |
| timestamp[s / ms / us / ns] | timestamp[s / ms / us / ns] (adopts) |
| duration[s / ms / us / ns] | duration[s / ms / us / ns] (adopts) |
| dictionary\<i32, utf8\> | dictionary\<i32, string\> (adopts; other index widths copy) |
| chunked or sliced columns | concatenated or rebased (copies) |

Chunked pyarrow tables adopt after `table.combine_chunks()`.
Timestamps and durations keep their counts and unit verbatim: a `timestamp[ns]` round trip is exact to the nanosecond, tz included.
Nanosecond fidelity needs pymizu from this version on both ends.
A peer without the capability (an older pymizu, or R) receives the microsecond form instead: values rounded to the double-second wire's precision, never an error.

Adopted buffers stay alive for the frame's lifetime: a frame from a 4 GB table pins it, like a numpy view.
`to_dict()`'s masked reads copy where the Arrow export does not.
When you want the exact producer type back, keep the default: a plain pyarrow or polars object pickles, container-exact.
On a pool, a `Frame` task argument and a frame returned from a task both ride the shared-memory region as on a channel (see [the pools guide](pools.qmd#frames-as-task-arguments)).
`Pool.map` does not take a frame as its element source.

## Sizing

`Channel.create()` takes `capacity` (ring slots, a power of two), `slot_size` (bytes per slot, so the inline payload budget is `slot_size - 16`), and `arena_size` (the channel's spill arena).
A value past the inline budget spills to the arena or a fresh shared-memory region.
`spin=True` selects pure-spin waiting: receivers never park to the OS.
This gives the lowest latency, at full CPU use while waiting.


### Task pools

A pool is a set of worker processes that divide submitted tasks among themselves.
`Pool.create()` spawns them.
Workers claim tasks from per-submitter injection rings and steal work from each other, so the load balances itself.
A submission is one shared-memory write plus at most one directed wake: no dispatcher process is in the loop.

```{python}
import pymizu

with pymizu.Pool.create(4) as pool:
    task = pool.submit(pow, 2, 16)
    result = task.collect(timeout=5)

result
```

`pool.submit()` returns a `Task` handle immediately.
`task.collect()` waits for the result.
The pool is a context manager: leaving the block stops it.
`pool.stop()` cancels the pending tasks, waits for the workers to exit cleanly, and releases the shared memory.
The pool's lifetime is bound to the creating process: dropping the handle shuts the pool down without the wait.

## Task callables

A task callable travels by pickle.
With the standard pickle it must be an importable reference, as with `multiprocessing`.
Installing cloudpickle lifts that restriction.
Task arguments follow the same rules as [channel values](channels.qmd#what-crosses-a-channel).
A buffer-protocol argument past the zero-copy floor (32 KB) arrives as a read-only view over the shared pages, not a writable copy.

## Frames as task arguments

A [`Frame`](`pymizu.Frame`) argument follows the same rule.
Past the zero-copy floor it crosses as one shared-memory region, and the worker's callable receives a region-backed frame, exactly as on a channel.
At sizes under the floor the argument pickles (the worker gets a copy-backed frame), as it does with a complex column (which has no Arrow type).
One region crosses per task: a second region-sized argument, another frame or a large buffer, sends the whole task down the pickle path, values exact.
Raw pyarrow and polars frames keep the pickle path (they round-trip container-exact).
Construct the `Frame` first to opt in.

## Outcomes

| Outcome | On collect |
|---|---|
| The task raised | `pymizu.TaskError`, with `remote_type` and `remote_traceback` |
| The task was cancelled before it ran | `pymizu.CancelledError` |
| The executing worker died | `pymizu.WorkerDiedError`, with `slot` and `pid` |
| The task's payload region vanished without a death verdict | `pymizu.TaskLostError` |
| The collect timeout expired | the `pymizu.TIMEOUT` sentinel |

A task error crosses as a plain description of type, message, and traceback, never a pickled exception object.
A worker's death is detected at OS notification latency, with no heartbeats or polling.
A dead worker fails exactly the tasks it had claimed.
The surviving workers take over the rest of its queue.

`task.cancel()` withdraws a task.
A task still queued is skipped.
A task already running completes, and its result is discarded.
Cancellation never interrupts executing code.

Submission itself can fail: `SubmitTimeoutError` when the submit queue stays full past `timeout`, `SlotsExhaustedError` when too many tasks are outstanding and uncollected, `StoppedError` on a stopped pool.

## Waiting on several tasks

`pool.collect_any()` waits on a list of task handles and returns `(index, value)` for the first to reach a terminal state.
The other handles stay collectible.
`pool.collect_all()` returns every value in input order once all are terminal.
Both take one overall `timeout` and return the `TIMEOUT` sentinel on expiry, consuming nothing:

```{python}
import time

with pymizu.Pool.create(2) as pool:
    tasks = [pool.submit(time.sleep, 0.5), pool.submit(len, "pymizu")]
    first = pool.collect_any(tasks, timeout=5)

first
```

```{python}
with pymizu.Pool.create(4) as pool:
    tasks = [pool.submit(pow, 2, i) for i in range(4)]
    rest = pool.collect_all(tasks, timeout=5)

rest
```

`pool.submit_batch()` submits one task per zero-argument callable in a single call.
Bind arguments with `functools.partial`.

## asyncio

`await task` — or `task.collect_async()` — is the event-driven collect: an `asyncio.Future` resolving with the task's result (or raising the flattened remote error), driven by the worker's notify signal rather than any blocking or polling.
Cancelling the Future detaches the waiter — a later blocking `collect()` works again — and never cancels the task (`task.cancel()` stays the explicit verb).
An enrolled task refuses blocking collects: one consumer per result.

`pool.collect_any_async(tasks)` and `pool.collect_all_async(tasks)` are the volume form: exactly one Future per collection, never one per task, resolved by a scan over the enrolled tasks on each wake.
Semantics mirror the sync verbs: collect-all resolves in input order and the first error by position rejects with `index`; collect-any resolves `(index, value)` for the first terminal task.
There are no timeout arguments: `asyncio.timeout` / `asyncio.wait_for` are the idiom.

```{python}
with pymizu.Pool.create(2) as pool:
    tasks = [pool.submit(pow, 2, i) for i in range(4)]
    rest = await pool.collect_all_async(tasks)

rest
```

`await pool.map(fn, x)` resolves a map handle the same way (see [Parallel map](map.qmd)).

## Nested tasks

Inside a task, `pymizu.current_pool()` returns the evaluating worker's own handle.
A nested submit pushes straight onto the worker's own work-stealing deque, with no ring and no wake.
A worker waiting on a nested result executes other work instead of sleeping.
Divide-and-conquer runs at fork/join cost and never deadlocks the pool:

```{python}
def fan_out(parts):
    pool = pymizu.current_pool()
    subs = [pool.submit(sum, part) for part in parts]
    return sum(t.collect() for t in subs)

parts = [range(0, 250), range(250, 500), range(500, 750), range(750, 1000)]

with pymizu.Pool.create(4) as pool:
    total = pool.submit(fan_out, parts).collect()

total
```

(This example needs cloudpickle: `fan_out` is not an importable reference.)

`Pool.map()` nests the same way: a task can map over its own pool.
See [Nested maps](map.qmd#nested-maps).

## A default pool for package code

Package code can read a process-wide default pool with `pymizu.default_pool()`.
`pymizu.set_default_pool(pool)` sets it (`None` clears it) and returns the previous default, so callers can save and restore.
The registry anchors the handle: a pool set as the default stays alive even after its variable is deleted.
`pymizu.using_pool(pool)` scopes the default to a `with` block and restores the previous default on exit, including on exception.

Resolve an optional pool in this order: an explicit `pool` argument, `current_pool()` inside a task, `default_pool()`, then your own fallback (sequential execution or an error).
The default never overrides `current_pool()`: inside a task the worker's own handle wins, so nested submission is never shadowed.
Setting a default checks the type only.
A stopped pool is accepted and fails at use time.
Handles from `Pool.attach()` are valid defaults.

```{python}
def run(x, pool=None):
    pool = pool or pymizu.current_pool() or pymizu.default_pool()
    if pool is None:
        raise RuntimeError("no pool: pass one, or set a default")
    return pool.submit(pow, x, 2).collect(timeout=30)

with pymizu.Pool.create(2) as pool:
    old = pymizu.set_default_pool(pool)
    squared = run(21)
    pymizu.set_default_pool(old)

squared
```

## Growing, shrinking, and joining a pool

A pool can grow and shrink while it runs.
`pool.spawn_workers(n)` adds workers, up to the `max_workers` limit set at create.
`pool.retire(slot)` asks a worker to exit cleanly: it finishes its current task, and the remaining workers take over anything still queued to it.
Only the pool's creator resizes it.
An attached submitter gets `MizuError`.

Other processes join a running pool as submitters with `Pool.attach(token)`.
The token is `pool.token` on the creator.
Attached submitters submit and collect exactly as the creator does.

## Observing a pool

Four read-only tools watch a running pool without disturbing it:

```python
pool.status()   # snapshot: worker states, queued tasks, result slots
pool.stats()    # cumulative counters since each participant joined
pool.dump()     # every slot in full detail — the first tool when a pool hangs
pool.trace(fn)  # fn(event, id) at each task lifecycle event; None removes it
```

## concurrent.futures interop

`pymizu.PoolExecutor` adapts a pool to the stdlib `concurrent.futures.Executor` interface.
Code written against `ProcessPoolExecutor` drops in: `submit` returns real `concurrent.futures.Future` objects.
`wait`, `as_completed`, `asyncio.wrap_future`, and the base `Executor.map` all work unchanged.
Futures resolve event-driven off the pool's notify endpoint — there is no drain thread and no polling bound.

```{python}
with pymizu.PoolExecutor.create(4) as ex:
    futures = [ex.submit(pow, 2, i) for i in range(4)]
    result = [f.result() for f in futures]

result
```

Wrap an existing pool with `PoolExecutor(pool)`.
Its lifetime then stays with its owner.
`PoolExecutor.create(workers, **pool_kwargs)` spawns an owned pool that `shutdown` stops.
A task's failure surfaces as `pymizu.TaskError`, and cancellation is advisory, like `Task.cancel`.
For bulk maps, `Pool.map` remains the faster path: the executor's `map` runs one task per element.

## Sizing a pool

`Pool.create()` sizing options: `max_workers` (the worker limit), `max_submitters`, `injection_cap` (each submitter's injection ring), `per_worker_cap` (each worker's deque), `result_slots` (outstanding uncollected tasks per submitter), and `slot_size` (bytes per queue entry and result slot, with a payload past the inline budget traveling in a fresh shared-memory region).


### Parallel map

`Pool.map(fn, x)` maps `fn` over `x` on the pool and returns the run's handle immediately — staging and submission happen at the call; collection happens on the handle.
It is not a loop over `pool.submit()`.
The function, its constant `args=`/`kwargs=`, and the data are written to shared memory once, and one *runner* task per worker divides up the elements:

```{python}
import pymizu

with pymizu.Pool.create(4) as pool:
    result = pool.map(abs, range(-5, 5)).collect()

result
```

## The map handle

A map collects exactly once, through the handle:

- `handle.collect(timeout=None)` blocks for the results: a list in input order, or with `template=` the gathered output area (a view with `collect="view"`). On `timeout` expiry the outstanding work is cancelled and the `TIMEOUT` sentinel is returned, never raised.
- `handle.collect_async()` — or simply `await handle` — resolves one `asyncio.Future` with the same results, driven by the workers' notify signals rather than any blocking or polling. Cancelling the Future only detaches the waiter (the handle collects blocking again); it never cancels the map. A streaming map (`stream=True`) collects blocking only: its sliding window needs the blocking turn loop.
- `handle.cancel()` stops the run: workers halt within about one batch and a pending collect — blocking or async — settles with `CancelledError`.

`Pool.map()`'s own `timeout=` bounds submission only: on its expiry the tasks already in are cancelled and `Pool.map()` itself returns the `TIMEOUT` sentinel (no handle).
A dropped uncollected handle cancels at garbage collection, so no run of `fn` is wasted.

## Scheduling

Runners self-schedule element batches off a shared cursor, and the batch size adapts.
A cheap `fn` runs in large batches at near-zero scheduling cost.
An expensive or uneven one takes smaller batches to keep the workers balanced.
`n_chunks=` sets the batch size directly.

A 1-D C-contiguous numpy array of float64, int32, int64, complex128, or uint8 crosses as raw bytes.
Each worker views the same array in shared memory and reads its elements directly, so `x` is never serialized:

```{python}
import numpy as np

with pymizu.Pool.create(4) as pool:
    result = pool.map(np.negative, np.arange(-5.0, 5.0)).collect()

result
```

## Unpacked elements (starmap)

`pool.starmap(fn, x)` unpacks each element of `x` as the call's positional arguments: `fn(*element, *args, **kwargs)`.
This is the `multiprocessing.Pool.starmap` convention:

```{python}
with pymizu.Pool.create(4) as pool:
    result = pool.starmap(pow, [(2, 10), (3, 3)]).collect()

result
```

For the multi-iterable shape of `concurrent.futures.Executor.map`, zip the iterables first: `pool.starmap(fn, zip(xs, ys))`.
Starmap needs Python workers: a `pymizu.call` spec raises `TypeError`.
Everything else (seeding, templates, streaming, the outcome taxonomy) is `Pool.map`'s.

## Reproducible randomness

`seed=` (an int or bytes) gives every element its own deterministic stream of the stdlib `random` module.
Element `i` is seeded from SHA-256 of the seed and `i`.
Results are identical for any chunking, worker count, or steal order:

```{python}
import random

def draw(i):
    return random.random()

with pymizu.Pool.create(4) as pool:
    a = pool.map(draw, range(100), seed=123).collect()
    b = pool.map(draw, range(100), seed=123, n_chunks=10).collect()
    print(a == b)
```

Pass `seed=(seed, offset)` to shift every element's stream by `offset` positions, for maps split across runs or processes.

`seed=` covers the stdlib `random` module only.
A task that draws from numpy calls `pymizu.current_rng()` inside the task.
This is the element's own `numpy.random.Generator`, derived from the same seed and memoized per element:

```{python}
def draw_np(i):
    return pymizu.current_rng().random()

with pymizu.Pool.create(4) as pool:
    a = pool.map(draw_np, range(100), seed=123).collect()
    b = pool.map(draw_np, range(100), seed=123, n_chunks=10).collect()
    print(a == b)
```

The legacy `np.random.*` module functions draw from the worker's shared global `RandomState`.
A seeded map does not make them deterministic: results depend on claim and steal order.
Other RNG universes (torch, jax) are out of scope.
A seeded map element must not nested-submit and collect.
Worker helping can run another map's batches mid-element and wipe the element's `current_rng()` stash.
The stdlib streams survive.

## Templates

A `template=` exemplar buffer (for example `np.empty(m, dtype=...)`) declares that every `fn` result is `m` values of that dtype.
Workers write their results straight into a shared `n x m` output area, so each result crosses the process boundary exactly once, never serialized.
Each result must be a matching buffer.
If `m == 1`, a plain Python scalar also works:

```{python}
import math

with pymizu.Pool.create(4) as pool:
    result = pool.map(math.log1p, [1.0, 2.0, 3.0], template=np.empty(1)).collect()

result
```

With a template, `collect="copy"` (the default) returns the output as one numpy array of shape `(n, m)`.
If `m == 1`, the shape is `(n,)`.
If numpy is not installed, the output is a memoryview.
`collect="view"` returns the shared output area itself as a read-only zero-copy view.
The shared memory is released when the view is garbage-collected.
The view exports `__arrow_c_array__`, so an Arrow consumer such as pyarrow or polars wraps it without numpy.
With numpy, reach the view through the result's `.base` (`arr.base` when `m == 1`, one link deeper when `m > 1`).
Without numpy, it is `mv.obj` on the memoryview.

## Streaming maps

With `stream=True`, the map never stages the whole of `x` into shared memory: it streams slices of `x` to workers as they take work, and the return value is unchanged.
Fixed slices ride ordinary chunk tasks under a sliding submit/collect window.
At most `min(n_chunks, 2 * live workers, free result slots)` tasks stay outstanding.
Shared-memory residency is bounded by `window x slice` instead of `sizeof(x)`.
The chunk count defaults to `min(len(x), 32 * live workers)`, and `n_chunks=` overrides it outright.

```{python}
with pymizu.Pool.create(4) as pool:
    result = pool.map(abs, range(-5, 5), stream=True).collect()

result
```

Everything else (ordering, `template=` and `collect=`, `seed=` invariance, the error taxonomy) is exactly the non-streaming map's.
Two trade-offs come with fixed chunks.
The adaptive batch sizing is lost: raise `n_chunks=` to mitigate skew.
Slices cross via the serialized tiers, so per-chunk staging costs an ordinary submit's serialization rather than the raw section's zero-copy slicing.
A streaming map with Python workers shares one description of the map with every worker, each read at most once.
A `pymizu.call` spec as `fn` streams too, over workers of any language.
Each chunk in the window stages a small shared region of its own: the slice plus the task that runs on it, one runner task per chunk.
The region is removed as soon as the chunk completes.
Shared-memory use stays bounded by the window plus a small per-worker cache rather than by `sizeof(x)`.
Each chunk also writes its results separately, so they cannot combine into one view.
`collect="view"` raises `TypeError` on a streaming spec map, as does preparing one with `pool.map_prepare()`.

## Prepared maps

`pool.map_prepare()` prepares a map once for repeated runs: the data is written to shared memory once, and each worker sets up the map once.
Each `m.run()` then returns the run's handle — costing only the task submissions and the collection — and reuses each worker's cached map context:

```{python}
with pymizu.Pool.create(4) as pool:
    with pool.map_prepare(abs, range(-5, 5)) as m:
        print(m.run().collect())
        print(m.run().collect())
```

`m.run(x=...)` replaces the data for that and later runs.
A raw-buffer replacement of the same dtype and length swaps in place at memcpy cost.
Any other replacement is written out fresh.
A streaming prepared map keeps `x` submitter-side, so a replacement of any shape re-slices.
Only a length change under `template=` uses fresh shared memory.
After a run collected with `collect="view"`, the next run uses fresh shared memory, because the previous output area belongs to the returned view.

## Nested maps

A map can nest: a task can start a map on the evaluating worker's own pool handle via `pymizu.current_pool()`.
The runner submissions push straight onto the worker's own deque, and the blocked collect executes its own runners while idle peers steal the rest: nested maps never deadlock the pool.

```{python}
def nested_total(parts):
    return sum(pymizu.current_pool().map(sum, parts).collect())

parts = [range(0, 250), range(250, 500), range(500, 750), range(750, 1000)]

with pymizu.Pool.create(4) as pool:
    total = pool.submit(nested_total, parts).collect()

total
```

(This example needs cloudpickle: `nested_total` is not an importable reference.)

The first nested map of a worker claims a submitter slot.
With the default `max_submitters = 8`, one slot is held by the creating process, so at most 7 workers can nest concurrently.
Raise `max_submitters` for wider nested fan-outs.
Do not nest inside the element of a seeded map.
Worker helping can run another map's batches mid-element and wipe the element's `current_rng()` stash.
See [Reproducible randomness](#reproducible-randomness).

## Outcomes

- An error raised by `fn` re-raises as `pymizu.TaskError` carrying the failing element's 0-based `index`.
  The map fails fast: the workers stop within about one batch.
- Worker death raises `pymizu.WorkerDiedError` carrying the lost element ranges as `lost` (0-based half-open `(lo, hi)` pairs, conservative).
- On `timeout=` expiry the outstanding work is cancelled and the `pymizu.TIMEOUT` sentinel is returned, never raised.
- Ctrl-C during a map cancels its outstanding tasks.

A map started inside a task runs on the worker's own pool handle, via `pymizu.current_pool()`.
A `pymizu.call` spec as `fn` is the cross-language map.
See [R interop](interop.qmd#mapping-over-a-foreign-pool).


### R interop

The process at the other end of a channel does not have to be Python.
[mizu](https://github.com/shikokuchuo/mizu) is the R counterpart of pymizu, built on the same shared-memory core.
A channel or pool connects the two languages as directly as two Python processes.

## R channel peers

A channel peer can be an R process with the `mizu` package installed.
Pass the peer program as R source, and set the launcher to `pymizu.r_launcher()`.
Here, an R process doubles the array you send:

```python
import numpy as np
import pymizu

ch = pymizu.Channel.create(
    """
library(mizu)
repeat {
  x <- mizu_recv(ch, timeout = 30)
  if (inherits(x, "mizu_sentinel")) break
  mizu_send(ch, x * 2)
}
""",
    launcher=pymizu.r_launcher(),
)

ch.send(np.array([1.5, 2.5, 3.5]))
print(ch.recv(timeout=5))            # [3. 5. 7.]
ch.close()
```

`r_launcher()` needs R and the `mizu` R package installed.
If either is missing, it raises `MizuError` before the channel is created.
Pass `rscript=` to use a specific R installation.

To spawn the peer differently, write your own launcher: one callable that takes the channel's token and starts the peer process.
The reverse direction works too: an R session spawns a Python peer with `mizu::mizu_py_launcher()`.

## Large data crosses without copying

Past a size floor, the data itself never moves.
pymizu writes it to shared memory once, and both sides read it in place: no serialization, no parsing, no further copies.
A large value you send is read by R directly over the shared pages.
A large value you receive arrives as a view over those pages, usually wrapped as a numpy array.
The peer below echoes back whatever it receives.
This channel stays open for the next few examples:

```python
ch = pymizu.Channel.create(
    """
library(mizu)
repeat {
  x <- mizu_recv(ch, timeout = 30)
  if (inherits(x, "mizu_sentinel")) break
  mizu_send(ch, x)
}
""",
    launcher=pymizu.r_launcher(),
)

big = np.random.default_rng(0).normal(size=1_250_000)  # ten megabytes
ch.send(big)
back = ch.recv(timeout=5)
np.array_equal(back, big)                              # True
```

Apart from that one write into the shared region, nothing moved at all.
R's echo came back as a reference with zero payload bytes.
If you send a received view onward, it crosses the same way: as the region's name alone.
The zero-copy tier covers every large layout: a flat array of an identity dtype (`float64`, `int32`, `int64`, `complex128`, `uint8`), an array of any shape and memory order, or a nested tree of lists, dicts, arrays, and strings.
A `datetime64` array crosses the same way, with `NaT` as `NA`, and a `list[str | None]` with `None` as `NA`.
A large data frame crosses as one shared region in both directions.
The floor is low: 256 KB for a flat contiguous array on a channel, and 32 KB everywhere else.[^byvalue]

Arrays of other numeric dtypes (booleans, the narrow ints, `uint64`, `float32`, ...) cross as a converted copy at any size.
The [dtype matrix](#the-dtype-matrix) has the per-dtype rule.
Without numpy, a received view is a plain buffer: `memoryview(view)` works, and `pa.array(view)` or `pl.from_arrow(view)` wrap the same pages without copying.
One exception: R's ALTREP vectors (say, `1:1e9`, stored as a recipe rather than data) cross by value, values exact.
A polars or pyarrow object that wraps a received frame keeps the frame's region alive until garbage collection, so delete these wrappers when you no longer need them.

[^byvalue]: Below the floor, a value crosses as one quick copy.
    At any size, these types cross: Python scalars (`bool`, `int`, `float`, `complex`, `str`, `None`), lists, dicts with `str` keys (a tuple arrives as a `list` in R), `bytes` (an R raw vector), and `datetime.date` / `datetime.datetime` scalars (length-1 dates and times in R).
    A few round trips change the container but never the values.
    A standalone R factor comes back as strings, a length-1 `NA` arrives as `None` and sent back is R's `NULL`, and an R logical vector with `NA`s comes back as integer.
    A `POSIXct` timezone normalizes to UTC (the instant is exact), and `difftime` units normalize to seconds.
    The bit patterns `-2147483648` (int32) and `-9223372036854775808` (int64) collide with R's own `NA` sentinels and read as `NA` in R.
    A `uint64` up to 2^63 - 1 converts exactly to `integer64`; past that it crosses as `NA` with a warning at send.

## What crosses

Small values round-trip just as exactly.
Each form below comes back from R unchanged:

```python
def echo(x):
    ch.send(x)
    return ch.recv(timeout=5)

x = np.array([1.5, 2.5, 3.5])
np.array_equal(echo(x), x)     # a float64 vector: True

m = np.arange(6, dtype=np.int32).reshape(2, 3)
np.array_equal(echo(m), m)     # an int32 matrix, dim included: True

echo("hello") == "hello"       # a string: True

d = np.array(["2024-01-01", "2024-01-03"], dtype="datetime64[D]")
np.array_equal(echo(d), d)     # dates keep their meaning: True
```

A data frame crosses as well, with its column types intact.
Here a polars frame with a categorical column; pandas and pyarrow frames work the same way:

```python
import polars as pl
df = pl.DataFrame({
    "total_bill": [16.99, 10.34, 21.01],
    "day": pl.Series(["Sun", "Sun", "Mon"], dtype=pl.Categorical),
})
ch.send(df)        # arrives in R as a data.frame, the categorical column as a factor
back = ch.recv(timeout=5)
back.names         # ('total_bill', 'day')
```

R sees the natural equivalent throughout: a vector for an array, a factor for a category column, a named list for a dict, a `Date` or `POSIXct` for a `datetime64` array.
An Arrow array arrives as an R vector, with Arrow nulls as R missing values: `ch.send(pa.array([1, None, 3]))`.
An ordered categorical (an R ordered factor, a polars `Enum`, an ordered pyarrow dictionary) crosses as an ordered factor in R, and the `Frame`'s Arrow export keeps the ordered flag, so pandas and polars see an ordered categorical and a round trip is `identical()`.
Labels cross with the values they name: a named R vector arrives as a [`DimArray`](`pymizu.DimArray`) carrying its names, and labels you attach in Python ride to R as `names` or `dimnames`.

## What doesn't cross

Some values have no portable home on the other side.
`pymizu` declines these at send time with a `DeclinedError` (a subclass of `TypeError`) that names the offending value, carried on its `path` and `reason` attributes.
The channel stays usable:

```python
import decimal
import pyarrow as pa

tbl = pa.table({"price": pa.array([decimal.Decimal("1.5")])})
try:
    ch.send(tbl)
except pymizu.DeclinedError as e:
    print(e)
    # pymizu: value is not portable to the peer (column 'price': an Arrow type
    # with no portable home (format 'd:2,1'))
ch.close()
```

From Python, also declined: Arrow nested, `date64`, `time`, and `interval` columns.
An Arrow `uint8` column *with nulls* is a `TypeError`: an R raw vector has no `NA`.

From R: a `POSIXlt` time, a character matrix, and S4 objects stay on R's side.
So does a named vector with a class or of character type (a named character vector, factor, `Date`, or `POSIXct`), since names ride only on the plain atomic vectors.
Among data frames: a data.table, a groupedData-style frame (a formula attribute), and a column that is itself a labeled matrix.

## pymizu.Frame

A `pymizu.Frame` is Python's home for an R `data.frame`: named columns with a row count.
`frame.to_dict()` returns the columns as a dict: numpy arrays when numpy is installed, memoryviews otherwise, strings and factors as `list[str | None]`.
`frame.names` and `frame.row_names` carry the metadata.

Any Arrow consumer takes a frame in one line: `pl.from_arrow(f)`, `pa.table(f)`, or `pd.DataFrame.from_arrow(f)`.
(With polars >= 2.0, a streaming producer gives a struct Series; `.struct.unnest()` restores the columns.)
Factors become dictionary columns, logicals Arrow `bool`, `Date` `date32`, and `POSIXct` `timestamp[us]`.

A `Frame` pickles, so it crosses Python-to-Python channels.
Sent to R, it becomes a `data.frame` again.

Frame metadata rides along verbatim: the full class vector of a data.frame (a tibble's `c("tbl_df", "tbl", "data.frame")`, any class ending in `"data.frame"`) and extra attributes cross opaquely, stashed undecoded on the `Frame`.
An unmodified echo restores them in R.
A frame rebuilt through pyarrow, polars, pandas, or `from_arrow` arrives in R as a plain `data.frame`.

## Mixed-language pools

A pool's workers can be R processes.
`pymizu.r_pool_launcher()` spawns the workers.
`pymizu.call()` builds a language-neutral task: a qualified `"pkg::fn"` name for R or `"mod.fn"` for Python, or a `source=` string, plus constant arguments:

```python
import numpy as np
import pymizu

pool = pymizu.Pool.create(4, launcher=pymizu.r_pool_launcher())
t = pool.submit(pymizu.call("stats::quantile", np.arange(101.0),
                            probs=np.array([0.25, 0.5, 0.75]), names=False))
print(t.collect())                  # [ 25.  50.  75.]
t2 = pool.submit(pymizu.call(source="y <- x * 2\ny + 1", x=20))
print(t2.collect())                 # 41
pool.stop()
```

Unnamed arguments become positional arguments, and keyword arguments become named ones, matching R's calling convention.
A `source=` task runs in a fresh environment with the keyword arguments bound as names and the positional arguments bound as `_1`, `_2`, ...
The result is the value of the last expression.

On an R-worker pool, submitting a plain Python callable raises an error naming `pymizu.call()`.
Task errors cross as `TaskError` with the R error type preserved in `remote_type`.
Arguments follow the same rules as channel payloads, zero-copy included: a large array crosses as one shared view, and an array that is already a received view crosses as its name alone.
A pool's workers all run one language: mixing R and Python workers in one pool is not supported.
The reverse direction mirrors this: `mizu::mizu_pool()` with `mizu::mizu_py_pool_launcher()`, and `mizu::mizu_submit_call()` with `mizu::mizu_call()` specs.

## Mapping over a foreign pool

To map an R function over Python data, pass a `pymizu.call()` spec as the `fn` of `Pool.map()` or `Pool.map_prepare()` on an R-worker pool.
`mizu::mizu_map()` mirrors this in R:

```python
pool = pymizu.Pool.create(4, launcher=pymizu.r_pool_launcher())
pool.map(pymizu.call("stats::median"), x).collect()
pool.map(pymizu.call(source="summary(mgcv::gam(x$y ~ s(x$x)))$r.sq"),
         folds, template=0.0).collect()   # each fold a dict; the element binds as x
pool.stop()
```

Each element becomes the spec's first positional argument, or binds as `x` for a `source=` spec.
Put constant arguments in the spec itself: the map's own `args=` and `kwargs=` stay empty.
`template=` and `collect=` work unchanged, and a per-element error crosses with its `index`.
Seeds must be integers within R's 32-bit range.
Each worker language derives its own random streams from the seed, so results are reproducible for a given worker language but not identical across languages.

## The dtype matrix

The exact correspondence between Python and R dtypes.
Conversion happens once, at send time.
The identity rows (float64, int32, int64, complex128, uint8) are a plain memcpy.

| Python sends | R receives | Notes |
|----|----|----|
| `bytes` / uint8 | raw | Arrow uint8 with nulls: `TypeError` (R raw has no NA) |
| int8 / int16 / uint16 | integer | widened, exact |
| int32 | integer | |
| uint32 | double | widened, exact |
| int64 | integer64 (bit64's layout) | bit-exact; `INT64_MIN` reads as `NA` |
| uint64 | integer64 (`NA` past 2^63 - 1, warned) | the frame column and `mizu_as_arrow()` are bit-exact |
| float16 / float32 / float64 | double | float16 and float32 widened, exact |
| bool | logical | |
| complex64 / complex128 | complex | (buffer protocol only; Arrow has no standard complex) |
| Arrow bool / numeric with nulls | logical / numeric with `NA` | the validity bitmap is honored, slices included |
| Arrow strings (utf8 / large_utf8 / string_view) | character | a Series as `list[str \| None]`; a column as a frame column |
| Arrow `date32` / `timestamp[u]` | `Date` / `POSIXct` | the zone name rides a frame column's metadata |
| Arrow `duration[u]` | `difftime` (secs) | the frame column exports back as `duration[us]` |
| timedelta64 (any non-calendar unit) | `difftime` (secs) | `NaT` reads as `NA`; the unit converts to seconds |
| Arrow dictionary-encoded | factor | an ordered dictionary (polars `Enum`) arrives as an ordered factor |
| Arrow ChunkedArray / Table / Series | vector / `data.frame` | chunks are concatenated |
| Arrow nested, decimal, time32/64, interval | — | `DeclinedError` at send time |

Missing values:

- R marks missing values with sentinels inside the data: `INT_MIN` for integer and logical, `INT64_MIN` for integer64, and a special NaN value (`NA_real_`) for double.
  Arrow nulls convert to these sentinels, so R sees correct `NA`s.
- Read from Python, the result depends on the accessor:
  - Copied reads (`list(...)`, `Frame.to_dict()`) give exact dtypes: a logical arrives as `bool_` when it has no `NA`, else int32.
    An integer arrives as int32 when `NA`-free, else float64 with NaN (every int32 value is exact in float64).
  - Zero-copy views keep the sentinel values in the buffer.
    `.to_numpy()` applies the same rule and reads NA-freeness off the region's validity section before any data scan.
  - int64 keeps its dtype in every numpy form and warns when an `NA` is detected on a copied read or `.to_numpy()`.
    Use `.to_arrow()` for an NA-exact read.
    A genuine Python `-2^63` value round-trips without a warning.
  - Every Arrow export of a logical is Arrow `bool` with a validity bitmap, whatever the path.
    Integer exports are int32 or int64 with a validity bitmap.
    A double exports as float64 with `NA_real_` slots as Arrow nulls (a genuine NaN stays a valid value).
- A genuine int32 value of `-2147483648` collides with the `NA` sentinel and reads as `NA` in R.
  Likewise an int64 value of `-9223372036854775808` reads as `NA_integer64_`.
  numpy sends stay silent.
  An Arrow int32 or int64 send with a validity bitmap warns once, to an R peer.
  A Python peer reads the identical columns exactly and stays silent.
- Python compute treats `NA_real_` as an ordinary NaN.
  Whether the exact NaN value survives arithmetic is platform-dependent.
  Do not rely on it.

Round trips are stable after the first hop, and a pass-through echo is bit-exact.
An untouched received view sent back crosses as a reference to its region, so no data moves and even the `NA_real_` value survives a relay.
A slice or a dtype view of a received view crosses by value.

| Python sends | R sees | Back in Python | Notes |
|----|----|----|----|
| uint8 | raw | uint8 | exact |
| int8 / int16 / uint16 | integer | int32 | widened, values exact |
| int32 | integer | int32 | exact |
| uint32 | double | float64 | exact |
| int64 | integer64 | int64 | exact |
| uint64 | integer64 (`NA` past 2^63 - 1, warned) | uint64 | the column and R's Arrow export are bit-exact |
| float16 | double | float64 | widened, values exact |
| float32 | double | float64 | widened, values exact |
| float64 | double | float64 | exact |
| bool | logical | `bool_` (int32 with NAs) | dtype lost |
| complex64 / complex128 | complex | complex128 | exact |
| timedelta64 | `difftime` | timedelta64[us] | unit normalizes to us, values exact |

| R sends | Python sees | Back in R | Notes |
|----|----|----|----|
| integer | int32 (float64 with NAs) | integer | exact |
| integer64 (bit64) | int64 (warns on `NA`) | integer64 | exact; a genuine Python `-2^63` arrives as `NA` |
| double | float64 | double | exact |
| raw | uint8 | raw | exact |
| complex | complex128 | complex | exact |
| logical | `bool_` (int32 with NAs) | **integer** | tag lost on a Python round trip (values kept); a whole view sent back stays logical |
| difftime | timedelta64[us] | difftime (secs) | units normalize to seconds, values exact; `NA` crosses as `NaT` |

Between two Python processes, on a channel or a pool, the conversions on this page never run.
The identity dtypes cross unchanged, and everything else pickles.
An exact Python-to-Python send of a non-identity dtype needs no work.
They apply only when the other end runs R.