---

TSqueue.h

Go to the documentation of this file.

#ifndef UTILITIES_THREADS_TSQUEUE_H
#define UTILITIES_THREADS_TSQUEUE_H
//
//  a thread safe queue
//
#include "Utilities/Threads/interface/Thread.h"
#include "Utilities/General/interface/MutexUtils.h"
#include <boost/thread/condition.hpp>
#include "Utilities/General/interface/GeneralVerbosity.h"

#include "Utilities/General/interface/CMSexception.h"
#include "Utilities/General/interface/UncatchedException.h"
#include "Utilities/General/interface/ClassName.h"
#include "boost/lexical_cast.hpp"

#include <queue>
#include <deque>

struct TSqueue_termination : public Thread::Terminate {};

template<class X>
class TSqueue {
  
public: 

  typedef TSqueue<X> self;

public:
  explicit TSqueue(size_t imax=1000, size_t imin=900) : 
    maxSize(imax), minSize(imin),
    draining_(false), terminate_(false), abort_(false) {
    if (minSize<1) minSize=1;
  }

  void reset() {
    draining_=terminate_=abort_=false;
  }
  
  void push(const X& x) {
    SimpleLockMutex gl(genLock);
    if (abort_) throw TSqueue_termination();
    while(cont.size()>maxSize) {
      notFull.wait(gl());
      if (abort_) throw TSqueue_termination();
    }
    cont.push(x);    
    if ( !(cont.size()<minSize )) { notEmpty.notify_all(); } 
  }
  
  X pop() {
    SimpleLockMutex gl(genLock);
    while(cont.empty()) {
      if (draining_) dodrain.notify_all();
      if (terminate_) throw TSqueue_termination();
      notEmpty.wait(gl());
    }
    X x = cont.front();
    cont.pop();
    if (cont.size()<minSize) notFull.notify_all(); 
    return x;
  }
  
  void terminate() {
    SimpleLockMutex gl(genLock);
    terminate_ = true;
    notEmpty.notify_all();
  }
  
  void abort() {
    SimpleLockMutex gl(genLock);
    abort_ = true;
    clear();
    notFull.notify_all(); 
    if (draining_) dodrain.notify_all();
  }

  void drain() {
    SimpleLockMutex gl(genLock);
    if (cont.empty()) return;
    draining_=true;
    notEmpty.notify_all();
    while(!cont.empty()) {
      dodrain.wait(gl());
      if (abort_) throw TSqueue_termination();
    }
  }
  size_t size() const {
    SimpleLockMutex gl(genLock);
    return cont.size();
  }

  void clear() {
    while(!cont.empty()) cont.pop();
  }


private:


  
  std::queue<X,std::deque<X> > cont;
  
  size_t maxSize;
  size_t minSize;
  
  mutable SimpleLockMutex::Mutex genLock;
  boost::condition notEmpty;
  boost::condition notFull;
  boost::condition dodrain;


  bool draining_;

  bool terminate_;
  bool abort_;
  
};

template<class X, class C>
class TSqueueConsumer : public Thread{

public: 
  
  typedef TSqueueConsumer<X,C> self;
  
public:
  
  typedef TSqueue<X>  Xqueue;
  
private:
  
  void run() {
      if (!GeneralVerbosity::silent())
          GeneralUtilities::cout << ">>>>> consumer " << ClassName<C>::name()
                    << " thread starts.." << thread_self_tid() <<  std::endl;
    
#ifdef __linux__
    // now outside...
    //    string mess("Thread "); mess += boost::lexical_cast<string>(thread_self_tid());
    //    LinuxElapsedTime totaltime(mess,cout); 
#endif // __linux__
    
    if (!GeneralVerbosity::silent())
      GeneralUtilities::cout << thread_self_tid() << " queue size is "<< inqueue.size() <<  std::endl;
    
    C consumer;
    
    try{  // this try logic is here to avoid double throw from consumer destructor...
      for(;;) {
        try {
          consumer(inqueue.pop());
        }
        catch ( Thread::Terminate ) {
          throw;
        }
        catch ( GenTerminate ) {
          if (!interactive_) throw;
          abort();
        }
        catch (...) {
          if (!interactive_) throw;
          if (!GeneralVerbosity::silent()) {
            GeneralUtilities::cout << "thread " 
                          << self_tid() << " caught exception " << std::endl;
          } 
          abort();
        } 
      }
    }
    catch ( Thread::Terminate ) {
      throw;
    }
    catch ( GenTerminate ) {
      throw;
    }
    catch ( cms::Exception & cexp) {
      if (!GeneralVerbosity::silent()) {
        GeneralUtilities::cout << "thread " 
                      << self_tid() << " caught  CMS/SEAL exception (" 
                      << className(cexp) 
                      << ") :" <<  std::endl;
        Genexception * ce = dynamic_cast<Genexception *>(&cexp);
        if (ce) ce->dump(GeneralUtilities::cout, true);
        else  GeneralUtilities::cout << cexp.what() <<  std::endl;
        GeneralUtilities::cout << " inform producer...." << std::endl;
      }
      UncatchedException uce(cexp);
      assert(&uce != 0); // Avoids warning message for unused variable uce.
      abort();
    }
    catch (...) {
      if (!GeneralVerbosity::silent()) {
        GeneralUtilities::cout << "thread " 
                      << self_tid() << " caught unknown exception " <<  std::endl;    
        GeneralUtilities::cout << " inform producer...." <<  std::endl;
      }
      UncatchedException uce;
      assert(&uce != 0); // Avoids warning message for unused variable uce.
      abort();
    }
    
    if (!GeneralVerbosity::silent())
      GeneralUtilities::cout << ">>>>> consumer " << ClassName<C>::name()
                    << " thread ends.." << thread_self_tid() << std::endl;
  }
  
  virtual void terminate(){
    if (!GeneralVerbosity::silent())
      GeneralUtilities::cout << ">>>>> consumer " << ClassName<C>::name()
                    << " thread terminate.." << thread_self_tid() << std:: endl;
  }
  
  virtual void abort() { 
    if (!GeneralVerbosity::silent())
      GeneralUtilities::cout << ">>>>> consumer " << ClassName<C>::name()
                    << " thread aborts.." << thread_self_tid() <<  std::endl;
    inqueue.abort(); 
  }
  
public:
  explicit TSqueueConsumer(size_t imax=1000, size_t imin=900):
    inqueue(imax,imin),  interactive_(false) {
    initialize();
  }
  
  ~TSqueueConsumer() {
    inqueue.terminate();
    finalize();
  }
  
  void reset() {  inqueue.reset();}
  
  void interactive() { interactive_=true;}
  
  // wait queue to drain
  void wait() { 
    if (!GeneralVerbosity::silent())
      GeneralUtilities::cout << "wait queue to drain" << std::endl;
    inqueue.drain();
  }
  
  void push(const X& x) { inqueue.push(x); }
  
  size_t size() const { return inqueue.size();}

    
  
protected:

  Xqueue inqueue;
  bool interactive_;
  
};

#endif // UTILITIES_THREADS_TSQUEUE_H

---