The arguments the function was called with. Changes to out and ref arguments will be visible here.

The return type of the function called by this Task. This can be void.

If the Task isn't started yet, execute it in the current thread. If it's done, return its return value, if any. If it's in progress, busy spin until it's done, then return the return value. If it threw an exception, rethrow that exception.

This function should be used when you expect the result of the Task to be available on a timescale shorter than that of an OS context switch.

If the Task isn't started yet, execute it in the current thread. If it's done, return its return value, if any. If it's in progress, wait on a condition variable. If it threw an exception, rethrow that exception.

This function should be used for expensive functions, as waiting on a condition variable introduces latency, but avoids wasted CPU cycles.

If this Task was not started yet, execute it in the current thread. If it is finished, return its result. If it is in progress, execute any other Task from the TaskPool instance that this Task was submitted to until this one is finished. If it threw an exception, rethrow that exception. If no other tasks are available or this Task was executed using executeInNewThread, wait on a condition variable.

Returns true if the Task is finished executing.

Create a new thread for executing this Task, execute it in the newly created thread, then terminate the thread. This can be used for future/promise parallelism. An explicit priority may be given to the Task. If one is provided, its value is forwarded to core.thread.Thread.priority. See std.parallelism.task for usage example.

Default constructor that initializes a TaskPool with totalCPUs - 1 worker threads. The minus 1 is included because the main thread will also be available to do work.

Allows for custom number of worker threads.

Implements a  parallel foreach loop over a range. This works by implicitly creating and submitting one Task to the TaskPool for each worker thread. A work unit is a set of consecutive elements of range to be processed by a worker thread between communication with any other thread. The number of elements processed per work unit is controlled by the workUnitSize parameter. Smaller work units provide better load balancing, but larger work units avoid the overhead of communicating with other threads frequently to fetch the next work unit. Large work units also avoid false sharing in cases where the range is being modified. The less time a single iteration of the loop takes, the larger workUnitSize should be. For very expensive loop bodies, workUnitSize should be 1. An overload that chooses a default work unit size is also available.

// Find the logarithm of every number from 1 to

// 10_000_000 in parallel.

auto logs = new double[10_000_000];

// Parallel foreach works with or without an index

// variable.  It can be iterate by ref if range.front

// returns by ref.


// Iterate over logs using work units of size 100.

foreach(i, ref elem; taskPool.parallel(logs, 100))
{
    elem = log(i + 1.0);
}

// Same thing, but use the default work unit size.

//

// Timings on an Athlon 64 X2 dual core machine:

//

// Parallel foreach:  388 milliseconds

// Regular foreach:   619 milliseconds

foreach(i, ref elem; taskPool.parallel(logs))
{
    elem = log(i + 1.0);
}

Eager parallel map. The eagerness of this function means it has less overhead than the lazily evaluated TaskPool.map and should be preferred where the memory requirements of eagerness are acceptable. functions are the functions to be evaluated, passed as template alias parameters in a style similar to std.algorithm.map. The first argument must be a random access range.

auto numbers = iota(100_000_000.0);

// Find the square roots of numbers.

//

// Timings on an Athlon 64 X2 dual core machine:

//

// Parallel eager map:                   0.802 s

// Equivalent serial implementation:     1.768 s

auto squareRoots = taskPool.amap!sqrt(numbers);

// Same thing, but make work unit size 100.

auto squareRoots = taskPool.amap!sqrt(numbers, 100);

// Same thing, but explicitly allocate an array

// to return the results in.  The element type

// of the array may be either the exact type

// returned by functions or an implicit conversion

// target.

auto squareRoots = new float[numbers.length];
taskPool.amap!sqrt(numbers, squareRoots);

// Multiple functions, explicit output range, and

// explicit work unit size.

auto results = new Tuple!(float, real)[numbers.length];
taskPool.amap!(sqrt, log)(numbers, 100, results);

A semi-lazy parallel  map that can be used for pipelining. The  map functions are evaluated for the first bufSize elements and stored in a buffer and made available to popFront. Meanwhile, in the background a second buffer of the same size is filled. When the first buffer is exhausted, it is swapped with the second buffer and filled while the values from what was originally the second buffer are read. This implementation allows for elements to be written to the buffer without the need for atomic operations or synchronization for each write, and enables the mapping function to be evaluated efficiently in parallel.

 map has more overhead than the simpler procedure used by amap but avoids the need to keep all results in memory simultaneously and works with non-random access ranges.

// Pipeline reading a file, converting each line

// to a number, taking the logarithms of the numbers,

// and performing the additions necessary to find

// the sum of the logarithms.


auto lineRange = File("numberList.txt").byLine();
auto dupedLines = std.algorithm.map!"a.idup"(lineRange);
auto nums = taskPool.map!(to!double)(dupedLines);
auto logs = taskPool.map!log10(nums);

double sum = 0;
foreach(elem; logs)
{
    sum += elem;
}

Given a source range that is expensive to iterate over, returns an input range that asynchronously buffers the contents of source into a buffer of bufSize elements in a worker thread, while making prevously buffered elements from a second buffer, also of size bufSize, available via the range interface of the returned object. The returned range has a length iff hasLength!S.  asyncBuf is useful, for example, when performing expensive operations on the elements of ranges that represent data on a disk or network.

import std.conv, std.stdio;

void main()
{
    // Fetch lines of a file in a background thread

    // while processing prevously fetched lines,

    // dealing with byLine's buffer recycling by

    // eagerly duplicating every line.

    auto lines = File("foo.txt").byLine();
    auto duped = std.algorithm.map!"a.idup"(lines);

    // Fetch more lines in the background while we

    // process the lines already read into memory

    // into a matrix of doubles.

    double[][] matrix;
    auto asyncReader = taskPool.asyncBuf(duped);

    foreach(line; asyncReader)
    {
        auto ls = line.split("\t");
        matrix ~= to!(double[])(ls);
    }
}

Given a callable object next that writes to a user-provided buffer and a second callable object empty that determines whether more data is available to write via next, returns an input range that asynchronously calls next with a set of size nBuffers of buffers and makes the results available in the order they were obtained via the input range interface of the returned object. Similarly to the input range overload of  asyncBuf, the first half of the buffers are made available via the range interface while the second half are filled and vice-versa.

// Fetch lines of a file in a background

// thread while processing prevously fetched

// lines, without duplicating any lines.

auto file = File("foo.txt");

void next(ref char[] buf)
{
    file.readln(buf);
}

// Fetch more lines in the background while we

// process the lines already read into memory

// into a matrix of doubles.

double[][] matrix;
auto asyncReader = taskPool.asyncBuf(&next, &file.eof);

foreach(line; asyncReader)
{
    auto ls = line.split("\t");
    matrix ~= to!(double[])(ls);
}

Parallel  reduce on a random access range. Except as otherwise noted, usage is similar to std.algorithm.reduce. This function works by splitting the range to be reduced into work units, which are slices to be reduced in parallel. Once the results from all work units are computed, a final serial reduction is performed on these results to compute the final answer. Therefore, care must be taken to choose the seed value appropriately.

Because the reduction is being performed in parallel, functions must be associative. For notational simplicity, let # be an infix operator representing functions. Then, (a # b) # c must equal a # (b # c). Floating point addition is not associative even though addition in exact arithmetic is. Summing floating point numbers using this function may give different results than summing serially. However, for many practical purposes floating point addition can be treated as associative.

Note that, since functions are assumed to be associative, additional optimizations are made to the serial portion of the reduction algorithm. These take advantage of the instruction level parallelism of modern CPUs, in addition to the thread-level parallelism that the rest of this module exploits. This can lead to better than linear speedups relative to std.algorithm.reduce, especially for fine-grained benchmarks like dot products.

An explicit seed may be provided as the first argument. If provided, it is used as the seed for all work units and for the final reduction of results from all work units. Therefore, if it is not the identity value for the operation being performed, results may differ from those generated by std.algorithm.reduce or depending on how many work units are used. The next argument must be the range to be reduced.

// Find the sum of squares of a range in parallel, using

// an explicit seed.

//

// Timings on an Athlon 64 X2 dual core machine:

//

// Parallel reduce:                     72 milliseconds

// Using std.algorithm.reduce instead:  181 milliseconds

auto nums = iota(10_000_000.0f);
auto sumSquares = taskPool.reduce!"a + b"(
    0.0, std.algorithm.map!"a * a"(nums)
);

// Find the sum of a range in parallel, using the first

// element of each work unit as the seed.

auto sum = taskPool.reduce!"a + b"(nums);

// Use a work unit size of 100.

auto sum2 = taskPool.reduce!"a + b"(nums, 100);

// Work unit size of 100 and explicit seed.

auto sum3 = taskPool.reduce!"a + b"(0.0, nums, 100);

// Find both the min and max of nums.

auto minMax = taskPool.reduce!(min, max)(nums);
assert(minMax[0] == reduce!min(nums));
assert(minMax[1] == reduce!max(nums));

Gets the index of the current thread relative to this TaskPool. Any thread not in this pool will receive an index of 0. The worker threads in this pool receive unique indices of 1 through this.size.

This function is useful for maintaining worker-local resources.

// Execute a loop that computes the greatest common

// divisor of every number from 0 through 999 with

// 42 in parallel.  Write the results out to

// a set of files, one for each thread.  This allows

// results to be written out without any synchronization.


import std.conv, std.range, std.numeric, std.stdio;

void main()
{
    auto filesHandles = new File[taskPool.size + 1];
    scope(exit) {
        foreach(ref handle; fileHandles) {
            handle.close();
        }
    }

    foreach(i, ref handle; fileHandles)
    {
        handle = File("workerResults" ~ to!string(i) ~ ".txt");
    }

    foreach(num; parallel(iota(1_000)))
    {
        auto outHandle = fileHandles[taskPool.workerIndex];
        outHandle.writeln(num, '\t', gcd(num, 42));
    }
}

Struct for creating worker-local storage. Worker-local storage is thread-local storage that exists only for worker threads in a given TaskPool plus a single thread outside the pool. It is allocated on the garbage collected heap in a way that avoids false sharing, and doesn't necessarily have global scope within any thread. It can be accessed from any worker thread in the TaskPool that created it, and one thread outside this TaskPool. All threads outside the pool that created a given instance of worker-local storage share a single slot.

Since the underlying data for this struct is heap-allocated, this struct has reference semantics when passed between functions.

The main uses cases for WorkerLocalStorageStorage are:

1. Performing parallel reductions with an imperative, as opposed to functional, programming style. In this case, it's useful to treat WorkerLocalStorageStorage as local to each thread for only the parallel portion of an algorithm.

2. Recycling temporary buffers across iterations of a parallel foreach loop.

// Calculate pi as in our synopsis example, but

// use an imperative instead of a functional style.

immutable n = 1_000_000_000;
immutable delta = 1.0L / n;

auto sums = taskPool.workerLocalStorage(0.0L);
foreach(i; parallel(iota(n)))
{
    immutable x = ( i - 0.5L ) * delta;
    immutable toAdd = delta / ( 1.0 + x * x );
    sums.get += toAdd;
}

// Add up the results from each worker thread.

real pi = 0;
foreach(threadResult; sums.toRange)
{
    pi += 4.0L * threadResult;
}

Range primitives for worker-local storage. The purpose of this is to access results produced by each worker thread from a single thread once you are no longer using the worker-local storage from multiple threads. Do not use this struct in the parallel portion of your algorithm.

The proper way to instantiate this object is to call WorkerLocalStorage.toRange. Once instantiated, this object behaves as a finite random-access range with assignable, lvalue elemends and a length equal to the number of worker threads in the TaskPool that created it plus 1.

Creates an instance of worker-local storage, initialized with a given value. The value is lazy so that you can, for example, easily create one instance of a class for each worker. For usage example, see the WorkerLocalStorage struct.

Signals to all worker threads to terminate as soon as they are finished with their current Task, or immediately if they are not executing a Task. Tasks that were in queue will not be executed unless a call to Task.workForce, Task.yieldForce or Task.spinForce causes them to be executed.

Use only if you have waitied on every Task and therefore know the queue is empty, or if you speculatively executed some tasks and no longer need the results.

Signals worker threads to terminate when the queue becomes empty.

If blocking argument is true, wait for all worker threads to terminate before returning. This option might be used in applications where task results are never consumed-- e.g. when TaskPool is employed as a rudimentary scheduler for tasks which communicate by means other than return values.

Returns the number of worker threads in the pool.

Put a Task object on the back of the task queue. The Task object may be passed by pointer or reference.

import std.file;

// Create a task.

auto t = task!read("foo.txt");

// Add it to the queue to be executed.

taskPool.put(t);

These properties control whether the worker threads are daemon threads. A daemon thread is automatically terminated when all non-daemon threads have terminated. A non-daemon thread will prevent a program from terminating as long as it has not terminated.

If any TaskPool with non-daemon threads is active, either stop or finish must be called on it before the program can terminate.

The worker treads in the TaskPool instance returned by the taskPool property are daemon by default. The worker threads of manually instantiated task pools are non-daemon by default.

These functions allow getting and setting the OS scheduling  priority of the worker threads in this TaskPool. They forward to core.thread.Thread. priority, so a given  priority value here means the same thing as an identical  priority value in core.thread.