parfor

nimony/lib/std/parfor.nim

Structured data parallelism: for i in a || b: x[i] = f(input[i]).

The || iterator is a for-loop plugin: at sem time the plugin rewrites the loop into a set of chunk runners submitted to the shared std/threadpool worker pool plus a structured join at the loop's closing. The join is the loop end -- there is no spawn, no Flowvar, no handle, and (by construction of the static race rules) no data race.

A chunk runner is a .passive coroutine, so an iteration that performs I/O parks its worker instead of pinning it -- the parallel for therefore covers I/O-bound maps, not just CPU-bound ones.

This module provides the runtime the lowering targets; see deps/parfor.nim for the plugin that emits calls into it.

type Workload = enum
  MixedBound = (0, "MixedBound")
  CpuBound = (1, "CpuBound")
  IoBound = (2, "IoBound")

func dollar`.Workload(e: Workload): string

type ParJoin = object
  remaining: int64

proc ensureParPool()
Start the worker pool the first time a parallel for runs (idempotent — defaultPool() is itself a lazy singleton, so subsequent calls, from this module or any other caller of defaultPool(), are a no-op).
proc parIterCount(a: int64; b: int64; step: int64): int64
Number of iterations of the inclusive strided range a, a+step, …, ≤ b (Nim's || yields a, a+step, …). Zero for an empty range or a non-positive step.
proc parGrain(iters: int64; chunkSize: int64): int64
Iterations per chunk. A positive chunkSize is honoured (the programmer tunes it to the body's cost-per-iteration), but coarsened up if it would need more than ParMaxChunks runners — so a tiny grain over a huge range stays pool-scheduled instead of spilling onto the submitting thread. 0 derives a grain that yields about ParDefaultChunks chunks (one per worker), adapting to the machine. Always >= 1 for a non-empty range, so a chunk is never empty.
proc parChunkCount(iters: int64; grain: int64): int64
Number of grain-sized chunks needed to cover iters iterations, ceil(iters / grain).
proc parChunkLo(grain: int64; k: int64): int64
First iteration index (inclusive) of chunk k: k * grain. Chunks are fixed-size half-open [lo, hi) ranges over the iteration-index space; the chunk runner maps each index j back to the value a + j*step.
proc parChunkHi(iters: int64; grain: int64; k: int64): int64
One past the last iteration index of chunk k: min((k+1)*grain, iters) (the final chunk is short when grain does not divide iters).
proc parBegin(j: var ParJoin; chunks: int64)
Arm the join for chunks outstanding runners. Call before submitting.
proc parChunkDone(j: ptr ParJoin)
Signal that one chunk runner has finished. Called at the tail of every chunk runner the plugin emits.
proc parWait(j: var ParJoin; workload: Workload)
Block until every chunk runner has finished. While waiting the thread acts as a temporary worker: it first helps drain pool tasks (poolHelp) so a chunk body that opens its own || (recursive fork-join) keeps making progress instead of deadlocking. When there is no CPU work to run it polls I/O (poolPollIo) so a join whose chunks are all parked on I/O still advances them — without that, nested joins where every worker is waiting would deadlock with no thread left polling the event loop.
CpuBound skips the I/O poll: such chunks never park, so there is nothing in the event loop and the joiner just spins for the in-flight CPU work.
proc parSubmit(c: Continuation; hint: int64)
Hand a chunk runner's continuation to the worker pool. Re-exported so the || plugin only needs symbols visible through import std/parfor.
hint is deliberately ignored — do not "fix" this. The plugin passes the chunk number, and spreading chunks across stripes by it sounds like the better scheduling decision. It is much worse, because it changes WHICH threadpool.submit runs: any hint but -1 takes the shared stripe and one lock acquisition per chunk, while -1 on a worker stages the chunk on that worker's private ring and hands a whole batch over on its next cycle. A nested || submits its chunks from inside workers, so every level of a fork-join pays that lock, and the workers end up stuck on it rather than running the work.
Measured (blackmius, nim-lang/nimony#2390, and reproduced here on a dfs(8, 8) recursive ||, 16.7M chunks): passing the hint 7.7s, -1 0.87s — a 9x difference, and his profile showed one worker having handled 113k of 114M tasks. Passing it only from non-workers (the main thread, which cannot stage) does not recover it either: 3.0s, because the main thread is a full participant through parWait's poolHelp and its locked submissions contend with every worker. On a flat top-level || — the case the spread was supposed to help — the two are indistinguishable (45ms either way): submit's try-the-other-stripes fallback and work stealing already balance it.
The parameter stays because the chunk number is the plugin's to say and a future scheduler may want it; what it is NOT is a stripe index.
iterator ||(a: int64; b: int64; step: int64; chunkSize: int64; workload: Workload): int64
Parallel range for loop. for i in a || b: x[i] = f(input[i]) runs the body for every i in the inclusive range a .. b across the worker pool, joining at the loop's closing — matching Nim's standard ||, which yields a, a+step, …, ≤ b. step is the iteration stride (default 1).
chunkSize is the grain: how many iterations each parallel runner handles. Tune it to the body's cost-per-iteration (a property you know) rather than the worker count (which varies between machines); the number of runners falls out as ceil(iters / chunkSize). 0 (the default) derives a grain giving about one chunk per worker. Pass it by name to skip step: for i in ||(a, b, chunkSize = 64): ….
workload hints at the body's typical cost (MixedBound / CpuBound / IoBound) so the join can wait efficiently — e.g. CpuBound skips the I/O poll. It is a hint only; every value is correct. Pass it by name: for i in ||(0, n, workload = CpuBound): ….
The body must write only x[i]-style outputs at the iteration index and must not read those outputs back; under that contract iterations are data-race free by construction. The plugin in deps/parfor.nim performs the rewrite.