# Parallel Iterators > ForkUnion for Rust For Rayon-style ergonomics, use the parallel iterator API with the `prelude`. Unlike Rayon, ForkUnion's parallel iterators don't depend on the global state and allow explicit control over the thread pool and scheduling strategy. For statically shaped workloads, the default static scheduling is more efficient: ```rust use forkunion as fu; use forkunion::prelude::*; let topology = fu::Topology::new().expect("Failed to detect hardware topology"); let mut pool = fu::spawn(&topology, 4); let mut data: Vec = (0..1000).collect(); (&data[..]) .into_par_iter() .with_pool(&mut pool) .for_each(|value, _task, _at| { println!("Value: {}", value); })?; ``` For dynamic work-stealing, use `with_schedule` with `DynamicScheduler`: ```rust (&mut data[..]) .into_par_iter() .with_schedule(&mut pool, DynamicScheduler) .for_each(|value, _task, _at| { *value *= 2; })?; ``` This easily composes with other iterator adaptors, like `map`, `filter`, and `zip`: ```rust (&data[..]) .into_par_iter() .filter(|&x| x % 2 == 0) .map(|x| x * x) .with_pool(&mut pool) .for_each(|value, _task, _at| { println!("Squared even: {}", value); })?; ``` For parallel reductions, ForkUnion provides Rayon-like convenience methods with automatic cache-aligned scratch allocation: ```rust let data: Vec = (0..1_000_000).map(|i| i as u64).collect(); // Sum all elements with automatic scratch allocation let total: u64 = (&data[..]) .into_par_iter() .with_pool(&mut pool) .sum()?; // Count elements matching a predicate let evens = (&data[..]) .into_par_iter() .filter(|&x| x % 2 == 0) .with_pool(&mut pool) .count()?; // Custom reduction (maximum) let maximum = (&data[..]) .into_par_iter() .with_pool(&mut pool) .reduce( || 0u64, // initial value |acc, value, _, _| *acc = (*acc).max(*value), // fold function |a, b| a.max(b) // combine function )?; ``` For manual control over scratch allocation, use `reduce_with_scratch`: ```rust // Cache-line aligned wrapper to prevent false sharing let mut scratch: Vec> = (0..pool.threads_count()).map(|_| CacheAligned(0)).collect(); let total = (&data[..]) .into_par_iter() .with_pool(&mut pool) .reduce_with_scratch( scratch.as_mut_slice(), |acc, value, _, _| acc.0 += *value, // fold |a, b| a.0 += b.0 // combine in-place )?; ``` Beyond reductions, the iterators offer short-circuiting searches - `find_first` and `find_last` for the deterministic lowest- or highest-index match, `find_any` for whichever match is found first with cooperative cancellation, and `any` / `all` for boolean predicates that stop the moment the answer is known. Fallible bodies get `for_each_fallible` and `fold_with_scratch_fallible`, which propagate the first error and signal the other workers to stop. Their two answers stay apart: the outer `Result` reports whether the dispatch ran, the inner one whether the caller's own work succeeded, so the caller's error type carries no constraint. This Rayon-style layer is __Rust-only__; C, C++, Zig, and Mojo expose the pool primitives `for_threads`, `for_n`, `for_n_dynamic`, and `for_slices` directly. Zig additionally mirrors Rust's `for_slices_mut` as `forSlicesMut`, which hands each thread a disjoint sub-slice rather than the `TasksRange` its indices would form. `use forkunion::StaticScheduler;` `use forkunion::DynamicScheduler;` `use forkunion::ParallelSchedule;` `use forkunion::ParallelIterator;` `use forkunion::ParallelIteratorExt;` `use forkunion::ParallelRunner;` `use forkunion::IntoParallelIterator;` `use forkunion::Map;` `use forkunion::Filter;` `use forkunion::ParallelSlice;` `use forkunion::ParallelSliceZip;` `use forkunion::ParallelSliceMut;` `use forkunion::ParallelRange;` `use forkunion::ParallelExactIter;` `use forkunion::CacheAligned;` `use forkunion::*;` ## StaticScheduler Scheduler that uses static chunk assignment. ```rust pub struct StaticScheduler; ``` Implements: `Clone`, `Copy`, `Debug`, `[ParallelSchedule](#ParallelSchedule)` ## * Runs `function` once per task index, each on the thread the schedule picks. ```rust fn dispatch(&self, pool: &mut ThreadPool, tasks: usize, function: F) -> Result<()> where F: Fn(usize, ThreadInDomain) + Sync, ``` ## dispatch_*_slices Runs `function` over contiguous runs. ```rust fn dispatch__slices(&self, pool: &mut ThreadPool, tasks: usize, function: F) -> Result<()> where F: Fn(TasksRange, ThreadInDomain) + Sync, ``` The count is a real run length under static scheduling and always 1 under dynamic, which has no batched entry point. ## DynamicScheduler Scheduler that enables dynamic work stealing. ```rust pub struct DynamicScheduler; ``` Implements: `Clone`, `Copy`, `Debug`, `[ParallelSchedule](#ParallelSchedule)` ## dispatch_slices Work-stealing hands out one task at a time, so every run is length 1 - the C core exposes no batched dynamic entry point. ```rust fn dispatch_slices(&self, pool: &mut ThreadPool, tasks: usize, function: F) -> Result<()> where F: Fn(TasksRange, ThreadInDomain) + Sync, ``` Use [`StaticScheduler`](#StaticScheduler) when the batching is what you want. ## ParallelSchedule How a parallel iterator's tasks are assigned to the threads of a pool. ```rust pub trait ParallelSchedule: Copy ``` Implementors: - `impl [ParallelSchedule](#ParallelSchedule) for [StaticScheduler](#StaticScheduler)` - `impl [ParallelSchedule](#ParallelSchedule) for [DynamicScheduler](#DynamicScheduler)` ## ParallelIterator A source of items that a pool consumes in parallel. ```rust pub trait ParallelIterator: Sized ``` Implementors: - `impl [ParallelIterator](#ParallelIterator) for [Map](#Map) where I: [ParallelIterator](#ParallelIterator), M: Fn(I::Item) -> U + Sync` - `impl [ParallelIterator](#ParallelIterator) for [Filter](#Filter) where I: [ParallelIterator](#ParallelIterator), P: Fn(&I::Item) -> bool + Sync` - `impl<'a, T> [ParallelIterator](#ParallelIterator) for [ParallelSlice](#ParallelSlice)<'a, T> where T: Sync` - `impl<'a, 'b, T, U> [ParallelIterator](#ParallelIterator) for [ParallelSliceZip](#ParallelSliceZip)<'a, 'b, T, U> where T: Sync, U: Sync` - `impl<'a, T> [ParallelIterator](#ParallelIterator) for [ParallelSliceMut](#ParallelSliceMut)<'a, T> where T: Send` - `impl [ParallelIterator](#ParallelIterator) for [ParallelRange](#ParallelRange)` - `impl [ParallelIterator](#ParallelIterator) for [ParallelExactIter](#ParallelExactIter) where T: Send, I: Fn(usize) -> T + Sync` ## Item What the consumer receives for every task. ```rust type Item; ``` ## * Returns the number of tasks. ```rust fn len(&self) -> usize ``` ## is_*_empty Returns true iff there are no tasks. ```rust fn is__empty(&self) -> bool ``` ## * Feeds every item to `consumer` along with its task index and executing thread. ```rust fn drive(self, pool: &mut ThreadPool, schedule: S, consumer: &F) -> Result<()> where S: ParallelSchedule, F: Fn(Self::Item, usize, ThreadInDomain) + Sync, ``` ## drive_* As [`drive`](#drive) under the static schedule. ```rust fn drive_static(self, pool: &mut ThreadPool, consumer: F) -> Result<()> where F: Fn(Self::Item, usize, ThreadInDomain) + Sync, ``` - `drive_static` - `drive_dynamic` - `drive_static` - `drive_dynamic` - `drive_static` - `drive_dynamic` - `drive_static` - `drive_dynamic` - `drive_static` - `drive_dynamic` - `drive_static` - `drive_dynamic` - `drive_static` - `drive_dynamic` - `drive_static` - `drive_dynamic` ## * Transforms every item with `mapper` before it reaches the consumer. ```rust fn map(self, mapper: M) -> Map where M: Fn(Self::Item) -> U + Sync, ``` ## * Keeps only the items for which `predicate` holds. ```rust fn filter

(self, predicate: P) -> Filter where P: Fn(&Self::Item) -> bool + Sync, ``` ## ParallelIteratorExt Binds a parallel iterator to the pool that will run it. ```rust pub trait ParallelIteratorExt: ParallelIterator + Sized ``` Implementors: - `impl [ParallelIteratorExt](#ParallelIteratorExt) for I` ## with_*_pool Runs on `pool` under the static schedule. ```rust fn with__pool<'pool>( self, pool: &'pool mut ThreadPool, ) -> ParallelRunner<'pool, Self, StaticScheduler> ``` ## with_*_schedule Runs on `pool` under the given schedule. ```rust fn with__schedule<'pool, S>( self, pool: &'pool mut ThreadPool, schedule: S, ) -> ParallelRunner<'pool, Self, S> where S: ParallelSchedule, ``` ## ParallelRunner A parallel iterator paired with the pool and schedule that will run it. ```rust pub struct ParallelRunner<'pool, I, S> ``` ## with_schedule Replaces the schedule. ```rust pub fn with_schedule(self, schedule: S2) -> ParallelRunner<'pool, I, S2> where S2: ParallelSchedule, ``` ## for_each Calls `function` for every item with its task index and executing thread. ```rust pub fn for_each(self, function: F) -> Result<()> where F: Fn(I::Item, usize, ThreadInDomain) + Sync, ``` ## fold_with_scratch Folds every item into the per-thread slot of `scratch` that runs it. ```rust pub fn fold_with_scratch(self, scratch: &mut [T], fold: F) -> Result<()> where T: Send, F: Fn(&mut T, I::Item, usize, ThreadInDomain) + Sync, ``` ## reduce_with_scratch Parallel reduction with caller-provided scratch buffer. ```rust pub fn reduce_with_scratch(self, scratch: &mut [T], fold: F, combine: C) -> Result where T: Send + Default, F: Fn(&mut T, I::Item, usize, ThreadInDomain) + Sync, C: Fn(&mut T, T), ``` Reduces items in parallel by folding into per-thread accumulators, then combining results on the caller thread. Uses cache-aligned scratch to prevent false sharing. Indexes by thread_index, which works with dynamic scheduling. #### Returns The final reduced value of type `T` Arguments: - `scratch` - Per-thread accumulators, at least `pool.threads_count()` long - `fold` - Function to accumulate items: `fn(&mut T, I::Item, usize, ThreadInDomain)` - `combine` - Function to merge two accumulators: `fn(T, T) -> T` Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = (0..1000).collect(); let mut scratch: Vec<CacheAligned<u64>> = (0..pool.threads_count()).map(|_| CacheAligned(0)).collect(); let total = (&data[..]).into_par_iter().with_pool(&mut pool) .reduce_with_scratch( scratch.as_mut_slice(), |acc, value, _, _| acc.0 += *value, |a, b| a.0 += b.0, ); ``` ## fold_with_scratch_fallible Fold with early-exit on error, using caller-provided scratch buffer. ```rust pub fn fold_with_scratch_fallible( self, scratch: &mut [T], fold: F, ) -> Result> where T: Send, F: Fn(&mut T, I::Item, usize, ThreadInDomain) -> Result<(), E> + Sync, E: Send, ``` Similar to [`fold_with_scratch`](#fold_with_scratch), but allows the fold function to return [`Result`](https://www.unum.cloud/docs/forkunion/rust/quickstart.md#Result). Stops processing on the first error. Scratch buffers are indexed by `thread_index`. #### Returns - `Ok(())` if all items were folded successfully - `Err(E)` with the first error encountered Arguments: - `scratch` - Per-thread accumulators, at least `pool.threads_count()` long - `fold` - Fallible fold function: `fn(&mut T, I::Item, usize, ThreadInDomain) → core::result::Result<(), E>` Examples: ``` use forkunion::*; fn checked_add(acc: &mut u64, value: &u64) -> core::result::Result<(), &'static str> { *acc = acc.checked_add(*value).ok_or("overflow")?; Ok(()) } let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = (1..100).collect(); let mut scratch: Vec<CacheAligned<u64>> = (0..pool.threads_count()).map(|_| CacheAligned(0)).collect(); let result = (&data[..]) .into_par_iter() .with_pool(&mut pool) .fold_with_scratch_fallible(scratch.as_mut_slice(), |acc, value, _, _| { checked_add(&mut acc.0, value) }); assert!(result.is_ok()); ``` ## for_each_fallible Executes a fallible operation on each item, stopping at the first error. ```rust pub fn for_each_fallible(self, function: F) -> Result> where F: Fn(I::Item, usize, ThreadInDomain) -> Result<(), E> + Sync, E: Send, ``` Uses cooperative cancellation: once an error occurs, no further items are processed. Items already "in flight" may still complete, but new items won't start processing. #### Returns - `Ok(())` if all items were processed successfully or were skipped after stop - `Err(E)` with the first error encountered #### Performance Overhead is one atomic load per item (~2% in compute-bound workloads). The atomic swap on error is negligible as it happens at most once. Examples: ``` use forkunion::*; fn validate(x: &u64) -> core::result::Result<(), &'static str> { if *x < 100 { Ok(()) } else { Err("value too large") } } let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = (0..50).collect(); let result = (&data[..]) .into_par_iter() .with_pool(&mut pool) .for_each_fallible(|x, _, _| validate(x)); assert!(result.is_ok()); ``` ## find_* Searches for the first element that matches a predicate, deterministically, by index. ```rust pub fn find_first

(self, predicate: P) -> Result> where I::Item: Send, P: Fn(&I::Item) -> bool + Sync, ``` - `find_first` - `find_last` Returns the element with the smallest `task_index` among all matches. #### Returns - `Some(item)` with the lowest index if any match was found - `None` if no item matched or the iterator was empty Returns the element with the largest `task_index` among all matches. #### Returns - `Some(item)` with the highest index if any match was found - `None` if no item matched or the iterator was empty Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = vec![10, 20, 30, 20, 10]; let found = (&data[..]) .into_par_iter() .with_pool(&mut pool) .find_first(|&&x| x == 20) .unwrap(); assert_eq!(found, Some(&20)); // Index 1, not 3 ``` Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = vec![10, 20, 30, 20, 10]; let found = (&data[..]) .into_par_iter() .with_pool(&mut pool) .find_last(|&&x| x == 20) .unwrap(); assert_eq!(found, Some(&20)); // Index 3, not 1 ``` ## find_any Searches for any element that matches a predicate, non-deterministically. ```rust pub fn find_any

(self, predicate: P) -> Result> where I::Item: Send, P: Fn(&I::Item) -> bool + Sync, ``` Uses cooperative cancellation: once a match is found, no further items are processed. If multiple items match, any one of them may be returned. #### Returns - `Some(item)` if a matching item was found - `None` if no item matched or the iterator was empty Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = (0..1000).collect(); let found = (&data[..]) .into_par_iter() .with_pool(&mut pool) .find_any(|&&x| x == 42) .unwrap(); assert_eq!(found, Some(&42)); ``` ## any Returns `true` if any item matches the predicate. ```rust pub fn any

(self, predicate: P) -> Result where I::Item: Send, P: Fn(&I::Item) -> bool + Sync, ``` Stops searching after the first match is found. Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = (0..1000).collect(); let has_large = (&data[..]) .into_par_iter() .with_pool(&mut pool) .any(|&&x| x > 500) .unwrap(); assert!(has_large); ``` ## all Returns `true` if all items match the predicate. ```rust pub fn all

(self, predicate: P) -> Result where I::Item: Send, P: Fn(&I::Item) -> bool + Sync, ``` Stops searching after the first non-match is found. Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = (0..100).collect(); let all_small = (&data[..]) .into_par_iter() .with_pool(&mut pool) .all(|&&x| x < 200) .unwrap(); assert!(all_small); ``` ## reduce Parallel reduction with a cache-aligned per-thread scratch accumulator. ```rust pub fn reduce(self, init: Init, fold: F, combine: C) -> Result where Init: Fn() -> T + Sync, T: Send + Sync + Default, F: Fn(&mut T, I::Item, usize, ThreadInDomain) + Sync, C: Fn(T, T) -> T, ``` Allocates one `CacheAligned` accumulator per thread in a node-local [`PinnedVec`](https://www.unum.cloud/docs/forkunion/rust/numa.md#PinnedVec), so threads fold into local memory before the serial combine. Nearly identical to Rayon's reduce API, just requires explicit pool. Arguments: - `init` - Function to create initial accumulator value - `fold` - Function to accumulate items: `fn(&mut T, I::Item, usize, ThreadInDomain)` - `combine` - Function to merge two accumulators: `fn(T, T) -> T` Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<u64> = (0..1000).collect(); let total = (&data[..]).into_par_iter().with_pool(&mut pool) .reduce(|| 0, |acc, value, _, _| *acc += *value, |a, b| a + b); ``` ## sum Sum all items in parallel with NUMA-aware local accumulators. ```rust pub fn sum(self) -> Result where T: Send + Sync + Default + Copy + AddAssign + Add, ``` Works for owned values, such as usize and u64, and for references, such as &u64. Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data = vec![1u64, 2, 3, 4, 5]; let sum: u64 = (&data[..]).into_par_iter().with_pool(&mut pool).sum().unwrap(); assert_eq!(sum, 15); ``` ## count Count all items in parallel with NUMA-aware local counters. ```rust pub fn count(self) -> Result ``` Examples: ``` use forkunion::*; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<usize> = (0..1000).collect(); let count = (&data[..]).into_par_iter().with_pool(&mut pool).count(); ``` ## IntoParallelIterator Conversion into a [`ParallelIterator`](#ParallelIterator). ```rust pub trait IntoParallelIterator ``` Implementors: - `impl<'a, T> [IntoParallelIterator](#IntoParallelIterator) for &'a [T] where T: Sync` - `impl<'a, T> [IntoParallelIterator](#IntoParallelIterator) for &'a mut [T] where T: Send` - `impl [IntoParallelIterator](#IntoParallelIterator) for Range` ## Item The item the resulting iterator yields. ```rust type Item; ``` ## Iter The resulting iterator. ```rust type Iter: ParallelIterator; ``` ## into_par_iter Converts `self` into a parallel iterator. ```rust fn into_par_iter(self) -> Self::Iter ``` ## Map A parallel iterator that transforms every item of its base. ```rust pub struct Map ``` Implements: `[ParallelIteratorExt](#ParallelIteratorExt)`, `[ParallelIterator](#ParallelIterator)` ## Filter A parallel iterator that keeps only the items of its base that satisfy a predicate. ```rust pub struct Filter ``` Implements: `[ParallelIteratorExt](#ParallelIteratorExt)`, `[ParallelIterator](#ParallelIterator)` ## ParallelSlice A parallel iterator over the elements of a shared slice. ```rust pub struct ParallelSlice<'a, T> ``` Implements: `[ParallelIteratorExt](#ParallelIteratorExt)`, `Clone`, `Copy`, `[ParallelIterator](#ParallelIterator)` ## new Wraps a slice. ```rust pub fn new(data: &'a [T]) -> Self ``` ## for_each_* Calls `function` for every element under the static schedule. ```rust pub fn for_each_static(self, pool: &mut ThreadPool, function: F) -> Result<()> where T: Sync, F: Fn(&'a T, usize, ThreadInDomain) + Sync, ``` - `for_each_static` - `for_each_dynamic` - `for_each_static` - `for_each_dynamic` ## zip Pairs the elements of two slices by position. ```rust pub fn zip<'b, U>(self, other: ParallelSlice<'b, U>) -> ParallelSliceZip<'a, 'b, T, U> ``` ## ParallelSliceZip A parallel iterator over the paired elements of two shared slices. ```rust pub struct ParallelSliceZip<'a, 'b, T, U> ``` Implements: `[ParallelIteratorExt](#ParallelIteratorExt)`, `[ParallelIterator](#ParallelIterator)` ## ParallelSliceMut A parallel iterator over the elements of an exclusive slice. ```rust pub struct ParallelSliceMut<'a, T> ``` Implements: `[ParallelIteratorExt](#ParallelIteratorExt)`, `[ParallelIterator](#ParallelIterator)` ## new Wraps a slice. ```rust pub fn new(data: &'a mut [T]) -> Self ``` ## ParallelRange A parallel iterator over a range of indices. ```rust pub struct ParallelRange ``` Implements: `[ParallelIteratorExt](#ParallelIteratorExt)`, `Clone`, `[ParallelIterator](#ParallelIterator)` ## new Wraps a range. ```rust pub fn new(range: Range) -> Self ``` ## ParallelExactIter A parallel iterator that builds each item from its task index. ```rust pub struct ParallelExactIter ``` Implements: `[ParallelIteratorExt](#ParallelIteratorExt)`, `[ParallelIterator](#ParallelIterator)` ## new Builds an iterator of [`len`](#len) items, each produced by `indexer` from its index. ```rust pub fn new(len: usize, indexer: I) -> Self ``` ## CacheAligned Cache-line aligned wrapper to prevent false sharing between threads. ```rust pub struct CacheAligned(pub T); ``` When multiple threads access separate data that resides on the same cache line, modifications by one thread invalidate the cache line for all others, causing performance degradation known as "false sharing". This wrapper ensures each wrapped value occupies its own [`DEFAULT_ALIGNMENT`](https://www.unum.cloud/docs/forkunion/rust/quickstart.md#DEFAULT_ALIGNMENT) bytes, eliminating false sharing at the cost of increased memory usage. - `0` `T` Examples: ``` use forkunion::{CacheAligned, ThreadPool, Topology}; let topology = Topology::new().unwrap(); let mut pool = ThreadPool::spawn(&topology, 4).unwrap(); let data: Vec<usize> = (0..1000).collect(); // Each thread gets its own cache-aligned accumulator let mut scratch: Vec<CacheAligned<usize>> = (0..pool.threads_count()).map(|_| CacheAligned(0)).collect(); // No false sharing during parallel reduction for value in &data { let tid = *value % pool.threads_count(); scratch[tid].0 += value; } let total: usize = scratch.iter().map(|a| a.0).sum(); ``` Implements: `Clone`, `Copy`, `Debug`, `Default` ## fold_with_scratch Folds every item of `iterator` into the per-thread slot of `scratch` that runs it. ```rust pub fn fold_with_scratch( pool: &mut ThreadPool, iterator: I, schedule: S, scratch: &mut [T], fold: F, ) -> Result<()> where I: ParallelIterator, S: ParallelSchedule, T: Send, F: Fn(&mut T, I::Item, usize, ThreadInDomain) + Sync, ```