pub struct StaticScheduler;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:
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<usize> = (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:
(&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:
(&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:
let data: Vec<u64> = (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:
// Cache-line aligned wrapper to prevent false sharing
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, // 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.
API
StructStaticSchedulerScheduler that uses static chunk assignment.2
*× 3Runs function once per task index, each on the thread the schedule picks.
Inherited from ParallelSchedule
fn dispatch<F>(&self, pool: &mut ThreadPool, tasks: usize, function: F) -> Result<()>
where
F: Fn(usize, ThreadInDomain) + Sync,dispatch_*_slices× 2Runs function over contiguous runs.
Inherited from ParallelSchedule
fn dispatch__slices<F>(&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.
StructDynamicSchedulerScheduler that enables dynamic work stealing.1
dispatch_slicesWork-stealing hands out one task at a time, so every run is length 1 - the C core exposes no batched dynamic entry point.
Inherited from ParallelSchedule
fn dispatch_slices<F>(&self, pool: &mut ThreadPool, tasks: usize, function: F) -> Result<()>
where
F: Fn(TasksRange, ThreadInDomain) + Sync,Use StaticScheduler when the batching is what you want.
TraitParallelScheduleHow a parallel iterator's tasks are assigned to the threads of a pool.
pub trait ParallelSchedule: CopyImplementors
impl ParallelSchedule for StaticSchedulerimpl ParallelSchedule for DynamicScheduler
TraitParallelIteratorA source of items that a pool consumes in parallel.7
pub trait ParallelIterator: SizedImplementors
impl<I, M, U> ParallelIterator for Map<I, M> where I: ParallelIterator, M: Fn(I::Item) -> U + Syncimpl<I, P> ParallelIterator for Filter<I, P> where I: ParallelIterator, P: Fn(&I::Item) -> bool + Syncimpl<'a, T> ParallelIterator for ParallelSlice<'a, T> where T: Syncimpl<'a, 'b, T, U> ParallelIterator for ParallelSliceZip<'a, 'b, T, U> where T: Sync, U: Syncimpl<'a, T> ParallelIterator for ParallelSliceMut<'a, T> where T: Sendimpl ParallelIterator for ParallelRangeimpl<T, I> ParallelIterator for ParallelExactIter<T, I> where T: Send, I: Fn(usize) -> T + Sync
ItemWhat the consumer receives for every task.
type Item;*× 8Returns the number of tasks.
fn len(&self) -> usizeis_*_empty× 8Returns true iff there are no tasks.
fn is__empty(&self) -> bool*× 8Feeds every item to consumer along with its task index and executing thread.
fn drive<S, F>(self, pool: &mut ThreadPool, schedule: S, consumer: &F) -> Result<()>
where
S: ParallelSchedule,
F: Fn(Self::Item, usize, ThreadInDomain) + Sync,drive_*× 16As drive under the static schedule.
fn drive_static<F>(self, pool: &mut ThreadPool, consumer: F) -> Result<()>
where
F: Fn(Self::Item, usize, ThreadInDomain) + Sync,| Symbol |
|---|
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 |
*× 8Transforms every item with mapper before it reaches the consumer.
fn map<M, U>(self, mapper: M) -> Map<Self, M>
where
M: Fn(Self::Item) -> U + Sync,*× 8Keeps only the items for which predicate holds.
fn filter<P>(self, predicate: P) -> Filter<Self, P>
where
P: Fn(&Self::Item) -> bool + Sync,TraitParallelIteratorExtBinds a parallel iterator to the pool that will run it.2
pub trait ParallelIteratorExt: ParallelIterator + SizedImplementors
impl<I: ParallelIterator> ParallelIteratorExt for I
with_*_pool× 8Runs on pool under the static schedule.
fn with__pool<'pool>(
self,
pool: &'pool mut ThreadPool,
) -> ParallelRunner<'pool, Self, StaticScheduler>with_*_schedule× 8Runs on pool under the given schedule.
fn with__schedule<'pool, S>(
self,
pool: &'pool mut ThreadPool,
schedule: S,
) -> ParallelRunner<'pool, Self, S>
where
S: ParallelSchedule,StructParallelRunnerA parallel iterator paired with the pool and schedule that will run it.13
pub struct ParallelRunner<'pool, I, S>with_scheduleReplaces the schedule.
pub fn with_schedule<S2>(self, schedule: S2) -> ParallelRunner<'pool, I, S2>
where
S2: ParallelSchedule,for_eachCalls function for every item with its task index and executing thread.
pub fn for_each<F>(self, function: F) -> Result<()>
where
F: Fn(I::Item, usize, ThreadInDomain) + Sync,fold_with_scratchFolds every item into the per-thread slot of scratch that runs it.
pub fn fold_with_scratch<T, F>(self, scratch: &mut [T], fold: F) -> Result<()>
where
T: Send,
F: Fn(&mut T, I::Item, usize, ThreadInDomain) + Sync,reduce_with_scratchParallel reduction with caller-provided scratch buffer.
pub fn reduce_with_scratch<T, F, C>(self, scratch: &mut [T], fold: F, combine: C) -> Result<T>
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 leastpool.threads_count()longfold- 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_fallibleFold with early-exit on error, using caller-provided scratch buffer.
pub fn fold_with_scratch_fallible<T, F, E>(
self,
scratch: &mut [T],
fold: F,
) -> Result<Result<(), E>>
where
T: Send,
F: Fn(&mut T, I::Item, usize, ThreadInDomain) -> Result<(), E> + Sync,
E: Send,Similar to fold_with_scratch, but allows the fold function to return Result.
Stops processing on the first error. Scratch buffers are indexed by thread_index.
Returns
Ok(())if all items were folded successfullyErr(E)with the first error encountered
Arguments
scratch- Per-thread accumulators, at leastpool.threads_count()longfold- 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_fallibleExecutes a fallible operation on each item, stopping at the first error.
pub fn for_each_fallible<F, E>(self, function: F) -> Result<Result<(), E>>
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 stopErr(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_*× 2Searches for the first element that matches a predicate, deterministically, by index.
pub fn find_first<P>(self, predicate: P) -> Result<Option<I::Item>>
where
I::Item: Send,
P: Fn(&I::Item) -> bool + Sync,| Symbol |
|---|
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 foundNoneif 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 foundNoneif 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
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_anySearches for any element that matches a predicate, non-deterministically.
pub fn find_any<P>(self, predicate: P) -> Result<Option<I::Item>>
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 foundNoneif 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));
anyReturns true if any item matches the predicate.
pub fn any<P>(self, predicate: P) -> Result<bool>
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);
allReturns true if all items match the predicate.
pub fn all<P>(self, predicate: P) -> Result<bool>
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);
reduceParallel reduction with a cache-aligned per-thread scratch accumulator.
pub fn reduce<T, Init, F, C>(self, init: Init, fold: F, combine: C) -> Result<T>
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<T> accumulator per thread in a node-local 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 valuefold- 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);
sumSum all items in parallel with NUMA-aware local accumulators.
pub fn sum<T>(self) -> Result<T>
where
T: Send + Sync + Default + Copy + AddAssign<I::Item> + Add<Output = T>,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);
countCount all items in parallel with NUMA-aware local counters.
pub fn count(self) -> Result<usize>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();
TraitIntoParallelIteratorConversion into a ParallelIterator.3
pub trait IntoParallelIteratorImplementors
impl<'a, T> IntoParallelIterator for &'a [T] where T: Syncimpl<'a, T> IntoParallelIterator for &'a mut [T] where T: Sendimpl IntoParallelIterator for Range<usize>
ItemThe item the resulting iterator yields.
type Item;IterThe resulting iterator.
type Iter: ParallelIterator<Item = Self::Item>;into_par_iterConverts self into a parallel iterator.
fn into_par_iter(self) -> Self::IterStructMapA parallel iterator that transforms every item of its base.
StructFilterA parallel iterator that keeps only the items of its base that satisfy a predicate.
StructParallelSliceA parallel iterator over the elements of a shared slice.3
newWraps a slice.
pub fn new(data: &'a [T]) -> Selffor_each_*× 4Calls function for every element under the static schedule.
pub fn for_each_static<F>(self, pool: &mut ThreadPool, function: F) -> Result<()>
where
T: Sync,
F: Fn(&'a T, usize, ThreadInDomain) + Sync,| Symbol |
|---|
for_each_static |
for_each_dynamic |
for_each_static |
for_each_dynamic |
zipPairs the elements of two slices by position.
pub fn zip<'b, U>(self, other: ParallelSlice<'b, U>) -> ParallelSliceZip<'a, 'b, T, U>StructParallelSliceZipA parallel iterator over the paired elements of two shared slices.
StructParallelSliceMutA parallel iterator over the elements of an exclusive slice.1
newWraps a slice.
pub fn new(data: &'a mut [T]) -> SelfStructParallelRangeA parallel iterator over a range of indices.1
newWraps a range.
pub fn new(range: Range<usize>) -> SelfStructParallelExactIterA parallel iterator that builds each item from its task index.1
newBuilds an iterator of len items, each produced by indexer from its index.
pub fn new(len: usize, indexer: I) -> SelfStructCacheAlignedCache-line aligned wrapper to prevent false sharing between threads.
pub struct CacheAligned<T>(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 bytes,
eliminating false sharing at the cost of increased memory usage.
Fields
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, DefaultFunctions
fold_with_scratchFolds every item of iterator into the per-thread slot of scratch that runs it.
pub fn fold_with_scratch<I, S, T, F>(
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,