Parallel map

pyrula.parallel.par_map runs a function over a collection in forked worker processes on macOS and Linux. Each process has its own GIL, so CPU-bound Python can run in parallel. The callback and inputs are inherited through the fork rather than pickled; only results return to the parent process.

Call it only when no other Python threads are alive. Native threads are not visible to that guard, so set timeout when a native client or numerical library may have started threads. See Scope for the complete constraints.

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 input determines the numeric dtype and output container. FloatList and IntList results are written to a shared-memory column without serializing each element.

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. Workers materialize range values locally. IList is accepted, but does not use the typed numeric path used by FloatList and IntList.

Typed results with result_type

result_type selects both the wire format and return container. It is mutually exclusive with result_codec.

from hashlib import sha256
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 d: sha256(d).digest(), 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

For model and dataclass results, encode validates each result in the worker process. A result that fails validation becomes an element error; the rest of the batch continues. (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]

Results that come back as JSON or pickle frames go through a shared region sized max(1 MiB, 128 bytes × n). If they don’t fit, the call raises RegionOverflowError; pass a larger region_size.

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
    region_size=None,      # bytes for object-path results; default max(1 MiB, 128 * n)
    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.

Callback state

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

Return any data that the parent process must collect.

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. Very cheap functions are dominated by per-call fork and dispatch overhead; use a comprehension or a numeric list operation for trivial arithmetic.

Versus other process pools

Threads don’t help CPU work (0.97x here, GIL-bound); the comparison is process pools. Same heavy fn, 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

The pools serialize every input and result on every call; par_map forks and returns typed results through shared memory. The gap grows with input size and narrows for tiny inputs.

Scope

Windows raises NotImplementedError. par_map enforces that no other Python threads are alive; pass unsafe_allow_threads=True only when no live thread holds a lock that a child may need. Each call forks new workers; there is no persistent pool.

The guard sees Python threads only. A native thread is invisible to it (a live pyrula-kafka client’s I/O threads, a numpy or BLAS thread pool), and a child that needs something such a thread held can hang, so set timeout when those are live. pyrula’s own collections, serde and records code keeps no thread a child could wait on.

Full signature and error types: Parallel API reference.