xmipp_threads.h

Go to the documentation of this file.

/***************************************************************************
 * Authors:     J.M. de la Rosa Trevin (jmdelarosa@cnb.csic.es)
 *
 *
 * Unidad de  Bioinformatica of Centro Nacional de Biotecnologia , CSIC
 *
 * 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
 *
 *  All comments concerning this program package may be sent to the
 *  e-mail address 'xmipp@cnb.csic.es'
 ***************************************************************************/

#ifndef THREADS_T
#define THREADS_T

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

class ThreadManager;
class ThreadArgument;
class Condition;

/* Prototype of functions for threads works. */
typedef void (*ThreadFunction) (ThreadArgument &arg);

//TODO (MARIANA) Please give more documentation and in a good structure e.g. @name (see args.h as example)


class Mutex
{
private:
    pthread_mutex_t mutex; //our pthread mutex

public:
    Mutex();

    virtual ~Mutex();

    virtual void lock();

    virtual void unlock();

    friend class Condition;
}
;//end of class Mutex

class Condition
{
private:
    Mutex *mutex; //our pthread mutex
    pthread_cond_t cond; //our pthread cond

public:
    Condition();

     ~Condition();

     void lock();

     void unlock();

     void wait();

     void signal();

     void broadcast();

}
;//end of class Condition

class Barrier
{
private:
    int needed; 
    int called; 
    Condition * condition; 

public:
    Barrier(int numberOfThreads);

    ~Barrier();

    void wait();

}
;//end of class Barrier

class Thread
{
private:
    pthread_t thId; 

public:
    Thread();

     virtual ~Thread();

     virtual void run() = 0;

     void start();
}
;//end of class Condition

void * _singleThreadMain(void * data);

void * _threadMain(void * data);

class ThreadManager
{
public:
    int threads; 
private:
    pthread_t * ids; 
    ThreadArgument * arguments; 
    Barrier * barrier; 
    ThreadFunction workFunction;
    bool started;
    void * workClass;

    void createThreads();

public:
    void setData(void * data, int nThread = -1);

    ThreadManager(int numberOfThreads, void * workClass = NULL);

    ~ThreadManager();

    void run(ThreadFunction function, void * data = NULL);

    void runAsync(ThreadFunction function, void * data = NULL);

    void wait();

    friend void * _threadMain(void * data);

    int getNumberOfThreads() {return threads;}
}
;//end of class ThreadManager

class ThreadArgument
{
private:
    ThreadManager * manager;
public:
    int thread_id; 
    int threads;   
    void * workClass; 
    void * data; // Pointer to void *

    ThreadArgument();
    ThreadArgument(int id, ThreadManager * manager = NULL, void * data = NULL);

    friend class ThreadManager;
    friend void * _threadMain(void * data);
    int getNumberOfThreads() {return manager->getNumberOfThreads();}
};

class ParallelTaskDistributor
{
protected:
    //How many tasks give in each request
    size_t blockSize;
    //The number of tasks that have been assigned
    size_t assignedTasks;

public:
    //The total number of tasks to be distributed
    size_t numberOfTasks;
    ParallelTaskDistributor(size_t nTasks, size_t bSize);

    virtual ~ParallelTaskDistributor()
    {}
    ;

    void clear();

    void setBlockSize(size_t bSize);

    int getBlockSize() const;

    bool getTasks(size_t &first, size_t &last); // False = no more jobs, true = more jobs
    /* This function set the number of completed tasks.
     * Usually this not need to be called. Its more useful
     * for restarting work, when usually the master detect
     * the number of tasks already done.
     */
    bool setAssignedTasks(size_t tasks);


protected:
    //Virtual functions that should be implemented in
    //subclasses, providing a mechanism of lock and
    //the specific way of distribute tasks.
    virtual void lock() = 0;
    virtual void unlock() = 0;
    virtual bool distribute(size_t &first, size_t &last) = 0;

}
;//class ParallelTaskDistributor

class ThreadTaskDistributor: public ParallelTaskDistributor
{
public:
    ThreadTaskDistributor(size_t nTasks, size_t bSize):ParallelTaskDistributor(nTasks, bSize)
    {}
    virtual ~ThreadTaskDistributor()
    {}
    ;
protected:
    Mutex mutex; 
    virtual void lock();
    virtual void unlock();
    virtual bool distribute(size_t &first, size_t &last);
public:
    virtual void reset() { setAssignedTasks(0); };
}
;//end of class ThreadTaskDistributor

typedef struct mybarrier_t
{
    int needed;
    int called;
    pthread_mutex_t mutex;
    pthread_cond_t cond;
}
barrier_t;

int barrier_init(barrier_t *barrier, int needed);
int barrier_destroy(barrier_t *barrier);
int barrier_wait(barrier_t *barrier);

#endif