JDQSimulator.cc

Go to the documentation of this file.

#include <string>
#include <iostream>
#include <iomanip>
#include <vector>
#include <limits>
#include <map>

#include "TRandom3.h"
#include "JMath/JRandom.hh"

#include "Jeep/JParser.hh"
#include "Jeep/JProperties.hh"
#include "Jeep/JTimer.hh"
#include "Jeep/JTimekeeper.hh"
#include "km3net-dataformat/online/JDAQ.hh"
#include "JDAQ/JDAQHeaderIO.hh"
#include "JDAQ/JDAQTimesliceIO.hh"
#include "JSupport/JMultipleFileScanner.hh"
#include "JSupport/JSupport.hh"
#include "JLang/JException.hh"
#include "JLang/JRedirectStream.hh"
#include "JRuncontrol/JDAQClient.hh"
#include "JIO/JByteArrayIO.hh"
#include "JNet/JSocket.hh"
#include "JNet/JHostname.hh"
#include "JNet/JSocketBlocking.hh"
#include "JTools/JRange.hh"


namespace KM3NETDAQ {


  using namespace JPP;


  /**
   * Data structure for configuration of JDQSimulator.
   */
  class JSource :
    public std::string
  {
  public:
    /**
     * Default constructor.
     */
    JSource() :
      std::string()
    {}


    /**
     * Read JSource from input stream.
     *
     * \param  in             input stream
     * \param  source         JSource
     * \return                input stream
     */
    friend inline std::istream& operator>>(std::istream& in, JSource& source)
    {
      int index;

      in >> index >> static_cast<std::string&>(source);

      return in;
    }
        

    /**
     * Write JSource to output stream.
     *
     * \param  out            output stream
     * \param  source         JSource
     * \return                output stream
     */
    friend inline std::ostream& operator<<(std::ostream& out, const JSource& source)
    {
      out << static_cast<const std::string&>(source);

      return out;
    }
  };
  

  /**
   * Data structure for configuration of JDataFilter.
   */
  class JTarget :
    public JSocketBlocking
  {
  public:
    /**
     * Default constructor.
     */
    JTarget() :
      JSocketBlocking()
    {}


    /**
     * Read JTarget from input stream.
     *
     * \param  in             input stream
     * \param  target         JTarget
     * \return                input stream
     */
    friend inline std::istream& operator>>(std::istream& in, JTarget& target)
    {
      using namespace std;
      using namespace JPP;

      int       index;
      JHostname hostname;

      if (in >> index >> hostname) {

        try {

          target.connect(hostname.hostname, hostname.port);
          
          target.setTcpNoDelay  (true);
          target.setReuseAddress(true);
          target.setKeepAlive   (true);
          target.setReceiveBufferSize(1024);
          target.setSendBufferSize   (1024*1024);
          target.setNonBlocking (false);
        }
        catch(const JException& error) {
          cout << error << endl;
          target.close();
        }
      }

      return in;
    }


    /**
     * Write JTarget to output stream.
     *
     * \param  out            output stream
     * \param  target         JTarget
     * \return                output stream
     */
    friend inline std::ostream& operator<<(std::ostream& out, const JTarget& target)
    {
      using namespace std;

      out << "TCP no-delay    " << target.getTcpNoDelay()        << endl;
      out << "Reuse address   " << target.getReuseAddress()      << endl;
      out << "Keep alive      " << target.getKeepAlive()         << endl;
      out << "Receive buffer  " << target.getReceiveBufferSize() << endl;
      out << "Send    buffer  " << target.getSendBufferSize()    << endl;
      out << "Non blocking    " << target.getNonBlocking()       << endl;

      return out;
    }
  };


  typedef JRange<int>                   range_type;     // frame index range
  typedef std::map<int, range_type>     map_type;       // module inactivity

  /**
   * Runcontrol client to simulate data queue.
   * In state running, this application will send raw data to the data filters
   * in a round robin way, based on the frame index.
   */
  class JDQSimulator :
    public JDAQClient
  {
  public:
    /**
     * Constructor.
     *
     * \param  name                 name of client
     * \param  server               name of command message server
     * \param  logger               pointer to logger
     * \param  level                debug level
     */
    JDQSimulator(const std::string& name,
                 const std::string& server,
                 JLogger*           logger,
                 const int          level) :
      JDAQClient(name, server, logger, level)
    {
      replaceEvent(RC_CMD, RC_DQSIMULATOR, ev_configure);
    }


    virtual void actionConfigure(int length, const char* buffer) override
    {
      using namespace std;
      using namespace KM3NETDAQ;
      using namespace JPP;


      JMultipleFileScanner<JDAQTimesliceL0> inputFile;
      Long64_t        numberOfEvents = 1;
      int             numberOfFrames = numeric_limits<int>::max();
      double          P              = 0.0;

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

      properties["source"]         = source;
      properties["target"]         = target;
      properties["snooze"]         = snooze;
      properties["inputFile"]      = inputFile;
      properties["numberOfEvents"] = numberOfEvents;
      properties["numberOfFrames"] = numberOfFrames;
      properties["probability"]    = P;
      
      properties.read(string(buffer, length));


      for (vector<JTarget>::iterator i = target.begin(); i != target.end(); ) {
        if (i->is_open())
          ++i;
        else
          i = target.erase(i);
      }

      if (inputFile.empty()) { JErrorStream(logger) << "No input files"; }
      if (target   .empty()) { JErrorStream(logger) << "No targets"; }

      if (!snooze.empty()) {

        ostringstream os;

        os << "snooze";

        for (map_type::const_iterator i = snooze.begin(); i != snooze.end(); ++i) {
          os << ' ' << setw(8) << i->first << " [" << FILL(6,'0') << i->second.getLowerLimit() << "," << FILL(6,'0') << i->second.getUpperLimit() << "]";
        }

        JNoticeStream(logger) << os.str();
      }

      const unsigned int index = distance(source.begin(), find(source.begin(), source.end(), getName()));

      int number_of_hits   = 0;
      int number_of_errors = 0;

      if (index < source.size()) {
      
        while (inputFile.hasNext()) {

          JDAQTimeslice* timeslice = inputFile.next();

          int i1 = (timeslice->size() * (index + 0)) / source.size();
          int i2 = (timeslice->size() * (index + 1)) / source.size();

          if (i2 - i1 > numberOfFrames) {
            i2 = i1 + numberOfFrames;
          }

          for (int i = i1; i != i2; ++i) {

            JDAQSuperFrame& frame = timeslice->at(i);

            for (JDAQSuperFrame::iterator hit = frame.begin(); hit != frame.end(); ++hit) {

              ++number_of_hits;

              if (gRandom->Rndm() <= P) {

                *hit = JDAQHit(hit->getPMT(), getRandom<JDAQHit::JTDC_t>(), hit->getToT());

                ++number_of_errors;
              }
            }
          }

          JDebugStream(logger) << "Processing timeslice: " << inputFile.getCounter() << " [" << i1 << "," << i2 << "]";
            
          data.push_back(JTimeslice(i2 - i1));
          
          for (int i = i1; i != i2; ++i) {
            data.rbegin()->at(i - i1) << timeslice->at(i);
          }
        }

        JNoticeStream(logger) << "Number of errors / hits " << number_of_errors << " / " << number_of_hits << " for P = " << P;

      } else {

        JErrorStream(logger) << "Source not found in configuration data: " << getName();
      }

      setClockInterval((long long int) (1e-3 * getFrameTime()));
    }


    virtual void actionReset(int length, const char* buffer) override
    {
      for (std::vector<JTarget>::iterator i = target.begin(); i != target.end(); ++i) {
        i->close();
      }

      target.clear();
      source.clear();
      data  .clear();
    }


    virtual void actionQuit(int length, const char* buffer) override
    {
      actionReset(0, NULL);
    }


    virtual void actionStart(int length, const char* buffer) override
    {
      numberOfSlices = 0;
      numberOfBytes  = 0;

      data.setRunNumber(getRunNumber());
      data.reset();

      timer.reset();

      resetClock();
    }


    virtual void actionStop(int length, const char* buffer) override
    {
      if (timer.usec_wall > 0) { JNoticeStream(logger) << "I/O "         << (int) (numberOfBytes   / timer.usec_wall) << " MB/s"; }
      if (numberOfSlices  > 0) { JNoticeStream(logger) << "Delay/slice " << (int) (getClockDelay() / numberOfSlices)  << " us"; }
    }


    virtual void actionRunning() override
    {
      if (!data.empty() && !target.empty()) {

        timer.start();

        try {
          
          JDAQPreamble         preamble;
          JDAQSuperFrameHeader header;
          
          for (JTimeslice::const_iterator frame = data->begin(); frame != data->end(); ++frame) {
            
            JIO::JByteArrayReader in(frame->data(), getSizeof<JDAQPreamble>() + getSizeof<JDAQSuperFrameHeader>());
              
            in >> preamble;
            in >> header;

            if (snooze.count(header.getModuleID()) == 0 || !snooze[header.getModuleID()](header.getFrameIndex())) {

              JSocketBlocking& socket = target[header.getFrameIndex() % target.size()];
            
              socket.write(frame->data(), frame->size());
            
              numberOfBytes += frame->size();
            }
          }
          
          numberOfSlices += 1;
        }
        catch(const JException& exception) {
          JErrorStream(logger) << exception;
        }
        
        data.next();

        timer.stop();
      }
    }

  private:
    std::vector<JSource>             source;
    std::vector<JTarget>             target;
    map_type                         snooze;


    /**
     * Memory management for sending of raw data.
     */
    typedef JIO::JByteArrayWriter    JFrame;
    typedef std::vector<JFrame>      JTimeslice; 

    class JData : 
      public std::vector<JTimeslice>
    {
    public:
      /**
       * Default constructor.
       */
      JData() :
        std::vector<JTimeslice>()
      {}


      /**
       * Set run number.
       *
       * \param  run_number           run number
       */
      void setRunNumber(int run_number) 
      {
        JDAQSuperFrameHeader header;

        for (iterator timeslice = begin(); timeslice != end(); ++timeslice) {

          for (JTimeslice::iterator frame = timeslice->begin(); frame != timeslice->end(); ++frame) {
            
            JIO::JByteArrayReader in(frame->data(), getSizeof<JDAQPreamble>() + getSizeof<JDAQSuperFrameHeader>());
            
            in.seekg(getSizeof<JDAQPreamble>());
            
            in >> header;
            
            header.setRunNumber(run_number);
            
            frame->seekp(getSizeof<JDAQPreamble>());
              
            *frame << header;
          }
        }
      }


      /**
       * Reset internal iterator to begin.
       */
      void reset() 
      {
        page = begin();
      }


      /**
       * Increment internal iterator.
       * When the internal iterator reaches the end of the data,
       * the frame indices of the data are increased and 
       * the internal iterator is reset to the begin of data.
       */
      void next() 
      {
        if (page != end() && ++page == end()) {

          JDAQSuperFrameHeader header;
        
          for (iterator timeslice = begin(); timeslice != end(); ++timeslice) {

            for (JTimeslice::iterator frame = timeslice->begin(); frame != timeslice->end(); ++frame) {
            
              JIO::JByteArrayReader in(frame->data(), getSizeof<JDAQPreamble>() + getSizeof<JDAQSuperFrameHeader>());
              
              in.seekg(getSizeof<JDAQPreamble>());

              in >> header;
              
              header.setFrameIndex(header.getFrameIndex() + this->size());
              header.setTimesliceStart(JDAQUTCExtended(getTimeOfFrame(header.getFrameIndex())));
            
              frame->seekp(getSizeof<JDAQPreamble>());
              
              *frame << header;
            }
          }

          reset();
        }
      }


      /**
       * Smart pointer operator.
       *
       * \return          current iterator
       */
      const_iterator operator->()
      { 
        return page; 
      }


    private:
      JData::const_iterator  page;
    };


    JData data;

    long long int numberOfSlices;    // total number of timeslices
    long long int numberOfBytes;     // total number of bytes

    JEEP::JTimer timer;
  };
}


/**
 * \file
 *
 * Program for real-time simulation of data queue.
 * \author rbruijn
 */
int main(int argc, char* argv[])
{
  using namespace std;

  string  server;
  string  logger;
  string  client_name;
  bool    use_cout;
  int     debug;

  try {

    JParser<> zap("Program for real-time simulation of data queue.");

    zap['H'] = make_field(server)          = "localhost";
    zap['M'] = make_field(logger)          = "localhost";
    zap['u'] = make_field(client_name)     = "JDQSimulator";
    zap['c'] = make_field(use_cout);
    zap['d'] = make_field(debug)           = 3;

    zap(argc, argv);
  }
  catch(const exception &error) {
    FATAL(error.what() << endl);
  }

  
  using namespace KM3NETDAQ;
  using namespace JPP;

  JLogger* out = NULL;

  if (use_cout)
    out = new JStreamLogger(cout);
  else
    out = new JControlHostLogger(logger);

  JDQSimulator simbad(client_name, server, out, debug);

  simbad.enter();
  simbad.run();
}