GenericTask.h

Go to the documentation of this file.

#ifndef GENERICTASK_H
#define GENERICTASK_H

#include <vector>
#include <unistd.h>
#include <sys/socket.h> // shutdown()
#include <sys/time.h>
#include <sys/types.h>

#include "carma/szautil/Debug.h"     
#include "carma/szautil/IoLock.h"
#include "carma/szautil/Exception.h" // Definition of Error macro
#include "carma/szautil/FdSet.h"
#include "carma/szautil/LogStream.h"
#include "carma/szautil/PipeQ.h"
#include "carma/szautil/Thread.h"

namespace sza {
  namespace util {
    
    template<class Msg>
      class GenericTask {
      
      public:
      
      void sendRestartMsg();
      
      void sendStopMsg();
      
      void sendHeartBeatMsg();
      
      virtual void fwdTaskMsg(Msg* msg);
      
      protected:
      
      GenericTask();
      
      GenericTask(Thread* thread);
      
      virtual ~GenericTask();
      
      void sendTaskMsg(Msg* msg);
      
      Thread* thread_;
      
      std::vector<Thread*> threads_;
      
      void startThreads(void* arg);

      void startThread(void* arg, unsigned order);
      
      unsigned getMinStartOrder();

      bool threadsNeedStarting();

      void cancelThreads();

      void cancelThread(unsigned order);
      
      unsigned getMinCancelOrder();

      bool threadsNeedCancelling();

      void pingThreads(void* arg);
      
      void raise(std::string name, int sigNo);
      
      Thread* getThread(std::string name);
      
      bool threadsAreRunning();
      
      PipeQ<Msg> msgq_;
      
      sza::util::FdSet fdSet_;
      
      // Fd utilities

      void shutdownConnection(int fd);

      virtual void serviceMsgQ(void);
      
      virtual void restart(void);
      
      virtual void run(void);
      
      virtual void processTaskMsg(bool* stop);
      
      virtual void processMsg(Msg* msg);
      
      virtual void respondToHeartBeat();

      //------------------------------------------------------------
      // Signals and timers
      //------------------------------------------------------------

      virtual void installTimer(Msg* msg);
      
      virtual void installSignal(Msg* msg);
      
      virtual void enableTimer(Msg* msg);
      
      virtual void addHandler(Msg* msg);
      
    }; // End class GenericTask
    
    // Template functions must be defined in the header file.
    
    template<class Msg>
      GenericTask<Msg>::GenericTask() 
      {
        thread_  = 0;
        
        // Register the message queue fd to be watched for readability
        
        fdSet_.zeroReadFdSet();
        fdSet_.registerReadFd(msgq_.fd());
      }
    
    template<class Msg>
      GenericTask<Msg>::GenericTask(Thread* thread) 
      {
        thread_  = thread;
        
        // Register the message queue fd to be watched for readability
        
        fdSet_.zeroReadFdSet();
        fdSet_.registerReadFd(msgq_.fd());
      }
    
    template<class Msg>
      GenericTask<Msg>::~GenericTask()
      {
        // Broadcast to other threads that this task is shutting
        // down.
        
        if(thread_ != 0)
          thread_->setRunState(false); 
        
        // Cancel all threads managed by this task.
        
        cancelThreads();
        
        // And delete any allocated memory.  NB: the default
        // destructor for vector<Thread*> will not delete memory
        // pointed to by its elements.
        
        for(unsigned ithread=0; ithread < threads_.size(); ithread++)
          if(threads_[ithread] != 0)
            delete threads_[ithread];
      }
    
    template<class Msg>
      void GenericTask<Msg>::restart(void) {}
    
    template<class Msg>
      void GenericTask<Msg>::run(void)
      {
        serviceMsgQ();
      }
    
    template<class Msg>
      void GenericTask<Msg>::serviceMsgQ(void) 
      {
        bool stop=false;
        int nready; // number of file descriptors ready for reading
        
        if(msgq_.fd() < 0)
          ThrowError("Received NULL file descriptor");
        
        // Loop, checking the message queue file descriptor for readability
        
        while(!stop && (nready=select(fdSet_.size(), fdSet_.readFdSet(), 
                                      NULL, NULL, NULL)) > 0) {
          
          // If no file descriptors were ready, throw an exception
          
          if(nready != 1)
            ThrowError("Error");

          DBPRINT(true, Debug::DEBUG2, "About to call processTaskMsg: "
                  << "nready = "  << nready);
                  
          
          processTaskMsg(&stop);
        };
      };
    
    template<class Msg>
      void GenericTask<Msg>::processTaskMsg(bool* stop)
      {
        Msg msg;
        
        msgq_.readMsg(&msg);

        switch (msg.genericMsgType_) {
        case Msg::HEARTBEAT: // Is this a heartbeat request?
          respondToHeartBeat();
          break;
        case Msg::RESTART:  // Is this a request to restart?
          restart();
          break;
        case Msg::STOP:     // Did we receive a request to shut
          // down?
          *stop = true;
          break;
        default: // Else forward this message to the task-specific
          // process method
          processMsg(&msg);
          break;
        }
      };
    
    template<class Msg>
      void GenericTask<Msg>::processMsg(Msg* msg) {};
    
    template<class Msg>
      void GenericTask<Msg>::respondToHeartBeat()
      {
        if(thread_ != 0)
          thread_->setRunState(true);
      }
    
    template<class Msg>
      void GenericTask<Msg>::sendRestartMsg() 
      {
        Msg msg;
        msg.genericMsgType_ = Msg::RESTART;
        
        // We use fwdTaskMsg() here instead of sendTaskMsg() so
        // that inheriting tasks can control their own messages
        // routing.
        
        fwdTaskMsg(&msg);
      }
    
    template<class Msg>
      void GenericTask<Msg>::sendStopMsg() 
      {
        Msg msg;
        msg.genericMsgType_ = Msg::STOP;
        
        // We use fwdTaskMsg() here instead of sendTaskMsg() so
        // that inheriting tasks can control their own messages
        // routing.
        
        fwdTaskMsg(&msg);
      };
    
    template<class Msg>
      void GenericTask<Msg>::sendHeartBeatMsg() 
      {
        Msg msg;
        msg.genericMsgType_ = Msg::HEARTBEAT;
        
        // We use fwdTaskMsg() here instead of sendTaskMsg() so
        // that inheriting tasks can control their own message
        // routing.
        
        fwdTaskMsg(&msg);
      };
    
    template<class Msg>
      void GenericTask<Msg>::sendTaskMsg(Msg* msg)
      { 
        msgq_.sendMsg(msg); 
      }
    
    template<class Msg>
      void GenericTask<Msg>::fwdTaskMsg(Msg* msg)
      { 
        msgq_.sendMsg(msg); 
      }
    
    template<class Msg>
      void GenericTask<Msg>::startThreads(void* arg)
      { 
        // Start threads in priority order

        while(threadsNeedStarting()) {
          unsigned order = getMinStartOrder();
          startThread(arg, order);
        }
      }

    template<class Msg>
      void GenericTask<Msg>::startThread(void* arg, unsigned order)
      { 
        // Search through the list in first-in, first-started order

        for(unsigned ithread=0; ithread < threads_.size(); ithread++) 
          if(!threads_[ithread]->isRunning() && threads_[ithread]->startOrder()==order) {

            threads_[ithread]->start(arg);

            // Check the error code from the start up function

            if(threads_[ithread]->wasError_) {
              ThrowError("Error in startup function for thread: " 
                         << threads_[ithread]->name());
            }
          }
      }

    template<class Msg>
      bool GenericTask<Msg>::threadsNeedStarting()
      {
        for(unsigned ithread=0; ithread < threads_.size(); ithread++) 
          if(!threads_[ithread]->isRunning())
            return true;
        return false;
      }

    template<class Msg>
      bool GenericTask<Msg>::threadsNeedCancelling()
      {
        for(unsigned ithread=0; ithread < threads_.size(); ithread++) 
          if(threads_[ithread]->isRunning())
            return true;
        return false;
      }

    template<class Msg>
      unsigned GenericTask<Msg>::getMinStartOrder()
      {
        bool first=true;
        unsigned minOrder=0;

        for(unsigned ithread=0; ithread < threads_.size(); ithread++) {
          if(!threads_[ithread]->isRunning()) {

            if(first) {
              minOrder = threads_[ithread]->startOrder();
                first = false;
            } else {
              minOrder = threads_[ithread]->startOrder() < minOrder ? 
                threads_[ithread]->startOrder() : minOrder;
            }
      }

        }   

        return minOrder;
      }

    template<class Msg>
      unsigned GenericTask<Msg>::getMinCancelOrder()
      {
        bool first=true;
        unsigned minOrder=0;

        for(unsigned ithread=0; ithread < threads_.size(); ithread++) {
          if(threads_[ithread]->isRunning()) {

            if(first) {
              minOrder = threads_[ithread]->cancelOrder();
              first = false;
            } else {
              minOrder = threads_[ithread]->cancelOrder() < minOrder ? 
                threads_[ithread]->cancelOrder() : minOrder;
            }
      }

        }

        return minOrder;
      }

    template<class Msg>
      void GenericTask<Msg>::cancelThreads()
      { 
        // Cancel threads in first-started, last-canceled order.
        
        DBPRINT(true, Debug::DEBUG7, "Managing " << threads_.size() 
                << " threads");

        while(threadsNeedCancelling()) {
          unsigned order = getMinCancelOrder();
          cancelThread(order);
        }

        DBPRINT(true, Debug::DEBUG7, "Leaving cancelThreads");
      }
    
    template<class Msg>
      void GenericTask<Msg>::cancelThread(unsigned order)
      { 
        // Cancel threads in first-started, last-canceled order.
        
        for(int ithread=threads_.size()-1; ithread >=0; ithread--) 
          if(threads_[ithread]->isRunning() && threads_[ithread]->cancelOrder() == order) 
            threads_[ithread]->cancel();
      }

    template<class Msg>
      void GenericTask<Msg>::pingThreads(void* arg)
      { 
        for(unsigned ithread=0; ithread < threads_.size(); ithread++) 
          if(threads_[ithread]->isPingable()) {
            threads_[ithread]->setRunState(false);
            threads_[ithread]->ping(arg);    
          }
      }
    
    template<class Msg>
      Thread* GenericTask<Msg>::getThread(std::string name)
      {
        for(unsigned ithread=0; ithread < threads_.size(); ithread++) {
          if(threads_[ithread]->matchName(name))
            return threads_[ithread];
        }
        
        ThrowError("No matching thread found");

        return 0;
      };
    
    
    // Raise a signal to a thread.
    
    template<class Msg>
      void GenericTask<Msg>::raise(std::string name, int sigNo)
      {
        for(unsigned ithread=0; ithread < threads_.size(); ithread++) {
          if(threads_[ithread]->matchName(name)) {
            threads_[ithread]->raise(sigNo);
            return;
          }
        }
        
        ReportError("No matching thread found");
      };
    
    
    // Check the running status of all threads

    template<class Msg>
      bool GenericTask<Msg>::threadsAreRunning()
      {
        for(unsigned ithread=0; ithread < threads_.size(); ithread++) {
          if(!threads_[ithread]->isRunning()) {
            ReportSimpleError("No heartbeat response from thread " 
                              << threads_[ithread]->strName());

            return false;
          } else {
            COUT("Thread " 
                 << threads_[ithread]->strName() 
                 << " is running.");
          }
        }
        return true;
      }
    
    //------------------------------------------------------------
    // Signals and timers
    //------------------------------------------------------------

    template<class Msg>
      void GenericTask<Msg>::installTimer(Msg* msg) {};
    
    template<class Msg>
      void GenericTask<Msg>::installSignal(Msg* msg) {};
    
    template<class Msg>
      void GenericTask<Msg>::enableTimer(Msg* msg) {};
    
    template<class Msg>
      void GenericTask<Msg>::addHandler(Msg* msg) {};

    // Utiliies
    
    template<class Msg>
      void GenericTask<Msg>::shutdownConnection(int fd)
    {
      if(fd >= 0) {
        fdSet_.clear(fd);
        ::shutdown(fd, 2);
        ::close(fd);
      }
    }
  }; // End namespace util
}; // End namespace sza

#endif // End #ifndef