ah-parallel.H File Reference

Parallel functional programming operations using ThreadPool. More...

#include <vector>
#include <atomic>
#include <optional>
#include <algorithm>
#include <numeric>
#include <type_traits>
#include <iterator>
#include <functional>
#include <thread_pool.H>

Include dependency graph for ah-parallel.H:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Classes
struct	Aleph::parallel_zip_detail::ContainerHolder< Container >
	Holder for converted containers (either pointer or unique_ptr to vector). More...

Namespaces
namespace	Aleph
	Main namespace for Aleph-w library functions.
namespace	Aleph::parallel_detail
namespace	Aleph::parallel_zip_detail

Functions
size_t	Aleph::parallel_detail::chunk_size (const size_t n, const size_t num_threads, const size_t min_chunk=64)
	Calculate optimal chunk size based on data size and thread count.
template<typename Container >
constexpr bool	Aleph::parallel_detail::has_random_access ()
	Check if container supports random access.
template<typename Container >
auto	Aleph::parallel_detail::ensure_random_access (const Container &c)
	For containers with random access, just return a pointer to it For non-random access, copy to vector.
template<typename T >
decltype(auto)	Aleph::parallel_detail::deref (T &&ptr)
	Get reference from pointer or unique_ptr.
template<typename ResultT = void, typename Container , typename Op >
auto	Aleph::pmaps (ThreadPool &pool, const Container &c, Op op, size_t chunk_size=0)
	Parallel map operation.
template<typename Container , typename Pred >
auto	Aleph::pfilter (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel filter operation.
template<typename T , typename Container , typename BinaryOp >
T	Aleph::pfoldl (ThreadPool &pool, const Container &c, T init, BinaryOp op, size_t chunk_size=0)
	Parallel left fold (reduce).
template<typename Container , typename Op >
void	Aleph::pfor_each (ThreadPool &pool, Container &c, Op op, size_t chunk_size=0)
	Parallel for_each operation.
template<typename Container , typename Op >
void	Aleph::pfor_each (ThreadPool &pool, const Container &c, Op op, size_t chunk_size=0)
	This is an overloaded member function, provided for convenience. It differs from the above function only in what argument(s) it accepts.Parallel for_each for const containers.
template<typename Container , typename Pred >
bool	Aleph::pall (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel all predicate (short-circuit).
template<typename Container , typename Pred >
bool	Aleph::pexists (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel exists predicate (short-circuit).
template<typename Container , typename Pred >
bool	Aleph::pnone (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel none predicate.
template<typename Container , typename Pred >
size_t	Aleph::pcount_if (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel count_if operation.
template<typename Container , typename Pred >
std::optional< size_t >	Aleph::pfind (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel find operation (returns index).
template<typename Container , typename Pred >
auto	Aleph::pfind_value (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel find with value return.
template<typename Container , typename T = std::decay_t<decltype(*std::begin(std::declval<Container>()))>>
T	Aleph::psum (ThreadPool &pool, const Container &c, T init=T{}, size_t chunk_size=0)
	Parallel sum of elements.
template<typename Container , typename T = std::decay_t<decltype(*std::begin(std::declval<Container>()))>>
T	Aleph::pproduct (ThreadPool &pool, const Container &c, T init=T{1}, size_t chunk_size=0)
	Parallel product of elements.
template<typename Container >
auto	Aleph::pmin (ThreadPool &pool, const Container &c, size_t chunk_size=0)
	Parallel minimum element.
template<typename Container >
auto	Aleph::pmax (ThreadPool &pool, const Container &c, size_t chunk_size=0)
	Parallel maximum element.
template<typename Container >
auto	Aleph::pminmax (ThreadPool &pool, const Container &c, size_t chunk_size=0)
	Parallel min and max elements.
template<typename Container , typename Compare = std::less<>>
void	Aleph::psort (ThreadPool &pool, Container &c, Compare cmp=Compare{}, const size_t min_parallel_size=1024)
	Parallel sort (in-place).
template<typename Container1 , typename Container2 , typename Op >
void	Aleph::pzip_for_each (ThreadPool &pool, const Container1 &c1, const Container2 &c2, Op op, size_t chunk_size=0)
	Parallel zip + for_each.
template<typename Container1 , typename Container2 , typename Op >
auto	Aleph::pzip_maps (ThreadPool &pool, const Container1 &c1, const Container2 &c2, Op op, size_t chunk_size=0)
	Parallel zip + map.
template<typename Container1 , typename Container2 , typename T , typename Op >
T	Aleph::pzip_foldl (ThreadPool &pool, const Container1 &c1, const Container2 &c2, T init, Op op, size_t chunk_size=0)
	Parallel zip + fold.
template<typename Container , typename Pred >
auto	Aleph::ppartition (ThreadPool &pool, const Container &c, Pred pred, size_t chunk_size=0)
	Parallel partition (stable).
template<typename... Holders, size_t... Is>
size_t	Aleph::parallel_zip_detail::min_holder_size_impl (const std::tuple< Holders... > &holders, std::index_sequence< Is... >)
	Get minimum size from tuple of holders - always O(1) per holder.
template<typename... Holders>
size_t	Aleph::parallel_zip_detail::min_holder_size (const std::tuple< Holders... > &holders)
template<typename... Holders, size_t... Is>
auto	Aleph::parallel_zip_detail::make_iterators_at (size_t offset, const std::tuple< Holders... > &holders, std::index_sequence< Is... >)
	Create tuple of iterators at given offset.
template<typename... Iters, size_t... Is>
void	Aleph::parallel_zip_detail::advance_all_iters (std::tuple< Iters... > &iters, std::index_sequence< Is... >)
	Advance all iterators in tuple.
template<typename... Iters, size_t... Is>
auto	Aleph::parallel_zip_detail::deref_all_iters (const std::tuple< Iters... > &iters, std::index_sequence< Is... >)
	Dereference all iterators and make tuple.
template<typename Op , typename... Containers>
void	Aleph::pzip_for_each_n (ThreadPool &pool, Op op, const Containers &... cs)
	Parallel for_each over N zipped containers (variadic).
template<typename Op , typename... Containers>
auto	Aleph::pzip_maps_n (ThreadPool &pool, Op op, const Containers &... cs)
	Parallel map over N zipped containers (variadic).
template<typename T , typename Op , typename Combiner , typename... Containers>
T	Aleph::pzip_foldl_n (ThreadPool &pool, T init, Op op, Combiner combiner, const Containers &... cs)
	Parallel fold/reduce over N zipped containers (variadic).
template<typename Pred , typename... Containers>
bool	Aleph::pzip_all_n (ThreadPool &pool, Pred pred, const Containers &... cs)
	Parallel all predicate over N zipped containers (variadic).
template<typename Pred , typename... Containers>
bool	Aleph::pzip_exists_n (ThreadPool &pool, Pred pred, const Containers &... cs)
	Parallel exists predicate over N zipped containers (variadic).
template<typename Pred , typename... Containers>
size_t	Aleph::pzip_count_if_n (ThreadPool &pool, Pred pred, const Containers &... cs)
	Parallel count over N zipped containers (variadic).
template<typename Container , typename Op >
void	Aleph::penumerate_for_each (ThreadPool &pool, Container &c, Op op, size_t chunk_size=0)
	Parallel for_each with index (enumerate).
template<typename Container , typename Op >
void	Aleph::penumerate_for_each (ThreadPool &pool, const Container &c, Op op, size_t chunk_size=0)
	This is an overloaded member function, provided for convenience. It differs from the above function only in what argument(s) it accepts.Parallel enumerate_for_each for const containers.
template<typename Container , typename Op >
auto	Aleph::penumerate_maps (ThreadPool &pool, const Container &c, Op op, size_t chunk_size=0)
	Parallel enumerate with map.
ThreadPool &	Aleph::parallel_default_pool ()
	Global default pool for parallel operations.

Detailed Description

Parallel functional programming operations using ThreadPool.

This file provides parallel versions of common functional operations like map, filter, fold, for_each, all, exists, etc. These operations leverage multi-core processors to accelerate data processing.

All functions in this file require a ThreadPool instance and operate on STL-compatible containers (std::vector, std::deque, std::array) or Aleph-w containers with random access (Array, DynArray).

Key Features

• Zero-copy where possible: Operations on references avoid copying
• Automatic chunking: Work is divided optimally across threads
• Short-circuit evaluation: pall, pexists, pfind stop early
• Exception propagation: Exceptions from any thread are captured

Available Operations

Function	Description	Time Complexity
pmaps	Parallel map	O(n/p)
pfilter	Parallel filter	O(n/p)
pfoldl	Parallel reduce	O(n/p + p)
pfor_each	Parallel for_each	O(n/p)
pall	Parallel all predicate	O(n/p) best, O(n) worst
pexists	Parallel exists	O(1) best, O(n/p) worst
pnone	Parallel none	O(n/p)
pcount_if	Parallel count	O(n/p)
pfind	Parallel find	O(1) best, O(n/p) worst
psum	Parallel sum	O(n/p + p)
pproduct	Parallel product	O(n/p + p)
pminmax	Parallel min/max	O(n/p + p)

Where n = elements, p = threads.

Usage Examples

#include <ah-parallel.H>
#include <vector>
 
using namespace Aleph;
 
ThreadPool pool(8);  // 8 worker threads
 
std::vector<int> data(1000000);
std::iota(data.begin(), data.end(), 0);
 
// Parallel map: square each element
auto squares = pmaps<long long>(pool, data, [](int x) {
  return static_cast<long long>(x) * x;
});
 
// Parallel filter: keep even numbers
auto evens = pfilter(pool, data, [](int x) { return x % 2 == 0; });
 
// Parallel sum
long long total = psum(pool, data);
 
// Parallel find
auto it = pfind(pool, data, [](int x) { return x == 500000; });

Thread Safety

• All functions are thread-safe with respect to the ThreadPool
• Input containers should not be modified during parallel operations
• Output containers are constructed locally and returned by value

Author

Leandro Rabindranath León

Definition in file ah-parallel.H.