JDAQDriver.hh

Go to the documentation of this file.

#ifndef __JRUNCONTROL__JDAQDRIVER__
#define __JRUNCONTROL__JDAQDRIVER__

#include <string>
#include <iostream>
#include <sstream>
#include <limits>
#include <unistd.h>

#include "JLang/JException.hh"
#include "JLang/JRedirectStream.hh"
#include "JLang/JWhiteSpacesFacet.hh"

#include "JRuncontrol/JDAQClient.hh"
#include "JRuncontrol/JClient.hh"
#include "JRuncontrol/JClientList.hh"


/**
 * \author mdejong
 */

namespace KM3NETDAQ {

  using JLANG::JException;
  using JLANG::JIOException;
  using JLANG::JControlHostException;


  /**
   * Simple driver for run control clients.
   * This class can be used to start a set of run control clients,
   * trigger events and eventually stop the clients.
   */
  class JDAQDriver :
    public JDAQClient
  {
  public:
    /**
     * Constructor.
     *
     * \param  name                 name of driver
     * \param  server               name of command message server
     * \param  logger               logger
     * \param  level                debug level 
     * \param  timeout_s            timeout_s [s]
     */
    JDAQDriver(const std::string& name,
               const std::string& server,
               JLogger*           logger,
               const int          level,
               const int          timeout_s) :
      JDAQClient(name, server, logger, level),
      timeout_s (timeout_s),
      is_alive  (false)
    {}


    /**
     * Enter the state machine.
     * The driver will subscribe to the ControlHost tags corresponding to born, died and
     * reply messages of the clients instead of the standard tags for run control commands.
     * The clients are started after the driver is ready to receive ControlHost messages.
     * In case of an error, a message is printed on the terminal and the state machine
     * is not entered.
     */
    virtual bool enter() override
    {
      using namespace std;
      using namespace JLANG;
      using namespace JNET;

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

        try {
          
          server->Subscribe(JSubscriptionAll(RC_REPLY)     + 
                            JSubscriptionAll(RC_LOG)       + 
                            JSubscriptionAll(DISPTAG_Born) + 
                            JSubscriptionAll(DISPTAG_Died));
          server->SendMeAlways();
          server->MyId(getFullName());

          // check alive of this driver

          for (int i = 0; i != timeout_s && !is_alive; ) {
            if (!update(true)) {
              sleep(1);
              ++i;
            }
          }
          
          if (is_alive) {

            {
              clientList.start();
              
              for (int i = 0; i != timeout_s; ++i) {

                while (update(true)) {}

                if (clientList.count() >= clientList.count(JClient::ACTIVE))
                  break;
                else
                  sleep(1);
              }
            }

            if (clientList.count() != clientList.count(JClient::ACTIVE)) {
              for (JClientList::const_iterator i = clientList.begin(); i != clientList.end(); ++i) {
                if (i->isActive() && i->getBorn() <= i->getDied()) {
                  JWarningStream(logger) << "client " << i->getFullName() << " did not start.";
                }
              }
            }

            return CHSM::machine::enter();

          } else {
            cerr << "Timeout at subscription." << endl;
          }
        }
        catch(const JControlHostException& error) {
          cerr << error << endl;
        }

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

      return false;
    }


    /**
     * Exit the state machine.
     *
     * This method waits for the clients to terminate using the died message generated by ControlHost.
     * In case of a timeout, the process is terminated by calling the method stop() of the corresponding client.
     */
    void actionExit()
    {
      using namespace std;
      using namespace JLANG;

      for (int i = 0; i != timeout_s; ++i) {

        while (update(true)) {}

        if (clientList.count() == 0)
          break;
        else
          sleep(1);
      }

      if (clientList.count() != 0) {
        JWarningStream(logger) << "Timeout at transition " << "exit(); forced stop.";
      }

      clientList.stop();
    }


    /**
     * Action when entering state.
     * This method waits for all clients to produce the enter state message.
     * In case of a timeout, no specific action is taken but an error message is logged.
     *
     * \param  state                entered state
     * \param  event                event that triggered transition
     */
    virtual void enterState(const CHSM::state& state, const CHSM::event& event) override
    {
      for (int i = 0; i != timeout_s && clientList.count(state) < clientList.count(JClient::ACTIVE) && clientList.count() != 0; ) {
        if (!update(true)) {
          sleep(1);
          ++i;
        }
      }

      if (clientList.count(state) < clientList.count(JClient::ACTIVE)) {
        JWarningStream(logger) << "Timeout at transition " << event.name() << " to state " << state.name(); 
      }
    }


    /**
     * Update client list with incoming ControlHost message.
     * This method receives and processes a message.
     * The client list is updated accordingly.
     * If the no-wait option is set to true, it returns in the absence of a pending message immediately.
     * The return value is then false.
     * If the no-wait option is set to false, it waits until the next message is received.
     *
     * \param  no_wait              wait option
     * \return                      true if message received; else false
     */
    bool update(const bool no_wait)
    {
      using namespace std;
      using namespace JNET;
      using JLANG::JWhiteSpacesFacet;

      try {

        string        tag;
        long long int length = 0;

        if ( no_wait && server->CheckHead(tag, length) <= 0) { return false; }
        if (!no_wait && server->WaitHead (tag, length) <  0) { return false; }
        
        char* data= new char[length];
        
        server->GetFullData(data, length);
        
        const string buffer(data, length);

        delete [] data;
        
        JDebugStream(logger) << "Got message " << tag << ' ' << buffer;
        
        if (tag == RC_LOG) {

          rc_log = buffer;

        } else if (buffer.find(getFullName()) != string::npos) {
          
          if      (tag == DISPTAG_Born)
            is_alive  = true;
          else if (tag == DISPTAG_Died)
            is_alive  = false;
          
        } else {

          JClientList::iterator i = clientList.find(buffer);
          
          if (i != clientList.end()) {

            i->update(tag, buffer);

          } else {

            JErrorStream(logger) << "Message fom illegal client " << buffer;

            try {

              if (tag == DISPTAG_Born ||
                  tag == DISPTAG_Died ||
                  tag == RC_REPLY) {

                string key, hostname, name;
                
                istringstream is(buffer);
      
                const locale loc(is.getloc(), new JWhiteSpacesFacet(is.getloc(), TOKEN_DELIMETER));
      
                is.imbue(loc);
                
                if (is >> key >> hostname >> name && key == RUN_CONTROL_CLIENT) {

                  JClient client(name, hostname);

                  client.update(tag, buffer);
                  client.setMode(JClient::ILLEGAL);
            
                  clientList.insert(client);
                  
                  JWarningStream(logger) << "Added illegal client " << client.getFullName();

                } else {
                  throw JIOException("JClient: Error reading " + buffer);
                }
              }
            }
            catch(const JException& error) {
              JErrorStream(logger) << error;
            }
          }
        }

        return true;
      }
      catch(const JControlHostException& error) {
        JErrorStream(logger) << error;
      }

      return false;
    }

    virtual void actionStart(int, const char*) override
    {
      rc_log = "";
    }

    virtual void actionStop(int, const char*) override
    {
      if (rc_log != "") 
        JNoticeStream(logger) << rc_log;
      else
        JErrorStream (logger) << "Missing message from JDataWriter with tag " << RC_LOG;
    }

    /**
     * Run driver with user input.
     */
    void run()
    {
      run(std::cin);
    }


    /**
     * Run driver.
     *
     * Example input format:
     * <pre>
     * # comment line.
     *
     * process <process name> <host name> [<start command>];
     *
     * # The following tokens in <start command> will be substituted: 
     * #   $HOST$    by  <host name>;
     * #   $NAME$    by  <process name>;
     * #   $SERVER$  by  JClient::SERVER;
     * #   $LOGGER$  by  JClient::LOGGER;
     * #   $ARGS$    by  part following '/' in <process name>;
     *
     * # enter state machine.
     *
     * enter
     *
     * # trigger event
     * # data can be provided online and mixed with data from a separate file (optional).
     * # multiple tags should be separated by a new line.
     *
     * event <event name> {
     *   [<tag 1> [data]]
     *   [<tag 2> [data][%<file name>%][data]]
     * }
     *
     * # optionally quit before end of input
     * [quit]
     *
     * # optionally kill processes that did not properly terminate.
     * [exit]
     * </pre>
     *
     * \param  in                   input stream
     */
    void run(std::istream& in)
    {
      using namespace std;
      
      for (string key; in >> key; ) {

        if (key[0] == '#') {

          in.ignore(numeric_limits<streamsize>::max(), '\n');

        } else if (key == "enter") {
          
          enter();
          
          if (!active()) {
            cerr << "State machine not entered; abort." << endl;
            return;
          }

        } else if (key == "exit") {
          
          timeout_s = 0;
          exit();
          
        } else if (key == "quit") {
          
          break;
          
        } else if (key == "sleep") {

          int sec;

          if (in >> sec) {
            sleep(sec);
          }
          
        } else if (key == "process") {

          string buffer;

          getline(in, buffer, ';');
          
          istringstream is(buffer);
          
          JClient client;
          
          if (is >> client) {

            client.setMode(JClient::ACTIVE);

            if (!clientList.insert(client).second) {
              JWarningStream(logger) << "Process already exists " << client;
            }

          } else {
            JErrorStream(logger) << "Error reading key word process.";
          }

        } else if (key == "event" || key == "event*") {

          JEvent_t   event;
          char       c;
          string     buffer;
          const char eol = '\n';
          
          if (in >> event >> c && c == '{' && getline(in, buffer, '}')) {

            if (clientList.count() != 0) {

              JDAQEvent_t* pev = findEvent(RC_CMD, event.getName());
            
              if (pev != NULL) {

                if (pev->active() || key == "event*") {

                  istringstream is(buffer);
                
                  for (string tag; is >> tag; ) {
                  
                    ostringstream os;

                    os << event << getTokenDelimeter();

                    copy(is, os, eol);

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

                  if (key != "event*") {
                    (*pev)(0, NULL);   // trigger driver
                  }

                } else {
                  JErrorStream(logger) << "Inactive event " << event;
                }

              } else {
                JErrorStream(logger) << "Unknown event " << event;
              }

            } else {
              JErrorStream(logger) << "No active client to trigger event.";
            }

          } else {
            JErrorStream(logger) << "Error reading key word event.";
          }

        } else if (key == "message") {

          string tag;
          string buffer;

          if (in >> tag && getline(in, buffer, ';'))
            server->PutFullString(tag, buffer);
          else
            JErrorStream(logger) << "Invalid message: <" << tag << "> \"" << buffer << "\"";
          
        } else if (key == "print") {

          for (JClientList::const_iterator i = clientList.begin(); i != clientList.end(); ++i) {
            JNoticeStream(logger) << i->getFullName()     << ' '
                                  << i->getStartCommand() << ' '
                                  << i->getAlive()        << ' '
                                  << i->getStatename();
          }

        } else if (key == "filter") {

          string client;
          string buffer;

          getline(in, buffer, ';');

          for (istringstream is(buffer); is >> client; ) {
            filter(client);
          }

        } else if (key == "sync") {

          synchronise();
          
        } else {

          JErrorStream(logger) << "Unknown key: " << key;

          in.ignore(numeric_limits<streamsize>::max(), '\n');
        }
      }
    }


    /**
     * Update client list with incoming ControlHost messages until the client list 
     * is synchronised with the current state or until the timeout.
     */
    void update()
    {
      using namespace std;

      const CHSM::parent& parent = static_cast<const CHSM::parent&>(Main.RunControl);

      for (CHSM::parent::const_iterator state = parent.begin(); state != parent.end(); ++state) {

        if (state->active()) {
          for (int i = 0; i != timeout_s && clientList.count(*state) < clientList.count(JClient::ACTIVE); ) {
            if (!update(true)) {
              sleep(1);
              ++i;
            }
          }
        }
      }
    }


    /**
     * Synchronise clients.
     */
    void synchronise()
    {
      using namespace std;

      const CHSM::parent& parent = static_cast<const CHSM::parent&>(Main.RunControl);

      for (CHSM::parent::const_iterator state = parent.begin(); state != parent.end(); ++state) {

        if (state->active()) {

          if (clientList.count(*state) < clientList.count(JClient::ACTIVE)) {

            JDebugStream(logger) << "Synchronising " << state->name();
            
            for (JClientList::iterator i = clientList.begin(); i != clientList.end(); ++i) {

              if (i->getMode() == JClient::ACTIVE) {
                
                if (!i->getAlive()) {
              
                  try {
                
                    string buffer;
                
                    if (JControlHost::WhereIs(JClient::SERVER, i->getFullName(), buffer) > 0) {

                      i->setAlive(true);

                      if (buffer.find(i->getHostname()) == string::npos) {
                        JErrorStream(logger) << i->getFullName() << " running on " << buffer << " but not alive.";
                      }
                    }
                  }
                  catch(const JControlHostException& error) {
                    JErrorStream(logger) << error;
                  }
                }

                if (i->getAlive() && i->getStatename() != state->name()) {
                  server->PutFullString(KM3NETDAQ::getUniqueTag(i->getHostname(), i->getName()), ev_check.name()); 
                }
              }
            }
            
            
            for (int i = 0; i != timeout_s && clientList.count(*state) < clientList.count(JClient::ACTIVE); ) {
              if (!update(true)) {
                sleep(1);
                ++i;
              }
            }
            
            if (clientList.count(*state) < clientList.count(JClient::ACTIVE)) {
              JWarningStream(logger) << "Timeout at synchronisation.";
            }
          }
        }
      }
    }


    /**
     * Filter client list by putting failing clients to sleep.
     * In this, only clients with names that contain the given character sequence are considered.
     *
     * \param  target               target name of client(s)
     */
    void filter(const std::string& target = "")
    {
      const CHSM::parent& parent = static_cast<const CHSM::parent&>(Main.RunControl);

      for (CHSM::parent::const_iterator state = parent.begin(); state != parent.end(); ++state) {

        if (state->active()) {

          for (JClientList::iterator i = clientList.begin(); i != clientList.end(); ++i) {

            if (target == "" || i->getName().find(target) != std::string::npos) {

              if (!i->getAlive() || i->getStatename() != state->name()) {

                JNoticeStream(logger) << "Put to sleep " << i->getFullName();

                i->setMode(JClient::SLEEP);
              }
            }
          }
        }
      }
    }
    
    int         timeout_s;      //!< timeout of state transitions [s]

  protected:

    JClientList clientList;
    bool        is_alive;
    std::string rc_log;

    /**
     * Copy data from input to output stream.
     * Tagged file names are recursively expanded.
     *
     * \param  in                   input  stream
     * \param  out                  output stream
     * \param  eol                  end of line
     */
    static void copy(std::istream& in, std::ostream& out, const char eol = '\n') 
    {
      using namespace std;

      string buffer; 

      if (getline(in, buffer, eol)) {

        for (string::size_type pos = 0; pos < buffer.length(); ) {
          
          string::size_type lpos = buffer.substr(pos).find(FILENAME_PREFIX);
          string::size_type rpos = buffer.substr(pos).find(FILENAME_POSTFIX);
          
          if (lpos != string::npos && 
              rpos != string::npos) {
            
            out << buffer.substr(pos, lpos);
            
            lpos += FILENAME_PREFIX.length();
            pos  += lpos;
            
            ifstream file(buffer.substr(pos, rpos - lpos).c_str());
            
            copy(file, out, '\0');
            
            rpos += FILENAME_POSTFIX.length();
            pos  += rpos - lpos;
            
          } else {
            
            out << buffer.substr(pos);
            
            pos  += buffer.substr(pos).length();
          }
        }
      }
    }
  };
}

#endif