root/trunk/parallelFuture/parallelFuture.d

/**A library that implements various high-level parallelism primitives, such as
 * parallel foreach over arbitrary ranges, parallel map and reduce,
 * futures, and a task pool.
 *
 * By:  David Simcha
 *
 * Examples:
 * ---
    // Create a TaskPool object with the default number of threads.
    auto poolInstance = new TaskPool();

    // Create some data to work on.
    uint[] numbers = new uint[1_000];

    // Fill in this array in parallel, using default block size.
    // Note:  Be careful when writing to adjacent elements of an array from
    // different threads, as this can cause word tearing bugs when
    // the elements aren't properly aligned or aren't the machine's native
    // word size.  In this case, though, we're ok.
    foreach(i; poolInstance.parallel( iota(0, numbers.length)) ) {
        numbers[i] = i;
    }

    // Make sure it works.
    foreach(i; 0..numbers.length) {
        assert(numbers[i] == i);
    }

    stderr.writeln("Done creating nums.");

    // Parallel foreach also works on non-random access ranges, albeit
    // less efficiently.
    auto myNumbers = filter!"a % 7 > 0"( iota(0, 1000));
    foreach(num; poolInstance.parallel(myNumbers)) {
        assert(num % 7 > 0 && num < 1000);
    }
    stderr.writeln("Done modulus test.");

    // Use parallel map to calculate the square of each element in numbers,
    // and make sure it's right.
    uint[] squares = poolInstance.map!"a * a"(numbers, 100);
    assert(squares.length == numbers.length);
    foreach(i, number; numbers) {
        assert(squares[i] == number * number);
    }
    stderr.writeln("Done squares.");

    // Sum up the array in parallel with the current thread.
    auto sumFuture = poolInstance.task!( reduce!"a + b" )(numbers);

    // Go off and do other stuff while that future executes:
    // Find the sum of squares of numbers.
    ulong sumSquares = 0;
    foreach(elem; numbers) {
        sumSquares += elem * elem;
    }

    // Ask for our result.  If the pool has not yet started working on
    // this task, spinWait() automatically steals it and executes it in this
    // thread.
    uint mySum = sumFuture.spinWait();
    assert(mySum == 999 * 1000 / 2);

    // We could have also computed this sum in parallel using parallel
    // reduce.
    auto mySumParallel = poolInstance.reduce!"a + b"(numbers);
    assert(mySum == mySumParallel);
    stderr.writeln("Done sums.");

    // Execute an anonymous delegate as a task.
    auto myTask = poolInstance.task({
        synchronized writeln("Our lives are parallel...Our lives are parallel.");
    });

    // Parallel foreach loops can also be nested:
    auto nestedOuter = "abcd"d;
    auto nestedInner =  iota(0, 5);

    foreach(letter; poolInstance.parallel(nestedOuter, 1)) {
        foreach(number; poolInstance.parallel(nestedInner, 1)) {
            synchronized writeln(cast(char) letter, number);
        }
    }

    // Block until all jobs are finished and then shut down the thread pool.
    poolInstance.waitStop();
 * ---
 *
 *
 * License:
 * Boost Software License - Version 1.0 - August 17th, 2003
 *
 * Permission is hereby granted, free of charge, to any person or organization
 * obtaining a copy of the software and accompanying documentation covered by
 * this license (the "Software") to use, reproduce, display, distribute,
 * execute, and transmit the Software, and to prepare derivative works of the
 * Software, and to permit third-parties to whom the Software is furnished to
 * do so, all subject to the following:
 *
 * The copyright notices in the Software and this entire statement, including
 * the above license grant, this restriction and the following disclaimer,
 * must be included in all copies of the Software, in whole or in part, and
 * all derivative works of the Software, unless such copies or derivative
 * works are solely in the form of machine-executable object code generated by
 * a source language processor.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
 * SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
 * FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
 * ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 */
module parallelfuture;

import core.thread, core.cpuid, std.algorithm, std.range, std.c.stdlib, std.stdio,
    std.contracts, std.functional, std.conv, std.math, core.memory, std.traits,
    std.typetuple;

import core.sync.condition, core.sync.mutex;

// Workaround for bug 3753.
version(Posix) {
    // Can't use alloca() because it can't be used with exception handling.
    // Use the GC instead even though it's slightly less efficient.
    void* alloca(size_t nBytes) {
        return GC.malloc(nBytes);
    }
} else {
    // Can really use alloca().
    import core.stdc.stdlib : alloca;
}

/* Atomics code.  The ASM routines were borrowed from Tango on an ad-hoc basis.
 * These snippets are likely not copyrightable.  Nonetheless,
 * these should be replaced with generic functions once D2 gets a decent
 * standard atomics lib.
 */
void atomicSetUbyte(ref ubyte stuff, ubyte newVal) {
    asm {
        mov EAX, stuff;
        mov DL, newVal;
        lock;
        xchg [EAX], DL;
    }
}

ubyte atomicReadUbyte(ref ubyte stuff) {
    asm {
        mov DL, 0;
        mov AL, 0;
        mov ECX, stuff;
        lock;
        cmpxchg [ECX], DL;
    }
}

bool atomicCasUbyte(ref ubyte stuff, ubyte testVal, ubyte newVal) {
    asm {
        mov DL, newVal;
        mov AL, testVal;
        mov ECX, stuff;
        lock;
        cmpxchg [ECX], DL;
        setz AL;
    }
}

void atomicIncUint(ref uint num) {
    asm {
        mov EAX, num;
        lock;
        inc int ptr [EAX];
        mov EAX, [EAX];
    }
}

//-----------------------------------------------------------------------------


/*--------------------- Generic helper functions, etc.------------------------*/
private template MapType(alias fun, R) {
    alias typeof(unaryFun!(fun)(ElementType!(R).init)) MapType;
}

private template ReduceType(alias fun, R, E) {
    alias typeof(binaryFun!(fun)(E.init, ElementType!(R).init)) ReduceType;
}

private template hasLength(R) {
    enum bool hasLength = __traits(compiles, R.init.length);
}

private template lValueElements(R) {
    enum bool lValueElements = lValueElementsImpl!(R).ret;
}

private template lValueElementsImpl(R) {
    private R myRange;
    enum ret = is(typeof(&(myRange.front())));
}

private void sleepMillisecond(long nMilliseconds) {
    Thread.sleep(nMilliseconds * 10_000);
}


private T* toHeap(T)(T object) {
    GC.BlkAttr gcFlags = (typeid(T).flags & 1) ?
                          cast(GC.BlkAttr) 0 :
                          GC.BlkAttr.NO_SCAN;
    T* myPtr = cast(T*) GC.malloc(T.sizeof, gcFlags);
    *myPtr = object;
    return myPtr;
}

//------------------------------------------------------------------------------
/* Various classes of task.  These use manual C-style polymorphism, the kind
 * with lots of structs and pointer casting.  This is because real classes
 * would prevent some of the allocation tricks I'm using and waste space on
 * monitors and vtbls for something that needs to be ultra-efficient.
 */

private enum TaskState : ubyte {
    notStarted,
    inProgress,
    done
}

private template BaseMixin(ubyte initTaskStatus) {
    AbstractTask* prev;
    AbstractTask* next;

    static if(is(typeof(&impl))) {
        void function(void*) runTask = &impl;
    } else {
        void function(void*) runTask;
    }

    Object exception;
    ubyte taskStatus = initTaskStatus;


    /* Kludge:  Some tasks need to re-submit themselves after they finish.
     * In this case, they will set themselves to TaskState.notStarted before
     * resubmitting themselves.  Setting this flag to false prevents the work
     * stealing loop from setting them to done.*/
    bool shouldSetDone = true;

    bool done() {
        if(atomicReadUbyte(taskStatus) == TaskState.done) {
            if(exception) {
                throw exception;
            }

            return true;
        }

        return false;
    }
}

// The base "class" for all of the other tasks.
private struct AbstractTask {
    mixin BaseMixin!(TaskState.notStarted);

    void job() {
        runTask(&this);
    }
}

private template AliasReturn(alias fun, T...) {
    alias AliasReturnImpl!(fun, T).ret AliasReturn;
}

private template AliasReturnImpl(alias fun, T...) {
    private T args;
    alias typeof(fun(args)) ret;
}

/**Calls a delegate or function pointer with args.  This is basically an
 * adapter that makes Task work with delegates, function pointers and
 * functors instead of just aliases.
 */
ReturnType!(F) runCallable(F, Args...)(F fpOrDelegate, Args args) {
    return fpOrDelegate(args);
}

/**A struct that encapsulates the information about a task, including
 * its current status, what pool it was submitted to, and its arguments.
 *
 * Note:  If a Task has been submitted to the pool, is being stored in a stack
 * frame, and has not yet finished, the destructor for this struct will
 * automatically call yieldWait() so that the task can finish and the
 * stack frame can be destroyed safely.
 *
 * Bugs:  Doesn't work with ref/out arguments.
 */
struct Task(alias fun, Args...) {
private:
    static void impl(void* myTask) {
        Task* myCastedTask = cast(typeof(this)*) myTask;
        static if(is(ReturnType == void)) {
            fun(myCastedTask.args);
        } else {
            myCastedTask.returnVal = fun(myCastedTask.args);
        }
    }
    mixin BaseMixin!(TaskState.notStarted) Base;

    TaskPool pool;  // For work stealing stuff.
    Args args;

    alias typeof(fun(args)) ReturnType;
    static if(!is(ReturnType == void)) {
        ReturnType returnVal;
    }

    void enforcePool() {
        enforce(this.pool !is null, "Job not submitted yet.");
    }

    this(Args args) {
        static if(args.length > 0) {
            this.args = args;
        }
    }

public:
    /**If the task isn't started yet, steal it and do it in this 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.
     *
     * This function should be used when you expect the result of the
     * task to be available relatively quickly, on a timescale shorter
     * than that of an OS context switch.
     */
    ReturnType spinWait() {
        enforcePool();

        this.pool.tryStealDelete( cast(AbstractTask*) &this);

        while(atomicReadUbyte(this.taskStatus) != TaskState.done) {}

        if(exception) {
            throw exception;
        }

        static if(!is(ReturnType == void)) {
            return this.returnVal;
        }
    }

    /**If the task isn't started yet, steal it and do it in this thread.
     * If it's done, return its return value, if any.  If it's in progress,
     * wait on a condition variable.
     *
     * This function should be used when you expect the result of the
     * task to take a while, as waiting on a condition variable
     * introduces latency, but results in negligible wasted CPU cycles.
     */
    ReturnType yieldWait() {
        enforcePool();
        this.pool.tryStealDelete( cast(AbstractTask*) &this);
        if(atomicReadUbyte(this.taskStatus) == TaskState.done) {

            static if(is(ReturnType == void)) {
                return;
            } else {
                return this.returnVal;
            }
        }

        pool.lock();
        scope(exit) pool.unlock();

        while(atomicReadUbyte(this.taskStatus) != TaskState.done) {
            pool.waitUntilCompletion();
        }

        if(exception) {
            throw exception;
        }

        static if(!is(ReturnType == void)) {
            return this.returnVal;
        }
    }

    /**If this task is not started yet, steal it and execute it in the current
     * thread.  If it is finished, return its result.  If it is in progress,
     * execute any other available tasks from the task pool until this one
     * is finished.  If no other tasks are available, yield wait.
     */
    ReturnType workWait() {
        enforcePool();
        this.pool.tryStealDelete( cast(AbstractTask*) &this);

        while(true) {
            if(done) {
                static if(is(ReturnType == void)) {
                    return;
                } else {
                    return this.returnVal;
                }
            }

            pool.lock();
            AbstractTask* job;
            try {
                // Locking explicitly and calling popNoSync() because
                // pop() waits on a condition variable if there are no jobs
                // in the queue.
                job = pool.popNoSync();
            } finally {
                pool.unlock();
            }

            if(job !is null) {

                version(verboseUnittest) {
                    stderr.writeln("Doing workWait work.");
                }

                pool.doJob(job);

                if(done) {
                    static if(is(ReturnType == void)) {
                        return;
                    } else {
                        return this.returnVal;
                    }
                }
            } else {
                version(verboseUnittest) {
                    stderr.writeln("Yield from workWait.");
                }

                return yieldWait();
            }
        }
    }

    ///
    bool done() {
        // Explicitly forwarded for documentation purposes.
        return Base.done();
    }

    ~this() {
        if(pool !is null && taskStatus != TaskState.done) {
            yieldWait();
        }
    }
}

/**Creates a task that calls an alias.
 *
 * Examples:
 * ---
 * auto pool = new TaskPool();
 * uint[] foo = someFunction();
 *
 * // Create a task to sum this array in the background.
 * auto myTask = task!( reduce!"a + b" )(foo);
 * pool.put(myTask);
 *
 * // Do other stuff.
 *
 * // Get value.  Steals the job and executes it in this thread if it
 * // hasn't been started by a worker thread yet.
 * writeln("Sum = ", myFuture.spinWait());
 * ---
 *
 * Note:
 * This method of creating tasks allocates on the stack and requires an explicit
 * submission to the pool.  It is designed for tasks that are to finish before
 * the function in which they are created returns.  If you want to escape the
 * Task object from the function in which it was created or prefer to heap
 * allocate and automatically submit to the pool, see pool.task().
 */
Task!(fun, Args) task(alias fun, Args...)(Args args) {
    alias Task!(fun, Args) RetType;
    return RetType(args);
}

/**Create a task that calls a function pointer, delegate, or functor.
 * This works for anonymous delegates.
 *
 * Examples:
 * ---
 * auto pool = new TaskPool();
 * auto myTask = task({
 *     stderr.writeln("I've completed a task.");
 * });
 *
 * pool.put(myTask);
 *
 * // Do other stuff.
 *
 * myTask.yieldWait();
 * ---
 *
 * Notes:
 * This method of creating tasks allocates on the stack and requires an explicit
 * submission to the pool.  It is designed for tasks that are to finish before
 * the function in which they are created returns.  If you want to escape the
 * Task object from the function in which it was created or prefer to heap
 * allocate and automatically submit to the pool, see pool.task().
 *
 * In the case of delegates, this function takes a scope delegate to prevent
 * the allocation of closures, since its intended use is for tasks that will
 * be finished before the function in which they're created returns.
 * pool.task() takes a non-scope delegate and will allow the use of closures.
 */
Task!(runCallable, TypeTuple!(F, Args))
task(F, Args...)(scope F delegateOrFp, Args args)
if(is(typeof(delegateOrFp(args)))) {
    alias typeof(return) RT;
    return RT(delegateOrFp, args);
}

private struct ParallelForeachTask(R, Delegate)
if(isRandomAccessRange!R && hasLength!R) {
    enum withIndex = ParameterTypeTuple!(Delegate).length == 2;

    static void impl(void* myTask) {
        auto myCastedTask = cast(ParallelForeachTask!(R, Delegate)*) myTask;
        foreach(i; myCastedTask.lowerBound..myCastedTask.upperBound) {

            static if(lValueElements!R) {
                static if(withIndex) {
                    myCastedTask.runMe(i, myCastedTask.myRange[i]);
                } else {
                    myCastedTask.runMe( myCastedTask.myRange[i]);
                }
            } else {
                auto valToPass = myCastedTask.myRange[i];
                static if(withIndex) {
                    myCastedTask.runMe(i, valToPass);
                } else {
                    myCastedTask.runMe(valToPass);
                }
            }
        }

        // Allow some memory reclamation.
        myCastedTask.myRange = R.init;
        myCastedTask.runMe = null;
    }

    mixin BaseMixin!(TaskState.done);

    TaskPool pool;

    // More specific stuff.
    size_t lowerBound;
    size_t upperBound;
    R myRange;
    Delegate runMe;

    void wait() {
        if(pool is null) {
            // Never submitted.  No need to wait.
            return;
        }

        pool.lock();
        scope(exit) pool.unlock();

        // No work stealing here b/c the function that waits on this task
        // wants to recycle it as soon as it finishes.
        while(!done()) {
            pool.waitUntilCompletion();
        }

        if(exception) {
            throw exception;
        }
    }
}

private struct ParallelForeachTask(R, Delegate)
if(!isRandomAccessRange!R || !hasLength!R) {
    enum withIndex = ParameterTypeTuple!(Delegate).length == 2;

    static void impl(void* myTask) {
        auto myCastedTask = cast(ParallelForeachTask!(R, Delegate)*) myTask;

        static ref ElementType!(R) getElement(T)(ref T elemOrPtr) {
            static if(is(typeof(*elemOrPtr) == ElementType!R)) {
                return *elemOrPtr;
            } else {
                return elemOrPtr;
            }
        }

        foreach(i, element; myCastedTask.elements) {
            static if(withIndex) {
                size_t lValueIndex = i + myCastedTask.startIndex;
                myCastedTask.runMe(lValueIndex, getElement(element));
            } else {
                myCastedTask.runMe(getElement(element));
            }
        }

        // Make memory easier to reclaim.
        myCastedTask.runMe = null;
    }

    mixin BaseMixin!(TaskState.done);

    TaskPool pool;

    // More specific stuff.
    alias ElementType!R E;
    Delegate runMe;

    static if(lValueElements!(R)) {
        E*[] elements;
    } else {
        E[] elements;
    }
    size_t startIndex;

    void wait() {
        if(pool is null) {
            // Never submitted.  No need to wait.
            return;
        }

        pool.lock();
        scope(exit) pool.unlock();

        // No work stealing here b/c the function that waits on this task
        // wants to recycle it as soon as it finishes.

        while(!done()) {
            pool.waitUntilCompletion();
        }

        if(exception) {
            throw exception;
        }
    }
}

private struct MapTask(alias fun, R) if(isRandomAccessRange!R && hasLength!R) {
    static void impl(void* myTask) {
        auto myCastedTask = cast(MapTask!(fun, R)*) myTask;

        foreach(i; myCastedTask.lowerBound..myCastedTask.upperBound) {
            myCastedTask.results[i] = uFun(myCastedTask.range[i]);
        }

        // Nullify stuff, make GC's life easier.
        myCastedTask.results = null;
        myCastedTask.range = R.init;
    }

    mixin BaseMixin!(TaskState.done);

    TaskPool pool;

    // More specific stuff.
    alias unaryFun!fun uFun;
    R range;
    alias ElementType!R E;
    alias typeof(uFun(E.init)) ReturnType;
    ReturnType[] results;
    size_t lowerBound;
    size_t upperBound;

    void wait() {
        if(pool is null) {
            // Never submitted.  No need to wait on it.
            return;
        }

        pool.lock();
        scope(exit) pool.unlock();

        // Again, no work stealing.

        while(!done()) {
            pool.waitUntilCompletion();
        }

        if(exception) {
            throw exception;
        }
    }
}

/**Struct for creating worker-local storage.  Worker-local storage is basically
 * thread-local storage that exists only for workers in a given pool, is
 * allocated on the 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 pool that created
 * it, or the thread that created the pool that created it.  Accessing from
 * other threads will result in undefined behavior.
 *
 * Since the underlying data for this struct is heap-allocated, this struct
 * has reference semantics when passed around.
 *
 * At a more concrete level, the main use case for WorkerLocal is performing
 * parallel reductions with an imperative, as opposed to functional,
 * programming style.  Therefore, it's useful to treat WorkerLocal as local
 * to each thread for only the parallel portion of an algorithm.
 *
 * Examples:
 * ---
 * auto pool = new TaskPool;
 * auto sumParts = pool.createWorkerLocal!(uint)();
 * foreach(i; pool.parallel(iota(0U, someLargeNumber))) {
 *     // Do complicated stuff.
 *     sumParts.get += resultOfComplicatedStuff;
 * }
 *
 * writeln("Sum = ", reduce!"a + b"(sumParts.toRange));
 * ---
 */
struct WorkerLocal(T) {
private:
    size_t subtractThis;
    size_t size;

    static immutable uint cacheLineSize;
    size_t elemSize;
    bool* stillThreadLocal;

    static this() {
        size_t lineSize = 0;
        foreach(cachelevel; datacache) {
            if(cachelevel.lineSize > lineSize && cachelevel.lineSize < uint.max) {
                lineSize = cachelevel.lineSize;
            }
        }

        cacheLineSize = lineSize;
    }

    static size_t roundToLine(size_t num) pure nothrow {
        if(num % cacheLineSize == 0) {
            return num;
        } else {
            return ((num / cacheLineSize) + 1) * cacheLineSize;
        }
    }

    void* data;

    void initialize(TaskPool pool) {
        subtractThis = pool.instanceStartIndex - 1;
        size = pool.size() + 1;
        stillThreadLocal = new bool;
        *stillThreadLocal = true;

        // Determines whether the GC should scan the array.
        auto blkInfo = (typeid(T).flags & 1) ?
                       cast(GC.BlkAttr) 0 :
                       GC.BlkAttr.NO_SCAN;

        immutable nElem = pool.size + 1;
        elemSize = roundToLine(T.sizeof);

        // The + 3 is to pad one full cache line worth of space on either side
        // of the data structure to make sure false sharing with completely
        // unrelated heap data is prevented, and to provide enough padding to
        // make sure that data is cache line-aligned.
        data = GC.malloc(elemSize * (nElem + 3), blkInfo) + elemSize;

        // Cache line align data ptr.
        data = cast(void*) roundToLine(cast(size_t) data);

        foreach(i; 0..nElem) {
            this.opIndex(i) = T.init;
        }
    }

    ref T opIndex(size_t index) {
        assert(index < size, text(index, '\t', uint.max));
        return *(cast(T*) (data + elemSize * index));
    }

    void opIndexAssign(T val, size_t index) {
        assert(index < size);
        *(cast(T*) (data + elemSize * index)) = val;
    }

public:
    /**Get the current thread's instance.  Returns by reference even though
     * ddoc refuses to say so.  Undefined behavior will result if the current
     * thread is not a worker in the pool that created this instance or the
     * thread that created the pool that created this instance.
     *
     * If assertions are enabled and toRange() has been called, then this
     * WorkerLocal instance is no longer worker-local and an assertion
     * failure will result when calling this method.  This is not checked
     * when assertions are disabled for performance reasons.
     */
    ref T get() @property {
        assert(*stillThreadLocal,
               "Cannot call get() on this instance of WorkerLocal because it" ~
               " is no longer worker-local."
        );
        auto index = TaskPool.threadIndex;

        // Assume it's the thread that created the pool.
        if(index < subtractThis) {
            index = subtractThis;
        }
        return opIndex(index - subtractThis);
    }

    /**Assign a value to the current thread's instance.  This function has
     * the same caveats as its overload.
     */
    void get(T val) @property {
        assert(*stillThreadLocal,
               "Cannot call get() on this instance of WorkerLocal because it" ~
               " is no longer worker-local."
        );

        auto index = TaskPool.threadIndex;

        // Assume it's the thread that created the pool.
        if(index < subtractThis) {
            index = subtractThis;
        }
        opIndexAssign(val, index - subtractThis);
    }

    /**Returns a range view of the values for all threads, which can be used
     * to do stuff with the results of each thread after running the parallel
     * part of your algorithm.  Do NOT use this method in the parallel portion
     * of your algorithm.
     *
     * Calling this function will also set a flag indicating
     * that this struct is no longer thread-local, and attempting to use the
     * get() method again will result in an assertion failure if assertions
     * are enabled.
     */
    WorkerLocalRange!T toRange() @property {
        if(*stillThreadLocal) {
            *stillThreadLocal = false;

            // Make absolutely sure results are visible to all threads.
            synchronized {}
        }

       return WorkerLocalRange!(T)(this);
    }
}

/**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.
 */
struct WorkerLocalRange(T) {
private:
    WorkerLocal!T workerLocal;

    size_t _length;
    size_t beginOffset;

    this(WorkerLocal!(T) wl) {
        this.workerLocal = wl;
        _length = wl.size;
    }

public:
    ///
    ref T front() @property {
        return this[0];
    }

    ///
    ref T back() @property {
        return this[_length - 1];
    }

    ///
    void popFront() {
        if(_length > 0) {
            beginOffset++;
            _length--;
        }
    }

    ///
    void popBack() {
        if(_length > 0) {
            _length--;
        }
    }

    ///
    typeof(this) save() @property {
        return this;
    }

    ///
    ref T opIndex(size_t index) {
        assert(index < _length);
        return workerLocal[index + beginOffset];
    }

    ///
    void opIndexAssign(T val, size_t index) {
        assert(index < _length);
        workerLocal[index] = val;
    }

    ///
    bool empty() @property {
        return length == 0;
    }

    ///
    size_t length() @property {
        return _length;
    }
}

/**The task pool class that is the workhorse of this library.
 */
class TaskPool {
private:
    Thread[] pool;
    AbstractTask* head;
    AbstractTask* tail;
    PoolState status = PoolState.running;  // All operations on this are done atomically.
    Condition workerCondition;
    Condition waiterCondition;
    Mutex mutex;

    // The instanceStartIndex of the next instance that will be created.
    __gshared static size_t nextInstanceIndex = 1;

    // The index of the current thread.
    static size_t threadIndex;

    // The index of the first thread in the next instance.
    immutable size_t instanceStartIndex;

    // The index that the next thread to be initialized in this pool will have.
    size_t nextThreadIndex;

    enum PoolState : ubyte {
        running,
        finishing,
        stopNow
    }

    void doJob(AbstractTask* job) {
        assert(job.taskStatus == TaskState.inProgress);
        assert(job.next is null);
        assert(job.prev is null);

        scope(exit) {
            lock();
            notifyWaiters();
            unlock();
        }

        try {
            job.job();
            if(job.shouldSetDone) {
                atomicSetUbyte(job.taskStatus, TaskState.done);
            }
        } catch(Object e) {
            job.exception = e;
            if(job.shouldSetDone) {
                atomicSetUbyte(job.taskStatus, TaskState.done);
            }
        }
    }

    void workLoop() {
        // Initialize thread index.
        synchronized(this) {
            threadIndex = nextThreadIndex;
            nextThreadIndex++;
        }

        while(atomicReadUbyte(status) != PoolState.stopNow) {
            AbstractTask* task = pop();
            if (task is null) {
                if(atomicReadUbyte(status) == PoolState.finishing) {
                    atomicSetUbyte(status, PoolState.stopNow);
                    return;
                }
            } else {
                doJob(task);
            }
        }
    }

    bool deleteItem(AbstractTask* item) {
        lock();
        auto ret = deleteItemNoSync(item);
        unlock();
        return ret;
    }

    bool deleteItemNoSync(AbstractTask* item)
    out {
        assert(item.next is null);
        assert(item.prev is null);
    } body {
        if(item.taskStatus != TaskState.notStarted) {
            return false;
        }
        item.taskStatus = TaskState.inProgress;

        if(item is head) {
            // Make sure head gets set properly.
            popNoSync();
            return true;;
        }
        if(item is tail) {
            tail = tail.prev;
            if(tail !is null) {
                tail.next = null;
            }
            item.next = null;
            item.prev = null;
            return true;
        }
        if(item.next !is null) {
            assert(item.next.prev is item);  // Check queue consistency.
            item.next.prev = item.prev;
        }
        if(item.prev !is null) {
            assert(item.prev.next is item);  // Check queue consistency.
            item.prev.next = item.next;
        }
        item.next = null;
        item.prev = null;
        return true;
    }

    // Pop a task off the queue.  Should only be called by worker threads.
    AbstractTask* pop() {
        lock();
        auto ret = popNoSync();
        if(ret is null && status == PoolState.running) {
            wait();
           // stderr.writeln("AWAKE");
        }
        unlock();
        return ret;
    }

    AbstractTask* popNoSync()
    out(returned) {
        /* If task.prev and task.next aren't null, then another thread
         * can try to delete this task from the pool after it's
         * alreadly been deleted/popped.
         */
        if(returned !is null) {
            assert(returned.next is null);
            assert(returned.prev is null);
        }
    } body {
        AbstractTask* returned = head;
        if (head !is null) {
            head = head.next;
            returned.prev = null;
            returned.next = null;
            returned.taskStatus = TaskState.inProgress;
        }
        if(head !is null) {
            head.prev = null;
        }

        return returned;
    }

    // Push a task onto the queue.
    void abstractPut(AbstractTask* task) {
        lock();
        abstractPutNoSync(task);
        unlock();
    }

    void abstractPutNoSync(AbstractTask* task)
    out {
        assert(tail.prev !is tail);
        assert(tail.next is null, text(tail.prev, '\t', tail.next));
        if(tail.prev !is null) {
            assert(tail.prev.next is tail, text(tail.prev, '\t', tail.next));
        }
    } body {
        task.next = null;
        if (head is null) { //Queue is empty.
            head = task;
            tail = task;
            tail.prev = null;
        } else {
            task.prev = tail;
            tail.next = task;
            tail = task;
        }
        notify();
    }

    // Same as trySteal, but also deletes the task from the queue so the
    // Task object can be recycled.
    bool tryStealDelete(AbstractTask* toSteal) {
        if( !deleteItem(toSteal) ) {
            return false;
        }

        toSteal.job();

        /* shouldSetDone should always be true except if the task re-submits
         * itself to the pool and needs to bypass this.*/
        if(toSteal.shouldSetDone == 1) {
            atomicSetUbyte(toSteal.taskStatus, TaskState.done);
        }

        return true;
    }

    size_t defaultBlockSize(size_t rangeLen) {
        if(this.size == 0) {
            return rangeLen;
        }

        immutable size_t twoSize = 2 * (this.size + 1);
        return (rangeLen / twoSize) + ((rangeLen % twoSize == 0) ? 0 : 1);
    }

    void lock() {
        mutex.lock();
    }

    void unlock() {
        mutex.unlock();
    }

    void wait() {
        workerCondition.wait();
    }

    void notify() {
        workerCondition.notify();
    }

    void notifyAll() {
        workerCondition.notifyAll();
    }

    void waitUntilCompletion() {
        waiterCondition.wait();
    }

    void notifyWaiters() {
        waiterCondition.notifyAll();
    }

public:

    /**Default constructor that initializes a TaskPool with
     * however many cores are on your CPU, minus 1 because the thread
     * that initialized the pool will also do work.
     *
     * Note:  Initializing a pool with zero threads (as would happen in the
     * case of a single-core CPU) is well-tested and does work.
     */
    this() {
        this(coresPerCPU - 1);
    }

    /**Allows for custom poolSize.*/
    this(size_t poolSize) {
        synchronized(TaskPool.classinfo) {
            instanceStartIndex = nextInstanceIndex;

            // The first worker thread to be initialized will have this index,
            // and will increment it.  The second worker to be initialized will
            // have this index plus 1.
            nextThreadIndex = instanceStartIndex;

            nextInstanceIndex += poolSize;
        }

        mutex = new Mutex(this);
        workerCondition = new Condition(mutex);
        waiterCondition = new Condition(mutex);

        pool = new Thread[poolSize];
        foreach(ref poolThread; pool) {
            poolThread = new Thread(&workLoop);
            poolThread.start();
        }
    }

    /**Gets the index of the current thread relative to this pool.  The thread
     * that created this pool receives an index of 0.  The worker threads in
     * this pool receive indices of 1 through poolSize.
     *
     * The worker index is useful mainly for maintaining worker-local storate.
     *
     * Bugs:  Subject to integer overflow errors if more than size_t.max threads
     *        are ever created during the course of a program's execution.  This
     *        will likely never be fixed because it's an extreme corner case
     *        on 32-bit and it's completely implausible on 64-bit.
     *
     *        Will silently return undefined results for threads that are not
     *        workers in this pool and did not create this pool.
     */
    final size_t workerIndex() {
        immutable rawInd = threadIndex;
        return (rawInd >= instanceStartIndex) ?
                (rawInd - instanceStartIndex + 1) : 0;
    }

    /**Create 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.
     */
    WorkerLocal!(T) createWorkerLocal(T)(lazy T initialVal = T.init) {
        WorkerLocal!(T) ret;
        ret.initialize(this);
        foreach(i; 0..size + 1) {
            ret[i] = initialVal;
        }
        synchronized {}  // Make sure updates are visible in all threads.
        return ret;
    }

    /**Kills pool immediately w/o waiting for jobs to finish.  Use only if you
     * have waitied on every job and therefore know there can't possibly be more
     * in queue, or if you speculatively executed a bunch of stuff and realized
     * you don't need those results anymore.
     *
     * Note:  Does not affect jobs that are already executing, only those
     * in queue.
     */
    void stop() {
        lock();
        scope(exit) unlock();
        atomicSetUbyte(status, PoolState.stopNow);
        notifyAll();
    }

    /// Waits for all jobs to finish, then shuts down the pool.
    void waitStop() {
        finish();
        foreach(t; pool) {
            t.join();
        }
    }

    /**Instructs worker threads to stop when the queue becomes empty, but does
     * not block.
     */
    void finish() {
        lock();
        scope(exit) unlock();
        atomicCasUbyte(status, PoolState.running, PoolState.finishing);
        notifyAll();
    }

    /// Returns the number of worker threads in the pool.
    final uint size() {
        return pool.length;
    }

    // Kept public for backwards compatibility, but not documented.
    // Using ref parameters is a nicer API and is made safe because the
    // d'tor for Task waits until the task is finished before destroying the
    // stack frame.  This function will eventually be made private and/or
    // deprecated.
    void put(alias fun, Args...)(Task!(fun, Args)* task) {
        task.pool = this;
        abstractPut( cast(AbstractTask*) task);
    }

    /**Put a task on the queue.
    *
    * Note:  While this function takes the address of variables that may
    * potentially be on the stack, it is safe and will be marked as @trusted
    * once SafeD is fully implemented.  Task objects include a destructor that
    * waits for the task to complete before destroying the stack frame that
    * they are allocated on.  Therefore, it is impossible for the stack
    * frame to be destroyed before the task is complete and out of the queue.
    */
    void put(alias fun, Args...)(ref Task!(fun, Args) task) {
        task.pool = this;
        abstractPut( cast(AbstractTask*) &task);
    }

    /**Convenience method that automatically creates a Task calling an alias on
     * the GC heap and submits it to the pool.  See examples for the
     * non-member function task().
     *
     * Returns:  A pointer to the Task object.
     */
    Task!(fun, Args)* task(alias fun, Args...)(Args args) {
        auto ret = toHeap(.task!(fun)(args));
        put(ret);
        return ret;
    }

    /**Convenience method that automatically creates a Task calling a delegate,
     * function pointer, or functor on the GC heap and submits it to the pool.
     * See examples for the non-member function task().
     *
     * Returns:  A pointer to the Task object.
     *
     * Note:  This function takes a non-scope delegate, meaning it can be
     * used with closures.  If you can't allocate a closure due to objects
     * on the stack that have scoped destruction, see the global function
     * task(), which takes a scope delegate.
     */
     Task!(runCallable, TypeTuple!(F, Args))*
     task(F, Args...)(F delegateOrFp, Args args)
     if(is(ReturnType!(F))) {
         auto ptr = toHeap(.task(delegateOrFp, args));
         put(ptr);
         return ptr;
     }

    /**Implements a parallel foreach loop over a range.  blockSize is the
     * number of elements to process in one work unit.
     *
     * Examples:
     * ---
     * auto pool = new TaskPool();
     *
     * uint[] squares = new uint[1_000];
     * foreach(i; pool.parallel( iota(0, foo.length), 100)) {
     *     // Iterate over foo using work units of size 100.
     *     squares[i] = i * i;
     * }
     * ---
     *
     * Note:  Since breaking from a loop that's being executed in parallel
     * doesn't make much sense, it is considered undefined behavior in this
     * implementation of parallel foreach.
     *
     */
    ParallelForeach!R parallel(R)(R range, size_t blockSize) {
        alias ParallelForeach!R RetType;
        return RetType(this, range, blockSize);
    }

    /**Parallel foreach with default block size.  For ranges that don't have
     * a length, the default is 512 elements.  For ranges that do, the default
     * is whatever number would create exactly twice as many work units as
     * we have worker threads.
     */
    ParallelForeach!R parallel(R)(R range) {
        static if(hasLength!R) {
            // Default block size is such that we would use 2x as many
            // slots as are in this thread pool.
            size_t blockSize = defaultBlockSize(range.length);
            return parallel(range, blockSize);
        } else {
            // Just use a really, really dumb guess if the user is too lazy to
            // specify.
            return parallel(range, 512);
        }
    }

    /**Parallel map.  Unlike std.algorithm.map, this is evaluated eagerly
     * because it wouldn't make sense to evaluate a parallel map lazily.
     *
     * fun is the function to be evaluated, range is the range to evaluate
     * this function on.  range must be a random access range with length
     * for now, though this restriction may be lifted in the future.
     * blockSize is the size of the work unit to submit to the thread pool,
     * in elements.  buf is an optional buffer to store the results in.
     * If none is provided, one is allocated.  If one is provided, it
     * must have the same length as range.
     *
     * Examples:
     * ---
     * auto pool = new TaskPool();
     *
     * real[] numbers = new real[1_000];
     * foreach(i, ref num; numbers) {
     *     num = i;
     * }
     *
     * // Find the squares of numbers[] using work units of size 100.
     * real[] squares = pool.map!"a * a"(numbers, 100);
     * ---
     */
    // Bug:  Weird behavior trying to do the overload the normal way.
    // Doing it with templating on the type of blockSize b/c that's the
    // only way it compiles.
    MapType!(fun, R)[] map(alias fun, R, I : size_t)(R range, I blockSize,
        MapType!(fun, R)[] buf = null) {
        immutable len = range.length;

        if(buf.length == 0) {
            // Create buffer without initializing contents.
            alias MapType!(fun, R) MT;
            GC.BlkAttr gcFlags = (typeid(MT).flags & 1) ?
                                  cast(GC.BlkAttr) 0 :
                                  GC.BlkAttr.NO_SCAN;
            auto myPtr = cast(MT*) GC.malloc(len * MT.sizeof, gcFlags);
            buf = myPtr[0..len];
        }
        enforce(buf.length == len,
            text("Can't use a user supplied buffer that's the wrong size.  ",
            "(Expected  :", len, " Got:  ", buf.length));
        if(blockSize > len) {
            blockSize = len;
        }

        // Handle as a special case:
        if(size == 0) {
            size_t index = 0;
            foreach(elem; range) {
                buf[index++] = unaryFun!(fun)(elem);
            }
            return buf;
        }

        alias MapTask!(fun, R) MTask;
        MTask[] tasks = (cast(MTask*) alloca(this.size * MTask.sizeof * 2))
                        [0..this.size * 2];
        tasks[] = MTask.init;

        size_t curPos;
        void useTask(ref MTask task) {
            task.lowerBound = curPos;
            task.upperBound = min(len, curPos + blockSize);
            task.range = range;
            task.results = buf;
            task.pool = this;
            curPos += blockSize;

            lock();
            atomicSetUbyte(task.taskStatus, TaskState.notStarted);
            abstractPutNoSync(cast(AbstractTask*) &task);
            unlock();
        }

        ubyte doneSubmitting = 0;

        Task!(runCallable, void delegate()) submitNextBatch;

        void submitJobs() {
            // Search for slots, then sleep.
            foreach(ref task; tasks) if(task.done) {
                useTask(task);
                if(curPos >= len) {
                    atomicSetUbyte(doneSubmitting, 1);
                    return;
                }
            }

            // Now that we've submitted all the worker tasks, submit
            // the next submission task.  Synchronizing on the pool
            // to prevent the stealing thread from deleting the job
            // before it's submitted.
            lock();
            atomicSetUbyte(submitNextBatch.taskStatus, TaskState.notStarted);
            abstractPutNoSync( cast(AbstractTask*) &submitNextBatch);
            unlock();
        }

        submitNextBatch = .task(&submitJobs);

        // The submitAndSteal mixin relies on the TaskPool instance
        // being called pool.
        TaskPool pool = this;

        mixin(submitAndSteal);

        return buf;
    }

    /**Parallel map with default block size.*/
    MapType!(fun, R)[] map(alias fun, R)(R range, MapType!(fun, R)[] buf = null) {
        size_t blkSize = defaultBlockSize(range.length);
        alias map!(fun, R, size_t) mapFun;

        return mapFun(range, blkSize, buf);
    }

    /**Parallel reduce.  For now, the range must offer random access and have
     * a length.  In the future, this restriction may be lifted.
     *
     * Note:  Because this operation is being carried out in parallel,
     * fun must be associative.  For notational simplicity, let # be an
     * infix operator representing fun.  Then, (a # b) # c must equal
     * a # (b # c).  This is NOT the same thing as commutativity.  Matrix
     * multiplication, for example, is associative but not commutative.
     *
     * Examples:
     * ---
     * // Find the max of an array in parallel.  Note that this is a toy example
     * // and unless the comparison function was very expensive, it would
     * // almost always be faster to do this in serial.
     *
     * auto pool = new TaskPool();
     *
     * auto myArr = somethingExpensiveToCompare();
     * auto myMax = pool.reduce!max(myArr);
     * ---
     */
    ReduceType!(fun, R, E)
    reduce(alias fun, R, E)(E startVal, R range, size_t blockSize = 0) {

        if(size == 0) {
            return std.algorithm.reduce!(fun)(startVal, range);
        }

        // Unlike the rest of the functions here, I can't use the Task object
        // recycling trick here because this has to work on non-commutative
        // operations.  After all the tasks are done executing, fun() has to
        // be applied on the results of these to get a final result, but
        // it can't be evaluated out of order.

        immutable len = range.length;
        if(blockSize == 0) {
            blockSize = defaultBlockSize(len);
        }

        if(blockSize > len) {
            blockSize = len;
        }

        immutable size_t nWorkUnits = (len / blockSize) +
            ((len % blockSize == 0) ? 0 : 1);
        assert(nWorkUnits * blockSize >= len);

        alias binaryFun!fun compiledFun;

        static E reduceOnRange
        (E startVal, R range, size_t lowerBound, size_t upperBound) {
            E result = startVal;
            foreach(i; lowerBound..upperBound) {
                result = compiledFun(result, range[i]);
            }
            return result;
        }

        alias Task!(reduceOnRange, E, R, size_t, size_t) RTask;
        RTask[] tasks;

        enum MAX_STACK = 512;
        immutable size_t nBytesNeeded = nWorkUnits * RTask.sizeof;

        if(nBytesNeeded < MAX_STACK) {
            tasks = (cast(RTask*) alloca(nBytesNeeded))[0..nWorkUnits];
            tasks[] = RTask.init;
        } else {
            tasks = new RTask[nWorkUnits];
        }

        size_t curPos = 0;
        void useTask(ref RTask task) {
            task.args[2] = curPos + 1; // lower bound.
            task.args[3] = min(len, curPos + blockSize);  // upper bound.
            task.args[1] = range;  // range
            task.args[0] = range[curPos];  // Start val.

            curPos += blockSize;
            put(task);
        }

        foreach(ref task; tasks) {
            useTask(task);
        }

        // Try to steal each of these.
        foreach(ref task; tasks) {
            tryStealDelete( cast(AbstractTask*) &task);
        }

        // Now that we've tried to steal every task, they're all either done
        // or in progress.  Wait on all of them.
        E result = startVal;
        foreach(ref task; tasks) {
            task.yieldWait();
            result = compiledFun(result, task.returnVal);
        }
        return result;
    }

    /**Parallel reduce with the first element of the range as the start value.*/
    ReduceType!(fun, R, ElementType!R)
    reduce(alias fun, R)(R range, size_t blockSize = 0) {
        enforce(!range.empty,
            "Cannot reduce an empty range with first element as start value.");
        auto startVal = range.front;
        range.popFront;
        return reduce!(fun, R, ElementType!R)(startVal, range, blockSize);
    }
}

// Where the magic happens.  This mixin causes tasks to be submitted lazily to
// the task pool.  Attempts are then made by the calling thread to steal
// them.
enum submitAndSteal = q{

    // See documentation for BaseMixin.shouldSetDone.
    submitNextBatch.shouldSetDone = false;

    // Submit first batch from this thread.
    submitJobs();

    while( !atomicReadUbyte(doneSubmitting) ) {
        // Try to steal parallel foreach tasks.
        foreach(ref task; tasks) {
            pool.tryStealDelete( cast(AbstractTask*) &task);
        }

        // All tasks in progress or done unless next
        // submission task started running.  Try to steal the submission task.
        pool.tryStealDelete(cast(AbstractTask*) &submitNextBatch);
    }

    // Steal one last time, after they're all submitted.
    foreach(ref task; tasks) {
        pool.tryStealDelete( cast(AbstractTask*) &task);
    }


    foreach(ref task; tasks) {
        task.wait();
    }
};

/*------Structs that implement opApply for parallel foreach.------------------*/
template randLen(R) {
    enum randLen = isRandomAccessRange!R && hasLength!R;
}

private enum string parallelApplyMixin = q{
    alias ParallelForeachTask!(R, typeof(dg)) PTask;

    // Handle empty thread pool as special case.
    if(pool.size == 0) {
        int res = 0;
        size_t index = 0;

        // The explicit ElementType!R in the foreach loops is necessary for
        // correct behavior when iterating over strings.
        static if(lValueElements!(R)) {
            foreach(ref ElementType!R elem; range) {
                static if(ParameterTypeTuple!(dg).length == 2) {
                    res = dg(index, elem);
                } else {
                    res = dg(elem);
                }
                index++;
            }
        } else {
            foreach(ElementType!R elem; range) {
                static if(ParameterTypeTuple!(dg).length == 2) {
                    res = dg(index, elem);
                } else {
                    res = dg(elem);
                }
                index++;
            }
        }
        return res;
    }

    PTask[] tasks = (cast(PTask*) alloca(pool.size * PTask.sizeof * 2))
                    [0..pool.size * 2];
    tasks[] = PTask.init;
    Task!(runCallable, void delegate()) submitNextBatch;

    static if(randLen!R) {

        immutable size_t len = range.length;
        size_t curPos = 0;

        void useTask(ref PTask task) {
            task.lowerBound = curPos;
            task.upperBound = min(len, curPos + blockSize);
            task.myRange = range;
            task.runMe = dg;
            task.pool = pool;
            curPos += blockSize;

            pool.lock();
            atomicSetUbyte(task.taskStatus, TaskState.notStarted);
            pool.abstractPutNoSync(cast(AbstractTask*) &task);
            pool.unlock();
        }

        void submitJobs() {
            // Search for slots to recycle.
            foreach(ref task; tasks) if(task.done) {
                useTask(task);
                if(curPos >= len) {
                    atomicSetUbyte(doneSubmitting, 1);
                    return;
                }
            }

            // Now that we've submitted all the worker tasks, submit
            // the next submission task.  Synchronizing on the pool
            // to prevent the stealing thread from deleting the job
            // before it's submitted.
            pool.lock();
            atomicSetUbyte(submitNextBatch.taskStatus, TaskState.notStarted);
            pool.abstractPutNoSync( cast(AbstractTask*) &submitNextBatch);
            pool.unlock();
        }

    } else {

        void useTask(ref PTask task) {
            task.runMe = dg;
            task.pool = pool;
            size_t copyIndex = 0;
            if(task.elements.length == 0) {
                task.elements.length = blockSize;
            }

            for(; copyIndex < blockSize && !range.empty; copyIndex++) {
                static if(lValueElements!R) {
                    task.elements[copyIndex] = &range.front();
                } else {
                    task.elements[copyIndex] = range.front;
                }
                range.popFront;
            }

            // We only actually change the array  size on the last task, when the
            // range is empty.
            task.elements = task.elements[0..copyIndex];

            pool.lock();
            task.startIndex = this.startIndex;
            this.startIndex += copyIndex;
            atomicSetUbyte(task.taskStatus, TaskState.notStarted);
            pool.abstractPutNoSync(cast(AbstractTask*) &task);
            pool.unlock();
        }


        void submitJobs() {
            // Search for slots to recycle.
            foreach(ref task; tasks) if(task.done) {
                useTask(task);
                if(range.empty) {
                    atomicSetUbyte(doneSubmitting, 1);
                    return;
                }
            }

            // Now that we've submitted all the worker tasks, submit
            // the next submission task.  Synchronizing on the pool
            // to prevent the stealing thread from deleting the job
            // before it's submitted.
            pool.lock();
            atomicSetUbyte(submitNextBatch.taskStatus, TaskState.notStarted);
            pool.abstractPutNoSync( cast(AbstractTask*) &submitNextBatch);
            pool.unlock();
        }

    }
    submitNextBatch = task(&submitJobs);

    mixin(submitAndSteal);

    return 0;
};

private struct ParallelForeach(R) {
    TaskPool pool;
    R range;
    size_t blockSize;
    size_t startIndex;
    ubyte doneSubmitting;

    alias ElementType!R E;

    int opApply(scope int delegate(ref E) dg) {
        mixin(parallelApplyMixin);
    }

    int opApply(scope int delegate(ref size_t, ref E) dg) {
        mixin(parallelApplyMixin);
    }
}

version(unittest) {
    // This was the only way I could get nested maps to work.
    __gshared TaskPool poolInstance;
}

// These unittests are intended to also function as an example of how to
// use this module.
unittest {
    foreach(attempt; 0..10)
    foreach(poolSize; [0, 4]) {
        // Create a TaskPool object with the default number of threads.
        poolInstance = new TaskPool(poolSize);

        // Create some data to work on.
        uint[] numbers = new uint[1_000];

        // Fill in this array in parallel, using default block size.
        // Note:  Be careful when writing to adjacent elements of an arary from
        // different threads, as this can cause word tearing bugs when
        // the elements aren't properly aligned or aren't the machine's native
        // word size.  In this case, though, we're ok.
        foreach(i; poolInstance.parallel( iota(0, numbers.length)) ) {
            numbers[i] = i;
        }

        // Make sure it works.
        foreach(i; 0..numbers.length) {
            assert(numbers[i] == i);
        }

        stderr.writeln("Done creating nums.");

        // Parallel foreach also works on non-random access ranges, albeit
        // less efficiently.
        auto myNumbers = filter!"a % 7 > 0"( iota(0, 1000));
        foreach(num; poolInstance.parallel(myNumbers)) {
            assert(num % 7 > 0 && num < 1000);
        }
        stderr.writeln("Done modulus test.");

        // Use parallel map to calculate the square of each element in numbers,
        // and make sure it's right.
        uint[] squares = poolInstance.map!"a * a"(numbers, 100);
        assert(squares.length == numbers.length);
        foreach(i, number; numbers) {
            assert(squares[i] == number * number);
        }
        stderr.writeln("Done squares.");

        // Sum up the array in parallel with the current thread.
        auto sumFuture = poolInstance.task!( reduce!"a + b" )(numbers);

        // Go off and do other stuff while that future executes:
        // Find the sum of squares of numbers.
        ulong sumSquares = 0;
        foreach(elem; numbers) {
            sumSquares += elem * elem;
        }

        // Ask for our result.  If the pool has not yet started working on
        // this task, spinWait() automatically steals it and executes it in this
        // thread.
        uint mySum = sumFuture.spinWait();
        assert(mySum == 999 * 1000 / 2);

        // We could have also computed this sum in parallel using parallel
        // reduce.
        auto mySumParallel = poolInstance.reduce!"a + b"(numbers);
        assert(mySum == mySumParallel);
        stderr.writeln("Done sums.");

        // Execute an anonymous delegate as a task.
        auto myTask = task({
            synchronized writeln("Our lives are parallel...Our lives are parallel.");
        });
        poolInstance.put(myTask);

        // Parallel foreach loops can also be nested, and can have an index
        // variable attached to the foreach loop.
        auto nestedOuter = "abcd";
        auto nestedInner =  iota(0, 10, 2);

        foreach(i, letter; poolInstance.parallel(nestedOuter, 1)) {
            foreach(j, number; poolInstance.parallel(nestedInner, 1)) {
                synchronized writeln
                    (i, ": ", letter, "  ", j, ": ", number);
            }
        }

        // Block until all jobs are finished and then shut down the thread pool.
        poolInstance.waitStop();
    }

    writeln("Press enter to go to next round of unittests.");
    readln();
}

// These unittests are intended more for actual testing and not so much
// as examples.
unittest {
    foreach(attempt; 0..10)
    foreach(poolSize; [0, 4]) {
        poolInstance = new TaskPool(poolSize);

        // Test indexing.
        stderr.writeln("Creator Raw Index:  ", poolInstance.threadIndex);
        assert(poolInstance.workerIndex() == 0);

        // Test worker-local storage.
        auto workerLocal = poolInstance.createWorkerLocal!(uint)(1);
        foreach(i; poolInstance.parallel(iota(0U, 1_000_000))) {
            workerLocal.get++;
        }
        assert(reduce!"a + b"(workerLocal.toRange) ==
            1_000_000 + poolInstance.size + 1);

        // Make sure work is reasonably balanced among threads.  This test is
        // non-deterministic and is more of a sanity check than something that
        // has an absolute pass/fail.
        uint[void*] nJobsByThread;
        foreach(thread; poolInstance.pool) {
            nJobsByThread[cast(void*) thread] = 0;
        }
        nJobsByThread[ cast(void*) Thread.getThis] = 0;

        foreach(i; poolInstance.parallel( iota(0, 1_000_000), 100 )) {
            atomicIncUint( nJobsByThread[ cast(void*) Thread.getThis() ]);
        }

        stderr.writeln("\nCurrent (stealing) thread is:  ",
            cast(void*) Thread.getThis());
        stderr.writeln("Workload distribution:  ");
        foreach(k, v; nJobsByThread) {
            stderr.writeln(k, '\t', v);
        }

        // Test whether map can be nested.
        real[][] matrix = new real[][](1000, 1000);
        foreach(i; poolInstance.parallel( iota(0, matrix.length) )) {
            foreach(j; poolInstance.parallel( iota(0, matrix[0].length) )) {
                matrix[i][j] = i * j;
            }
        }

        // Get around weird bugs having to do w/ sqrt being an intrinsic:
        static real mySqrt(real num) {
            return sqrt(num);
        }

        static real[] parallelSqrt(real[] nums) {
            return poolInstance.map!mySqrt(nums);
        }

        real[][] sqrtMatrix = poolInstance.map!parallelSqrt(matrix);

        foreach(i, row; sqrtMatrix) {
            foreach(j, elem; row) {
                real shouldBe = sqrt( cast(real) i * j);
                assert(approxEqual(shouldBe, elem));
                sqrtMatrix[i][j] = shouldBe;
            }
        }

        auto saySuccess = task({
            stderr.writeln(
                "Success doing matrix stuff that involves nested pool use.");
        });
        poolInstance.put(saySuccess);
        saySuccess.workWait();

        // A more thorough test of map, reduce:  Find the sum of the square roots of
        // matrix.

        static real parallelSum(real[] input) {
            return poolInstance.reduce!"a + b"(input);
        }

        auto sumSqrt = poolInstance.reduce!"a + b"(
            poolInstance.map!parallelSum(
                sqrtMatrix
            )
        );

        assert(approxEqual(sumSqrt, 4.437e8));
        stderr.writeln("Done sum of square roots.");

        // Test whether tasks work with function pointers.
        auto nanTask = poolInstance.task(&isNaN, 1.0L);
        assert(nanTask.spinWait == false);

        if(poolInstance.size > 0) {
            // Test work waiting.
            static void uselessFun() {
                foreach(i; 0..1_000_000) {}
            }

            auto uselessTasks = new typeof(task(&uselessFun))[1000];
            foreach(ref uselessTask; uselessTasks) {
                uselessTask = task(&uselessFun);
            }
            foreach(ref uselessTask; uselessTasks) {
                poolInstance.put(uselessTask);
            }
            foreach(ref uselessTask; uselessTasks) {
                uselessTask.workWait();
            }
        }

        // Test the case of non-random access + ref returns.
        int[] nums = [1,2,3,4,5];
        static struct RemoveRandom {
            int[] arr;

            ref int front() { return arr.front; }
            void popFront() { arr.popFront(); }
            bool empty() { return arr.empty; }
        }

        auto refRange = RemoveRandom(nums);
        foreach(ref elem; poolInstance.parallel(refRange)) {
            elem++;
        }
        assert(nums == [2,3,4,5,6]);
        stderr.writeln("Nums:  ", nums);

        poolInstance.stop();
    }
}

version(unittest) {
    void main() {}
}