thread_pool.h

#ifndef THREAD_POOL
#define THREAD_POOL


//#ifdef _WIN32
//#else
#include <pthread.h>
#include <semaphore.h>
#include "thread_management.h"
//#endif


#include <iostream>
#include <stdlib.h>
#include <vector>

#include <map>



//using namespace std;



namespace skepu
{

typedef void *(*ThreadFunc)(void *);

struct WorkerThread 
{
    ThreadFunc job_func;
//  void    *(*job_func)(void *);   /* function to call */
    void    *job_arg;       /* its argument */
    
    unsigned virtual executeThis()
    {
        job_func(job_arg);
        return 0;
    }
    
    WorkerThread(ThreadFunc job, void *arg): job_func(job), job_arg(arg)
    {
        
        }
    
    ~WorkerThread()
    {   
        job_func = NULL;
        job_arg = NULL;
    }
};

class ThreadPool
{
public:
    ThreadPool();
    ThreadPool(int maxThreadsTemp);
    virtual ~ThreadPool();
    
    void destroyPool(int maxPollMilliSecs=100);
    void finishAll(int maxPollMilliSecs=100);

    bool assignWork(WorkerThread *worker);
    bool fetchWork(WorkerThread **worker);

    void initializeThreads();
    
    static void *threadExecute(void *param);
    
    static pthread_mutex_t mutexSync;
    static pthread_mutex_t mutexWorkCompletion;
    
    
private:
    int maxThreads;
    
    pthread_cond_t  condCrit;
    sem_t availableWork;
    sem_t availableThreads;

    std::vector<WorkerThread *> workerQueue;
    
    int topIndex;
    int bottomIndex;
    
    int incompleteWork;

    
    int queueSize;

};



pthread_mutex_t ThreadPool::mutexSync = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t ThreadPool::mutexWorkCompletion = PTHREAD_MUTEX_INITIALIZER;


ThreadPool::ThreadPool()
{
//
//  #ifdef SKEPU_CUDA
//    {
//      int numDevices = 0;
//      cudaError_t err;
//      err = cudaGetDeviceCount(&numDevices);
//  
//      ThreadPool(numDevices);
//    } 
//      
//  #else
    
        ThreadPool(2);
    
//  #endif
}

ThreadPool::ThreadPool(int maxThreads)
{
   if (maxThreads < 1)  maxThreads=1;
  
   //mutexSync = PTHREAD_MUTEX_INITIALIZER;
   //mutexWorkCompletion = PTHREAD_MUTEX_INITIALIZER; 
   
   pthread_mutex_lock(&mutexSync);
   this->maxThreads = maxThreads;
   this->queueSize = maxThreads;
   //workerQueue = new WorkerThread *[maxThreads];
   workerQueue.resize(maxThreads, NULL);
   topIndex = 0;
   bottomIndex = 0;
   incompleteWork = 0;
   sem_init(&availableWork, 0, 0);
   sem_init(&availableThreads, 0, queueSize);
   pthread_mutex_unlock(&mutexSync);
}

void ThreadPool::initializeThreads()
{
   for(int i = 0; i<maxThreads; ++i)
    {
        pthread_t tempThread;
        pthread_create(&tempThread, NULL, &ThreadPool::threadExecute, (void *) this ); 
         //threadIdVec[i] = tempThread;
   }

}

ThreadPool::~ThreadPool()
{
   workerQueue.clear();
}


void ThreadPool::finishAll(int maxPollMilliSecs)
{
    while( incompleteWork>0 )
    {
            //cout << "Work is still incomplete=" << incompleteWork << endl;
        usleep(maxPollMilliSecs);
    }
}


void ThreadPool::destroyPool(int maxPollMilliSecs)
{
    while( incompleteWork>0 )
    {
            //cout << "Work is still incomplete=" << incompleteWork << endl;
        usleep(maxPollMilliSecs);
    }
    //std::cout << "All Done!! Wow! That was a lot of work!" << endl;
    sem_destroy(&availableWork);
    sem_destroy(&availableThreads);
        pthread_mutex_destroy(&mutexSync);
        pthread_mutex_destroy(&mutexWorkCompletion);

}


bool ThreadPool::assignWork(WorkerThread *workerThread)
{
                  //  std::cerr << "STOP INSIDE** \n";
        pthread_mutex_lock(&mutexWorkCompletion);
        incompleteWork++;
        //cout << "assignWork...incomapleteWork=" << incompleteWork << endl;
    pthread_mutex_unlock(&mutexWorkCompletion);
    
    sem_wait(&availableThreads);
    
    pthread_mutex_lock(&mutexSync);
        //workerVec[topIndex] = workerThread;
        workerQueue[topIndex] = workerThread;
          //      std::cout << "Assigning Worker[] Address:[" << workerThread << "] to Queue index [" << topIndex << "]\n";
        if(queueSize !=1 )
            topIndex = (topIndex+1) % (queueSize);
        sem_post(&availableWork);
    pthread_mutex_unlock(&mutexSync);
    return true;
}

bool ThreadPool::fetchWork(WorkerThread **workerArg)
{
    sem_wait(&availableWork);

    pthread_mutex_lock(&mutexSync);
        WorkerThread * workerThread = workerQueue[bottomIndex];
                workerQueue[bottomIndex] = NULL;
        *workerArg = workerThread;
        if(queueSize !=1 )
            bottomIndex = (bottomIndex+1) % (queueSize);
        sem_post(&availableThreads);
    pthread_mutex_unlock(&mutexSync);
    return true;
}

void *ThreadPool::threadExecute(void *param)
{
    WorkerThread *worker = NULL;
    
    while(((ThreadPool *)param)->fetchWork(&worker))
    {
        if(worker)
                {                                   
            worker->executeThis();
                        delete worker;
                        worker = NULL;
                }

        pthread_mutex_lock( &(((ThreadPool *)param)->mutexWorkCompletion) );
                //cout << "Thread " << pthread_self() << " has completed a Job !" << endl;
        ((ThreadPool *)param)->incompleteWork--;
        pthread_mutex_unlock( &(((ThreadPool *)param)->mutexWorkCompletion) );
    }
    return 0;
}

}


#endif