WorkerGroup.h

#ifndef WORKERGROUP_H
#define WORKERGROUP_H
#include <list>
#include <vector>
#include <algorithm>

#include <boost/thread.hpp>
#include <boost/timer.hpp>

#include "Sct/LogicErrors.h"
#include "Sct/ISUtilities.h"
#include "Sct/SctNames.h"
#include "IS/ServiceStatus.h"

namespace SctService{

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

    unsigned queueSize() const;
    unsigned nWorkers() const;
    unsigned long nDone() const;
    unsigned busy() const;
    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() const;
    void setSleepTime(unsigned msec);
    unsigned getSleepTime() const {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);    

    /*
      Set First In First Out (if true) or First In Last Out (if false).
      FILO is more performant, since less disk swap is necessary, 
      but a bit less intuitive for the user!
    */
    void setFifo(bool val);
    /*
      get FIFO (true) or FILO (false) state.
    */
    bool isFifo() const;
    virtual ~WorkerGroup() {;} 
    
    WorkerGroup() : m_run(true), m_pause(false), m_sleeptime(1), m_reportInterval(1000), 
      m_nDone(0), m_fifo(true) {}

    void reportTo(std::string isname);
    void reportInterval(int msec);
protected:
    class Worker {
    public:
    Worker(WorkerGroup<DataType>& group) : group(group), m_busy(false), m_paused(false) {
    }
    
    Worker(const Worker& w) : group(w.group), m_busy(false), m_paused(false) {}

    void operator() () {
        group.threadMain(*this);
    }
    WorkerGroup<DataType>& group;
    bool busy();
    bool paused();
    void setBusy(bool v);
    void setPaused(bool v);
    //Worker& operator=(const Worker& w)
    private:
    boost::mutex m_statusMutex;      
    volatile bool m_busy;            
    volatile bool m_paused;          
    };

    class Reporter {
    public:
      Reporter(WorkerGroup<DataType>& group) : group(group){ 
      }
      void operator() (){
    group.reportThreadMain(*this);
      }
      WorkerGroup<DataType>& group;
      ServiceStatus last;
    };

    virtual void reportThreadMain(Reporter& reporter) throw();
    
    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;
    typedef typename WorkerStore::const_iterator ConstIterator;
    
     WorkerStore m_workers;

    std::list<DataType> m_queue;
    

    boost::thread_group m_thread_group;
    
    volatile bool m_run;  
    volatile bool m_pause; 
    volatile unsigned m_sleeptime; 
    std::string m_reportTo; 
    volatile int m_reportInterval;   
    
    mutable boost::mutex m_queueAccess;  
    mutable boost::mutex m_vectorAccess; 
    mutable boost::recursive_mutex m_reportAccess; 
    volatile unsigned long m_nDone;   
    volatile double m_time;   
    friend class Worker;
    friend class Reporter;
    void addTime(double t);            
    bool m_fifo;                       

    DataType unlockedPopFrontOrBack();
};


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

template <class T>
inline 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.setPaused(true);
        wait(m_sleeptime);
    }
    worker.setPaused(false);
    if (!m_run) return;                     // if no longer running, stop.
    
    if (!popAndWork(worker)) {
        waitForData();    // if queue is empty, sleep.
    }
    }
    return;
}

template <class T>
inline void WorkerGroup<T>::reportThreadMain(Reporter& reporter) throw() { 
  bool problemsToMrs=false;
  bool first=true;
  std::cout << "Status to: [" << m_reportTo << "]" << std::endl;
  while (m_run) {
    while (m_pause && m_run) {
      wait(m_reportInterval);
    }
    if (!m_run) return; 
    wait(m_reportInterval);
    {
      boost::recursive_mutex::scoped_lock lock(m_reportAccess);
      if (!lock.locked()) {
    continue;
      }
      bool modified = (reporter.last.queue != queueSize()||
               reporter.last.done != nDone() || first);
      if (first) first=false;
      if (modified){
    reporter.last.averageTimeEach=0.;
    reporter.last.timeLeft=0.;
    reporter.last.queue = queueSize();
    reporter.last.done  = nDone();
    if (reporter.last.done!=0) {
      reporter.last.averageTimeEach = static_cast<double>(m_time)/static_cast<double>(reporter.last.done);
    }
    unsigned nw=nWorkers();
    if (nw!=0) {
      reporter.last.timeLeft = reporter.last.queue*reporter.last.averageTimeEach / static_cast<double>(nw);
    }
    if (m_reportTo!="") {
      try {Sct::ISUtilities::addOrUpdateOrThrow(m_reportTo, reporter.last,
                            __FILE__, __LINE__, MRS_DIAGNOSTIC);
      }catch(Sct::Throwable& e){
        if (!problemsToMrs){
          problemsToMrs=true;
          e.sendToMrs(MRS_DIAGNOSTIC);
        }
      }
    }
      }
    }
  }
  return;
}

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

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

template <class T>
inline T WorkerGroup<T>::unlockedPopFrontOrBack(){
  if (m_queue.empty()) throw Sct::IllegalStateError("Queue is emtpy!", __FILE__, __LINE__);
  if (m_fifo){
    DataType data=m_queue.front();
    m_queue.pop_front();
    return data;
  }else{
    DataType data=m_queue.back();
    m_queue.pop_back();
    return data;
  }
}

template <class T>
inline T WorkerGroup<T>::pop(){
    boost::mutex::scoped_lock lock(m_queueAccess);
    if (lock.locked()) {
      return unlockedPopFrontOrBack();
    } else throw Sct::IllegalStateError("Couldn't get queue lock", __FILE__, __LINE__);
}

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

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

template <class W>    
inline unsigned long WorkerGroup<W>::nDone() const{
    return m_nDone;
}

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

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

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

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

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

    return npaused;
}

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

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

template <class W>
inline 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>
inline void WorkerGroup<W>::waitForData(){
    wait(getSleepTime());
}

template <class T>
void WorkerGroup<T>::reportTo(std::string isname){
  boost::recursive_mutex::scoped_lock lock(m_reportAccess);
  m_reportTo=isname;
}

template <class T>
inline void WorkerGroup<T>::reportInterval(int msec){
  boost::recursive_mutex::scoped_lock lock(m_reportAccess);
  if (msec>100){
    m_reportInterval=msec;
  }
}

template <class W>
inline void WorkerGroup<W>::addTime(double t){
  boost::recursive_mutex::scoped_lock lock(m_reportAccess);
  m_time+=t;
  ++m_nDone;
}

template <class W>
inline void WorkerGroup<W>::setFifo(bool val){
  boost::mutex::scoped_lock lock(m_queueAccess);
  if (!lock.locked()) throw Sct::IllegalStateError("Could not get status lock",__FILE__, __LINE__);
  m_fifo=val;
}

template <class W>
inline bool WorkerGroup<W>::isFifo()const{
  boost::mutex::scoped_lock lock(m_queueAccess);
  if (!lock.locked()) throw Sct::IllegalStateError("Could not get status lock",__FILE__, __LINE__);
  return m_fifo;
}



template <class W>
inline bool WorkerGroup<W>::Worker::busy(){
  boost::mutex::scoped_lock lock(m_statusMutex);
  if (!lock.locked()) throw Sct::IllegalStateError("Could not get status lock",__FILE__, __LINE__);
  return m_busy;
}

template <class W>
inline bool WorkerGroup<W>::Worker::paused(){
  boost::mutex::scoped_lock lock(m_statusMutex);
  if (!lock.locked()) throw Sct::IllegalStateError("Could not get status lock",__FILE__, __LINE__);
  return m_paused;
}

template <class W>
void WorkerGroup<W>::Worker::setBusy(bool val){
  boost::mutex::scoped_lock lock(m_statusMutex);
  if (!lock.locked()) throw Sct::IllegalStateError("Could not get status lock",__FILE__, __LINE__);
  m_busy=val;
}

template <class W>
inline void WorkerGroup<W>::Worker::setPaused(bool val){
  boost::mutex::scoped_lock lock(m_statusMutex);
  if (!lock.locked()) throw Sct::IllegalStateError("Could not get status lock",__FILE__, __LINE__);
  m_paused=val;
}

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

---