cthreadpool.cpp

Go to the documentation of this file.

/***************************************************************************
 *   Copyright (C) 2004-2006 by Ilya A. Volynets-Evenbakh                  *
 *   ilya@total-knowledge.com                                              *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program is distributed in the hope that it will be useful,       *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 *   GNU General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, write to the                         *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/
#include "cthreadpool.h"
#include <iostream>
#include <sptk3/CGuard.h>

namespace container {

CThreadPool::CThreadPool(int numThreads,int queueLen)
    : m_maxQueueLen(queueLen)
    , m_curQueueLen(0)
    , m_running(true)
{
    sptk::CGuard guard(m_tlLock);
    for(int i=0;i<numThreads;i++){
        Thread* t=new Thread(this);
        m_threads.push_front(t);
        t->run();
    }
}


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


void CThreadPool::queue(CThreadPool::Task* t)
{
    sptk::CGuard guard(m_queueLock);
    while(m_curQueueLen>=m_maxQueueLen) {
        m_queueLock.waitForSignal();
    }
    m_queue.push_front(t);
    m_curQueueLen++;
#ifdef DEBUG_TP
    std::cerr<<__PRETTY_FUNCTION__<<": queue length: "<<m_curQueueLen<<std::endl;
#endif
    m_queueLock.broadcastSignalNoLock();
}

CThreadPool::Task* CThreadPool::getTask()
{
    sptk::CGuard guard(m_queueLock);
    while(m_curQueueLen==0 && m_running) { //Empty
        m_queueLock.waitForSignal();
    }
    CThreadPool::Task* t=0;
    if(m_running) {
        tasklist_t::iterator i=m_queue.end();
        i--;// Presume if curQueueLen !=0, we have some elements...
        t=*i;
        m_queue.erase(i);
        m_curQueueLen--;
    }
#ifdef DEBUG_TP
    std::cerr<<__PRETTY_FUNCTION__<<": queue length: "<<m_curQueueLen<<std::endl;
#endif
    m_queueLock.broadcastSignalNoLock();
    return t;
}

void CThreadPool::Thread::threadFunction()
{
    CThreadPool::Task* t;
    m_running=true;
    do {
#ifdef DEBUG_TP
        std::cerr<<__PRETTY_FUNCTION__<<": thread "<<(void*)this<<" is waiting for a task"<<std::endl;
#endif
        t=m_pool->getTask();
#ifdef DEBUG_TP
        std::cerr<<__PRETTY_FUNCTION__<<": thread "<<(void*)this<<" got task "<<(void*)t<<std::endl;
#endif
        if(t) {
            t->run();
            delete t;
        }
#ifdef DEBUG_TP
        std::cerr<<__PRETTY_FUNCTION__<<": thread "<<(void*)this<<" finished task "<<(void*)t<<std::endl;
#endif
    } while(t);
}

}


void container::CThreadPool::stop()
{
    sptk::CGuard guard(m_tlLock);
    m_queueLock.lock();
    m_running=false;
    m_queueLock.unlock();
    m_queueLock.broadcastSignalNoLock();
    do {
        m_tlLock.waitForSignal();
    } while(!m_threads.empty());
}


void container::CThreadPool::threadDone(CThreadPool::Thread* thread)
{
    sptk::CGuard guard(m_tlLock);
    for(threadlist_t::iterator it=m_threads.begin();
        it!=m_threads.end(); it++)
    {
        if(*it==thread) {
            //delete *it;
            m_threads.erase(it);
            break;
        }
    }
    m_tlLock.sendSignalNoLock();
}


void container::CThreadPool::Thread::onThreadExit()
{
    m_pool->threadDone(this);
    delete this;
}