lib/thread.h

#ifndef __THREAD_H__
#define __THREAD_H__

#include <stdint.h>
#include <queue>
#include <map>
#include <set>

#include "stm_export.h"
#include "threadutils.h"

namespace thread {

class Condition;
class Farmer;
class ThreadPool;


class STM_EXPORT Job
{
public:
    Job() 
        : m_processed(false)
    {
    }

    virtual ~Job()
    {
    }

    bool Processed(void) const
    {
        return m_processed;
    }

    void DoProcess(void)
    {
        Process();
        m_processed = true;
    }

protected:
    virtual void Process(void) = 0;
    
protected:
    bool m_processed;
};


class STM_EXPORT JobQueue : public Fifo<Job>
{
};


class STM_EXPORT WorkerThread : public Thread
{
public:
    WorkerThread(JobQueue &);
    ~WorkerThread();

    void Push(Job *);
    void Terminate(void);

protected:
    virtual void *MainLoop(void);

    volatile bool m_terminate;
    JobQueue m_input;
    JobQueue &m_output;
};


class STM_EXPORT ThreadPool : public Thread
{
public:

static ThreadPool *GetInstance(void);
static void Shutdown(void);
static bool CanUsePool(void);

    void Push(Job *, Farmer *);

    void RegisterFarmer(Farmer *);
    void UnRegisterFarmer(Farmer *);

    void SetNumThreads(size_t);

private:
    bool _CanUsePool(void);
    virtual void *MainLoop(void);
    
    ThreadPool();
    ~ThreadPool();
    void Resize(size_t);
    void Terminate(void);
    void Join(void);

    void AssignJobs(void);
    void TrimThreads(void);

    STM_THREAD_TYPE           m_thread;
    
    Mutex                     m_lock;
    bool                      m_allowJobs;

    std::map<Job *, Farmer *> m_destinations;
    std::map<Job *, size_t>   m_jobs;

    std::set<Farmer *>        m_farmers;
    std::set<size_t>          m_activeThreads;
    std::set<size_t>          m_idleThreads;
    std::map<size_t, WorkerThread*> m_threads;

    std::queue<Job *>         m_pending;
    size_t                    m_targetThreadCount;
    size_t                    m_nextThreadId;
    JobQueue                  m_queue;
    volatile bool             m_terminate;
};


class STM_EXPORT Farmer
{
    friend class WorkerThread;
    friend class ThreadPool;
public:
    Farmer();
    virtual ~Farmer();

    void Push(Job *);
    void Process(void);
protected:
    virtual void ProcessCompletedJob(Job *);
    JobQueue &GetPending(void) {return m_pending;}
    JobQueue &GetComplete(void){return m_complete;}
    bool GetTerminate(void) const {return m_terminate;}

    size_t GetNumSubmitted(void) const{return m_numSubmitted;}
    size_t GetNumCompleted(void) const{return m_numCompleted;}

private:

    JobQueue m_pending, m_complete;
    size_t m_numSubmitted, m_numCompleted;
    bool m_terminate;
    ThreadPool *m_pool;
};


}
#endif

---