Worker Pools

util::WorkerPool runs independent callables on owned, joinable threads. It supports standard futures without an event loop and optional typed completion callbacks on one owner loop. Pools are independent of Core’s internal I/O and coroutine workers, so application work can wait for those services without consuming their execution capacity.

Procedural Work

submit(work) returns std::future<R> for one value or void result. map(values, work) returns one future carrying an ordered List<R>; forEach(values, work) returns one void future. Call get() to consume the result or rethrow its operational error. Discarding a future does not cancel work; retain and observe it when failures matter.

util/WorkerPool/Procedural.cpp

/// Parallel population projections with one ordered result list.
/// WorkerPool owns its threads and drains accepted work on destruction. A map call
/// snapshots the input and returns one standard future for the complete operation.
void procedural() {
    const auto workers = el::WorkerPool::create(el::WorkerPoolOptions{}.setWorkerCount(2));
    const auto populations = el::List<double>{100.0, 250.0, 500.0};

    auto projected = workers->map(populations, [](const double &initial, el::StopToken stop) -> double {
        auto population = initial;
        for (auto year = 0; year < 20 && !stop.isStopRequested(); ++year) {
            population += 0.15 * population * (1.0 - population / 1000.0);
        }
        return population;
    });

    // Results retain input order even when individual projections finish out of order.
    el::stdOut()->printLine("Habitat: "_el, "duingebied"_el);
    for (const auto population : projected.get()) {
        el::stdOut()->printLine(el::StringFormat{"Projected population: {:.1f}"_el}.build(population));
    }
}

Output

Habitat: duingebied
Projected population: 675.9
Projected population: 874.4
Projected population: 958.5

A callable may take no arguments or a StopToken; collection callables take a const item reference and optionally a final token. The token-taking form wins when both forms are callable. Single jobs support move-only callables and result values; reference results are excluded. Move-only completion callbacks are supported as well.

Bulk Operations

Bulk calls copy a finite readable range into owned storage before returning. They never retain borrowed iterators or views. Input elements and mapped results must be copy-constructible values. Workers invoke one shared callable through const access, so the caller must synchronize any shared mutable state referenced by that callable.

map() preserves input order, independent of execution order. forEach() performs side effects and does not modify the original collection. Empty inputs complete successfully; event completion remains deferred.

One bulk call consumes one admission slot. Items are claimed incrementally without per-item futures or queue entries, and other operations can receive workers between claims. The first operational error stops the batch from starting more items; running items finish before its error is published. Other operations continue. Failed and cancelled batches never expose partial result lists.

Optional Owner-Loop Delivery

WorkerPool is an EventSource with a stable WorkerPoolEventEditor. Use WorkerPool::create() for a pool whose owner is selected by its first event use. To select the owner explicitly, obtain its Utilities frontend and create the pool inside that loop:

loop->invoke([loop, this, options]() -> void {
    _workers = loop->get<el::Utilities>().createWorkerPool(options);
    _workers->events().onFinal([this]() -> void { finishWorkflow(); });
});

Events::get() returns a reference, so the factory call uses a dot. As with System::createSubprocess(), creation through the frontend must run on its owner loop. Each call creates an independent pool; the frontend does not retain it. Retain the returned pool while its work and completion handlers are needed.

Like Subprocess, the first events() call binds the current loop permanently. Callback-form submissions bind it implicitly when needed. Perform binding, handler editing, and callback-form submission inside that owner loop. Procedural submissions remain thread-safe and can coexist with event operations.

submit(work, completed[, failed]), map(values, work, completed[, failed]), and forEach(values, work, completed[, failed]) deliver through callbacks instead of futures. The successful callback receives a result by value, or no argument for void. Every callback executes on the owner loop, never on a worker.

util/WorkerPool/EventDelivery.cpp

/// Deliver parallel projections on one event loop without waiting inside its callbacks.
/// Keep the pool alive until its final event; quit closes admission while accepted
/// computations and their completion callbacks finish normally.
void eventDelivery() {
    const auto loop = el::EventLoop::create();
    auto workers = el::WorkerPoolPtr{};

    loop->invoke([&]() -> void {
        workers = loop->get<el::Utilities>().createWorkerPool(el::WorkerPoolOptions{}.setWorkerCount(2));
        workers->events().onFinal([&]() -> void { loop->quit(); });
        workers->map(
            el::List<double>{100.0, 250.0, 500.0},
            [](const double &initial) -> double {
                auto population = initial;
                for (auto year = 0; year < 20; ++year) {
                    population += 0.15 * population * (1.0 - population / 1000.0);
                }
                return population;
            },
            [](el::List<double> populations) -> void {
                el::stdOut()->printLine("Habitat: "_el, "duingebied"_el);
                for (const auto population : populations) {
                    el::stdOut()->printLine(el::StringFormat{"Projected population: {:.1f}"_el}.build(population));
                }
            });
        workers->quit();
    });

    loop->run();
    workers->join();
}

Output

Habitat: duingebied
Projected population: 675.9
Projected population: 874.4
Projected population: 958.5

The editor’s onError(std::exception_ptr) is the fallback for operational task failures. A per-operation failed callback overrides it. Without either handler, the exception is rethrown during dispatch and follows the owner’s configured error policy. Exceptions from completion handlers also follow that loop’s normal policy.

Cancellation invokes the operation’s failed handler with WorkerCancelledError when supplied; otherwise it invokes onCancelled() or is silently ignored. onIdle() reports that accepted work and outcome callbacks have drained. New submissions suppress stale idle notifications. onFinal() runs once after closed-pool workers finish and preceding outcomes have been delivered. Binding does not replay past idle or final transitions.

Each on...() replaces one handler; an empty handler clears it. Source and target access follow the common Event System editor rules. Retain the source while callbacks are wanted; handlers must not strongly capture their source. Pool destruction suppresses callbacks that have not begun.

Completion delivery reserves one internal continuation per pool, independent of ordinary event-queue limits and overflow policies. Initial binding can fail with EventQueueFullError if no continuation can be reserved. Owner-loop shutdown suppresses undelivered outcomes and releases their admission without delaying computations or procedural waits.

Stopping and Lifetime

The pool owns a StopSource and supplies its read-only StopToken to cooperative work. See Cooperative Stopping. stopToken() also exposes the token directly. Stopping never forcibly interrupts a running callable or a blocking system call.

  • wait() waits for queued/running work, release of its work captures, and future publication; the pool stays open.

  • quit() closes admission and drains accepted work without requesting cancellation or waiting.

  • stop() closes admission, requests the shared token, and cancels unfinished operations without waiting.

  • join() waits for worker termination without initiating shutdown.

  • Destruction closes admission, drains accepted work, and joins workers; it never dispatches callbacks.

quit() and stop() are idempotent. stop() can escalate graceful shutdown. A closed pool cannot restart. Success committed before stop() remains success, including results waiting for event delivery. Failure and cancellation select one synchronized first terminal decision. Running items still release their captures before that decision becomes observable through a future or completion callback.

Blocking waits do not pump the owner loop or wait for callback delivery. They can therefore finish with undelivered event results still consuming admission. Workers must not synchronously wait for dependent work on their own pool. Calling wait() or join() from a pool worker raises LogicFailure; releasing the final pool owner on its own worker terminates. Do not create a dependency where a joining thread is required to complete a worker’s blocking request.

Capacity and Failures

WorkerPoolOptions is copied at creation. workerCount() defaults to zero, selecting hardware concurrency with a minimum of one. The effective count is available from the pool and stays fixed. maximumPendingTasks() defaults to zero for unlimited admission. A finite limit counts whole operations until future publication or event dispatch/discard; it does not limit snapshot or result byte sizes. Submissions never wait for capacity.

WorkerPoolFullError reports exhausted configured admission. WorkerPoolClosedError reports a closed pool or closed event delivery. Rejected work never executes. isAccepting() is a synchronized snapshot of procedural admission, independent of capacity and event-owner shutdown.

Only err::RuntimeError exceptions escaping work are transported. Other exceptions, including programming and resource failures, terminate at the worker boundary. Allocation and native-thread creation failures during submission or construction propagate after cleanup; they are not converted into operational task failures. Snapshot-copy errors occur synchronously before admission. Destruction and joins do not consume future results or rethrow task errors.

Interface

class Utilities

Event-loop frontend for utility event sources.

See: Worker Pools

Subclassed by erbsland::util::impl::UtilitiesBackend

Public Functions

virtual WorkerPoolPtr createWorkerPool(const WorkerPoolOptions &options = {}) = 0

Create an open pool bound to this loop; call on its owner loop.

Parameters:

options – Concurrency and operation limits, copied into the pool.

Throws:

  • LogicFailure – If called outside this frontend’s owner loop.

  • EventQueueFullError – If the owner cannot reserve completion delivery capacity.

Returns:

An independently owned pool with event delivery bound to this loop.

Public Static Functions

static inline constexpr event::EventBackendId backendId() noexcept

Get the utilities backend identifier.

Returns:

The built-in utilities backend identifier.

class WorkerCancelledError : public erbsland::err::RuntimeError

A worker operation was cancelled before committing its result.

class WorkerPool : public erbsland::event::EventSource

An independently owned worker pool with procedural results and optional owner-loop completion.

See: Worker Pools

Public Functions

~WorkerPool() override

Drain accepted work and join workers, suppressing callbacks that have not begun.

unsigned int workerCount() const noexcept

Get the effective fixed worker count.

bool isAccepting() const noexcept

Test whether new procedural operations can be admitted, independently of capacity.

StopToken stopToken() const noexcept

Get the shared cooperative stop token.

void wait()

Wait for work and its captures without closing admission or dispatching callbacks.

Must not be called by one of this pool’s workers.

void quit() noexcept

Close admission and drain accepted operations without waiting.

void stop() noexcept

Close admission, cancel unfinished operations, and signal running callables without waiting.

void join()

Join workers without initiating shutdown or dispatching callbacks.

Must not be called by one of this pool’s workers.

template<typename tFunction>
auto submit(tFunction work)

Submit one callable, optionally accepting a StopToken, and return its standard future.

Template Parameters:

tFunction – The movable work type, returning a non-reference value or void.

Parameters:

work – The callable, retained until execution or cancellation.

Throws:
Returns:

A future carrying the value, RuntimeError, or WorkerCancelledError.

template<typename tFunction, typename tCompleted>
void submit(tFunction work, tCompleted completed, WorkerPoolEventEditor::ErrorFn failed = {})

Submit one callable and deliver its outcome on the current owner loop.

Template Parameters:

  • tFunction – The movable work type.

  • tCompleted – The completion callable type.

Parameters:

  • work – The callable, optionally accepting a StopToken.

  • completed – Successful recipient of the result by value, or no argument for void.

  • failed – Optional operational-error/cancellation recipient overriding pool handlers.

Throws:

  • WorkerPoolFullError – If the configured admission limit is reached.

  • WorkerPoolClosedError – If admission or owner-loop delivery has closed.

  • EventQueueFullError – If initial owner binding cannot reserve delivery capacity.

template<std::ranges::input_range tRange, typename tFunction>
auto map(tRange &&values, tFunction work)

Transform a finite range in parallel, preserving input order in one aggregate future.

Template Parameters:

  • tRange – The finite readable range type with copyable values.

  • tFunction – The const-invocable work type producing copyable, non-reference values.

Parameters:

  • values – Values copied into owned storage before this call returns.

  • work – Work accepting a const item reference and optionally a StopToken.

Throws:
Returns:

A future carrying List of mapped results, or the first failure/cancellation.

template<std::ranges::input_range tRange, typename tFunction, typename tCompleted>
void map(tRange &&values, tFunction work, tCompleted completed, WorkerPoolEventEditor::ErrorFn failed = {})

Transform a finite range and deliver one ordered result list on the owner loop.

Template Parameters:

  • tRange – The finite readable range type with copyable values.

  • tFunction – The const-invocable work type producing copyable values.

  • tCompleted – The successful completion callable type.

Parameters:

  • values – Values copied into owned storage before this call returns.

  • work – Work accepting a const item reference and optionally a StopToken.

  • completed – Recipient of the ordered result list by value.

  • failed – Optional operation-specific error/cancellation recipient.

Throws:

  • WorkerPoolFullError – If the configured admission limit is reached.

  • WorkerPoolClosedError – If admission or event delivery has closed.

  • EventQueueFullError – If initial owner binding cannot reserve delivery capacity.

template<std::ranges::input_range tRange, typename tFunction>
auto forEach(tRange &&values, tFunction work)

Process a finite range in parallel and return one void future.

Template Parameters:

  • tRange – The finite readable range type with copyable values.

  • tFunction – The concurrently const-invocable void work type.

Parameters:

  • values – Values copied into owned storage; the caller’s range is not modified.

  • work – Work accepting a const item reference and optionally a StopToken.

Throws:
Returns:

A future completed after all running items release their captures.

template<std::ranges::input_range tRange, typename tFunction, typename tCompleted>
void forEach(tRange &&values, tFunction work, tCompleted completed, WorkerPoolEventEditor::ErrorFn failed = {})

Process a finite range and deliver one completion on the owner loop.

Template Parameters:

  • tRange – The finite readable range type with copyable values.

  • tFunction – The concurrently const-invocable void work type.

  • tCompleted – The successful completion callable type.

Parameters:

  • values – Values copied into owned storage; the caller’s range is not modified.

  • work – Work accepting a const item reference and optionally a StopToken.

  • completed – No-argument recipient of successful completion.

  • failed – Optional operation-specific error/cancellation recipient.

Throws:

  • WorkerPoolFullError – If the configured admission limit is reached.

  • WorkerPoolClosedError – If admission or event delivery has closed.

  • EventQueueFullError – If initial owner binding cannot reserve delivery capacity.

virtual WorkerPoolEventEditor &events() override

Access the stable editor owned by this source.

Must be called on the owner event loop.

Returns:

The source-owned editor.

Public Static Functions

static WorkerPoolPtr create(const WorkerPoolOptions &options = {})

Create an open pool; no event loop is required.

Parameters:

options – Concurrency and operation limits, copied into the pool.

Returns:

A shared pool owner, which must not be finally released on one of its own workers.

class WorkerPoolClosedError : public erbsland::err::RuntimeError

Work was submitted after the pool or its event delivery closed.

class WorkerPoolEventEditor : public erbsland::event::impl::CommonEventEditor

Source-owned single-handler configuration for a worker pool’s owner-loop notifications.

See: Worker Pools

Public Types

using ErrorFn = std::function<void(std::exception_ptr)>

An operational failure recipient.

Public Functions

WorkerPoolEventEditor &onError(ErrorFn callback)

Replace the fallback operational-error handler; empty rethrows on the owner loop.

Parameters:

callback – The replacement handler.

WorkerPoolEventEditor &onCancelled(event::EventCallback callback)

Replace the fallback per-operation cancellation notification; empty ignores cancellation.

Parameters:

callback – The replacement handler.

WorkerPoolEventEditor &onIdle(event::EventCallback callback)

Replace the notification after work and pending outcomes drain.

Parameters:

callback – The replacement handler.

WorkerPoolEventEditor &onFinal(event::EventCallback callback)

Replace the final notification after shutdown and outcome delivery.

Parameters:

callback – The replacement handler.

class WorkerPoolFullError : public erbsland::err::RuntimeError

The configured worker-pool admission limit was reached.

class WorkerPoolOptions

Worker concurrency and operation admission limits, copied when a pool is created.

Public Functions

inline unsigned int workerCount() const noexcept

Get the worker count; zero selects the hardware concurrency, with a minimum of one.

inline WorkerPoolOptions &setWorkerCount(unsigned int value) noexcept

Set the worker count; zero selects automatic concurrency.

inline std::size_t maximumPendingTasks() const noexcept

Get the pending-operation limit; zero means unlimited.

inline WorkerPoolOptions &setMaximumPendingTasks(std::size_t value) noexcept

Set the limit including results awaiting event delivery; a bulk operation counts once.

Complete demo sources

util/WorkerPool/Procedural.cpp

// Copyright (c) 2026 Tobias Erbsland - https://erbsland.dev
// SPDX-License-Identifier: Apache-2.0

#include "WorkerPoolDemos.hpp"

#include <erbsland/List.hpp>
#include <erbsland/StopToken.hpp>
#include <erbsland/StringFormat.hpp>
#include <erbsland/WorkerPool.hpp>

namespace demo {

/// Parallel population projections with one ordered result list.
/// WorkerPool owns its threads and drains accepted work on destruction. A map call
/// snapshots the input and returns one standard future for the complete operation.
void procedural() {
    const auto workers = el::WorkerPool::create(el::WorkerPoolOptions{}.setWorkerCount(2));
    const auto populations = el::List<double>{100.0, 250.0, 500.0};

    auto projected = workers->map(populations, [](const double &initial, el::StopToken stop) -> double {
        auto population = initial;
        for (auto year = 0; year < 20 && !stop.isStopRequested(); ++year) {
            population += 0.15 * population * (1.0 - population / 1000.0);
        }
        return population;
    });

    // Results retain input order even when individual projections finish out of order.
    el::stdOut()->printLine("Habitat: "_el, "duingebied"_el);
    for (const auto population : projected.get()) {
        el::stdOut()->printLine(el::StringFormat{"Projected population: {:.1f}"_el}.build(population));
    }
}

}

util/WorkerPool/EventDelivery.cpp

// Copyright (c) 2026 Tobias Erbsland - https://erbsland.dev
// SPDX-License-Identifier: Apache-2.0

#include "WorkerPoolDemos.hpp"

#include <erbsland/EventLoop.hpp>
#include <erbsland/List.hpp>
#include <erbsland/StringFormat.hpp>
#include <erbsland/Utilities.hpp>

namespace demo {

/// Deliver parallel projections on one event loop without waiting inside its callbacks.
/// Keep the pool alive until its final event; quit closes admission while accepted
/// computations and their completion callbacks finish normally.
void eventDelivery() {
    const auto loop = el::EventLoop::create();
    auto workers = el::WorkerPoolPtr{};

    loop->invoke([&]() -> void {
        workers = loop->get<el::Utilities>().createWorkerPool(el::WorkerPoolOptions{}.setWorkerCount(2));
        workers->events().onFinal([&]() -> void { loop->quit(); });
        workers->map(
            el::List<double>{100.0, 250.0, 500.0},
            [](const double &initial) -> double {
                auto population = initial;
                for (auto year = 0; year < 20; ++year) {
                    population += 0.15 * population * (1.0 - population / 1000.0);
                }
                return population;
            },
            [](el::List<double> populations) -> void {
                el::stdOut()->printLine("Habitat: "_el, "duingebied"_el);
                for (const auto population : populations) {
                    el::stdOut()->printLine(el::StringFormat{"Projected population: {:.1f}"_el}.build(population));
                }
            });
        workers->quit();
    });

    loop->run();
    workers->join();
}

}