Parallel map

Experimental

pyrula.parallel is experimental. The behavior described here is stable, but the API may still shift before it’s promoted into the core namespace.

pyrula.parallel.par_map runs a function over a collection across a pool of forked worker processes. Threads can’t help CPU-bound Python (one GIL, one running bytecode at a time); par_map sidesteps that by using processes, each with its own GIL. Unlike multiprocessing, it never pickles the inputs or the function. They cross to the workers for free through fork copy-on-write, so closures and lambdas just work, and only the results come back.

from pyrula import FloatList
from pyrula.parallel import par_map

def price_option(strike):
    # some heavy, pure numeric work
    ...

strikes = FloatList([float(s) for s in range(10_000)])
prices = par_map(price_option, strikes)   # FloatList in, FloatList out

The dtype is inferred from the input, and the output container matches it. No serialization happens on a FloatList/IntList: results are written straight into a shared memory column and handed back with zero per-element boxing.

Typed vs object input

Input type picks the path and the return type:

from pyrula import FloatList, IntList
from pyrula.parallel import par_map

par_map(lambda x: x * 2.0, FloatList([1.0, 2.0, 3.0]))   # -> FloatList
par_map(lambda n: n * n,   IntList([1, 2, 3]))           # -> IntList
par_map(lambda r: {"id": r["id"]}, [{"id": 1}, {"id": 2}])  # list -> list

FloatList and IntList take the fast numeric path (f64 / i64 columns). A plain list takes the object path: each result is encoded with result_codec, which is "json" by default (fast, restricted to JSON-native types) or "pickle" for arbitrary picklable results. Inputs and the function are never pickled either way.

# json is native-types only; a tuple result errors per element
par_map(lambda r: (r, r), rows, result_codec="json")     # element errors
# pickle keeps the type
par_map(lambda r: (r, r), rows, result_codec="pickle")   # [(r, r), ...]

tuple and range also take the object path, and range is free to pass: each integer is materialized fresh in the worker, so there’s nothing to copy. An IList works too, as a convenience, though it isn’t the copy-on-write win a FloatList is.

Typed results with result_type

Instead of a codec, you can declare the type of the results. result_type picks the wire format and the return container in one move, and it’s mutually exclusive with result_codec.

from pydantic import BaseModel
from pyrula.parallel import par_map

par_map(lambda x: x * 1.5, range(10_000), result_type=float)  # -> FloatList
par_map(lambda i: sha256(i), docs,        result_type=bytes)  # -> BytesList
par_map(lambda i: f"row-{i}", ids,        result_type=str)    # -> list[str]

class Row(BaseModel):
    id: int
    score: float

rows = par_map(build_row, ids, result_type=Row)   # -> list[Row], validated

The point of a model or dataclass result_type isn’t just the return type, it’s where the validation runs. encode validates each result, and encode runs in the forked workers, so you validate a batch across all your cores. A result that doesn’t satisfy the contract becomes that element’s error, not a crash; the rest of the batch is unaffected. (The parent re-validates on the way back in, so the parallel win is the validation, not the final decode.) A dataclass result_type uses pickle, so it must be a module-level class, not one defined inside a function.

Per-worker setup

initializer runs once in each worker right after it forks, before any element. Use it to reseed a random generator, open a per-worker handle, or warm a cache:

import os, random
par_map(fn, data, initializer=lambda: random.seed(os.urandom(16)))

Forked workers inherit the parent’s memory, including RNG state. CPython reseeds the standard random module for you on fork, but a generator you seeded yourself (or numpy’s global) is not reseeded, so every worker would otherwise draw the same “random” stream. initializer is where you fix that. If it raises, that worker dies and the call raises WorkerDiedError.

Errors

One bad element never kills the batch. on_error chooses how failures surface once the whole batch has run:

from pyrula.parallel import par_map, ParMapError

# "raise" (default): finish the batch, then raise with the failures collected
try:
    prices = par_map(price_option, strikes)
except ParMapError as e:
    e.errors      # [(index, traceback), ...] sorted by index
    e.partial     # the results that succeeded (error slots hold a sentinel)

# "either": return an aligned list of (ok, value_or_traceback)
results = par_map(price_option, strikes, on_error="either")
good = [v for ok, v in results if ok]

Options

par_map(
    fn, data,
    workers=None,          # default min(8, cpu_count - 1)
    chunk_size=None,       # default max(4, n // (workers * 8))
    timeout=None,          # seconds, whole-call wall clock
    on_error="raise",      # or "either"
    result_codec="json",   # object path only; or "pickle" (exclusive with result_type)
    result_type=None,      # float/int/bytes/str/dataclass/pydantic model
    initializer=None,      # callable run once per worker after fork
    initargs=(),           # positional args for initializer
)

timeout bounds the whole call and kills any stragglers if it’s exceeded. Small inputs (under 32 elements, or workers=1) skip forking entirely and run in-process with the same semantics, so par_map is safe to call unconditionally.

The one rule: fn must be pure

fn runs in a forked child. Any mutation it makes, to a global, a closed-over variable, self, or shared state, happens in the child and is silently lost. Only the return value comes back.

seen = []
par_map(lambda x: seen.append(x), data)   # `seen` stays empty in the parent

If you need to collect something, return it and read it from the results.

Benchmarks

10,000 elements, 8 workers on a 10-core machine, versus a serial list comprehension. Reproducible from benchmarks/core/par_map.py.

Workload	Serial	par_map (8)
Heavy numeric fn (FloatList)	1.00x	7.4x
Object path, heavy fn (list[dict])	1.00x	7.1x
Validate + build model (result_type=Model)	1.00x	6.8x
Cheap fn over 10M floats (FloatList)	1.00x	4.3x
Trivial fn (result_type=bytes, ~0.5µs each)	1.00x	0.36x

Speedup tracks the per-element cost: heavier functions get closer to linear (0.7 x workers and up). Very cheap functions are dominated by per-call fork and dispatch overhead, so par_map is for real CPU work, not trivial arithmetic where a plain comprehension or a numeric list op is already faster.

Versus other process pools

The honest comparison isn’t threads (the GIL makes them a non-starter for CPU work, measured at 0.97x here) — it’s a real process pool. Same heavy fn, same 10k inputs, 8 workers:

Tool	Speedup
par_map	7.1x	forks, pickles nothing
multiprocessing.Pool (warm)	6.1x	pool reused across calls
ProcessPoolExecutor	5.2x	pickles fn + every element
multiprocessing.Pool (cold)	5.7x	pickles fn + every element
joblib (loky backend)	5.1x	pickles fn + every element
ThreadPoolExecutor	0.97x	GIL-bound

par_map beats the process pools even when they’re kept warm (workers reused, no startup cost), because it never pickles the inputs or the results — they cross via fork copy-on-write, and typed results come back through shared memory. The other tools serialize all 10,000 inputs into the workers and 10,000 results back out on every call. The gap widens as inputs get bigger or less picklable (closures, large arrays); it narrows for tiny inputs where pickling is cheap.

Scope

Experimental, and fork-based (macOS and Linux; Windows raises NotImplementedError). It must be called with no other threads alive, which it enforces; pass unsafe_allow_threads=True only when you know no live thread holds a lock the children need. There is no persistent pool yet: each call forks fresh workers.