JDataFilter.cc

Go to the documentation of this file.

 
#include <string>
#include <iostream>
#include <iomanip>
#include <list>
#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 "JDAQ/JDAQTags.hh"
#include "JDAQ/JDAQEventIO.hh"
#include "JDAQ/JDAQTimesliceIO.hh"
#include "km3net-dataformat/online/JDAQAbstractPreamble.hh"
#include "JDAQ/JDAQPreambleIO.hh"
#include "JDAQ/JDAQFrameIO.hh"
#include "JDAQ/JDAQSuperFrameIO.hh"
#include "JDAQ/JDAQSummarysliceIO.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/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 "JTools/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
   * JDataFilter::setSelect and 
   * JDataFilter::actionSelect.
   *
   * In state <tt>Running</tt>, all incoming data are buffered, referred to as "queue".\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 JDataFilter::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 the corresponding pre-set limit:\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, 
   * upto the original value set at JDataFilter::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 director:
   *   - 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 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/script JConvert.cc/JConvert.sh can be used to readily convert multiple input files to a single output file. 
   *
   * The script JDOMandDataFilter.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(JException& exception) {}
        }
      }
 
      /**
       * 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()
    {}
 
 
    virtual void actionExit()
    { 
      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) 
    { 
      JDebugStream(logger) << "actionInit() " << std::string(buffer,length);
 
      try {
 
        JDebugStream(logger) << "Start server.";
 
        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)
    { 
      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;
    
      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["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;
 
      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." ;  
      }
    
      // trigger parameters
    
      parameters.set(getMaximalDistance(detector));
    
      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));
   
      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."; }
     
      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)
    {
      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)
    { 
      using namespace std;
 
      if (!timeslices.empty()) {
 
        JNoticeStream(logger) << "Flushing " << timeslices.size() << " slices.";
 
        for (list<JDAQTimesliceL0>::const_iterator i = timeslices.begin(); i != timeslices.end(); ++i) {
          queueSize -= getSizeof(*i);
        }
 
        timeslices.clear();
      }
 
      { // force clearance of memory
 
        std::list<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)
    {
      timer.start();    
    }
 
 
    virtual void actionStop(int length, const char* buffer) 
    {
      typeout();
 
      datawriter.reset();
    }
 
 
    virtual void actionReset(int length, const char* buffer)
    {
      serversocket.reset();
    }
  
 
    virtual void actionQuit(int length, const char* buffer)
    {
      datawriter.reset();
    }
 
 
    virtual void setSelect(JFileDescriptorMask& mask) const
    {
      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)
    {
      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 JSocketException& 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.front().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;
 
            JErrorStream(logErrorIncomplete) << "Timeout -> processed incomplete timeslice: " 
                                             << "Frame index = "       << pending_slice.getFrameIndex() << ';'
                                             << "Size of timeslice = " << pending_slice.size()          << ';'
                                             << "Queue depth = "       << timeslices.size()             << ';'
                                             << "Queue size = "        << queueSize;
          
            if (!timeslices.empty()) {
              
              JErrorStream(logger) << "Adjusting expected frames per timeslice from " << frames_per_slice << " to " << pending_slice.size();
              
              frames_per_slice = pending_slice.size();
            }
          }
        }
 
        timeslices.pop_front();
      }
    }
  
 
    virtual void actionRunning()
    {
      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;
      }
    
      list<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 {
 
        if (parameters.writeSummary()) {
          this->put(JDAQSummaryslice(timeslice));
        }
      
        if (parameters.trigger3DMuon.enabled    || 
            parameters.trigger3DShower.enabled  || 
            parameters.triggerMXShower.enabled  || 
            parameters.writeL0.prescale         ||
            parameters.writeL1.prescale         ||
            parameters.writeL2.prescale         ||
            parameters.writeSN.prescale         ||
            c_buffer.is_enabled()) {
          
          timesliceRouter->configure(timeslice);
      
          JTimeslice_t    timesliceL0(timeslice.getDAQChronometer());
          JTimeslice_t    timesliceL1(timeslice.getDAQChronometer());
          JTimeslice_t    timesliceL2(timeslice.getDAQChronometer());
          JTimeslice_t    timesliceSN(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 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));
 
              // 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()) {
            this->put(timesliceTX);
          }
        
          // Trigger
        
          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 JTriggerException& error) {
        JErrorStream(logger) << "Error = "       << error.what()              << ";"
                             << "run = "         << timeslice.getRunNumber()  << ";"
                             << "frame index = " << timeslice.getFrameIndex() << ";"
                             << "time slicee not correctly processed!";
      }
    }
    
 
    /**
     * Report status to message logger.
     */
    void typeout()
    { 
      timer.stop();
 
      const double T_us = (double) timer.usec_wall;
    
      JNoticeStream(logger) << "Elapsed real (wall) time [s] "          << T_us / 1e6;
      JNoticeStream(logger) << "Elapsed user CPU time [s] "             << (double) timer.usec_ucpu/ 1e6;
      JNoticeStream(logger) << "Elapsed system CPU time [s] "           << (double) timer.usec_scpu/ 1e6;
      try {
        JNoticeStream(logger) << "Elapsed real time per time slice [ms] " << Qt.getMean() * 1.0e-3  << " +/- " << Qt.getDeviation() * 1.0e-3;
      }
      catch(const std::exception&) {}
      JNoticeStream(logger) << "Number of packets received/discarded "  << number_of_packets_received << "/" << number_of_packets_discarded;
      JNoticeStream(logger) << "Number of events/MB sent "              << numberOfEvents             << "/" << numberOfBytes/1e6;
 
      if (number_of_packets_received > 0) {
        JNoticeStream(logger) << "Number of reads/packet " << (double) number_of_reads / (double) number_of_packets_received;
      }
    
      JNoticeStream(logger) << "Current queue depth/size " << timeslices.size() << "/" << queueSize;
      JNoticeStream(logger) << "Current number of frames per slice expected: " << frames_per_slice;
    
      JNoticeStream(logger) << "Number of timeslices processed total/incomplete " << numberOfTimeslicesProcessed << "/" << numberOfIncompleteTimeslicesProcessed;
 
      if (numberOfTimeslicesProcessed > 0) {
        JNoticeStream(logger) << "Real time per timeslice [ms] "       << timer.usec_wall / 1000 / numberOfTimeslicesProcessed;
        JNoticeStream(logger) << "User CPU  time per timeslice [ms] "  << timer.usec_ucpu / 1000 / numberOfTimeslicesProcessed;
        JNoticeStream(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) {
        JNoticeStream(logger) << "Performance factor (inaccurate estimate): "                << T_us / processedSlicesTime_us;
      }
      if (processedDetectorTime_us > 0) {
        JNoticeStream(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)
    {
      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 JControlHostException& error) {
        JErrorStream(logger) << error;
      }
 
      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::list<JDAQTimesliceL0>       timeslices;        //!< buffer with pending time slice data
    int                              current_slice_index;
    unsigned int                     frames_per_slice;
    unsigned int                     maximum_frames_per_slice;
  
    // trigger 
 
    JTriggerParameters               parameters;
    JSinglePointer<JTimesliceRouter> timesliceRouter;
 
    JDetector                        detector;
    JSinglePointer<JModuleRouter>    moduleRouter;
    JSinglePointer<JBuildL1_t>       buildL1; 
    JSinglePointer<JBuildL2_t>       buildL2; 
    JSinglePointer<JBuildL2_t>       buildSN; 
 
    JSinglePointer<JTrigger3DMuon>   trigger3DMuon;
    JSinglePointer<JTrigger3DShower> trigger3DShower;
    JSinglePointer<JTriggerMXShower> triggerMXShower;
 
    JMessageScheduler logErrorRun;
    JMessageScheduler logErrorDetector;
    JMessageScheduler logErrorIndex;
    JMessageScheduler logErrorIncomplete;
 
    // 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();
}