ThreadPool.h

Go to the documentation of this file.

#ifndef __THREAD_POOL_H
#define __THREAD_POOL_H

#include "RakMemoryOverride.h"
#include "DS_Queue.h"
#include "SimpleMutex.h"
#include "Export.h"
#include "RakThread.h"
#include "SignaledEvent.h"

#ifdef _MSC_VER
#pragma warning( push )
#endif

class ThreadDataInterface
{
public:
        ThreadDataInterface() {}
        virtual ~ThreadDataInterface() {}

        virtual void* PerThreadFactory(void *context)=0;
        virtual void PerThreadDestructor(void* factoryResult, void *context)=0;
};
template <class InputType, class OutputType>
struct RAK_DLL_EXPORT ThreadPool
{
        ThreadPool();
        ~ThreadPool();

        bool StartThreads(int numThreads, int stackSize, void* (*_perThreadInit)()=0, void (*_perThreadDeinit)(void*)=0);

        // Alternate form of _perThreadDataFactory, _perThreadDataDestructor
        void SetThreadDataInterface(ThreadDataInterface *tdi, void *context);

        void StopThreads(void);

        void AddInput(OutputType (*workerThreadCallback)(InputType, bool *returnOutput, void* perThreadData), InputType inputData);

        void AddOutput(OutputType outputData);

        bool HasOutput(void);

        bool HasOutputFast(void);

        bool HasInput(void);

        bool HasInputFast(void);

        OutputType GetOutput(void);

        void Clear(void);

        void LockInput(void);

        void UnlockInput(void);

        unsigned InputSize(void);

        InputType GetInputAtIndex(unsigned index);

        void RemoveInputAtIndex(unsigned index);

        void LockOutput(void);
        
        void UnlockOutput(void);

        unsigned OutputSize(void);

        OutputType GetOutputAtIndex(unsigned index);

        void RemoveOutputAtIndex(unsigned index);

        void ClearInput(void);

        void ClearOutput(void);

        bool IsWorking(void);

        int NumThreadsWorking(void);

        bool WasStarted(void);

        // Block until all threads are stopped.
        bool Pause(void);

        // Continue running
        void Resume(void);

protected:
        // It is valid to cancel input before it is processed.  To do so, lock the inputQueue with inputQueueMutex,
        // Scan the list, and remove the item you don't want.
        SimpleMutex inputQueueMutex, outputQueueMutex, workingThreadCountMutex, runThreadsMutex;

        void* (*perThreadDataFactory)();
        void (*perThreadDataDestructor)(void*);

        // inputFunctionQueue & inputQueue are paired arrays so if you delete from one at a particular index you must delete from the other
        // at the same index
        DataStructures::Queue<OutputType (*)(InputType, bool *, void*)> inputFunctionQueue;
        DataStructures::Queue<InputType> inputQueue;
        DataStructures::Queue<OutputType> outputQueue;

        ThreadDataInterface *threadDataInterface;
        void *tdiContext;

        
        template <class ThreadInputType, class ThreadOutputType>
        friend RAK_THREAD_DECLARATION(WorkerThread);

        /*
#ifdef _WIN32
        friend unsigned __stdcall WorkerThread( LPVOID arguments );
#else
        friend void* WorkerThread( void* arguments );
#endif
        */

        bool runThreads;
        int numThreadsRunning;
        int numThreadsWorking;
        SimpleMutex numThreadsRunningMutex;

        SignaledEvent quitAndIncomingDataEvents;
};

#include "ThreadPool.h"
#include "RakSleep.h"
#ifdef _WIN32
#else
#include <unistd.h>
#endif

#ifdef _MSC_VER
#pragma warning(disable:4127)
#pragma warning( disable : 4701 )  // potentially uninitialized local variable 'inputData' used
#endif

template <class ThreadInputType, class ThreadOutputType>
RAK_THREAD_DECLARATION(WorkerThread)
/*
#ifdef _WIN32
unsigned __stdcall WorkerThread( LPVOID arguments )
#else
void* WorkerThread( void* arguments )
#endif
*/
{
        bool returnOutput;
        ThreadPool<ThreadInputType, ThreadOutputType> *threadPool = (ThreadPool<ThreadInputType, ThreadOutputType>*) arguments;
        ThreadOutputType (*userCallback)(ThreadInputType, bool *, void*);
        ThreadInputType inputData;
        ThreadOutputType callbackOutput;

        userCallback=0;

        void *perThreadData;
        if (threadPool->perThreadDataFactory)
                perThreadData=threadPool->perThreadDataFactory();
        else if (threadPool->threadDataInterface)
                perThreadData=threadPool->threadDataInterface->PerThreadFactory(threadPool->tdiContext);
        else
                perThreadData=0;

        // Increase numThreadsRunning
        threadPool->numThreadsRunningMutex.Lock();
        ++threadPool->numThreadsRunning;
        threadPool->numThreadsRunningMutex.Unlock();

        while (1)
        {
#ifdef _WIN32
                if (userCallback==0)
                {
                        threadPool->quitAndIncomingDataEvents.WaitOnEvent(INFINITE);
                }               
#endif

                threadPool->runThreadsMutex.Lock();
                if (threadPool->runThreads==false)
                {
                        threadPool->runThreadsMutex.Unlock();
                        break;
                }
                threadPool->runThreadsMutex.Unlock();

                threadPool->workingThreadCountMutex.Lock();
                ++threadPool->numThreadsWorking;
                threadPool->workingThreadCountMutex.Unlock();

                // Read input data
                userCallback=0;
                threadPool->inputQueueMutex.Lock();
                if (threadPool->inputFunctionQueue.Size())
                {
                        userCallback=threadPool->inputFunctionQueue.Pop();
                        inputData=threadPool->inputQueue.Pop();
                }
                threadPool->inputQueueMutex.Unlock();

                if (userCallback)
                {
                        callbackOutput=userCallback(inputData, &returnOutput,perThreadData);
                        if (returnOutput)
                        {
                                threadPool->outputQueueMutex.Lock();
                                threadPool->outputQueue.Push(callbackOutput, __FILE__, __LINE__ );
                                threadPool->outputQueueMutex.Unlock();
                        }                       
                }

                threadPool->workingThreadCountMutex.Lock();
                --threadPool->numThreadsWorking;
                threadPool->workingThreadCountMutex.Unlock();
        }

        // Decrease numThreadsRunning
        threadPool->numThreadsRunningMutex.Lock();
        --threadPool->numThreadsRunning;
        threadPool->numThreadsRunningMutex.Unlock();
        
        if (threadPool->perThreadDataDestructor)
                threadPool->perThreadDataDestructor(perThreadData);
        else if (threadPool->threadDataInterface)
                threadPool->threadDataInterface->PerThreadDestructor(perThreadData, threadPool->tdiContext);

        return 0;
}
template <class InputType, class OutputType>
ThreadPool<InputType, OutputType>::ThreadPool()
{
        runThreads=false;
        numThreadsRunning=0;
        threadDataInterface=0;
        tdiContext=0;
        numThreadsWorking=0;

}
template <class InputType, class OutputType>
ThreadPool<InputType, OutputType>::~ThreadPool()
{
        StopThreads();
        Clear();
}
template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::StartThreads(int numThreads, int stackSize, void* (*_perThreadDataFactory)(), void (*_perThreadDataDestructor)(void *))
{
        (void) stackSize;

        runThreadsMutex.Lock();
        if (runThreads==true)
        {
                runThreadsMutex.Unlock();
                return false;
        }
        runThreadsMutex.Unlock();

        quitAndIncomingDataEvents.InitEvent();

        perThreadDataFactory=_perThreadDataFactory;
        perThreadDataDestructor=_perThreadDataDestructor;

        runThreadsMutex.Lock();
        runThreads=true;
        runThreadsMutex.Unlock();

        numThreadsWorking=0;
        unsigned threadId = 0;
        (void) threadId;
        int i;
        for (i=0; i < numThreads; i++)
        {
                int errorCode = RakNet::RakThread::Create(WorkerThread<InputType, OutputType>, this);
                if (errorCode!=0)
                {
                        StopThreads();
                        return false;
                }
        }
        // Wait for number of threads running to increase to numThreads
        bool done=false;
        while (done==false)
        {
                RakSleep(50);
                numThreadsRunningMutex.Lock();
                if (numThreadsRunning==numThreads)
                        done=true;
                numThreadsRunningMutex.Unlock();
        }

        return true;
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::SetThreadDataInterface(ThreadDataInterface *tdi, void *context)
{
        threadDataInterface=tdi;
        tdiContext=context;
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::StopThreads(void)
{
        runThreadsMutex.Lock();
        if (runThreads==false)
        {
                runThreadsMutex.Unlock();
                return;
        }

        runThreads=false;
        runThreadsMutex.Unlock();

        // Wait for number of threads running to decrease to 0
        bool done=false;
        while (done==false)
        {
                quitAndIncomingDataEvents.SetEvent();

                RakSleep(50);
                numThreadsRunningMutex.Lock();
                if (numThreadsRunning==0)
                        done=true;
                numThreadsRunningMutex.Unlock();
        }

        quitAndIncomingDataEvents.CloseEvent();
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::AddInput(OutputType (*workerThreadCallback)(InputType, bool *returnOutput, void* perThreadData), InputType inputData)
{
        inputQueueMutex.Lock();
        inputQueue.Push(inputData, __FILE__, __LINE__ );
        inputFunctionQueue.Push(workerThreadCallback, __FILE__, __LINE__ );
        inputQueueMutex.Unlock();

        quitAndIncomingDataEvents.SetEvent();
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::AddOutput(OutputType outputData)
{
        outputQueueMutex.Lock();
        outputQueue.Push(outputData, __FILE__, __LINE__ );
        outputQueueMutex.Unlock();
}
template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::HasOutputFast(void)
{
        return outputQueue.IsEmpty()==false;
}
template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::HasOutput(void)
{
        bool res;
        outputQueueMutex.Lock();
        res=outputQueue.IsEmpty()==false;
        outputQueueMutex.Unlock();
        return res;
}
template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::HasInputFast(void)
{
        return inputQueue.IsEmpty()==false;
}
template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::HasInput(void)
{
        bool res;
        inputQueueMutex.Lock();
        res=inputQueue.IsEmpty()==false;
        inputQueueMutex.Unlock();
        return res;
}
template <class InputType, class OutputType>
OutputType ThreadPool<InputType, OutputType>::GetOutput(void)
{
        // Real output check
        OutputType output;
        outputQueueMutex.Lock();
        output=outputQueue.Pop();
        outputQueueMutex.Unlock();
        return output;
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::Clear(void)
{
        runThreadsMutex.Lock();
        if (runThreads)
        {
                runThreadsMutex.Unlock();
                inputQueueMutex.Lock();
                inputFunctionQueue.Clear(__FILE__, __LINE__);
                inputQueue.Clear(__FILE__, __LINE__);
                inputQueueMutex.Unlock();

                outputQueueMutex.Lock();
                outputQueue.Clear(__FILE__, __LINE__);
                outputQueueMutex.Unlock();
        }
        else
        {
                inputFunctionQueue.Clear(__FILE__, __LINE__);
                inputQueue.Clear(__FILE__, __LINE__);
                outputQueue.Clear(__FILE__, __LINE__);
        }
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::LockInput(void)
{
        inputQueueMutex.Lock();
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::UnlockInput(void)
{
        inputQueueMutex.Unlock();
}
template <class InputType, class OutputType>
unsigned ThreadPool<InputType, OutputType>::InputSize(void)
{
        return inputQueue.Size();
}
template <class InputType, class OutputType>
InputType ThreadPool<InputType, OutputType>::GetInputAtIndex(unsigned index)
{
        return inputQueue[index];
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::RemoveInputAtIndex(unsigned index)
{
        inputQueue.RemoveAtIndex(index);
        inputFunctionQueue.RemoveAtIndex(index);
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::LockOutput(void)
{
        outputQueueMutex.Lock();
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::UnlockOutput(void)
{
        outputQueueMutex.Unlock();
}
template <class InputType, class OutputType>
unsigned ThreadPool<InputType, OutputType>::OutputSize(void)
{
        return outputQueue.Size();
}
template <class InputType, class OutputType>
OutputType ThreadPool<InputType, OutputType>::GetOutputAtIndex(unsigned index)
{
        return outputQueue[index];
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::RemoveOutputAtIndex(unsigned index)
{
        outputQueue.RemoveAtIndex(index);
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::ClearInput(void)
{
        inputQueue.Clear(__FILE__,__LINE__);
        inputFunctionQueue.Clear(__FILE__,__LINE__);
}

template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::ClearOutput(void)
{
        outputQueue.Clear(__FILE__,__LINE__);
}
template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::IsWorking(void)
{
        bool isWorking;
//      workingThreadCountMutex.Lock();
//      isWorking=numThreadsWorking!=0;
//      workingThreadCountMutex.Unlock();

//      if (isWorking)
//              return true;

        // Bug fix: Originally the order of these two was reversed.
        // It's possible with the thread timing that working could have been false, then it picks up the data in the other thread, then it checks
        // here and sees there is no data.  So it thinks the thread is not working when it was.
        if (HasOutputFast() && HasOutput())
                return true;

        if (HasInputFast() && HasInput())
                return true;

        // Need to check is working again, in case the thread was between the first and second checks
        workingThreadCountMutex.Lock();
        isWorking=numThreadsWorking!=0;
        workingThreadCountMutex.Unlock();

        return isWorking;
}

template <class InputType, class OutputType>
int ThreadPool<InputType, OutputType>::NumThreadsWorking(void)
{
        return numThreadsWorking;
}

template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::WasStarted(void)
{
        bool b;
        runThreadsMutex.Lock();
        b = runThreads;
        runThreadsMutex.Unlock();
        return b;
}
template <class InputType, class OutputType>
bool ThreadPool<InputType, OutputType>::Pause(void)
{
        if (WasStarted()==false)
                return false;

        workingThreadCountMutex.Lock();
        while (numThreadsWorking>0)
        {
                RakSleep(30);
        }
        return true;
}
template <class InputType, class OutputType>
void ThreadPool<InputType, OutputType>::Resume(void)
{
        workingThreadCountMutex.Unlock();
}

#ifdef _MSC_VER
#pragma warning( pop )
#endif

#endif