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 "gcp/util/common/Command.h"
#include "gcp/util/common/Debug.h"     
#include "gcp/util/common/IoLock.h"
#include "gcp/util/common/Exception.h" // Definition of Error macro
#include "gcp/util/common/FdSet.h"
#include "gcp/util/common/LogStream.h"
#include "gcp/util/common/PipeQ.h"
#include "gcp/util/common/Thread.h"

namespace gcp {
  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_;
      
      gcp::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);
      
      //------------------------------------------------------------
      // A vector of commands used by this task
      //------------------------------------------------------------

      std::vector<Command*> commands_;

      struct timeval* timeOut_;

      TimeVal commandTimeOut_;

      void stepCommands();
      
    }; // End class GenericTask
    
    // Template functions must be defined in the header file.
    
    template<class Msg>
      GenericTask<Msg>::GenericTask() 
      {
    thread_  = 0;
    timeOut_ = 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;
    timeOut_ = 0;
    
    // 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); 
    
    DBPRINT(false, Debug::DEBUG7, "About to cancel threads");
    
    // 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];
    // Delete any commands too

    for(unsigned iCommand=0; iCommand < commands_.size(); iCommand++)
      if(commands_[iCommand] != 0)
        delete commands_[iCommand];
      }
    
    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);
    
    DBPRINT(true, Debug::DEBUG2, "Inside processTaskMsg: " 
        << msg.genericMsgType_);
    
    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)
      { 
    //  DBPRINT(true, Debug::DEBUG10, "Inside sendTaskMsg for thread: "
    //      << thread_->strName());
    
    try {
      DBPRINT(true, Debug::DEBUG7, "Sending a message " << pthread_self());
      msgq_.sendMsg(msg); 
      DBPRINT(true, Debug::DEBUG7, "Sending a message: done " << pthread_self());
    } catch(Exception& err) {
      COUT("Caught an exception in task instantiated by thread: " << thread_->name());
      
      // Rethrow
      
      throw err;
    }
      }
    
    template<class Msg>
      void GenericTask<Msg>::fwdTaskMsg(Msg* msg)
      { 
    DBPRINT(true, Debug::DEBUG2, "Inside fwdTaskMsg");

    try {
      msgq_.sendMsg(msg); 
    } catch(Exception& err) {
      COUT("Caught an exception in task instantiated by thread: " << thread_->name());
      
      // Rethrow
      
      throw err;
    }
      }
    
    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(int 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_) {
          gcp::util::LogStream errStr;
          errStr.appendMessage(true, "Error in thread startup function\n");
          throw Error(errStr);
        }
      }
      }
    
    template<class Msg>
      bool GenericTask<Msg>::threadsNeedStarting()
      {
    for(int ithread=0; ithread < threads_.size(); ithread++) 
      if(!threads_[ithread]->isRunning())
        return true;
    return false;
      }
    
    template<class Msg>
      bool GenericTask<Msg>::threadsNeedCancelling()
      {
    for(int 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;
    
    for(int 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;
    
    for(int 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(int 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(int ithread=0; ithread < threads_.size(); ithread++) {
      if(threads_[ithread]->matchName(name))
        return threads_[ithread];
    }
    
    ThrowError("No matching thread found");
      };
    
    
    // Raise a signal to a thread.
    
    template<class Msg>
      void GenericTask<Msg>::raise(std::string name, int sigNo)
      {
    for(int 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(int ithread=0; ithread < threads_.size(); ithread++) {
      if(!threads_[ithread]->isRunning()) {
        ReportError("Thread " 
            << threads_[ithread]->strName() 
            << " is not running!");
        
        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) {};
    
    // Utilities
    
    template<class Msg>
      void GenericTask<Msg>::shutdownConnection(int fd)
      {
    if(fd >= 0) {
      fdSet_.clear(fd);
      ::shutdown(fd, 2);
      ::close(fd);
    }
    
      }
    
    template<class Msg>
      void GenericTask<Msg>::stepCommands() {

      // On entry, set the timeout to NULL.  If any command is still
      // active, we will set the timeout pointing to the timeout
      // returned by the command.  This is so that inheritors can use
      // timeOut_ without worrying about setting it back to NULL when
      // all commands are finished

      timeOut_ = 0;
      bool first=true;
      for(unsigned iCommand=0; iCommand < commands_.size(); iCommand++) {

    // If any command is still active, execute its next
    // instruction, and record the timeout.

    if(commands_[iCommand]->active()) {
      commands_[iCommand]->executeNextInstruction(commandTimeOut_, first);
      timeOut_ = commandTimeOut_.timeVal();

      if(first)
        first = false;
    }
      }
    }
    
  } // End namespace util
} // End namespace gcp


#endif // End #ifndef