---

WorkerGroup.h

Go to the documentation of this file.

#ifndef WORKERGROUP_H
#define WORKERGROUP_H
#include <queue>
#include <vector>
#include <algorithm>
#include <boost/thread.hpp>

#include "Sct/LogicErrors.h"

namespace SctService{

template <class T>
class WorkerGroup : private boost::noncopyable {
public:
    typedef T DataType;
    
    void go(unsigned nWorker=1);    

    unsigned queueSize();
    unsigned nWorkers();
    unsigned busy();
    void stop();
    /*
     * Raise flag to pause all worker threads.
     * @param true=pause, false=unpause/play/restart 
     */
    void setPaused(bool paused=true);
    void join();
    unsigned paused();
    void setSleepTime(unsigned msec);
    unsigned getSleepTime() {return m_sleeptime;}
    /*
     * Pop an object off the front of the queue. @return T to the object
     * if the queue isn't empty, otherwise, T()
     */
    DataType pop();
    void push(DataType ob);    
    
    virtual ~WorkerGroup() {;} 
    
    WorkerGroup() : m_run(true), m_pause(false), m_sleeptime(1) {}

protected:
    //Inner class to hold thread specific stuff
    class Worker {
    public:
    Worker(WorkerGroup<DataType>& group) : group(group), m_busy(false), m_paused(false) {}
    void operator() () {
        group.threadMain(*this);
    }
    WorkerGroup<DataType>& group;
    volatile bool m_busy; 
    volatile bool m_paused; 
    };
    
    virtual void threadMain(Worker& worker) throw();
    virtual bool popAndWork(Worker& worker);
    virtual void waitForData();
    virtual void work(DataType data) throw() = 0;
    void wait(int msec);
    void addWorker(Worker& worker);
    
    typedef typename std::vector<Worker*> WorkerStore;
    typedef typename WorkerStore::iterator Iterator;    
    
     WorkerStore m_workers;

    std::queue<DataType> m_queue;
    

    boost::thread_group m_thread_group;
    
    volatile bool m_run;  
    volatile bool m_pause; 
    volatile unsigned m_sleeptime; 
    
    boost::mutex m_queueAccess; 
    boost::mutex m_vectorAccess; 
    friend class Worker;
};


template <class T>
void WorkerGroup<T>::go(unsigned nWorker){    
    // create nWorker new threads
    for (unsigned i=0; i<nWorker; ++i) {
    m_thread_group.create_thread(Worker(*this));
    }
}

template <class T>
void WorkerGroup<T>::threadMain(Worker& worker) throw() { 
    //As the worker is owned by some boost thing, we just need to add it.
    addWorker(worker);
    
    while (m_run) {                        // if !m_run, thread will finish.
    while (m_pause && m_run) {     // while paused, and not stopped, wait in sleep loop.
        worker.m_paused = true;
        wait(m_sleeptime);
    }
    worker.m_paused = false;
    if (!m_run) return;                     // if no longer running, stop.
    
    if (!popAndWork(worker)) {
        waitForData();    // if queue is empty, sleep.
    }
    }
    return;
}

template<class T>
void WorkerGroup<T>::addWorker(Worker& worker) {
    boost::mutex::scoped_lock lock(m_vectorAccess);
    m_workers.push_back(&worker);
}

template <class T>
void WorkerGroup<T>::push(DataType ob){
    boost::mutex::scoped_lock lock(m_queueAccess);
    if (lock.locked()) m_queue.push(ob);
    else throw Sct::IllegalStateError("Couldn't get queue lock", __FILE__, __LINE__);
}

template <class T>
T WorkerGroup<T>::pop(){
    boost::mutex::scoped_lock lock(m_queueAccess);
    if (lock.locked()) {
    if (m_queue.size()!=0) {
        DataType a=m_queue.front();
        m_queue.pop();
        return a;
    }
    return DataType();
    } else throw Sct::IllegalStateError("Couldn't get queue lock", __FILE__, __LINE__);
}

template <class W>
bool WorkerGroup<W>::popAndWork(Worker& worker){
    boost::mutex::scoped_lock lock(m_queueAccess);
    if (lock.locked()) {    
    if (m_queue.size()!=0) {
        DataType data=m_queue.front();
        m_queue.pop();
        lock.unlock();          //Don't hold onto the lock for too long!
        worker.m_busy = true;            // set the busy flag
        work(data);                  // call the function to do work.
        worker.m_busy = false;           // unset the busy flag 
        return true;
    } else return false;
    } else throw Sct::IllegalStateError("Couldn't get queue lock", __FILE__, __LINE__);
}

template <class W>    
unsigned WorkerGroup<W>::queueSize(){
    return m_queue.size();
}

template <class W>
unsigned WorkerGroup<W>::nWorkers(){
    return m_workers.size();
}

template <class W>
void WorkerGroup<W>::stop(){
    m_run = false;
    join();
}

template <class W>
void WorkerGroup<W>::join(){
    m_thread_group.join_all();
}

template <class W>
void WorkerGroup<W>::setPaused(bool p){
    m_pause = p;
}

template <class W>
unsigned WorkerGroup<W>::paused(){
    unsigned npaused=0;
    for (Iterator it=m_workers.begin(); it!=m_workers.end(); ++it) 
    if ( (*it)->m_paused ) ++npaused;

    return npaused;
}

template <class W>
unsigned WorkerGroup<W>::busy(){
    unsigned nbusy=0;
    for (Iterator it=m_workers.begin(); it!=m_workers.end(); ++it) {
    if ( (*it)->m_busy ) ++nbusy;
    }
    return nbusy;
}

template <class W>
void WorkerGroup<W>::setSleepTime(unsigned t){
    m_sleeptime = t;
}

template <class W>
void WorkerGroup<W>::wait(int wait_time) {
    boost::xtime xt;
    boost::xtime_get(&xt, boost::TIME_UTC);
    xt.nsec += wait_time*1000000;
    boost::thread::sleep(xt);
}

template <class W>
void WorkerGroup<W>::waitForData(){
    wait(getSleepTime());
}


} // end of namepace SctService
#endif //#ifndef WORKERGROUP_H

---