JDataFilter.cc

Go to the documentation of this file.

#include <string>
#include <iostream>
#include <iomanip>
#include <deque>
#include <limits>
#include <algorithm>
#include <unistd.h>

#include "JLang/JSinglePointer.hh"
#include "JLang/JLangToolkit.hh"
#include "Jeep/JParser.hh"
#include "Jeep/JProperties.hh"
#include "Jeep/JTimer.hh"
#include "Jeep/JTimekeeper.hh"
#include "Jeep/JMessage.hh"
#include "Jeep/JPrint.hh"
#include "Jeep/JeepToolkit.hh"
#include "JLogger/JMessageScheduler.hh"
#include "km3net-dataformat/online/JDAQ.hh"
#include "km3net-dataformat/online/JDAQAbstractPreamble.hh"
#include "km3net-dataformat/online/JDAQPreamble.hh"
#include "km3net-dataformat/online/JDAQSuperFrame.hh"
#include "km3net-dataformat/online/JDAQFrame.hh"
#include "JDAQ/JDAQTags.hh"
#include "JDAQ/JDAQEventIO.hh"
#include "JDAQ/JDAQTimesliceIO.hh"
#include "JDAQ/JDAQSummarysliceIO.hh"
#include "JDetector/JDetector.hh"
#include "JTrigger/JHit.hh"
#include "JTrigger/JHitToolkit.hh"
#include "JTrigger/JSuperFrame1D.hh"
#include "JTrigger/JSuperFrame2D.hh"
#include "JTrigger/JTimeslice.hh"
#include "JTrigger/JHitL0.hh"
#include "JTrigger/JHitL1.hh"
#include "JTrigger/JBuildL1.hh"
#include "JTrigger/JBuildL2.hh"
#include "JTrigger/JTrigger3DShower.hh"
#include "JTrigger/JTriggerMXShower.hh"
#include "JTrigger/JTrigger3DMuon.hh"
#include "JTrigger/JTriggerNB.hh"
#include "JTrigger/JTriggerBits.hh"
#include "JTrigger/JEventOverlap.hh"
#include "JTrigger/JTimesliceRouter.hh"
#include "JTrigger/JTriggeredEvent.hh"
#include "JTrigger/JTimesliceL1.hh"
#include "JTrigger/JTriggerParameters.hh"
#include "JTrigger/JTriggerToolkit.hh"
#include "JTrigger/JChecksum.hh"
#include "JRuncontrol/JDAQClient.hh"
#include "JRuncontrol/JRuncontrolToolkit.hh"
#include "JNet/JControlHost.hh"
#include "JNet/JControlHostObjectOutput.hh"
#include "JNet/JSocket.hh"
#include "JNet/JSocketChannel.hh"
#include "JNet/JServerSocket.hh"
#include "JPhysics/JConstants.hh"
#include "JTools/JQuantile.hh"
#include "JSupport/JSupport.hh"
#include "JSupport/JTreeRecorder.hh"
#include "JSupport/JMeta.hh"
#include "JSystem/JTime.hh"
#include "JSystem/JSystemToolkit.hh"
#include "JSystem/JNetwork.hh"


namespace JNET {

  /**
   * Get size of packeet.
   *
   * \param  preamble             DAQ data preamble
   * \return                      size [B]
   */
  template<>
  inline int getSizeOfPacket(const KM3NETDAQ::JDAQAbstractPreamble& preamble)
  { 
    return preamble.getLength();
  }
}

namespace KM3NETDAQ {

  using namespace JPP;

  /**
   * Main class for real-time filtering of data.
   * 
   * This class implements the action methods for each transition of the state machine.\n
   * When the state machine is entered, data are continually collected using 
   * custom implementation of virtual methods
   * setSelect and 
   * actionSelect.
   *
   * In state <tt>Running</tt>, all incoming data from the DataQueue.cc are buffered, referred to as "queue".\n
   * These data are formatted as a JDAQSuperFrame and aggregated into a JDAQTimeslice.\n
   * When a JDAQTimeslice is complete, 
   * it processed in the same way as applications JTriggerProcessor.cc and JTriggerEfficiency.cc.\n
   * The completeness of a time slice is defined by the parameter <tt>frames_per_slice</tt>
   * which, together with other parameters, is parsed in method actionConfigure.\n
   * <pre>
   *   numberOfFramesPerSlice = frames_per_slice = 1;
   * </pre>
   * Note that this parameter can change during operation (see below).
   *
   * A timeout may occur when the total amount of data or the number of incomplete time slices 
   * in the queue exceeds one of the corresponding pre-set limits:\n
   * <pre>
   *   queueSize  = <maximal queue size [B]>;
   *   queueDepth = <maximal queue depth>;
   * </pre>
   * Note that these values apply per JDataFilter.\n
   *
   * When a timeout occurs,
   * the parameter <tt>frames_per_slice</tt> is set to the number of frames in the oldest time slice.\n
   * When data are received with a frame index below the current frame index,
   * the parameter <tt>frames_per_slice</tt> is incremented by one, 
   * up to the original value set at actionConfigure.\n
   *
   * In addition, a circular buffer based on a ROOT TTree for each JDAQTimeslice type can be maintained.\n
   * The following parameters can be used to configure the circular buffer.\n
   * <pre>
   *   path     = <write directory for circular buffer>;
   *   c_sizeL0 = <L0 buffer size>;
   *   c_sizeL1 = <L1 buffer size>;
   *   c_sizeL2 = <L2 buffer size>;
   *   c_sizeSN = <SN buffer size>;
   * </pre>
   *
   * Note that when path is set to a valid directory:
   *   - There will always be a file created which is closed when the state machine is exited;
   *   - The file name is <tt><path>/KM3NeT_YYYY-MM-DD_<tag>[_<index>].root</tt>;
   *   - L0, L1, L2 and SN buffer sizes are per JDataFilter;
   *   - The tag KM3NETDAQ::RC_ALERT defined in JDAQTags.hh has tentatively been reserved to trigger the closure of the file following an external alert;
   *  -  After closure of the file, a new file will be opened if path is set;
   *
   * In this, <tt>YYYY</tt>, <tt>MM</tt> and <tt>DD</tt> are the year, month and day at the time of
   * the configuration or alert and <tt><tag></tt> corresponds to the unique JNET::JTag of the JDataFilter process.\n 
   * The <tt><index></tt> is optional and will only be used to prevent overwriting of an existing file
   * (e.g. when there is more than one alert on a given day).
   *
   * The application JConvert.cc can be used to readily convert multiple input files to a single output file. 
   *
   * The script JDataFilter.sh can be used to test this application.
   */
  class JDataFilter :
    public JDAQClient
  {
  public:

    typedef JSocketInputChannel<JDAQAbstractPreamble>  JSocketInputChannel_t;
    typedef std::vector<JSocketInputChannel_t>         JChannelList_t;

    typedef double        hit_type;
    typedef JSuperFrame1D<hit_type>   JSuperFrame1D_t;
    typedef JSuperFrame2D<hit_type>   JSuperFrame2D_t;
    typedef JTimeslice   <hit_type>   JTimeslice_t;
    typedef JBuildL1     <hit_type>   JBuildL1_t;
    typedef JBuildL2     <hit_type>   JBuildL2_t;

    /**
     * Circular buffer.
     */
    struct JCircularBuffer_t :
      public JTreeRecorder<JDAQTimesliceTypes_t>
    {
      using JTreeRecorder<JDAQTimesliceTypes_t>::open;

      /**
       * Default constructor.
       */
      JCircularBuffer_t()
      {
        disable();
      }
      
      /**
       * Open file.
       *
       * \param  path                 directory
       * \param  tag                  unique process tag
       */
      void open(const std::string& path, const JTag& tag)
      {
        using namespace std;

        const JDateAndTime cal;

        for (int i = 0; !this->is_open() && i != MAXIMUM_FILE_NUMBER; ++i) {

          ostringstream os;

          os << getFullPath(path) 
             << "KM3NeT" 
             << "_"  << cal.getYear() << '-' << FILL(2,'0') << cal.getMonth() << '-' << FILL(2,'0') << cal.getDay()
             << "_"  << tag;

          if (i != 0) {
            os << "_" << i;
          }

          os << ".root";

          try {
            this->open(os.str().c_str());
          }
          catch(const exception& error) {}
        }
      }

      /**
       * Disable writing.
       */
      void disable()
      {
        sizeL0 = 0;
        sizeL1 = 0;
        sizeL2 = 0;
        sizeSN = 0;
      }

      /**
       * Check whether writing of data is enabled.
       *
       * \return                      true if writing enabled; else false
       */
      bool is_enabled() const
      {
        return (sizeL0 > 0 ||
                sizeL1 > 0 ||
                sizeL2 > 0 ||
                sizeSN > 0);
      }

      /**
       * Write circular buffer to output stream.
       *
       * \param  out                  output stream
       * \param  object               circular buffer
       * \return                      output stream
       */
      friend inline std::ostream& operator<<(std::ostream& out, const JCircularBuffer_t& object)
      {
        if (object.is_open())
          out << object.getFile()->GetName();
        else
          out << "void";

        out << ' ';
        
        return out << object.sizeL0 << '/'
                   << object.sizeL1 << '/'
                   << object.sizeL2 << '/'
                   << object.sizeSN << '/';
      }
      
      Long64_t sizeL0;  //!< Number of L0 time slices
      Long64_t sizeL1;  //!< Number of L1 time slices
      Long64_t sizeL2;  //!< Number of L2 time slices
      Long64_t sizeSN;  //!< Number of SN time slices
    };


    /**
     * Sort DAQ process by index.
     *
     * \param  first                first  DAQ process
     * \param  second               second DAQ process
     * \return                      true if index of first DAQ process less than that of second; else false
     */
    static inline bool processIndexSorter(const JDAQProcess& first, const JDAQProcess& second)
    {
      return first.index < second.index;
    }


    /**
     * Constructor.
     *
     * \param  name                 name of client
     * \param  server               name of command message server
     * \param  hostname             name of data server
     * \param  logger               pointer to logger
     * \param  level                debug level
     * \param  port                 server port
     * \param  backlog              server backlog
     * \param  buffer_size          server buffer
     * \param  path                 directory for writing circular buffer
     */
    JDataFilter(const std::string& name,
                const std::string& server,
                const std::string& hostname,
                JLogger*           logger,
                const int          level,
                const int          port, 
                const int          backlog,
                const int          buffer_size,
                const std::string& path) :
      JDAQClient (name,server,logger,level),
      hostname   (hostname),
      port       (port),
      backlog    (backlog),
      buffer_size(buffer_size),
      path       (path)
    {
      replaceEvent(RC_CMD, RC_DFILTER, ev_configure); /// replace tag to receive dfilter specific messages in configure

      addSubscription(JSubscriptionAll(RC_ALERT));
      
      totalCPURAM          = getRAM();   
      current_slice_index  = -1;
      reporting            = false;
    }


    virtual void actionEnter() override
    {}


    virtual void actionExit() override
    { 
      if (c_buffer.is_open()) {

        JNoticeStream(logger) << "Close circular buffer " << c_buffer;

        c_buffer.close();
      }

      datawriter.reset();
    }


    virtual void actionInit(int length, const char* buffer) override
    { 
      JDebugStream(logger) << "actionInit() " << std::string(buffer,length);

      try {

        JDebugStream(logger) << "Start server.";

        if (serversocket.is_valid()) {
          serversocket->shutdown();
        }

        serversocket.reset(new JServerSocket(port,backlog));
      }   
      catch(const std::exception& error) {
        JErrorStream(logger) << "Error \"" << error.what() << "\"; trigger ev_error.";
        ev_error();
      }
    }


    virtual void actionConfigure(int length, const char* buffer) override
    { 
      using namespace std;
    
      JDebugStream(logger) << "actionConfigure() " << endl << std::string(buffer,length);
  
      long long int update_s = 10;
      long long int logger_s =  5;

      parameters .reset();
      dataFilters.clear();
      dataQueues .clear();

      reporting = false;
      dumpCount = 0;
      dumpLimit = numeric_limits<int>::max();

      detector.clear();
      
      JProperties properties(JEquationParameters("=", ";", "", ""), 0);

      properties["dataWriter"]             = hostname;
      properties["numberOfFramesPerSlice"] = frames_per_slice = 1;
      properties["detector"]               = detector;
      properties["triggerParameters"]      = parameters;
      properties["queueSize"]              = maxQueueSize  = (totalCPURAM - GIGABYTE);  // leave 1 GB left ...
      properties["queueDepth"]             = maxQueueDepth = 20;                        // to be optimized
      properties["frameIndex"]             = maximal_frame_index = 100000;
      properties["logger_s"]               = logger_s;
      properties["update_s"]               = update_s;
      properties["JDataFilter"]            = dataFilters;
      properties["DataQueue"]              = dataQueues;
      properties["path"]                   = path;
      properties["c_sizeL0"]               = c_buffer.sizeL0;
      properties["c_sizeL1"]               = c_buffer.sizeL1;
      properties["c_sizeL2"]               = c_buffer.sizeL2;
      properties["c_sizeSN"]               = c_buffer.sizeSN;
      properties["dumpLimit"]              = dumpLimit;

      try {
        properties.read(string(buffer, length));
      }
      catch(const exception& error) {
        JErrorStream(logger) << error.what();
      }
    
      if (update_s <=  0) { update_s = 1; }
      if (logger_s <=  0) { logger_s = 1; }

      setClockInterval(update_s * 1000000LL);

      hostname = trim(hostname);

      if (hostname != "") 
        datawriter.reset(new JControlHost_t(hostname));
      else
        throw JException("Undefined data writer host name.");
    
      maximum_frames_per_slice = frames_per_slice;
    
      // process processlist
    
      if (dataFilters.empty()) {
        JNoticeStream(logger) << "No DataFilters in process list, or no process list. "
                              << "Assuming that this process is the only process on this CPU and setting parameters accordingly.";
      }
    
      sort(dataFilters.begin(), dataFilters.end(), processIndexSorter);
    
      unsigned int numberOfDataFiltersOnThisMachine = 0;
      vector<JDAQProcess>::iterator thisProcess     = dataFilters.end();

      const vector<string> IP = getListOfIPaddresses();

      {
        JNoticeStream notice(logger);

        notice << "My IP addresses:";

        for (vector<string>::const_iterator i = IP.begin(); i != IP.end(); ++i) {
          notice << ' ' << *i;
        }
      }
      
      for (vector<JDAQProcess>::iterator i = dataFilters.begin(); i != dataFilters.end(); ++i) {

        JDebugStream(logger) << "Test IP address \"" << i->hostname << "\" " << (find(IP.begin(), IP.end(), i->hostname) != IP.end());
        
        if (find(IP.begin(), IP.end(), i->hostname) != IP.end()) {

          numberOfDataFiltersOnThisMachine++;
        
          if (i->port == this->port) {
            thisProcess = i;
          }
        }
      }
    
      if (numberOfDataFiltersOnThisMachine == 0) {
        JNoticeStream(logger) << "Zero data filters on this machine according to process list (if it exists). "
                              << "Assuming one datafilter on this machine.";
        numberOfDataFiltersOnThisMachine = 1;
      }
    
      if (thisProcess == dataFilters.end()) {
        JErrorStream(logger) << "This process cannot be found in the process list. Why do I exist?";    
      }
    
      if (thisProcess != dataFilters.end() && thisProcess->index != getName()) {
        JErrorStream(logger) << "Mismatch between given process names: "
                             << "I am called " << getName()
                             << ", but in the process list I am referred to as " << thisProcess->index;
      }
    
      if (dataFilters.begin() == thisProcess || dataFilters.empty()) {                    // set reporting
        reporting = true;
      }

      if (maxQueueSize > (totalCPURAM-GIGABYTE)/(numberOfDataFiltersOnThisMachine)) {     // check maximum size of queue
        
        maxQueueSize  =  (totalCPURAM-GIGABYTE)/(numberOfDataFiltersOnThisMachine);

        JNoticeStream(logger) << "Maximum queue size is too large given the number of processes on this machine. "
                              << "Queue size reduced to " 
                              << maxQueueSize << " bytes." ;  
      }
    
      // detector

      if (parameters.disableHighRateVeto) {

        JNoticeStream(logger) << "Disabling high-rate veto of all PMTs.";

        detector.setPMTStatus(HIGH_RATE_VETO_DISABLE);
      }

      // trigger parameters
    
      parameters.set(getMaximalDistance(detector));
    
      triggerNB      .reset(new JTriggerNB      (parameters));
      trigger3DMuon  .reset(new JTrigger3DMuon  (parameters));
      trigger3DShower.reset(new JTrigger3DShower(parameters));
      triggerMXShower.reset(new JTriggerMXShower(parameters, detector));
    
      moduleRouter.reset(new JModuleRouter(detector));
   
      if (reporting) {
        JNoticeStream(logger) << "This data filter process will report.";
        JNoticeStream(logger) << "Number of modules: "    << (*moduleRouter)->size();
        JDebugStream (logger) << "Trigger parameters: "   << parameters;
        JDebugStream (logger) << "Detector description: " << endl << detector;
        JNoticeStream(logger) << "Update period [s]: "    << getClockInterval();
      }
    
      timesliceRouter.reset(new JTimesliceRouter(parameters.numberOfBins));

      // set L1, L2 and SN builders
    
      buildL1.reset(new JBuildL1_t(parameters));
      buildL2.reset(new JBuildL2_t(parameters.L2));
      buildSN.reset(new JBuildL2_t(parameters.SN));
      buildNB.reset(new JBuildL2_t(parameters.NB));
   
      if (buildL1.get() == NULL) { JErrorStream(logger) << "Failed to allocate buildL1."; }
      if (buildL2.get() == NULL) { JErrorStream(logger) << "Failed to allocate buildL2."; }
      if (buildSN.get() == NULL) { JErrorStream(logger) << "Failed to allocate buildSN."; }
      if (buildNB.get() == NULL) { JErrorStream(logger) << "Failed to allocate buildSN."; }
     
      logErrorRun        = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));
      logErrorDetector   = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));
      logErrorIndex      = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));
      logErrorIncomplete = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));

      if (c_buffer.is_enabled()) {

        if (!c_buffer.is_open()) {

          c_buffer.open(path, getUniqueTag());

          if (c_buffer.is_open()) {
            
            JNoticeStream(logger) << "Created circular buffer " << c_buffer;

            putObject(c_buffer.getFile(), meta);
            
          } else {
            
            JErrorStream (logger) << "Failed to create circular buffer in directory <" << path << ">; disable functionality.";

            c_buffer.disable();
          }

        } else {

          JNoticeStream(logger) << "Continue using circular buffer " << c_buffer;
        }

        if (c_buffer.sizeL0 > 0) { static_cast<JTreeWriterObjectOutput<JDAQTimesliceL0>&>(c_buffer).SetCircular(c_buffer.sizeL0); }
        if (c_buffer.sizeL1 > 0) { static_cast<JTreeWriterObjectOutput<JDAQTimesliceL1>&>(c_buffer).SetCircular(c_buffer.sizeL1); }
        if (c_buffer.sizeL2 > 0) { static_cast<JTreeWriterObjectOutput<JDAQTimesliceL2>&>(c_buffer).SetCircular(c_buffer.sizeL2); }
        if (c_buffer.sizeSN > 0) { static_cast<JTreeWriterObjectOutput<JDAQTimesliceSN>&>(c_buffer).SetCircular(c_buffer.sizeSN); }

      } else {

        if (c_buffer.is_open()) {
          c_buffer.close();
        }
      }
    }


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

      if (reporting) {
        JNoticeStream(logger) << "Start run " << getRunNumber();
      }
    
      timeslices.clear();

      current_slice_index = -1;
      queueSize           =  0;
    
      numberOfEvents                        = 0;
      numberOfBytes                         = 0;
      numberOfTimeslicesProcessed           = 0;
      numberOfIncompleteTimeslicesProcessed = 0;
    
      number_of_packets_received  = 0;
      number_of_packets_discarded = 0;
      number_of_bytes_received    = 0;
      number_of_reads             = 0;

      minFrameNumber = numeric_limits<int>::max();
      maxFrameNumber = numeric_limits<int>::min();

      // Reset global trigger counter.

      JDAQTriggerCounter::reset();

      logErrorRun       .reset();
      logErrorDetector  .reset();
      logErrorIndex     .reset();
      logErrorIncomplete.reset();
    
      timer.reset();
      timer.start();

      Qt.reset();

      // send trigger parameters to the datawriter
      
      ostringstream os;
    
      os << getRunNumber() << ' ' << parameters;
    
      datawriter->PutFullString(IO_TRIGGER_PARAMETERS, os.str());
    }


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

      if (!timeslices.empty()) {

        JNoticeStream(logger) << "Flushing " << timeslices.size() << " slices.";

        for (deque<JDAQTimesliceL0>::const_iterator i = timeslices.begin(); i != timeslices.end(); ++i) {
          queueSize -= getSizeof(*i);
        }

        timeslices.clear();
      }

      { // force clearance of memory

        deque<JDAQTimesliceL0> buffer;
        
        timeslices.swap(buffer);
      }

      if (queueSize != 0) {
        JWarningStream(logger) << "Pending data in queue " << queueSize << " [B]";
      }

      current_slice_index = -1;
      queueSize           =  0;

      timer.stop();    
    }


    virtual void actionContinue(int length, const char* buffer) override
    {
      timer.start();    
    }


    virtual void actionStop(int length, const char* buffer) override
    {
      typeout();

      datawriter.reset();
    }


    virtual void actionReset(int length, const char* buffer) override
    {
      if (serversocket.is_valid()) {
        serversocket->shutdown();
      }

      serversocket.reset();
    }
  

    virtual void actionQuit(int length, const char* buffer) override
    {
      datawriter.reset();
    }


    virtual void setSelect(JFileDescriptorMask& mask) const override
    {
      if (serversocket.is_valid()) {
        mask.set(*serversocket);
      }
    
      for (JChannelList_t::const_iterator channel = channelList.begin(); channel != channelList.end(); ++channel) {      
        if (!channel->isReady()) {
          mask.set(channel->getFileDescriptor());   
        }
      }
    }


    virtual void actionSelect(const JFileDescriptorMask& mask) override
    {
      using namespace std;

      for (JChannelList_t::iterator channel = channelList.begin(); channel != channelList.end(); ) {
      
        try {
          
          if (mask.has(channel->getFileDescriptor())) {
            channel->read();
          }

          if (channel->isReady()) {

            number_of_packets_received += 1;
            number_of_reads            += channel->getCounter();
            number_of_bytes_received   += channel->size();
            
            if (isRunning()) {

              updateFrameQueue(channel);

            } else {

              JErrorStream(logErrorRun) << "Receiving data while not running.";

              number_of_packets_discarded += 1;
            }

            channel->reset();
          }

          ++channel;
        }
        catch(const exception& error) {

          JNoticeStream(logger) << "Disconnecting channel " << channel->getFileDescriptor() << ' ' << error.what(); 

          channel->shutdown();

          channel = channelList.erase(channel);
        }
      }
     

      if (serversocket.is_valid()) {

        if (mask.has(*serversocket)) {

          JSocket socket;
        
          socket.accept(serversocket->getFileDescriptor());
        
          //socket.setSendBufferSize   (buffer_size);
          socket.setReceiveBufferSize(buffer_size);
        
          socket.setKeepAlive  (true);
          socket.setNonBlocking(true);
        
          JNoticeStream(logger) << "New channel" << '[' << socket.getFileDescriptor() << ']';
        
          channelList.push_back(JSocketInputChannel_t(socket));
        }
      }


      if (!timeslices.empty() && ((timeslices[0].size()      >=  frames_per_slice) ||
                                  (timeslices.size()         >=  2u                &&
                                   timeslices[1].size()      >=  frames_per_slice) ||
                                  (timeslices.size()         >=  maxQueueDepth)    ||
                                  (queueSize                 >=  maxQueueSize))) {
      
        const JDAQTimesliceL0& pending_slice  = timeslices.front();
        queueSize                            -= getSizeof(pending_slice);
      
        current_slice_index = pending_slice.getFrameIndex();
        minFrameNumber      = min(minFrameNumber, pending_slice.getFrameIndex());
        maxFrameNumber      = max(maxFrameNumber, pending_slice.getFrameIndex());
      
        if (pending_slice.size() > frames_per_slice) {
        
          JErrorStream(logger) << "More frames in timeslice than expected "
                               << pending_slice.size() << " > " << frames_per_slice;
        
          if (pending_slice.size() <= maximum_frames_per_slice) { 
          
            JErrorStream(logger) << "Adjusting expected frames per timeslice to " << pending_slice.size();
          
            frames_per_slice = pending_slice.size();
          }
        }
      
        if (!pending_slice.empty()) {

          const localtime_t t0 = getLocalTime();

          processTimeSlice(pending_slice);

          const localtime_t t1 = getLocalTime();

          numberOfTimeslicesProcessed += 1;

          Qt.put(t1 - t0);
        
          if (pending_slice.size() < frames_per_slice) {

            numberOfIncompleteTimeslicesProcessed += 1;

            ostringstream error;

            error << "Timeout -> processed incomplete timeslice: " 
                  << "Frame index = "       << pending_slice.getFrameIndex() << ';'
                  << "Size of timeslice = " << pending_slice.size()          << ';'
                  << "Queue depth = "       << timeslices.size()             << ';'
                  << "Queue size = "        << queueSize                     << ';';
            
            if ((timeslices.size()         >=  2u                &&
                 timeslices[1].size()      >=  frames_per_slice)) {

              error << " intermittent problem -> continues as-is";

            } else {
              
              error << " adjusting expected frames per timeslice from " << frames_per_slice << " to " << pending_slice.size();
              
              frames_per_slice = pending_slice.size();
            }

            JErrorStream(logErrorIncomplete) << error.str();
          }
        }

        timeslices.pop_front();
      }
    }
  

    virtual void actionRunning() override
    {
      if (reporting) {
        typeout();
      }
    }


    /**
     * Update queue with data frames.
     *
     * Note that any discarded data will be reported.
     *
     * \param  channel              incoming data channel
     */
    void updateFrameQueue(const JChannelList_t::const_iterator channel)
    {
      using namespace std;
    
      JByteArrayReader in(channel->data(), channel->size()); 
   
      JDAQPreamble         preamble;
      JDAQSuperFrameHeader header;
               
      in >> preamble;
      in >> header;    
    
      if (preamble.getLength() != channel->size()) {
      
        JErrorStream(logErrorRun) << "Size of received data does not match size reported by preamble: " 
                                  << "preamble.getLength() = " << preamble.getLength() << ';'
                                  << "channel->size(): "       << channel->size()      << ';';

        number_of_packets_discarded += 1;
       
        return;
      }
    
      if (header.getRunNumber() != getRunNumber()) {

        JErrorStream(logErrorRun) << "Run number "              << header.getRunNumber()
                                  << " unequal to current run " << getRunNumber()
                                  << " -> Dropping frame.";

        number_of_packets_discarded += 1;
      
        return;
      }
    
      if (header.getFrameIndex() <= current_slice_index) {
      
        JErrorStream(logErrorIndex) << "Frame index " << header.getFrameIndex() << " <= " << current_slice_index
                                    << " -> Dropping frame.";
      
        number_of_packets_discarded += 1;
      
        if (frames_per_slice < maximum_frames_per_slice) {
        
          frames_per_slice++;

          JErrorStream(logErrorIndex) << "Increase number of frames expected to: " << frames_per_slice;
        } 
      
        return;
      }
    
      if (current_slice_index != -1 && header.getFrameIndex() > current_slice_index + maximal_frame_index) {
      
        JErrorStream(logErrorIndex) << "Frame index " << header.getFrameIndex() << " > " << current_slice_index + maximal_frame_index 
                                    << " -> Dropping frame.";
      
        number_of_packets_discarded += 1;
      
        return;
      }
    
      deque<JDAQTimesliceL0>::iterator timesliceIterator = timeslices.begin();

      while (timesliceIterator != timeslices.end() && timesliceIterator->getFrameIndex() < header.getFrameIndex()) {
        ++timesliceIterator;
      }
    
      if (timesliceIterator != timeslices.end() && timesliceIterator->getFrameIndex() == header.getFrameIndex()) {

        // The corresponding time slice already exists
        
      } else {

        // This is the first frame of this time slice; insert a new time slice in the buffer at the right position in the list

        timesliceIterator = timeslices.insert(timesliceIterator, JDAQTimesliceL0());
        
        timesliceIterator->setDAQChronometer(header.getDAQChronometer());

        queueSize += getSizeof(*timesliceIterator);
      }
   
      timesliceIterator->push_back(JDAQSuperFrame(header));
      
      in >> static_cast<JDAQFrame&>(*(timesliceIterator->rbegin()));

      queueSize += getSizeof(*timesliceIterator->rbegin());
    }
  

    /**
     * Process time slice.
     *
     * \param  timeslice            time slice
     */
    void processTimeSlice(const JDAQTimesliceL0& timeslice)
    {
      using namespace std;

      try {

        timesliceRouter->configure(timeslice);

        if (parameters.writeSummary()) {
          this->put(JDAQSummaryslice(timeslice));
        }
      
        if (parameters.trigger3DMuon.enabled    || 
            parameters.trigger3DShower.enabled  || 
            parameters.triggerMXShower.enabled  || 
            parameters.triggerNB.enabled        || 
            parameters.writeL0.prescale         ||
            parameters.writeL1.prescale         ||
            parameters.writeL2.prescale         ||
            parameters.writeSN.prescale         ||
            c_buffer.is_enabled()) {

          JTimeslice_t    timesliceL0(timeslice.getDAQChronometer());
          JTimeslice_t    timesliceL1(timeslice.getDAQChronometer());
          JTimeslice_t    timesliceL2(timeslice.getDAQChronometer());
          JTimeslice_t    timesliceSN(timeslice.getDAQChronometer());
          JTimeslice_t    timesliceNB(timeslice.getDAQChronometer());
          JDAQTimeslice   timesliceTX(timeslice.getDAQChronometer());

          for (JDAQTimesliceL0::const_iterator frame = timeslice.begin(); frame != timeslice.end(); ++frame) {
          
            if (moduleRouter->hasModule(frame->getModuleID())) {

              if (!checksum(*frame)) {

                JErrorStream(logger) << "Invalid data at "
                                     << "run = "         << timeslice.getRunNumber()  << ";"
                                     << "frame index = " << timeslice.getFrameIndex() << ";"
                                     << "module = "      << frame->getModuleID()      << ";"
                                     << "discard" << (dumpCount < dumpLimit ? " and dump" : "");

                timesliceTX.push_back(*frame);

                continue;
              }

              const JModule&   module = moduleRouter->getModule(frame->getModuleID()); 
              JSuperFrame2D_t& buffer = JSuperFrame2D_t::demultiplex(*frame, module);
      
              // Apply high-rate veto
            
              buffer.applyHighRateVeto(parameters.highRateVeto_Hz);

              // L0

              timesliceL0.push_back(JSuperFrame1D_t(buffer));

              // Nano-beacon trigger

              if (parameters.triggerNB.enabled) {

                JSuperFrame2D_t::iterator __end = partition(buffer.begin(), buffer.end(), parameters.triggerNB.pmts);

                if (buffer.begin() != __end) {

                  timesliceNB.push_back(JSuperFrame1D_t(frame->getDAQChronometer(),
                                                        frame->getModuleIdentifier(),
                                                        module.getPosition()));

                  JSuperFrame1D_t zbuf;

                  (*buildL1)(buffer.begin(), __end , back_inserter(zbuf));

                  (*buildNB)(buffer.begin() , __end, zbuf, back_inserter(*timesliceNB.rbegin()));
                }
              }

              // L1
      
              timesliceL1.push_back(JSuperFrame1D_t(frame->getDAQChronometer(),
                                                    frame->getModuleIdentifier(),
                                                    module.getPosition()));
            
              (*buildL1)(*timesliceL0.rbegin(), back_inserter(*timesliceL1.rbegin()));
            
              // L2
            
              timesliceL2.push_back(JSuperFrame1D_t(frame->getDAQChronometer(),
                                                    frame->getModuleIdentifier(),
                                                    module.getPosition()));
      
              (*buildL2)(buffer, *timesliceL1.rbegin(), back_inserter(*timesliceL2.rbegin()));
            
              // SN
            
              timesliceSN.push_back(JSuperFrame1D_t(frame->getDAQChronometer(),
                                                    frame->getModuleIdentifier(),
                                                    module.getPosition()));
      
              (*buildSN)(buffer, *timesliceL1.rbegin(), back_inserter(*timesliceSN.rbegin()));
      
            } else {

              JErrorStream(logErrorDetector) << "No detector information for module " << frame->getModuleID(); 
            }
          }

          if (!timesliceTX.empty()) {

            if (dumpCount < dumpLimit) {

              this->put(timesliceTX);

              dumpCount += 1;
            }
          }
        
          // Trigger
        
          if (parameters.triggerNB.enabled) {      

            const JTriggerInput  trigger_input(timesliceNB);

            for (JTriggerInput::const_iterator hit = trigger_input.begin(); hit != trigger_input.end(); ++hit) {

              if (parameters.triggerNB.write()) {

                JTriggeredEvent tev(timesliceNB.getDAQChronometer(),
                                    getTriggerMask(triggerNB->getTriggerBit()),
                                    *hit, 
                                    *timesliceRouter, 
                                    *moduleRouter, 
                                    parameters.TMaxLocal_ns, 
                                    parameters.triggerNB.DMax_m,
                                    getTimeRange(parameters.triggerNB));

                this->put(tev);
              }
            }
          }

          JTriggerInput  trigger_input(timesliceL2);
          JTriggerOutput trigger_output;
          
          (*trigger3DMuon)  (trigger_input, back_inserter(trigger_output));
          (*trigger3DShower)(trigger_input, back_inserter(trigger_output));
          (*triggerMXShower)(trigger_input, timesliceL0, back_inserter(trigger_output));
        
          trigger_output.merge(JEventOverlap(parameters.TMaxEvent_ns));

          for (JTriggerOutput::const_iterator event = trigger_output.begin(); event != trigger_output.end(); ++event) {
            
            const JTriggeredEvent object(*event, *timesliceRouter, *moduleRouter, parameters.TMaxLocal_ns, getTimeRange(parameters));

            if (this->put(object)) {        
              numberOfEvents += 1;
            }
          }

          if (parameters.writeL1() || c_buffer.sizeL1 > 0) {
      
            const JTimesliceL1<JDAQTimesliceL1> object(timesliceL1, *timesliceRouter, *moduleRouter, parameters.TMaxLocal_ns);

            if (parameters.writeL1)  { this->put(object);  }
            if (c_buffer.sizeL1 > 0) { c_buffer.put(object); }
          }

          if (parameters.writeL2() || c_buffer.sizeL2 > 0) {
      
            const JTimesliceL1<JDAQTimesliceL2> object(timesliceL2, *timesliceRouter, *moduleRouter, parameters.L2.TMaxLocal_ns);

            if (parameters.writeL2)  { this->put(object); }
            if (c_buffer.sizeL2 > 0) { c_buffer.put(object); }
          }

          if (parameters.writeSN() || c_buffer.sizeSN > 0) {
      
            const JTimesliceL1<JDAQTimesliceSN> object(timesliceSN, *timesliceRouter, *moduleRouter, parameters.SN.TMaxLocal_ns);

            if (parameters.writeSN)  { this->put(object); }
            if (c_buffer.sizeSN > 0) { c_buffer.put(object); }
          }

          if (parameters.writeL0() || c_buffer.sizeL0 > 0) {
        
            if (parameters.writeL0)  { this->put(timeslice); }
            if (c_buffer.sizeL0 > 0) { c_buffer.put(timeslice); }
          }
        }

      } catch(const exception& error) {

        JErrorStream(logger) << "Error = "       << error.what()              << ";"
                             << "run = "         << timeslice.getRunNumber()  << ";"
                             << "frame index = " << timeslice.getFrameIndex() << ";"
                             << "time slice not correctly processed;"
                             << "discard" << (dumpCount < dumpLimit ? " and dump" : "");

        if (dumpCount < dumpLimit) {

          this->put(static_cast<const JDAQTimeslice&>(timeslice));

          dumpCount += 1;
        }
      }

      timesliceRouter->reset();
    }
    

    /**
     * Report status to message logger.
     */
    void typeout()
    { 
      timer.stop();

      const double T_us = (double) timer.usec_wall;
    
      JStatusStream(logger) << "Elapsed real (wall) time [s] "          << T_us / 1e6;
      JStatusStream(logger) << "Elapsed user CPU time [s] "             << (double) timer.usec_ucpu/ 1e6;
      JStatusStream(logger) << "Elapsed system CPU time [s] "           << (double) timer.usec_scpu/ 1e6;
      try {
        JStatusStream(logger) << "Elapsed real time per time slice [ms] " << Qt.getMean() * 1.0e-3  << " +/- " << Qt.getDeviation() * 1.0e-3;
      }
      catch(const std::exception&) {}
      JStatusStream(logger) << "Number of packets received/discarded "  << number_of_packets_received << "/" << number_of_packets_discarded;
      JStatusStream(logger) << "Number of events/MB sent "              << numberOfEvents             << "/" << numberOfBytes/1e6;

      if (number_of_packets_received > 0) {
        JStatusStream(logger) << "Number of reads/packet " << (double) number_of_reads / (double) number_of_packets_received;
      }
    
      JStatusStream(logger) << "Current queue depth/size " << timeslices.size() << "/" << queueSize;
      JStatusStream(logger) << "Current number of frames per slice expected: " << frames_per_slice;
    
      JStatusStream(logger) << "Number of timeslices processed total/incomplete " << numberOfTimeslicesProcessed << "/" << numberOfIncompleteTimeslicesProcessed;

      if (numberOfTimeslicesProcessed > 0) {
        JStatusStream(logger) << "Real time per timeslice [ms] "       << timer.usec_wall / 1000 / numberOfTimeslicesProcessed;
        JStatusStream(logger) << "User CPU  time per timeslice [ms] "  << timer.usec_ucpu / 1000 / numberOfTimeslicesProcessed;
        JStatusStream(logger) << "System CPU time per timeslice [ms] " << timer.usec_scpu / 1000 / numberOfTimeslicesProcessed;
      }
    
      const double processedSlicesTime_us   =  numberOfTimeslicesProcessed      * getFrameTime() / 1000;
      const double processedDetectorTime_us = (maxFrameNumber - minFrameNumber) * getFrameTime() / 1000;
    
      if (processedSlicesTime_us   > 0) {
        JStatusStream(logger) << "Performance factor (inaccurate estimate): "                << T_us / processedSlicesTime_us;
      }
      if (processedDetectorTime_us > 0) {
        JStatusStream(logger) << "Performance factor whole detector (inaccurate estimate): " << T_us / processedDetectorTime_us;
      }

      timer.start();
    }


    /**
     * Tagged action to handle alerts.
     *
     * \param  tag                  tag
     * \param  length               number of characters
     * \param  buffer               message
     */
    virtual void actionTagged(const JTag& tag, int length, const char* buffer) override
    {
      using namespace std;
      
      JDebugStream(logger) << "Received message <" << tag.toString() << "> \"" << string(buffer, length) << "\"";

      if (tag == RC_ALERT) {

        if (c_buffer.is_open()) {

          JNoticeStream(logger) << "Close circular buffer " << c_buffer;

          c_buffer.close();
        }

        if (c_buffer.is_enabled()) {
          
          c_buffer.open(path, getUniqueTag());

          if (c_buffer.is_open()) {
            
            JNoticeStream(logger) << "Created circular buffer " << c_buffer;
          
            putObject(c_buffer.getFile(), meta);
            
          } else {
            
            JErrorStream (logger) << "Failed to create circular buffer in directory <" << path << ">; disable functionality.";

            c_buffer.disable();
          }
        }

      } else {

        JWarningStream(logger) << "Tag <" << tag.toString() << "> not implemented";
      }
    }

    JMeta meta;                                 //!< meta data

    static const int MAXIMUM_FILE_NUMBER = 100; //!< maximum file number for overwrite protection.

  private:
  
    typedef JControlHostObjectOutput<JDAQTypes_t>  JControlHost_t;
    JSinglePointer<JControlHost_t> datawriter;     //!< controlhost of data server (to which data writer should be connected)
    std::string                    hostname;       //!< host name of data server

    /**
     * Auxiliary method to send object to data server.
     *
     * \param  object       object to be sent
     * \return              true if sent; else false
     */
    template<class T>
    bool put(const T& object) 
    {
      try {

        datawriter->put(object);

        numberOfBytes += getSizeof(object);

        return true;
      }
      catch(const std::exception& error) {
        JErrorStream(logger) << error.what();
      }

      return false;
    }


    int    port;                                        //!< server socket port
    int    backlog;
    int    buffer_size;

    JSinglePointer<JServerSocket>      serversocket;    //!< server for data queue connections
    JChannelList_t                     channelList;     //!< connections to data queue

    JTimer    timer;  
    JQuantile Qt;
  
    std::deque<JDAQTimesliceL0>      timeslices;        //!< buffer with pending time slice data
    int                              current_slice_index;
    unsigned int                     frames_per_slice;
    unsigned int                     maximum_frames_per_slice;
    int                              maximal_frame_index;
  
    // trigger 
 
    JTriggerParameters               parameters;
    JSinglePointer<JTimesliceRouter> timesliceRouter;

    JDetector                        detector;
    JSinglePointer<JModuleRouter>    moduleRouter;
    JSinglePointer<JBuildL1_t>       buildL1; 
    JSinglePointer<JBuildL2_t>       buildL2; 
    JSinglePointer<JBuildL2_t>       buildSN; 
    JSinglePointer<JBuildL2_t>       buildNB; 

    JSinglePointer<JTriggerNB>       triggerNB;
    JSinglePointer<JTrigger3DMuon>   trigger3DMuon;
    JSinglePointer<JTrigger3DShower> trigger3DShower;
    JSinglePointer<JTriggerMXShower> triggerMXShower;

    JMessageScheduler logErrorRun;
    JMessageScheduler logErrorDetector;
    JMessageScheduler logErrorIndex;
    JMessageScheduler logErrorIncomplete;

    int dumpCount;
    int dumpLimit;

    // process management

    std::vector<JDAQProcess> dataFilters;
    std::vector<JDAQProcess> dataQueues;

    // memory management

    long long int totalCPURAM;
    unsigned int  maxQueueDepth;
    long long int maxQueueSize;
    long long int queueSize;

    // statistics

    bool reporting;
  
    long long int numberOfEvents;
    long long int numberOfBytes;
    long long int numberOfTimeslicesProcessed;
    long long int numberOfIncompleteTimeslicesProcessed;
  
    int  minFrameNumber;
    int  maxFrameNumber;
  
    // temporary

    long long int number_of_packets_received;
    long long int number_of_packets_discarded; 
    long long int number_of_reads;
    long long int number_of_bytes_received;

    // circular buffer

    JCircularBuffer_t c_buffer;
    std::string       path;                             //!< directory for writing circular buffer
  };
}

/**
 * \file
 *
 * Application for real-time filtering of data.
 * For more information, see KM3NETDAQ::JDataFilter.
 *
 * \author rbruijn and mdejong
 */
int main(int argc, char* argv[])
{
  using namespace std;
  using namespace JPP;
  using namespace KM3NETDAQ;
  
  string  server;
  string  logger;
  string  hostname;
  string  client_name;
  int     port;
  int     backlog;
  int     buffer_size;
  bool    use_cout;
  string  path;
  int     debug;

  
  try {

    JParser<> zap("Application for real-time filtering of data.");
   
    zap['H'] = make_field(server)          = "localhost";
    zap['M'] = make_field(logger)          = "localhost";
    zap['D'] = make_field(hostname)        = "";
    zap['u'] = make_field(client_name)     = "JDataFilter";
    zap['P'] = make_field(port);
    zap['q'] = make_field(backlog)         =   1024;
    zap['s'] = make_field(buffer_size)     = 32 * MEGABYTE; // TCP buffer of 32 MB
    zap['c'] = make_field(use_cout);
    zap['p'] = make_field(path)            = "/tmp/";
    zap['d'] = make_field(debug)           =      0;
    
    zap(argc, argv);
  }
  catch(const exception& error) {
    FATAL(error.what() << endl);
  }


  JLogger* out = NULL;
  
  if (use_cout)
    out = new JStreamLogger(cout);
  else
    out = new JControlHostLogger(logger);
    
  JDataFilter dfilter(client_name, 
                      server, 
                      hostname,
                      out, 
                      debug, 
                      port, 
                      backlog, 
                      buffer_size,
                      path);

  dfilter.meta = JMeta(argc, argv);

  dfilter.enter();
  dfilter.run();
}