ForkJoin

Safe-by-construction fork/join parallelism via ``concurrent.futures``. Raw thread pools are easy to misuse in three ways this library closes: 1. **Forgotten joins / double joins (linearity).** A ``Future`` is a multiplicity-1 handle: you must ``join`` it exactly once. Dropping a future leaks unfinished work; joining twice races on a spent handle. Both are compile errors (``LinearUnusedError`` / ``LinearUsedTooManyTimesError``). 2. **Leaked pools (linearity).** A ``Pool`` is likewise linear: every ``pool`` must eventually reach ``shutdown``. Forgetting to shut down leaves worker threads alive; shutting down twice is rejected. 3. **Bogus worker counts and lost sizes (refinements).** Pool creation demands ``workers > 0`` (and a small upper bound). ``par_map`` proves the output length equals the input length. ``halves`` proves both pieces are non-empty and cover the original array — the invariant that makes divide-and-conquer parallel recursion safe to type-check. Linearity also prevents *data races on unique buffers*: a linear ``Array`` cannot be captured by two forked thunks unless you ``Array.copy`` first, so parallel tasks never alias the same unique resource. Narrative guide: ``docs/forkjoin.md``. HTML reference: ``aeon --doc libraries/ForkJoin.ae``. ── Opaque handles ─────────────────────────────────────────────────────── Thread pool. Linear: create with ``pool``, thread through ``fork`` / ``fork2`` / ``invoke``, finish with ``shutdown``.
Imports
open Array;
Table of Contents

Types

Pool

(type) linear
Thread pool. Linear: create with ``pool``, thread through ``fork`` / ``fork2`` / ``invoke``, finish with ``shutdown``.
linear type Pool

Future

(type) linear
Pending asynchronous result. Linear: must be ``join``ed exactly once.
linear type Future a

Forked

(type) linear
Result of ``fork``: packages the (still-open) pool with the new future so both can be projected and the pool threaded onward.
linear type Forked a

Forked2

(type) linear
Result of ``fork2``: pool plus the two independent futures.
linear type Forked2 a b

Invoked

(type) linear
Result of ``invoke``: pool plus the computed value.
linear type Invoked a

Joined2

(type) linear
Result of ``join2`` / ``parallel2``: an linear pair of values.
linear type Joined2 a b

Halves

(type) linear
Result of ``halves``: two contiguous non-empty pieces of an array.
linear type Halves a

Functions

pool

Create a thread pool.
def pool (workers : Int | workers > 0 && workers ≤ 64) : {p : Pool | pool_workers p = workers}

shutdown

Shut the pool down, waiting for in-flight work. Consumes the handle.
def shutdown (1 p : Pool) : Unit

fork

Submit ``task`` (a ``Unit -> a`` thunk) to ``p``. Consumes the pool and returns a ``Forked`` wrapper; project with ``forked_pool`` / ``forked_future``.
def fork (1 p : Pool) (task : (_ : Unit) → a) : {f : (Forked a) | forked_workers f = pool_workers p}

forked_pool

Recover the pool from a ``fork`` result, with its worker count intact.
def forked_pool (1 f : (Forked a)) : {p : Pool | pool_workers p = forked_workers f}

forked_future

Recover the linear future from a ``fork`` result. Bind it to ``let 1``.
def forked_future (1 f : (Forked a)) : (Future a)

fork2

Submit two independent thunks. Project with ``forked2_pool`` / ``forked2_left`` / ``forked2_right``.
def fork2 (1 p : Pool) (left : (_ : Unit) → a) (right : (_ : Unit) → b) : {f : (Forked2 a b) | forked2_workers f = pool_workers p}

forked2_pool

def forked2_pool (1 f : (Forked2 a b)) : {p : Pool | pool_workers p = forked2_workers f}

forked2_left

def forked2_left (1 f : (Forked2 a b)) : (Future a)

forked2_right

def forked2_right (1 f : (Forked2 a b)) : (Future b)

invoke

Run ``task`` on the pool and wait; keep the pool for further work.
def invoke (1 p : Pool) (task : (_ : Unit) → a) : {i : (Invoked a) | invoked_workers i = pool_workers p}

invoked_pool

def invoked_pool (1 i : (Invoked a)) : {p : Pool | pool_workers p = invoked_workers i}

invoked_value

def invoked_value (1 i : (Invoked a)) : a

join

Block until ``future`` completes; consumes the future.
def join (1 future : (Future a)) : a

join_timeout

Like ``join`` but aborts after a positive timeout (seconds).
def join_timeout (seconds : Float | seconds > 0.0) (1 future : (Future a)) : a

join2

Wait for both futures and return their results as a ``Joined2``.
def join2 (1 left : (Future a)) (1 right : (Future b)) : (Joined2 a b)

joined_fst

def joined_fst (1 j : (Joined2 a b)) : a

joined_snd

def joined_snd (1 j : (Joined2 a b)) : b

parallel2

Create a pool, run two thunks in parallel, join both, shut down.
def parallel2 (workers : Int | workers > 0 && workers ≤ 64) (left : (_ : Unit) → a) (right : (_ : Unit) → b) : (Joined2 a b)

par_map

Map ``f`` over a linear array using a temporary pool of ``workers`` threads. Preserves length: the result has the same ``size`` as the input.
def par_map (_inst2330 : (Copy a)) (workers : Int | workers > 0 && workers ≤ 64) (f : (x : a | p x) → {w : b | q w}) (1 xs : (Array a)) : {ys : (Array b) | size ys = size xs}

halves

Split a non-trivial linear array into two non-empty contiguous halves. The refinements guarantee coverage: ``left + right = original`` and both sides have size ≥ 1 — exactly the precondition recursive parallel forks need.
def halves (1 xs : (Array a) | size xs ≥ 2) : {h : (Halves a) | half_left_size h + half_right_size h = size xs && half_left_size h ≥ 1 && half_right_size h ≥ 1}

half_left

Left half; length pinned to ``half_left_size``.
def half_left (1 h : (Halves a)) : {xs : (Array a) | size xs = half_left_size h && size xs ≥ 1}

half_right

Right half; length pinned to ``half_right_size``.
def half_right (1 h : (Halves a)) : {xs : (Array a) | size xs = half_right_size h && size xs ≥ 1}

with_forked

Unpack every owned field in one consumption.
def with_forked (1 token : (Forked a)) (1 consume : (1 pool : Pool | pool_workers pool = forked_workers token) → (1 future : (Future a)) → c) : c

with_forked2

Unpack every owned field in one consumption.
def with_forked2 (1 token : (Forked2 a b)) ( 1 consume : (1 pool : Pool | pool_workers pool = forked2_workers token) → (1 left : (Future a)) → (1 right : (Future b)) → c ) : c

with_invoked

Unpack every owned field in one consumption.
def with_invoked (_inst2331 : (Copy a)) (1 token : (Invoked a)) (1 consume : (1 pool : Pool | pool_workers pool = invoked_workers token) → (value : a) → c) : c

with_invoked_owned

Unpack every owned field in one consumption. Transfer linear elements instead of copying observations.
def with_invoked_owned (1 token : (Invoked a)) (1 consume : (1 pool : Pool | pool_workers pool = invoked_workers token) → (1 value : a) → c) : c

with_joined

Unpack every owned field in one consumption.
def with_joined (_inst2332 : (Copy a)) (_inst2333 : (Copy b)) (1 token : (Joined2 a b)) (1 consume : (left : a) → (right : b) → c) : c

with_joined_owned

Unpack every owned field in one consumption. Transfer linear elements instead of copying observations.
def with_joined_owned (1 token : (Joined2 a b)) (1 consume : (1 left : a) → (1 right : b) → c) : c

with_halves

Unpack every owned field in one consumption.
def with_halves (1 token : (Halves a)) ( 1 consume : (1 left : (Array a) | size left = half_left_size token && size left ≥ 1) → (1 right : (Array a) | size right = half_right_size token && size right ≥ 1) → c ) : c

Uninterpreted

Functions declared as def f ... = uninterpreted: only their signature is known to the verifier; they have no body.

pool_workers

uninterpreted
Worker count of a pool / forked wrapper.
def pool_workers : (p : Pool) → Int

forked_workers

uninterpreted
def forked_workers : (f : (Forked a)) → Int

forked2_workers

uninterpreted
def forked2_workers : (f : (Forked2 a b)) → Int

invoked_workers

uninterpreted
def invoked_workers : (i : (Invoked a)) → Int

half_left_size

uninterpreted
Sizes carried by ``halves`` so the split covers the original array.
def half_left_size : (h : (Halves a)) → Int

half_right_size

uninterpreted
def half_right_size : (h : (Halves a)) → Int