btTaskScheduler.cpp

Go to the documentation of this file.

#include "LinearMath/btMinMax.h"
#include "LinearMath/btAlignedObjectArray.h"
#include "LinearMath/btThreads.h"
#include "LinearMath/btQuickprof.h"
#include <stdio.h>
#include <algorithm>



#if BT_THREADSAFE

#include "btThreadSupportInterface.h"

#if defined( _WIN32 )

#define WIN32_LEAN_AND_MEAN

#include <windows.h>

#endif


typedef unsigned long long btU64;
static const int kCacheLineSize = 64;

void btSpinPause()
{
#if defined( _WIN32 )
    YieldProcessor();
#endif
}


struct WorkerThreadStatus
{
    enum Type
    {
        kInvalid,
        kWaitingForWork,
        kWorking,
        kSleeping,
    };
};


ATTRIBUTE_ALIGNED64(class) WorkerThreadDirectives
{
    static const int kMaxThreadCount = BT_MAX_THREAD_COUNT;
    // directives for all worker threads packed into a single cacheline
    char m_threadDirs[kMaxThreadCount];

public:
    enum Type
    {
        kInvalid,
        kGoToSleep,         // go to sleep
        kStayAwakeButIdle,  // wait for not checking job queue
        kScanForJobs,       // actively scan job queue for jobs
    };
    WorkerThreadDirectives()
    {
        for ( int i = 0; i < kMaxThreadCount; ++i )
        {
            m_threadDirs[ i ] = 0;
        }
    }

    Type getDirective(int threadId)
    {
        btAssert(threadId < kMaxThreadCount);
        return static_cast<Type>(m_threadDirs[threadId]);
    }

    void setDirectiveByRange(int threadBegin, int threadEnd, Type dir)
    {
        btAssert( threadBegin < threadEnd );
        btAssert( threadEnd <= kMaxThreadCount );
        char dirChar = static_cast<char>(dir);
        for ( int i = threadBegin; i < threadEnd; ++i )
        {
            m_threadDirs[ i ] = dirChar;
        }
    }
};

class JobQueue;

ATTRIBUTE_ALIGNED64(struct) ThreadLocalStorage
{
    int m_threadId;
    WorkerThreadStatus::Type m_status;
    int m_numJobsFinished;
    btSpinMutex m_mutex;
    btScalar m_sumResult;
    WorkerThreadDirectives * m_directive;
    JobQueue* m_queue;
    btClock* m_clock;
    unsigned int m_cooldownTime;
};


struct IJob
{
    virtual void executeJob(int threadId) = 0;
};

class ParallelForJob : public IJob
{
    const btIParallelForBody* m_body;
    int m_begin;
    int m_end;

public:
    ParallelForJob( int iBegin, int iEnd, const btIParallelForBody& body )
    {
        m_body = &body;
        m_begin = iBegin;
        m_end = iEnd;
    }
    virtual void executeJob(int threadId) BT_OVERRIDE
    {
        BT_PROFILE( "executeJob" );

        // call the functor body to do the work
        m_body->forLoop( m_begin, m_end );
    }
};


class ParallelSumJob : public IJob
{
    const btIParallelSumBody* m_body;
    ThreadLocalStorage* m_threadLocalStoreArray;
    int m_begin;
    int m_end;

public:
    ParallelSumJob( int iBegin, int iEnd, const btIParallelSumBody& body, ThreadLocalStorage* tls )
    {
        m_body = &body;
        m_threadLocalStoreArray = tls;
        m_begin = iBegin;
        m_end = iEnd;
    }
    virtual void executeJob( int threadId ) BT_OVERRIDE
    {
        BT_PROFILE( "executeJob" );

        // call the functor body to do the work
        btScalar val = m_body->sumLoop( m_begin, m_end );
#if BT_PARALLEL_SUM_DETERMINISTISM
        // by truncating bits of the result, we can make the parallelSum deterministic (at the expense of precision)
        const float TRUNC_SCALE = float(1<<19);
        val = floor(val*TRUNC_SCALE+0.5f)/TRUNC_SCALE;  // truncate some bits
#endif
        m_threadLocalStoreArray[threadId].m_sumResult += val;
    }
};


ATTRIBUTE_ALIGNED64(class) JobQueue
{
    btThreadSupportInterface* m_threadSupport;
    btCriticalSection* m_queueLock;
    btSpinMutex m_mutex;

    btAlignedObjectArray<IJob*> m_jobQueue;
    char* m_jobMem;
    int m_jobMemSize;
    bool m_queueIsEmpty;
    int m_tailIndex;
    int m_headIndex;
    int m_allocSize;
    bool m_useSpinMutex;
    btAlignedObjectArray<JobQueue*> m_neighborContexts;
    char m_cachePadding[kCacheLineSize];  // prevent false sharing

    void freeJobMem()
    {
        if ( m_jobMem )
        {
            // free old
            btAlignedFree(m_jobMem);
            m_jobMem = NULL;
        }
    }
    void resizeJobMem(int newSize)
    {
        if (newSize > m_jobMemSize)
        {
            freeJobMem();
            m_jobMem = static_cast<char*>(btAlignedAlloc(newSize, kCacheLineSize));
            m_jobMemSize = newSize;
        }
    }

public:

    JobQueue()
    {
        m_jobMem = NULL;
        m_jobMemSize = 0;
        m_threadSupport = NULL;
        m_queueLock = NULL;
        m_headIndex = 0;
        m_tailIndex = 0;
        m_useSpinMutex = false;
    }
    ~JobQueue()
    {
                exit();
    }
        void exit()
    {
                freeJobMem();
        if (m_queueLock && m_threadSupport)
        {
            m_threadSupport->deleteCriticalSection(m_queueLock);
            m_queueLock = NULL;
                        m_threadSupport = 0;
        }
        }

    void init(btThreadSupportInterface* threadSup, btAlignedObjectArray<JobQueue>* contextArray)
    {
        m_threadSupport = threadSup;
        if (threadSup)
        {
            m_queueLock = m_threadSupport->createCriticalSection();
        }
        setupJobStealing(contextArray, contextArray->size());
    }
    void setupJobStealing(btAlignedObjectArray<JobQueue>* contextArray, int numActiveContexts)
    {
        btAlignedObjectArray<JobQueue>& contexts = *contextArray;
        int selfIndex = 0;
        for (int i = 0; i < contexts.size(); ++i)
        {
            if ( this == &contexts[ i ] )
            {
                selfIndex = i;
                break;
            }
        }
        int numNeighbors = btMin(2, contexts.size() - 1);
        int neighborOffsets[ ] = {-1, 1, -2, 2, -3, 3};
        int numOffsets = sizeof(neighborOffsets)/sizeof(neighborOffsets[0]);
        m_neighborContexts.reserve( numNeighbors );
        m_neighborContexts.resizeNoInitialize(0);
        for (int i = 0; i < numOffsets && m_neighborContexts.size() < numNeighbors; i++)
        {
            int neighborIndex = selfIndex + neighborOffsets[i];
            if ( neighborIndex >= 0 && neighborIndex < numActiveContexts)
            {
                m_neighborContexts.push_back( &contexts[ neighborIndex ] );
            }
        }
    }

    bool isQueueEmpty() const {return m_queueIsEmpty;}
    void lockQueue()
    {
        if ( m_useSpinMutex )
        {
            m_mutex.lock();
        }
        else
        {
            m_queueLock->lock();
        }
    }
    void unlockQueue()
    {
        if ( m_useSpinMutex )
        {
            m_mutex.unlock();
        }
        else
        {
            m_queueLock->unlock();
        }
    }
    void clearQueue(int jobCount, int jobSize)
    {
        lockQueue();
        m_headIndex = 0;
        m_tailIndex = 0;
        m_allocSize = 0;
        m_queueIsEmpty = true;
        int jobBufSize = jobSize * jobCount;
        // make sure we have enough memory allocated to store jobs
        if ( jobBufSize > m_jobMemSize )
        {
            resizeJobMem( jobBufSize );
        }
        // make sure job queue is big enough
        if ( jobCount > m_jobQueue.capacity() )
        {
            m_jobQueue.reserve( jobCount );
        }
        unlockQueue();
        m_jobQueue.resizeNoInitialize( 0 );
    }
    void* allocJobMem(int jobSize)
    {
        btAssert(m_jobMemSize >= (m_allocSize + jobSize));
        void* jobMem = &m_jobMem[m_allocSize];
        m_allocSize += jobSize;
        return jobMem;
    }
    void submitJob( IJob* job )
    {
        btAssert( reinterpret_cast<char*>( job ) >= &m_jobMem[ 0 ] && reinterpret_cast<char*>( job ) < &m_jobMem[ 0 ] + m_allocSize );
        m_jobQueue.push_back( job );
        lockQueue();
        m_tailIndex++;
        m_queueIsEmpty = false;
        unlockQueue();
    }
    IJob* consumeJobFromOwnQueue()
    {
        if ( m_queueIsEmpty )
        {
            // lock free path. even if this is taken erroneously it isn't harmful
            return NULL;
        }
        IJob* job = NULL;
        lockQueue();
        if ( !m_queueIsEmpty )
        {
            job = m_jobQueue[ m_headIndex++ ];
            btAssert( reinterpret_cast<char*>( job ) >= &m_jobMem[ 0 ] && reinterpret_cast<char*>( job ) < &m_jobMem[ 0 ] + m_allocSize );
            if ( m_headIndex == m_tailIndex )
            {
                m_queueIsEmpty = true;
            }
        }
        unlockQueue();
        return job;
    }
    IJob* consumeJob()
    {
        if (IJob* job = consumeJobFromOwnQueue())
        {
            return job;
        }
        // own queue is empty, try to steal from neighbor
        for (int i = 0; i < m_neighborContexts.size(); ++i)
        {
            JobQueue* otherContext = m_neighborContexts[ i ];
            if ( IJob* job = otherContext->consumeJobFromOwnQueue() )
            {
                return job;
            }
        }
        return NULL;
    }
};


static void WorkerThreadFunc( void* userPtr )
{
    BT_PROFILE( "WorkerThreadFunc" );
    ThreadLocalStorage* localStorage = (ThreadLocalStorage*) userPtr;
    JobQueue* jobQueue = localStorage->m_queue;

    bool shouldSleep = false;
    int threadId = localStorage->m_threadId;
    while (! shouldSleep)
    {
        // do work
        localStorage->m_mutex.lock();
        while ( IJob* job = jobQueue->consumeJob() )
        {
            localStorage->m_status = WorkerThreadStatus::kWorking;
            job->executeJob( threadId );
            localStorage->m_numJobsFinished++;
        }
        localStorage->m_status = WorkerThreadStatus::kWaitingForWork;
        localStorage->m_mutex.unlock();
        btU64 clockStart = localStorage->m_clock->getTimeMicroseconds();
        // while queue is empty,
        while (jobQueue->isQueueEmpty())
        {
            // todo: spin wait a bit to avoid hammering the empty queue
            btSpinPause();
            if ( localStorage->m_directive->getDirective(threadId) == WorkerThreadDirectives::kGoToSleep )
            {
                shouldSleep = true;
                break;
            }
            // if jobs are incoming,
            if ( localStorage->m_directive->getDirective( threadId ) == WorkerThreadDirectives::kScanForJobs )
            {
                clockStart = localStorage->m_clock->getTimeMicroseconds(); // reset clock
            }
            else
            {
                for ( int i = 0; i < 50; ++i )
                {
                    btSpinPause();
                    btSpinPause();
                    btSpinPause();
                    btSpinPause();
                    if (localStorage->m_directive->getDirective( threadId ) == WorkerThreadDirectives::kScanForJobs || !jobQueue->isQueueEmpty())
                    {
                        break;
                    }
                }
                // if no jobs incoming and queue has been empty for the cooldown time, sleep
                btU64 timeElapsed = localStorage->m_clock->getTimeMicroseconds() - clockStart;
                if (timeElapsed > localStorage->m_cooldownTime)
                {
                    shouldSleep = true;
                    break;
                }
            }
        }
    }
        {
                BT_PROFILE("sleep");
                // go sleep
                localStorage->m_mutex.lock();
                localStorage->m_status = WorkerThreadStatus::kSleeping;
                localStorage->m_mutex.unlock();
        }
}


class btTaskSchedulerDefault : public btITaskScheduler
{
    btThreadSupportInterface* m_threadSupport;
    WorkerThreadDirectives* m_workerDirective;
    btAlignedObjectArray<JobQueue> m_jobQueues;
    btAlignedObjectArray<JobQueue*> m_perThreadJobQueues;
    btAlignedObjectArray<ThreadLocalStorage> m_threadLocalStorage;
    btSpinMutex m_antiNestingLock;  // prevent nested parallel-for
    btClock m_clock;
    int m_numThreads;
    int m_numWorkerThreads;
    int m_numActiveJobQueues;
    int m_maxNumThreads;
    int m_numJobs;
    static const int kFirstWorkerThreadId = 1;
public:

    btTaskSchedulerDefault() : btITaskScheduler("ThreadSupport")
    {
        m_threadSupport = NULL;
        m_workerDirective = NULL;
    }

    virtual ~btTaskSchedulerDefault()
    {
        waitForWorkersToSleep();

                for ( int i = 0; i < m_jobQueues.size(); ++i )
        {
            m_jobQueues[i].exit();
        }

        if (m_threadSupport)
        {
            delete m_threadSupport;
            m_threadSupport = NULL;
        }
        if (m_workerDirective)
        {
            btAlignedFree(m_workerDirective);
            m_workerDirective = NULL;
        }
    }

    void init()
    {
        btThreadSupportInterface::ConstructionInfo constructionInfo( "TaskScheduler", WorkerThreadFunc );
        m_threadSupport = btThreadSupportInterface::create( constructionInfo );
        m_workerDirective = static_cast<WorkerThreadDirectives*>(btAlignedAlloc(sizeof(*m_workerDirective), 64));

        m_numWorkerThreads = m_threadSupport->getNumWorkerThreads();
        m_maxNumThreads = m_threadSupport->getNumWorkerThreads() + 1;
        m_numThreads = m_maxNumThreads;
        // ideal to have one job queue for each physical processor (except for the main thread which needs no queue)
        int numThreadsPerQueue = m_threadSupport->getLogicalToPhysicalCoreRatio();
        int numJobQueues = (numThreadsPerQueue == 1) ? (m_maxNumThreads-1) : (m_maxNumThreads / numThreadsPerQueue);
        m_jobQueues.resize(numJobQueues);
        m_numActiveJobQueues = numJobQueues;
        for ( int i = 0; i < m_jobQueues.size(); ++i )
        {
            m_jobQueues[i].init( m_threadSupport, &m_jobQueues );
        }
        m_perThreadJobQueues.resize(m_numThreads);
        for ( int i = 0; i < m_numThreads; i++ )
        {
            JobQueue* jq = NULL;
            // only worker threads get a job queue
            if (i > 0)
            {
                if (numThreadsPerQueue == 1)
                {
                    // one queue per worker thread
                    jq = &m_jobQueues[ i - kFirstWorkerThreadId ];
                }
                else
                {
                    // 2 threads share each queue
                    jq = &m_jobQueues[ i / numThreadsPerQueue ];
                }
            }
            m_perThreadJobQueues[i] = jq;
        }
        m_threadLocalStorage.resize(m_numThreads);
        for ( int i = 0; i < m_numThreads; i++ )
        {
            ThreadLocalStorage& storage = m_threadLocalStorage[i];
            storage.m_threadId = i;
            storage.m_directive = m_workerDirective;
            storage.m_status = WorkerThreadStatus::kSleeping;
            storage.m_cooldownTime = 100; // 100 microseconds, threads go to sleep after this long if they have nothing to do
            storage.m_clock = &m_clock;
            storage.m_queue = m_perThreadJobQueues[i];
        }
        setWorkerDirectives( WorkerThreadDirectives::kGoToSleep ); // no work for them yet
        setNumThreads( m_threadSupport->getCacheFriendlyNumThreads() );
    }

    void setWorkerDirectives(WorkerThreadDirectives::Type dir)
    {
        m_workerDirective->setDirectiveByRange(kFirstWorkerThreadId, m_numThreads, dir);
    }

    virtual int getMaxNumThreads() const BT_OVERRIDE
    {
        return m_maxNumThreads;
    }

    virtual int getNumThreads() const BT_OVERRIDE
    {
        return m_numThreads;
    }

    virtual void setNumThreads( int numThreads ) BT_OVERRIDE
    {
        m_numThreads = btMax( btMin(numThreads, int(m_maxNumThreads)), 1 );
        m_numWorkerThreads = m_numThreads - 1;
        m_numActiveJobQueues = 0;
        // if there is at least 1 worker,
        if ( m_numWorkerThreads > 0 )
        {
            // re-setup job stealing between queues to avoid attempting to steal from an inactive job queue
            JobQueue* lastActiveContext = m_perThreadJobQueues[ m_numThreads - 1 ];
            int iLastActiveContext = lastActiveContext - &m_jobQueues[0];
            m_numActiveJobQueues = iLastActiveContext + 1;
            for ( int i = 0; i < m_jobQueues.size(); ++i )
            {
                m_jobQueues[ i ].setupJobStealing( &m_jobQueues, m_numActiveJobQueues );
            }
        }
        m_workerDirective->setDirectiveByRange(m_numThreads, BT_MAX_THREAD_COUNT, WorkerThreadDirectives::kGoToSleep);
    }

    void waitJobs()
    {
        BT_PROFILE( "waitJobs" );
        // have the main thread work until the job queues are empty
        int numMainThreadJobsFinished = 0;
        for ( int i = 0; i < m_numActiveJobQueues; ++i )
        {
            while ( IJob* job = m_jobQueues[i].consumeJob() )
            {
                job->executeJob( 0 );
                numMainThreadJobsFinished++;
            }
        }

        // done with jobs for now, tell workers to rest (but not sleep)
        setWorkerDirectives( WorkerThreadDirectives::kStayAwakeButIdle );

        btU64 clockStart = m_clock.getTimeMicroseconds();
        // wait for workers to finish any jobs in progress
        while ( true )
        {
            int numWorkerJobsFinished = 0;
            for ( int iThread = kFirstWorkerThreadId; iThread < m_numThreads; ++iThread )
            {
                ThreadLocalStorage* storage = &m_threadLocalStorage[iThread];
                storage->m_mutex.lock();
                numWorkerJobsFinished += storage->m_numJobsFinished;
                storage->m_mutex.unlock();
            }
            if (numWorkerJobsFinished + numMainThreadJobsFinished == m_numJobs)
            {
                break;
            }
            btU64 timeElapsed = m_clock.getTimeMicroseconds() - clockStart;
            btAssert(timeElapsed < 1000);
            if (timeElapsed > 100000)
            {
                break;
            }
            btSpinPause();
        }
    }

    void wakeWorkers(int numWorkersToWake)
    {
        BT_PROFILE( "wakeWorkers" );
        btAssert( m_workerDirective->getDirective(1) == WorkerThreadDirectives::kScanForJobs );
        int numDesiredWorkers = btMin(numWorkersToWake, m_numWorkerThreads);
        int numActiveWorkers = 0;
        for ( int iWorker = 0; iWorker < m_numWorkerThreads; ++iWorker )
        {
            // note this count of active workers is not necessarily totally reliable, because a worker thread could be
            // just about to put itself to sleep. So we may on occasion fail to wake up all the workers. It should be rare.
            ThreadLocalStorage& storage = m_threadLocalStorage[ kFirstWorkerThreadId + iWorker ];
            if (storage.m_status != WorkerThreadStatus::kSleeping)
            {
                numActiveWorkers++;
            }
        }
        for ( int iWorker = 0; iWorker < m_numWorkerThreads && numActiveWorkers < numDesiredWorkers; ++iWorker )
        {
            ThreadLocalStorage& storage = m_threadLocalStorage[ kFirstWorkerThreadId + iWorker ];
            if (storage.m_status == WorkerThreadStatus::kSleeping)
            {
                m_threadSupport->runTask( iWorker, &storage );
                numActiveWorkers++;
            }
        }
    }

    void waitForWorkersToSleep()
    {
        BT_PROFILE( "waitForWorkersToSleep" );
        setWorkerDirectives( WorkerThreadDirectives::kGoToSleep );
        m_threadSupport->waitForAllTasks();
        for ( int i = kFirstWorkerThreadId; i < m_numThreads; i++ )
        {
            ThreadLocalStorage& storage = m_threadLocalStorage[i];
            btAssert( storage.m_status == WorkerThreadStatus::kSleeping );
        }
    }

    virtual void sleepWorkerThreadsHint() BT_OVERRIDE
    {
        BT_PROFILE( "sleepWorkerThreadsHint" );
        // hint the task scheduler that we may not be using these threads for a little while
        setWorkerDirectives( WorkerThreadDirectives::kGoToSleep );
    }

    void prepareWorkerThreads()
    {
        for ( int i = kFirstWorkerThreadId; i < m_numThreads; ++i )
        {
            ThreadLocalStorage& storage = m_threadLocalStorage[i];
            storage.m_mutex.lock();
            storage.m_numJobsFinished = 0;
            storage.m_mutex.unlock();
        }
        setWorkerDirectives( WorkerThreadDirectives::kScanForJobs );
    }

    virtual void parallelFor( int iBegin, int iEnd, int grainSize, const btIParallelForBody& body ) BT_OVERRIDE
    {
        BT_PROFILE( "parallelFor_ThreadSupport" );
        btAssert( iEnd >= iBegin );
        btAssert( grainSize >= 1 );
        int iterationCount = iEnd - iBegin;
        if ( iterationCount > grainSize && m_numWorkerThreads > 0 && m_antiNestingLock.tryLock() )
        {
            typedef ParallelForJob JobType;
            int jobCount = ( iterationCount + grainSize - 1 ) / grainSize;
            m_numJobs = jobCount;
            btAssert( jobCount >= 2 );  // need more than one job for multithreading
            int jobSize = sizeof( JobType );

            for (int i = 0; i < m_numActiveJobQueues; ++i)
            {
                m_jobQueues[i].clearQueue( jobCount, jobSize );
            }
            // prepare worker threads for incoming work
            prepareWorkerThreads();
            // submit all of the jobs
            int iJob = 0;
            int iThread = kFirstWorkerThreadId;  // first worker thread
            for ( int i = iBegin; i < iEnd; i += grainSize )
            {
                btAssert( iJob < jobCount );
                int iE = btMin( i + grainSize, iEnd );
                JobQueue* jq = m_perThreadJobQueues[ iThread ];
                btAssert(jq);
                btAssert((jq - &m_jobQueues[0]) < m_numActiveJobQueues);
                void* jobMem = jq->allocJobMem(jobSize);
                JobType* job = new ( jobMem ) ParallelForJob( i, iE, body );  // placement new
                jq->submitJob( job );
                iJob++;
                iThread++;
                if ( iThread >= m_numThreads )
                {
                    iThread = kFirstWorkerThreadId;  // first worker thread
                }
            }
            wakeWorkers( jobCount - 1 );

            // put the main thread to work on emptying the job queue and then wait for all workers to finish
            waitJobs();
            m_antiNestingLock.unlock();
        }
        else
        {
            BT_PROFILE( "parallelFor_mainThread" );
            // just run on main thread
            body.forLoop( iBegin, iEnd );
        }
    }
    virtual btScalar parallelSum( int iBegin, int iEnd, int grainSize, const btIParallelSumBody& body ) BT_OVERRIDE
    {
        BT_PROFILE( "parallelSum_ThreadSupport" );
        btAssert( iEnd >= iBegin );
        btAssert( grainSize >= 1 );
        int iterationCount = iEnd - iBegin;
        if ( iterationCount > grainSize && m_numWorkerThreads > 0 && m_antiNestingLock.tryLock() )
        {
            typedef ParallelSumJob JobType;
            int jobCount = ( iterationCount + grainSize - 1 ) / grainSize;
            m_numJobs = jobCount;
            btAssert( jobCount >= 2 );  // need more than one job for multithreading
            int jobSize = sizeof( JobType );
            for (int i = 0; i < m_numActiveJobQueues; ++i)
            {
                m_jobQueues[i].clearQueue( jobCount, jobSize );
            }

            // initialize summation
            for ( int iThread = 0; iThread < m_numThreads; ++iThread )
            {
                m_threadLocalStorage[iThread].m_sumResult = btScalar(0);
            }

            // prepare worker threads for incoming work
            prepareWorkerThreads();
            // submit all of the jobs
            int iJob = 0;
            int iThread = kFirstWorkerThreadId;  // first worker thread
            for ( int i = iBegin; i < iEnd; i += grainSize )
            {
                btAssert( iJob < jobCount );
                int iE = btMin( i + grainSize, iEnd );
                JobQueue* jq = m_perThreadJobQueues[ iThread ];
                btAssert(jq);
                btAssert((jq - &m_jobQueues[0]) < m_numActiveJobQueues);
                void* jobMem = jq->allocJobMem(jobSize);
                JobType* job = new ( jobMem ) ParallelSumJob( i, iE, body, &m_threadLocalStorage[0] );  // placement new
                jq->submitJob( job );
                iJob++;
                iThread++;
                if ( iThread >= m_numThreads )
                {
                    iThread = kFirstWorkerThreadId;  // first worker thread
                }
            }
            wakeWorkers( jobCount - 1 );

            // put the main thread to work on emptying the job queue and then wait for all workers to finish
            waitJobs();

            // add up all the thread sums
            btScalar sum = btScalar(0);
            for ( int iThread = 0; iThread < m_numThreads; ++iThread )
            {
                sum += m_threadLocalStorage[ iThread ].m_sumResult;
            }
            m_antiNestingLock.unlock();
            return sum;
        }
        else
        {
            BT_PROFILE( "parallelSum_mainThread" );
            // just run on main thread
            return body.sumLoop( iBegin, iEnd );
        }
    }
};



btITaskScheduler* btCreateDefaultTaskScheduler()
{
    btTaskSchedulerDefault* ts = new btTaskSchedulerDefault();
    ts->init();
    return ts;
}

#else // #if BT_THREADSAFE

btITaskScheduler* btCreateDefaultTaskScheduler()
{
    return NULL;
}

#endif // #else // #if BT_THREADSAFE