ThreadPool.cpp

Go to the documentation of this file.

#include "ThreadPool.h"
#include <iostream>

using std::cout ;
using std::endl ;
using std::cerr;

WorkerMainLoop::WorkerMainLoop(WorkerState *st,
                               boost::condition &c1, boost::mutex &m1,
                               boost::condition &c2, boost::mutex &m2)
        : cond1(c1), mtx1(m1), cond2(c2), mtx2(m2)
{
    state = st;
    state->toBeFinished = false;
}

void WorkerMainLoop::operator() ()
{
    boost::mutex::scoped_lock l1(mtx1);
    boost::mutex::scoped_lock l2(mtx2);
    l1.unlock();
    cond1.notify_one();
    while (!state->toBeFinished) {
        cond2.wait(l2); // <----- this is where the thread waits for a job to be started
        l1.lock();
        l1.unlock();
        cond1.notify_one();
        if (state->job && !state->toBeFinished) {
            (*state->job).start();
        }
    }
}

WorkerMainLoop::~WorkerMainLoop()
{}

WorkerState::WorkerState()
{
    job = 0;
}

void WorkerState::finish()
{
    toBeFinished = true ;
}

void WorkerState::assignJob(ThreadPoolJob &j)
{
    job = &j;
}

boost::function0<void> _f_;

ThreadPool::ThreadPool(int maxNumThreads) : _maxNumThreads(maxNumThreads)
{
    threads.resize(_maxNumThreads);
    workerstates.resize(_maxNumThreads);
    cond1.resize(_maxNumThreads);
    cond2.resize(_maxNumThreads);
    mtx1.resize(_maxNumThreads);
    mtx2.resize(_maxNumThreads);

    for (int i = 0 ; i < _maxNumThreads ; i++) {
        cond1[i] = new condition ;
        cond2[i] = new condition ;
        mtx1[i] = new mutex;
        mtx2[i] = new mutex;
        workerstates[i] = new WorkerState();
        WorkerMainLoop worker(workerstates[i], *cond1[i], *mtx1[i], *cond2[i], *mtx2[i]);
        boost::function0<void> f;
        f = worker ;
        mutex::scoped_lock l(*mtx1[i]);
        try {
             threads[i] = new thread( f );
        } catch (boost::thread_resource_error &) {
            cout << "ERROR: Cannot start another thread" << endl;
        }
        cond1[i]->wait(l);
        mutex::scoped_lock l2(*mtx2[i]);
    }
}

int ThreadPool::dispatch(int id, ThreadPoolJob &f)
{
    workerstates[id]->assignJob(f);
    boost::mutex::scoped_lock l1(*mtx1[id]);
    cond2[id]->notify_one();
    cond1[id]->wait(l1);
    return 0;
}

void ThreadPool::waitAll()
{
    for (int i = 0 ; i < _maxNumThreads ; i++ )
        boost::mutex::scoped_lock l2(*mtx2[i]);
}


ThreadPool::~ThreadPool()
{
    for (int i = 0 ; i < _maxNumThreads ; i++ ) {
        workerstates[i]->finish();
        cond2[i]->notify_one();
        threads[i]->join();
        delete workerstates[i];
        delete threads[i];
        delete cond1[i];
        delete cond2[i];
        delete mtx1[i];
        delete mtx2[i];
    }
}

---