JDAQClient.hh

Go to the documentation of this file.

#ifndef __JRUNCONTROL__JDAQCLIENT__
#define __JRUNCONTROL__JDAQCLIENT__

#include <string>
#include <iostream>
#include <sstream>
#include <iomanip>
#include <exception>
#include <typeinfo>
#include <ctype.h>

#include "JSystem/JNetwork.hh"
#include "JNet/JControlHost.hh"
#include "JNet/JControlHostServer.hh"
#include "JNet/JSelectReader.hh"
#include "JLang/JString.hh"
#include "JLang/JSharedPointer.hh"
#include "JLang/JAbstractFile.hh"
#include "JLang/JFileDescriptorMask.hh"
#include "JLang/JTimeval.hh"
#include "JLang/JRedirectStream.hh"
#include "JLang/JNullStream.hh"
#include "Jeep/JParser.hh"
#include "Jeep/JProperties.hh"
#include "Jeep/JArgs.hh"
#include "Jeep/JTimekeeper.hh"
#include "JLogger/JMessageLogger.hh"
#include "JLogger/JStreamLogger.hh"
#include "JLogger/JControlHostLogger.hh"
#include "JLogger/JMessageStream.hh"
#include "JDAQ/JDAQTags.hh"
#include "JRuncontrol/JRuncontrolToolkit.hh"
#include "JRuncontrol/JEventTable.hh"
#include "JRuncontrol/JDAQCHSM.hh"


/**
 * \author mdejong
 */

namespace KM3NETDAQ {

  using namespace JLOGGER;
  using JNET::JControlHost;
  using JNET::JControlHostServer;
  using JNET::JSelectReader;
  using JLANG::JControlHostException;
  using JLANG::JAbstractFile;
  using JLANG::JFileDescriptorMask;
  using JLANG::JAbstractFile;
  using JLANG::JTimeval;
  using JLANG::JSharedPointer;
  using JLANG::JRedirectStream;
  using JEEP::JArgs;
  using JEEP::JTimekeeper;


  /**
   * Run control client base class.
   *
   * This base class implements the protocol for the communication 
   * with the central run control program.
   * This protocol is based on JNET::JControlHost tags and CHSM event names.
   * The method run can be used to process JNET::JControlHost messages
   * and accordingly update the state machine.
   *
   * For client specific events with a designated JNET::JControlHost tag,
   * the corresponding entry in the event table should be replaced.
   * This should be done in the constructor of the derived class
   * using method JDAQClient::replaceEvent.
   *
   * The method JDAQClient::addParameter can be used to add a parameter of
   * the derived class to the list that is parsed in method enter.
   *
   * The method JDAQClient::filter can be overwritten so that a specific action
   * is made before the corresponding message is processed by the state machine.
   * The message is ignored if this method returns true,
   * else it is normally processed.
   *
   * The method JDAQClient::setSelect can be used to set the file descriptor mask
   * of the select call and the method actionSelect can be used to take 
   * client specific actions following the select call.
   *
   * If the clock interval is non-zero, the method JDAQClient::actionRunning is 
   * repeatedly called when this client machine is in state <tt>Running</tt>.
   * The clock interval can be set using method JDAQClient::setClockInterval.
   */
  class JDAQClient :
    public JDAQStateMachine
  {
  public:
    /**
     * Constructor.
     *
     * \param  name                 name of client
     */
    JDAQClient(const std::string& name) :
      JDAQStateMachine(name),
      parser(),
      event_number(-1)
    {
      configure();
    }


    /**
     * Constructor.
     *
     * \param  name                 name of client
     * \param  server               name of command message server
     * \param  logger               pointer to logger
     * \param  level                debug level   
     */
    JDAQClient(const std::string& name,
               const std::string& server,
               JLogger*           logger,
               const int          level) :
      JDAQStateMachine(name),
      parser(),
      event_number(-1)
    {
      this->logger = JMessageLogger(logger, name, level);

      try {
        this->server.reset(new JControlHost(server));
      }
      catch(JControlHostException& error) {
        JErrorStream(this->logger) << error;
      }

      configure();
    }


    /**
     * Enter the state machine.
     * This overloaded method enter reproduces the constructor.
     * All necessary input is parsed from the list of arguments.
     * In case of an error, the state machine is not entered.
     * 
     * \param  args                 array  of command line arguments
     */
    virtual bool enter(const JArgs& args)
    {
      using namespace std;

      string  server;
      string  logger;
      int     level;
      bool    use_cout;

      try {
        
        parser['H'] = make_field(server)    = "localhost";
        parser['M'] = make_field(logger)    = "localhost";
        parser['d'] = make_field(level)     = 0;
        parser['c'] = make_field(use_cout);
        
        if (parser.read(args) != 0) {
          return false;
        }
      }
      catch(const exception &error) {
        cerr << error.what() << endl;
        return false;
      }

      try {

        JLogger* out = NULL;

        if (use_cout)
          out = new JStreamLogger(cout);
        else
          out = new JControlHostLogger(logger);
        
        this->logger = JMessageLogger(out, getName(), level);

        this->server.reset(new JControlHost(server));

        return enter();
      } 
      catch(const JControlHostException& error) {
        cerr << error << endl;
        return false;
      }
    }


    /**
     * Enter the state machine.
     * This method activates the subscription to JNET::JControlHost messages.
     * In case of an error, the state machine is not entered.
     */
    virtual bool enter()
    {
      using namespace std;

      if (server.is_valid() && logger.is_valid()) {

        try {
          
          server->Subscribe(getSubscription(eventTable));
          server->SendMeAlways();
          server->MyId(getFullName());
          
          return CHSM::machine::enter();
        }
        catch(const JControlHostException& error) {
          JErrorStream(logger) << error;
        }

      } else {
        cerr << "Message server or logger not properly initialised." << endl;
      }

      return false;
    }


    /**
     * Exit the state machine.
     * This method releases the various resources.
     */
    virtual bool exit()
    {
      try { 
        if (server.is_valid()) server.reset(NULL);
      }
      catch(const JControlHostException& error) {
      }

      try {
        if (logger.is_valid()) logger.reset(NULL);
      }
      catch(const JControlHostException& error) {
      }

      return CHSM::machine::exit();
    }


    /**
     * Get full name of this run control client.
     *
     * \return                      full name
     */
    const std::string& getFullName() const
    {
      return full_name;
    }


    /**
     * Get unique tag of this run control client.
     *
     * \return                      unique tag
     */
    const JNET::JTag& getUniqueTag() const
    {
      return unique_tag;
    }


    /**
     * Get last event number.
     *
     * \return                      event number
     */
    int getEventNumber() const
    {
      return event_number;
    }


    /**
     * Check if this client is in runnig state.
     *
     * \return                      true if running; else false
     */
    bool isRunning() const
    {
      return Main.RunControl.Operational.Running.active();
    }


    /**
     * Replace tag of given event in event table.
     *
     * \param  oldTag               old tag
     * \param  newTag               new tag
     * \param  event                event
     */
    void replaceEvent(const JNET::JTag& oldTag, 
                      const JNET::JTag& newTag, 
                      JDAQEvent_t&      event) 
    {
      eventTable.replace(oldTag, newTag, event);
    }


    /**
     * Find event in event table.
     *
     * \param  tag                  tag
     * \param  event_name           event name
     * \return                      pointer to event or NULL
     */
    JDAQEvent_t* findEvent(const JNET::JTag& tag, const std::string& event_name) 
    {
      JEventTable::const_iterator i = eventTable.find(tag, event_name);
            
      if (i != eventTable.end())
        return i->second;
      else
        return NULL;
    }


    /**
     * Add parameter to parser used in method enter().
     *
     * \param  option               option
     * \param  parameter            parameter
     */
    template<class T>
    void addParameter(const char option, T& parameter)
    {
      parser[option] = make_field(parameter);
    }


    /**
     * Add parameter to parser used in method enter().
     *
     * \param  option               option
     * \param  parameter            parameter
     * \param  value                default value
     */
    template<class T>
    void addParameter(const char option, T& parameter, const T& value)
    {
      parser[option] = make_field(parameter) = value;
    }


    /**
     * Get total delay time.
     *
     * \return                 delay time [us]
     */
    long long int getClockDelay() const
    {
      return clock.getDelay();
    }


    /**
     * Get hostname.
     *
     * \return                 host name
     */
    const std::string& getHostname() const
    {
      return hostname;
    }


    /**
     * Get interval time.
     *
     * \return                 interval time [us]
     */
    long long int getClockInterval() const
    {
      return clock.getInterval();
    }


    /**
     * Set interval time.
     *
     * \param  interval_us     interval time [us]
     */
    void setClockInterval(const long long int interval_us)
    {
      clock.setInterval(interval_us);
    }


    /**
     * Reset clock.
     */
    void resetClock()
    {
      clock.reset();
    }


    /**
     * Set the file descriptor mask for the select call.
     */
    void setSelect() 
    {
      select.reset();
      
      setSelect(select.getReaderMask());

      select.setReaderMask(*server);
    }


    /**
     * Set the file descriptor mask for the select call.
     * This implementation does nothing but may be redefined by the derived class.
     *
     * \param  mask                 file descriptor mask
     */
    virtual void setSelect(JFileDescriptorMask& mask) const
    {}


    /**
     * Action method following last select call.
     * This implementation does nothing but may be redefined by the derived class.
     *
     * \param  mask                 file descriptor mask
     */
    virtual void actionSelect(const JFileDescriptorMask& mask)
    {}


    /**
     * This method is repeatedly called when this client machine is in state Running
     * and the clock interval time is non-zero.
     * This implementation does nothing but may be redefined by the derived class.
     * Care has to be taken so that the time needed to execute this method should be
     * less than the specified clock interval time (see method setClockInterval()).
     */
    virtual void actionRunning()
    {}


    /**
     * This method is called at <tt>ev_input</tt>.
     *
     * \param  length               length of data
     * \param  buffer               pointer to data
     */
    virtual void actionInput(int length, const char* buffer)
    {
      using namespace std;

      JProperties properties(JEquationParameters("=", ";", "", ""), 1);

      int level = this->logger.getLevel();

      properties["debug"] = level;

      properties.read(string(buffer, length));

      this->logger.setLevel(level);
    }


    /**
     * Filter message.
     * The filter method can be overloaded so that a specific action is made
     * before the corresponding message is processed by the state machine.
     * The message is ignored if true is returned, else it is normally processed.
     *
     * \param  tag                  tag
     * \param  length               number of characters
     * \param  buffer               message
     * \return                      skip message or not
     */
    virtual bool filter(const std::string& tag, int length, const char* buffer)
    {
      return false;
    }


    /**
     * Run as run control client following command messages via JNET::JControlHost.
     * This method can be called once the state machine is entered.
     * It returns when the state machine is exited.
     * If the clock interval is non-zero, the method actionRunning() is 
     * repeatedly called when this client machine is in state Running.
     * The file descriptor mask can be set to interrupt the timeout of 
     * the select call and clock method wait() in this calling sequence 
     * (see methods setSelect() and actionSelect()).
     */
    void run()
    {
      using namespace std;

      while (active()) {

        try {

          setSelect();

          if (select(JTimeval(TIMEOUT_S,0)) > 0) {
            
            if (select.hasReaderMask(*server)) {
              update();
            }
            
            actionSelect(select.getReaderMask());

          } else {

            continue;
          }


          if (isRunning() && clock.getInterval() != 0LL) {

            long long int numberOfCalls = 0;
            
            clock.reset();
            
            do {
              
              ++numberOfCalls;
              
              setSelect();

              if (clock.wait(select.getReaderMask())) {
                
                if (select.hasReaderMask(*server)) {
                  update();
                }

                actionSelect(select.getReaderMask());

              } else {

                try {
                  actionRunning();
                }
                catch(const exception& error) {
                  logger.error(error.what());
                }             
              }

            } while (isRunning());
            
            if (numberOfCalls != 0) {
              JNoticeStream(logger) << "Delay per call " << clock.getDelay() / numberOfCalls / 1000 << " ms";
            }
          }
        }
        catch(const exception& error) {
          JErrorStream(logger) << "method run(): " << error.what();
        }
      }
    }


    /**
     * Run for ever.
     * This method can be used when the run control client is started before the run control
     * (e.g. at boot time of the host processor).
     * This method should be called before the state machine is entered.
     * It launches a server which accepts a JNET::JControlHost connection from 
     * a designated application e.g. the JDAQClientStarter.cc program.
     * The state machine is entered using the available data in the JNET::JControlHost message. 
     * After the state machine is exited, it accepts a new a JNET::JControlHost connection.
     *
     * \param  port                 port number
     */
    void run(const int port)
    {
      JControlHostServer server(port);

      for ( ; ; ) {

        JControlHost* ps = server.AcceptClient();
        
        ps->Connected();

        JNET::JPrefix prefix;

        ps->WaitHead(prefix);

        const int length = prefix.getSize();

        char* buffer = new char[length];
        
        ps->GetFullData(buffer, length);        
        ps->PutFullData(prefix.toString(), buffer, length);

        delete ps;

        enter(JArgs(std::string(buffer, length)));

        delete [] buffer;
        
        run();

        exit();
      }
    }


    /**
     * Run client with commands from input stream (e.g for debugging).
     *
     * Example input format:
     * <pre>
     * <tag> <event name>[\#data];
     * <tag> <event name>[\#data];
     * </pre>
     *
     * \param  in                   input stream
     */
    void run(std::istream& in)
    {
      using std::string;

      string tag;
      string buffer;

      while (in >> tag && getline(in, buffer, ';')) {
        update(tag, buffer.length(), buffer.data());
      }
    }


    static const int TIMEOUT_S = 1;        //!< time out of update [s]

  protected:

    JSharedPointer<JControlHost> server;   //!< message server
    JMessageLogger               logger;   //!< message logger

  private:
    /**
     * Update state machine.
     * This method waits for a message from JNET::JControlHost server.
     */
    void update() 
    {
      JNET::JPrefix prefix;
      
      server->WaitHead(prefix);
      
      const int length = prefix.getSize();
      
      char* buffer = new char[length];
      
      server->GetFullData(buffer, length);
      
      if (!filter(prefix.toString(), length, buffer)) {
        update(prefix.toString(), length, buffer);
      }

      delete [] buffer;
    }


    /**
     * Update state machine.
     *
     * \param  tag                  tag
     * \param  length               number of characters
     * \param  buffer               message
     */
    void update(const JNET::JTag& tag, int length, const char* buffer)
    {
      using namespace std;
      using namespace JLANG;

      string::size_type pos = 0;

      while (pos != (string::size_type) length && TOKEN_DELIMETER.find(*(buffer + pos)) == string::npos) {
        ++pos;
      }

      JEvent_t event = JEvent_t::toValue(string(buffer, pos));
      
      JString event_name(event.getName());

      event_number = event.getNumber();
      event_name   = event_name.trim();     // remove white spaces, if any

      while (pos != (string::size_type) length && TOKEN_DELIMETER.find(*(buffer + pos)) != string::npos) {
        ++pos;
      }


      JEventTable::const_iterator i = eventTable.find(tag, event_name);
      
      if (i != eventTable.end()) {

        // redirect all I/O

        JErrorStream    error(logger);

        JRedirectStream redirect[] = { JRedirectStream(cin,  JLANG::null),
                                       JRedirectStream(cout, JLANG::null),
                                       JRedirectStream(cerr, error) };

        if (redirect[0] &&
            redirect[1] &&
            redirect[2]) {

          (*(i->second))(length - pos, buffer + pos);
        }
        
      } else {

        JErrorStream(logger) << "Unknown key <" << JEventTable::getKey(tag,event_name) << ">";
      }
    }


    /**
     * Configure client.
     * This method is used to setup the event table. 
     */
    void configure()
    {
      using namespace std;
      using namespace JNET;

      hostname   = JSYSTEM::getHostname();

      full_name  = KM3NETDAQ::getFullName (hostname, getName());
      unique_tag = KM3NETDAQ::getUniqueTag(hostname, getName());
      
      for (JTag buffer[] = { RC_CMD, unique_tag, DISPTAG_UNDEFINED }, *tag = buffer; *tag != DISPTAG_UNDEFINED; ++tag) {

        eventTable.insert(*tag, ev_init);
        eventTable.insert(*tag, ev_configure);
        eventTable.insert(*tag, ev_start);
        eventTable.insert(*tag, ev_pause);
        eventTable.insert(*tag, ev_continue);
        eventTable.insert(*tag, ev_stop);
        eventTable.insert(*tag, ev_reset);
        eventTable.insert(*tag, ev_quit);
        eventTable.insert(*tag, ev_off);
        
        eventTable.insert(*tag, ev_check);
        eventTable.insert(*tag, ev_input);

        eventTable.insert(*tag, ev_recover);
      }

      JControlHost::Throw(true);

      setClockInterval(TIMEOUT_S * 1000000LL);
    }


    /**
     * Action when entering state.
     * This method provides for the hand shaking with the run control program.
     *
     * \param  state                entered state
     * \param  event                event that triggered transition
     */
    void enterState(const CHSM::state& state, 
                    const CHSM::event& event)
    {
      std::ostringstream os;
      
      os << getFullName()
         << getTokenDelimeter() 
         << JEvent_t(event.name(), event_number)
         << getTokenDelimeter() 
         << getStateName(state.name());

      server->PutFullString(RC_REPLY, os.str());
    }


    /**
     * This method is called at <tt>ev_check</tt> and reports a system check by mimicing an enter state action.
     *
     * \param  length               number of characters
     * \param  buffer               message
     */
    virtual void actionCheck(int length, const char* buffer)
    {
      if (Main.RunControl.Error.active()) {

        enterState(Main.RunControl.Error, ev_check);

      } else {
        
        for (CHSM::parent::iterator state = Main.RunControl.Operational.begin(); state != Main.RunControl.Operational.end(); ++state) {

          if (state->active()) {

            // mimic enter state

            enterState(*state, ev_check);
          }
        }
      }
    }


    /**
     * The method to execute the action.
     *
     * \param  __action             pointer to action method
     * \param  __event              event that triggered the action
     */
    void execute(action __action, const CHSM::event& __event)
    {
      try {

        const JDAQStateMachine::ev_daq_event& event = dynamic_cast<const JDAQStateMachine::ev_daq_event&>(__event);

        (this->*__action)(event->length, event->buffer);
      }
      catch(const std::exception& error) {
        logger.error(error.what());
        logger.error("Trigger ev_error.");
        ev_error();
      }
    }


    JEventTable          eventTable;      //!< event table
    std::string          hostname;
    std::string          full_name;
    JNET::JTag           unique_tag;
    JTimekeeper          clock;           //!< central clock
    JSelectReader        select;          //!< select call
    JParser<>            parser;          //!< parser method enter()
    int                  event_number;    //!< number of last event
  };
}

#endif