eve.rt.pool

Thread Pool for Parallel Task Execution

This module provides a thread pool implementation for executing tasks across multiple worker threads. It integrates with the EVE runtime layer to enable efficient parallel and concurrent execution patterns.

The thread pool manages a fixed number of worker threads that pull tasks from a shared work queue. This amortizes thread creation overhead and provides backpressure when the system is overloaded.

Copyright

Copyright © 2025 Dejan Lekic

License

BSD-3-Clause
alias SubmitStatus Task
enum PoolState SubmitErrorReason
struct SubmitError
class ThreadPool
fn defaultWorkerCount

Types 6

enumPoolState : ubyte

State of the thread pool.

CREATEDPool has been created but not yet started.
RUNNINGPool is running and accepting tasks.
STOPPINGPool is shutting down; no new tasks accepted.
STOPPEDPool has fully stopped; all workers have exited.
enumSubmitErrorReason : ubyte

Reason a task submission to the pool failed.

Retryability and fatality:

  • REJECTED — Fatal: the pool is shutting down or stopped.
  • QUEUE_FULL — Retryable after the queue drains.

UNSPECIFIED = 0Not a real domain error; the default-constructed payload value.
REJECTEDPool is not accepting new tasks (stopping or stopped).
QUEUE_FULLTask queue is full (for bounded pools).
structSubmitError

Recoverable failure of a task submission.

Fields
SubmitErrorReason reasonWhy the submission failed.
int errnoReserved; OS errors do not apply to pool submissions.
aliasSubmitStatus = Status!SubmitError
aliasTask = void delegate() @safe

A task to be executed by the thread pool.

Tasks are simple delegates that take no arguments and return nothing. Any results should be communicated through captured variables, futures, or channels.

classThreadPool

A pool of worker threads for executing tasks in parallel.

The ThreadPool manages a collection of worker threads that execute submitted tasks. Tasks are queued and executed in approximate FIFO order, though no strict ordering is guaranteed due to concurrent execution.

Example: 

auto pool = ThreadPool.create(4);  // 4 worker threads
pool.start();

pool.submit(() @safe {
   // Task code here
});

pool.shutdown();
pool.awaitTermination();

Fields
private Thread[] workers
private Task[] taskQueue
private size_t queueHead
private size_t queueTail
private size_t queueCount
private size_t queueCapacity
private Mutex mutex
private Condition workAvailable
private Condition queueNotFull
private PoolState state_
private size_t activeWorkers_
private ulong tasksCompleted_
private ulong tasksSubmitted_
private size_t numWorkers_
private bool waitForTasks_
Methods
ThreadPool create(size_t numWorkers = 0, size_t queueCapacity = 1024) static @trustedCreate a new thread pool with the specified number of workers.
void start() @trustedStart the thread pool, spawning worker threads.
SubmitStatus submit(Task task) @trustedSubmit a task for execution by the thread pool.
SubmitStatus trySubmit(Task task) @trustedTry to submit a task without blocking.
void shutdown(bool waitForTasks = true) @trustedInitiate graceful shutdown of the thread pool.
void awaitTermination() @trustedWait for all worker threads to terminate.
PoolState state() @property const @trusted nothrow @nogcGet the current state of the pool.
size_t numWorkers() @property const pure @safe nothrow @nogcGet the number of worker threads in the pool.
size_t activeWorkers() @property const @trusted nothrow @nogcGet the number of currently active (executing) workers.
size_t pendingTasks() @property @trustedGet the number of tasks waiting in the queue.
ulong completedTasks() @property const @trusted nothrow @nogcGet the total number of tasks that have been completed.
ulong submittedTasks() @property const @trusted nothrow @nogcGet the total number of tasks that have been submitted.
private void workerLoop() @trusted
Constructors
this()
this(size_t numWorkers, size_t queueCapacity)

Functions 1

fnsize_t defaultWorkerCount() @trustedGet the default number of worker threads based on CPU count.