ThreadPool.cpp

Go to the documentation of this file.

#include "ThreadPool.h"

namespace Support
{
    ThreadPool::ThreadPoolWorker::ThreadPoolWorker(ThreadPool& threadPool) : _threadPool(threadPool) 
    {
        _waitTimeActive.sec = WorkerActiveWaitTimeInSeconds;
        _waitTimeIdle.sec = WorkerIdleWaitTimeInSeconds;
    }

    void ThreadPool::ThreadPoolWorker::operator()()
    {
        while(true)
        {
            //how long we wait after the task depends on if we have another
            //even if we have another, it's good to sleep to surrender the CPU from time to time
            if(_threadPool.consumeTask())
                boost::thread::sleep(_waitTimeActive);
            else
                boost::thread::sleep(_waitTimeIdle);
        }
    }

    void ThreadPool::ThreadPoolManager::operator()()
    {
        while(true)
        {
            _threadPool.checkGrowPool();
            boost::thread::sleep(_waitTime);
        }
    }

    bool ThreadPool::SafeDequeue(ThreadPoolTaskBase** item)
    {
        boost::lock_guard<boost::mutex> guard(_taskMutex);

        if(_tasks.size() == 0)
        {
            *item = NULL;
            return false;
        }
        else
        {
            *item = _tasks.front();
            _tasks.pop_front();
            return true;
        }
    }

    void ThreadPool::createWorkerThread()
    {
        ThreadPoolWorker worker(*this);
        boost::thread* newThread = new boost::thread(worker);
        //add thread provides some thread-safety
        _threads.add_thread(newThread);
    }

    void ThreadPool::createManagerThread()
    {
        ThreadPoolManager manager(*this);
        boost::thread* newThread = new boost::thread(manager);
        _threads.add_thread(newThread);
    }

    void ThreadPool::checkGrowPool()
    {
        if((taskCount() > _newThreadThreshold) && (_threads.size() < _maxThreadCount))
            createWorkerThread();
    }
    
    bool ThreadPool::consumeTask()
    {
        ThreadPoolTaskBase* task;
        if(SafeDequeue(&task))
        {
            task->execute();
            if(task->getDelete())
                delete task;
            return true;
        }
        else
            return false;
    }

    //max threads is also +1 to accomodate a manager thread that is created to spawn other threads

    ThreadPool::ThreadPool() : 
        _maxThreadCount(DefaultMaxThreadsInThreadPool + 1), 
        _newThreadThreshold(DefaultTaskCountThresholdForNewThread) 
    {
        createManagerThread();
        createWorkerThread();
    }

    ThreadPool::ThreadPool(size_t numThreads)
    {
        for(size_t i = 0; i < numThreads; ++i)
            createWorkerThread();
    }

    ThreadPool::ThreadPool(size_t maxThreads, size_t growThreshold) :
        _maxThreadCount(maxThreads + 1), 
        _newThreadThreshold(growThreshold) 
    {
        createManagerThread();
        createWorkerThread();
    }

    ThreadPool::~ThreadPool()
    {
        stop();
    }

    void ThreadPool::stop()
    {
        //cleans up both the remaining tasks and the thread group
        clearTasks();
        _threads.interrupt_all();
        _threads.join_all();
    }

    void ThreadPool::queueTask(ThreadPoolTaskBase* task)
    {
        boost::lock_guard<boost::mutex> guard(_taskMutex);
        _tasks.push_back(task);
    }

    size_t ThreadPool::taskCount()
    {
        boost::lock_guard<boost::mutex> guard(_taskMutex);
        return _tasks.size();
    }

    void ThreadPool::clearTasks()
    {
        boost::lock_guard<boost::mutex> guard(_taskMutex);

        for(std::list<ThreadPoolTaskBase*>::iterator it=_tasks.begin(),end=_tasks.end();
            it!=end;
            ++it)
        {
            if((*it)->getDelete())
                delete *it;
        }

        _tasks.clear();
    }
}