Threads and Concurrency

Threading utilities (thread pools, worker pools, timed event queues) and synchronization helpers. More...

Files
file	concurrency_utils.H
	Modern synchronization helpers for channels, shared state, and small producer-consumer queues.
file	experimental_async.H
	Opt-in experimental coroutine-friendly bridge over ThreadPool.
file	thread_pool.H
	A modern, efficient thread pool for parallel task execution.
file	timeoutQueue.H
	Priority queue for scheduling timed events.
file	useCondVar.H
	Legacy wrapper class for POSIX condition variables.
file	useMutex.H
	Legacy RAII-style wrapper for POSIX mutexes.
file	worker_pool.H
	General-purpose worker thread pool for parallel task execution.

Classes
class	UseCondVar
	Legacy wrapper class for POSIX condition variables. More...
class	UseMutex
	Legacy RAII-style mutex lock guard. More...

Functions
	UseCondVar::UseCondVar (pthread_cond_t *c, pthread_mutex_t *m)
	Construct wrapper for condition variable and mutex.

Detailed Description

Threading utilities (thread pools, worker pools, timed event queues) and synchronization helpers.

This module groups Aleph-w facilities related to concurrency and thread-based execution.

It covers:

• thread_pool.H (Aleph::ThreadPool): modern task-based thread pool returning std::future.
• thread_pool.H (Aleph::TaskGroup): small structured-concurrency helper on top of ThreadPool.
• thread_pool.H (Aleph::CancellationSource, Aleph::CancellationToken): cooperative cancellation primitives.
• thread_pool.H (Aleph::ParallelOptions): shared tuning/cancellation options for parallel helpers.
• thread_pool.H (parallel_invoke, pscan, pexclusive_scan, pmerge): foundational parallel building blocks for composing higher-level algorithms.
• concurrency_utils.H (Aleph::bounded_channel, Aleph::synchronized, Aleph::rw_synchronized, Aleph::spsc_queue): modern coordination helpers for channels, shared state, and SPSC handoff.
• worker_pool.H (WorkersSet): worker-function pool for processing queued void* work items.
• timeoutQueue.H (TimeoutQueue): timed event scheduler running a background thread.
• useMutex.H (UseMutex): legacy POSIX mutex wrapper.
• useCondVar.H (UseCondVar): legacy POSIX condition variable wrapper.

ThreadPool (task-based)

See thread_pool.H and the runnable example thread_pool_example.cc.

Typical usage pattern:

#include <thread_pool.H>
 
Aleph::ThreadPool pool(4);
 
auto f = pool.enqueue([](int x) { return x * x; }, 5);
int v = f.get();
(void) v;

Structured task groups:

#include <thread_pool.H>
 
Aleph::ThreadPool pool(4);
Aleph::TaskGroup group(pool);
 
group.launch([] { /* task A */ });
group.launch([] { /* task B */ });
group.wait();

Cooperative cancellation:

Aleph::CancellationSource cancel;
auto token = cancel.token();
 
pool.enqueue([token] {
  while (not token.stop_requested())
    do_some_work_step();
});
 
cancel.request_cancel();

Parallel algorithms can now be configured through ParallelOptions, including min_size, chunk_size, max_tasks, pool, and cancel_token.

This applies to the main helpers in ah-parallel.H, including map/filter/fold, parallel sort, scan/merge wrappers, zip, and enumerate operations. Existing ThreadPool& overloads remain available as compatibility wrappers.

If cancel_token is signaled and the algorithm observes it, the operation throws Aleph::operation_canceled.

Foundational building blocks now available in thread_pool.H:

Aleph::parallel_invoke(options, [] { /* branch A */ }, [] { /* branch B */ });
Aleph::pscan(first, last, out, std::plus<int>{}, options);
Aleph::pexclusive_scan(first, last, out, 0, std::plus<int>{}, options);
Aleph::pmerge(a_first, a_last, b_first, b_last, out, std::less<int>{}, options);

Modern coordination helpers in concurrency_utils.H:

#include <concurrency_utils.H>
 
Aleph::bounded_channel<int> ch(16);
Aleph::synchronized<std::vector<int>> items(std::in_place);
Aleph::rw_synchronized<int> shared_sum(0);
Aleph::spsc_queue<int> handoff(32);
 
ch.send(10);
if (auto item = ch.recv())
  items.with_lock([&](auto &out) { out.push_back(*item); });
 
shared_sum.with_write_lock([](int &value) { value += 5; });
const int snapshot = shared_sum.with_read_lock([](const int &value) {
  return value;
});
(void) snapshot;
 
handoff.try_push(1);
auto popped = handoff.try_pop();

bounded_channel<T> also offers cancellation-aware blocking operations:

Aleph::CancellationSource cancel;
Aleph::bounded_channel<int> jobs(8);
 
try
  {
    jobs.send(42, cancel.token());
    auto item = jobs.recv(cancel.token());
    (void) item;
  }
catch (const Aleph::operation_canceled &)
  {
    // stop blocked send/recv cleanly
  }

UseMutex and UseCondVar are still available for backward compatibility with older POSIX-style code, but new code should generally prefer concurrency_utils.H and standard C++ synchronization primitives.

See also the runnable example Examples/concurrency_utils_example.cc.

Work stealing was evaluated for this phase and intentionally deferred; see docs/work_stealing_backend_note.md for the design note and rationale.

TimeoutQueue (timed events)

See timeoutQueue.H and the runnable example timeoutQueue_example.C.

Notes

• Most components here are thread-safe by design (e.g. ThreadPool::enqueue()), but always check the specific header documentation.
• Some higher-level algorithms (e.g. shortest paths) are generally not thread-safe unless you use separate instances.

Function Documentation

UseCondVar()

UseCondVar::UseCondVar ( pthread_cond_t *  c, pthread_mutex_t *  m  )

inline

Construct wrapper for condition variable and mutex.

Precondition

c != nullptr && m != nullptr

Parameters

c	Pointer to initialized condition variable
m	Pointer to initialized mutex (must be held when waiting)

Exceptions

None

Complexity O(1)

Exception safety No-throw

Definition at line 129 of file useCondVar.H.

References UseCondVar::cond, m, and UseCondVar::mutex.