Quickstart

ForkUnion is dead-simple to use! There is no nested parallelism, exception handling, or “future promises”; they are banned. A pool runs one dispatch at a time and refuses any other with the contended status, including one from inside its own callbacks. The thread pool itself has a few core operations:

• spawn to initialize worker threads, and
• for_threads to launch a blocking callback on all threads.

Higher-level APIs for index-addressable tasks are also available:

• for_n - for individual evenly-sized tasks,
• for_n_dynamic - for individual unevenly-sized tasks,
• for_slices - for contiguous runs of evenly-sized tasks, one per worker.

For additional flow control and tuning, following helpers are available:

• nap - to park workers on the CPU’s address monitor, waking with no syscall,
• sleep(microseconds) - to block workers in the kernel for long idle periods, with 0 waiting for the next dispatch,
• terminate - to kill the threads before the destructor is called,
• unsafe_for_threads - to broadcast a callback without blocking, returning a generation token,
• unsafe_join - to block until the completion of a broadcasted generation,
• is_complete - to poll a generation token for completion without blocking.

Every dispatch is identified by a generation token. On caller_exclusive_k pools you can dispatch, overlap your own work, poll is_complete, and join - the classic poll-then-join pattern. On caller_inclusive_k pools the calling thread owes one slice of the work, which only runs inside unsafe_join, so completion can’t be reached by polling alone. The same rule shapes the RAII guard returned by for_threads and, in C++, the for_n family: the workers start at the guard’s construction on every pool, and on inclusive pools the caller’s own slice runs at join or destruction.

On caller_exclusive_k pools thread 0 may also fork from inside its own slice, at any depth: such a fork waits for every other slice of the one enclosing it, runs on all threads, and returns complete. So unsafe_for_threads can hand a whole multi-step call to thread 0, which forks the rest of the pool as it goes, while the caller only polls a status word the call stores at its end. That is how one manager thread keeps several exclusive pools busy with no queue and no extra thread. A join on such a pool waits for whatever dispatch runs, since the call may end before its submitter stores the token.

In C++ it provides NUMA extensions. That includes the colocated_pool analog of the flat_pool and the domain_allocator_t for allocating memory in a specific memory domain. Those are out-of-the-box compatible with the higher-level APIs. Most interestingly, for Big Data applications, a higher-level distributed_pool class addresses and balances the work across all compute domains.

domain_allocator_t is backed by linux_numa_allocator_t, freebsd_numa_allocator_t, or windows_numa_allocator_t.

To integrate into your Mojo project, let Pixi build the package from source — there is no channel to add and no linker flag to pass:

[workspace]
preview = ["pixi-build"]

[dependencies]
forkunion = {git = "https://github.com/ashvardanian/ForkUnion", tag = "v4.0.0-rc.1"}

Vendoring works too and needs no manifest: Mojo resolves a source package ahead of a compiled one, so mojo run -I <path-to>/mojo is enough once libforkunion.so is on the loader path.

Then import and use in your code:

from forkunion import ComputeDomain, Library, Pool, SyncMutPointer, TasksRange, ThreadInDomain, Topology

@fieldwise_init
struct Output(ImplicitlyCopyable, TrivialRegisterPassable):
    """The output array every kernel below writes into."""

    var values: SyncMutPointer[Int64]

def double(task: Int, at: ThreadInDomain, mut output: Output):
    output.values.at(task) = Int64(task * 2)

def announce(thread_index: Int, compute_domain_index: Int, mut output: Output):
    print(t"thread {thread_index} on compute domain {compute_domain_index}")

def sweep(tasks: TasksRange, at: ThreadInDomain, mut output: Output):
    for task in tasks:
        output.values.at(task) += 1

def main() raises:
    var library = Library()
    var topology = Topology(library)
    var pool = Pool(topology, threads=4, name="demo")

    var results = List[Int64](length=1000, fill=0)
    var output = Output(SyncMutPointer(results.unsafe_ptr().unsafe_origin_cast[MutUntrackedOrigin]()))

    pool.for_threads[announce](output)          # once per worker, OpenMP-style parallel
    pool.for_n[double](1000, output)            # evenly split, OpenMP-style parallel for
    pool.for_n_dynamic[double](1000, output)    # claimed as threads free up, for uneven work
    pool.for_slices[sweep](1000, output)        # one contiguous run per worker
    print(t"results[7] = {results[7]}")         # 15: doubled to 14, then swept once

The callback is a compile-time parameter and the state it needs is a separate runtime argument. Mojo cannot hand a capturing closure to C at all, so a callback carries no captures — the scratch is the only channel, and its type and origin are inferred from the argument, leaving a call site to name only the work. Fallible calls raise Error and nothing else, and a call raises if and only if it can fail - a refusal is never folded into an Optional the caller has to interpret.

The non-blocking dispatch is a context manager, so the join cannot be forgotten and the pool cannot be torn down while its workers are still running:

with pool.unsafe_for_threads[work](scratch) as dispatch:
    while not dispatch.is_complete(): # exclusive pools only
        prepare_next_batch()

The Topology handle reports the hardware topology, to size and place work:

var topology = Topology(library)
for index in range(topology.compute_domains_count()):
    var domain = ComputeDomain(index)
    var cores = topology.logical_cores_count_in(domain)
    var level = topology.compute_level_in(domain)
    print(t"domain {index}: {cores} cores, level {level}, allocate on {topology.local_memory_of(domain)}")

Handles release themselves, so there is no defer to forget — though Mojo destroys a value after its last use rather than at the end of the scope, which is earlier than a Zig defer fires.

Unlike max.algorithm.parallelize, which takes every hardware thread and offers no way to ask for fewer, ForkUnion’s pool width is an argument. On a machine you are sharing, that is the difference between a speedup and a slowdown.