Parallel

Parallel map-reduce over RRB.Vec.

Three public entry points:

pmap     (input, f)           -- transform every element, collect in order
preduce  (input, zero, f, g)  -- transform + reduce with explicit zero/merge
pmap_n   (input, f, workers)  -- like pmap  but with explicit worker count
preduce_n(input, zero, f, g, workers) -- like preduce with explicit count

Convenience wrappers built on preduce: psum, psum_float, pcount, pany, pall

All functions follow the same structure:

chunk(input, workers) — split into N slices
task_spawn per slice — each worker transforms its slice
fold_left over tasks — await and merge results left-to-right

Static partitioning means each worker processes a fixed slice. Imbalanced workloads (elements that vary widely in processing time) may leave fast workers idle while slow ones finish. This is a known v1 limitation.

Interpreter vs. compiled

Tasks run eagerly (single-threaded) in the interpreter, giving correct but sequential results. Real multi-core parallelism requires compiling with forge build or march --compile.

Quick start

let nums  = RRB.range(1, 1_000_001)
let total = Parallel.psum(nums)                         -- 500_000_500_000
let evens = Parallel.pcount(nums, fn n -> n % 2 == 0)  -- 500_000
let sq    = Parallel.pmap(nums, fn n -> n * n)          -- Vec of squares

See docs/cookbook/parallel-data.md for full examples.

Functions

fnpallpall(v, pred): Bool#

True if pred holds for every element. Evaluates all chunks (no early exit in v1).

Examples

march> Parallel.pall(RRB.from_list([2, 4, 6]), fn n -> n % 2 == 0)
true

fnpanypany(v, pred): Bool#

True if pred holds for at least one element. Evaluates all chunks even after finding a match (no short-circuit in v1).

Examples

march> Parallel.pany(RRB.from_list([1, 3, 4, 7]), fn n -> n % 2 == 0)
true

fnpcountpcount(v, pred): Int#

Count elements satisfying pred in parallel.

Examples

march> Parallel.pcount(RRB.range(1, 11), fn n -> n % 2 == 0)
5

fnpmappmap(input, f)#

Map f over every element of input in parallel, collecting results in a new Vec in the same order as the input. Uses System.cpu_count() workers. For a custom worker count, use pmap_n.

Examples

march> RRB.to_list(Parallel.pmap(RRB.range(1, 6), fn n -> n * 2))
[2, 4, 6, 8, 10]
march> RRB.to_list(Parallel.pmap(RRB.empty(), fn n -> n * 2))
[]

fnpmap_npmap_n(input, f, workers: Int)#

Like pmap but with an explicit workers count. Passing a larger value creates finer chunks for heterogeneous per-element workloads. The count is clamped to between 1 and the input length, so 0, a negative count or one larger than the input is accepted; an empty input spawns no task.

Examples

march> RRB.to_list(Parallel.pmap_n(RRB.range(1, 6), fn n -> n * n, 2))
[1, 4, 9, 16, 25]
march> RRB.to_list(Parallel.pmap_n(RRB.from_list([1, 2]), fn n -> n + 1, 0))
[2, 3]

fnpreducepreduce(input, zero, f, merge)#

Map f over every element then reduce with merge. Uses zero as the starting accumulator for each chunk (it must be the identity for merge). Uses System.cpu_count() workers. For a custom count, use preduce_n.

Contract: merge must be associative and zero must be its identity.

Reproducibility note: the input is split into System.cpu_count() chunks, so the grouping of merge applications depends on the backend's CPU/domain count. If merge is only approximately associative (e.g. floating-point +.), the result is not bit-identical across backends — do not golden-compare such a preduce. Integer merges (+, max, ||) are exactly associative and are reproducible.

An empty input returns zero.

Examples

Sum of squares, and a maximum (over non-negative values, so 0 is the identity):

march> Parallel.preduce(RRB.range(1, 11), 0, fn n -> n * n, fn (a, b) -> a + b)
385
march> Parallel.preduce(RRB.from_list([3, 9, 4]), 0, fn n -> n, fn (a, b) -> if a > b do a else b end)
9
march> Parallel.preduce(RRB.empty(), 0, fn n -> n, fn (a, b) -> a + b)
0

fnpreduce_npreduce_n(input, zero, f, merge, workers: Int)#

Like preduce but with an explicit workers count, clamped as in pmap_n. zero seeds every chunk, so it must be the identity of merge or it is counted once per chunk.

Examples

march> Parallel.preduce_n(RRB.range(1, 101), 0, fn n -> n, fn (a, b) -> a + b, 4)
5050
march> Parallel.preduce_n(RRB.from_list(["a", "b", "c"]), "", fn s -> s, fn (a, b) -> a ++ b, 2)
"abc"

fnpsumpsum(v): Int#

Parallel sum of an integer Vec.

Examples

march> Parallel.psum(RRB.range(1, 6))
15

fnpsum_floatpsum_float(v): Float#

Parallel sum of a float Vec.

Not bit-reproducible across backends. Floating-point addition is not associative, and the number of parallel chunks depends on the worker count, which differs between the interpreter (Domain.recommended_domain_count) and the compiled runtime (physical CPU count). The same input can therefore sum to slightly different values on different backends — do not golden-compare the result of psum_float (or any preduce with a non-associative merge).

Examples

march> (Parallel.psum_float(RRB.from_list([1.0, 2.5, 3.5]))) == 7.0
true