JDataFilter.cc

Go to the documentation of this file.

#include <string>
#include <iostream>
#include <iomanip>
#include <list>
#include <limits>
#include <algorithm>

#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 "JLogger/JMessageScheduler.hh"

#include "JDAQ/JDAQ.hh"
#include "JDAQ/JDAQTags.hh"
#include "JDAQ/JDAQEvent.hh"
#include "JDAQ/JDAQTimeslice.hh"
#include "JDAQ/JDAQAbstractPreamble.hh"
#include "JDAQ/JDAQPreamble.hh"
#include "JDAQ/JDAQSuperFrame.hh"
#include "JDAQ/JDAQFrame.hh"
#include "JDAQ/JDAQSummaryslice.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 "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 "JSupport/JSupport.hh"

#include "JSystem/JSystemToolkit.hh"
#include "JSystem/JNetwork.hh"


using namespace KM3NETDAQ;
using namespace JPP;
using namespace std;
//using JTOOLS::make_range;


bool processIndexSorter(const JDAQProcess &a, const JDAQProcess& b)
{
  return a.index<b.index;
}

namespace JNET {

  template<>
  inline int getSizeOfPacket (const KM3NETDAQ::JDAQAbstractPreamble& preamble) 
  { 
    return preamble.getLength();
  }
}

typedef JNET::JSocketInputChannel<KM3NETDAQ::JDAQAbstractPreamble>  JSocketInputChannel_t;
typedef std::vector<JSocketInputChannel_t>                          JChannelList_t;
typedef double        hit_type;

typedef vector       <hit_type>   JFrameL1_t;
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;


class JDataFilter :
  public JDAQClient
{
public:
  /**
   * 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
   */
  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) :
    JDAQClient(name,server,logger,level),
    hostname  (hostname),
    parameters()
  {
    this->port        = port;
    this->backlog     = backlog;
    this->buffer_size = buffer_size;

    replaceEvent(RC_CMD, RC_DFILTER, ev_configure); /// replace tag to receive dfilter specific messages in configure
    
    totalCPURAM       = getRAM();   
    
    current_slice_index  = -1;
  
    reporting = false;
  }


  virtual void actionEnter()
  {     
    try {

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

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


  virtual void actionExit()
  { 
    JDebugStream(logger) << "Stop server.";

    serversocket.reset();
    datawriter  .reset();
  }


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


  virtual void actionConfigure(int length, const char* buffer)
  { 

    // todo:
    
    // 3) receive and set site id
   
    
    using namespace std;
    using namespace JPP;
    
    JDebugStream(logger) << "actionConfigure() " << std::string(buffer,length);
  
    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["JDataFilter"]            = dataFilters;
    properties["DataQueue"]              = dataQueues;
    properties["pmtIdCheck"]             = checkForInvalidPMTs = false;
    properties["maxHitsFrame"]           = maxHitsFrame  = 31*2000;                    // default is 20 kHz

    try {
      properties.read(string(buffer, length));
    }
    catch(const exception& error) {
      JErrorStream(logger) << error.what();
    }
    
    hostname = trim(hostname);

    if (hostname != "") 
      datawriter.reset(new JControlHost_t(hostname));  // may throw an error
    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();
    
    for(vector<JDAQProcess>::iterator i = dataFilters.begin(); i != dataFilters.end(); ++i) {
      
      if(i->hostname == getHostname() ||
         i->hostname == getIPaddress()) {

        numberOfDataFiltersOnThisMachine++;
        
        if (i->port == this->port) {
          thisProcess = i;
        }
      }
    }
    
    if (numberOfDataFiltersOnThisMachine == 0) {
      JNoticeStream(logger) << "Zero datafilters 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 JDataFilter process will report.";
      JNoticeStream(logger) << "Number of modules: "    << (*moduleRouter)->size();
      JNoticeStream(logger) << "Trigger parameters: "   << parameters;
      JNoticeStream(logger) << "Detector description: " << endl << detector;
      JNoticeStream(logger) << "Clock interval: "       << 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));
    logErrorInvalidPMTs = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));
  }


  virtual void actionStart(int length, const char* buffer)
  { 
    if (reporting) {
      JNoticeStream(logger) << "Start run " << getRunNumber();
    }
    
    timeslices.clear();

    current_slice_index = -1;
    queueSize           =  0;
    
    numberOfEvents                        = 0;
    numberOfBytes                         = 0;
    numberOfTimeslicesProcessed           = 0;
    numberOfIncompleteTimeslicesProcessed = 0;
    
    numberOfFramesWithInvalidPMTs         = 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.

    KM3NETDAQ::JDAQTriggerCounter::reset();

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

    // 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)
  { 
    if (!timeslices.empty()) {

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

      for (list<JDAQTimesliceL0>::const_iterator i = timeslices.begin(); i != timeslices.begin(); ++i) {
        queueSize -= i->getSize();
      }

      timeslices.clear();
    }

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

    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();
  }


  virtual void actionReset(int length, const char* buffer)
  {}
  

  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)
  {
    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)) {
      
      JDAQTimesliceL0 pending_slice  = timeslices.front();
      queueSize                     -= pending_slice.getSize();
      timeslices.pop_front();
      
      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();
        
        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()) {

        processTimeSlice(pending_slice);

        numberOfTimeslicesProcessed += 1;
        
        if (pending_slice.size() < frames_per_slice) {
          numberOfIncompleteTimeslicesProcessed += 1;
          JErrorStream(logErrorIncomplete) << "Timeout -> processed incomplete timeslice " 
                                           << "Timeslice frameindex " << 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();
          }
        }
      }
    }
  }
  

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


  void updateFrameQueue(const JChannelList_t::const_iterator channel)
  {
    JIO::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) << "FrameIndex " << header.getFrameIndex()  <<" already processed, dropping frame.";
      
      number_of_packets_discarded += 1;
      
      if(frames_per_slice<maximum_frames_per_slice) {
        
        frames_per_slice++;

        JErrorStream(logErrorIndex) << "Increased number of frames expected to: " << frames_per_slice;
      } 
      
      return;
    }
    
    list<JDAQTimesliceL0>::iterator timesliceIterator;
     
    for (timesliceIterator  = timeslices.begin();
         timesliceIterator != timeslices.end() && timesliceIterator->getFrameIndex() < header.getFrameIndex();
         ++timesliceIterator) ;
    
    if (timesliceIterator != timeslices.end() && timesliceIterator->getFrameIndex() == header.getFrameIndex()) {
      
    } else {

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

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

      queueSize += timesliceIterator->getSize();
    }
    
   
    if (checkForInvalidPMTs) {
     
      JDAQSuperFrame frame(header);
      
      in >> static_cast<JDAQFrame&> (frame);
        
      if(frame.size()>maxHitsFrame) {
        JErrorStream(logger) << "Frame size exceeds limit. Size: " << frame.size()<< " Frame discarded.";
        return;
      }
      JDAQSuperFrame::const_iterator hit = frame.begin() ;
      
      //  std::vector<unsigned int> prevHitT(30,0);
      
      for (; hit!=frame.end(); ++hit) {
        
        if(hit->getPMT()>30) //||
          // hit->getT()<prevHitT[hit->getPMT()])
          {
            break;
          }
        
        //   prevHitT[hit->getPMT()]=hit->getT();
      }
      
      if(hit!=frame.end()) {
        numberOfFramesWithInvalidPMTs +=1;
        JErrorStream(logErrorInvalidPMTs)<< "Frame with invalid PMT id or wrong time ordering discarded. Total discarded: "<< numberOfFramesWithInvalidPMTs;
      } 
      else 
        {
        
          
          timesliceIterator->push_back(frame);
          queueSize += timesliceIterator->rbegin()->getSize();
        }  
    }
    else {
      
      
      timesliceIterator->push_back(JDAQSuperFrame(header));
      
      in >> static_cast<JDAQFrame&>(*(timesliceIterator->rbegin()));
      queueSize += timesliceIterator->rbegin()->getSize();
    }
    
  }
  

  void processTimeSlice(const JDAQTimesliceL0& pending_slice)
  {
    //JDebugStream(logger) << "Processing slice: " 
    //                   << pending_slice.getRunNumber()  << " " 
    //                   << pending_slice.getFrameIndex() << " " 
    //                   << pending_slice.size();
   
    try {

      if (parameters.writeSummary()) {
      
        try {

          JDAQSummaryslice object(pending_slice);

          datawriter->put(object);

          numberOfBytes  += object.getSize();
        }
        catch(const JControlHostException& error) {
          JErrorStream(logger) << error;
        }
      }
      
      timesliceRouter->configure(pending_slice);
      
      if (parameters.trigger3DMuon.enabled    || 
          parameters.trigger3DShower.enabled  || 
          parameters.triggerMXShower.enabled  || 
          parameters.writeL1.prescale         ||
          parameters.writeL2.prescale         ||
          parameters.writeSN.prescale) {
        
      
        JTimeslice_t    timesliceL0(pending_slice.getDAQChronometer());
        JTimeslice_t    timesliceL1(pending_slice.getDAQChronometer());
        JTimeslice_t    timesliceL2(pending_slice.getDAQChronometer());
        JTimeslice_t    timesliceSN(pending_slice.getDAQChronometer());
        
        for (JDAQTimesliceL0::const_iterator super_frame = pending_slice.begin(); super_frame != pending_slice.end(); ++super_frame) {
          
          if (moduleRouter->hasModule(super_frame->getModuleID())) {
          
            const JModule&   module = moduleRouter->getModule(super_frame->getModuleID()); 
            JSuperFrame2D_t& buffer = JSuperFrame2D_t::demultiplex(*super_frame, module);
      
            // Apply high-rate veto
            
            buffer.applyHighRateVeto(parameters.highRateVeto_Hz);

            // L0

            timesliceL0.push_back(JSuperFrame1D_t(buffer));

            // L1

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

            JErrorStream(logErrorDetector) << "No detector information for module " << super_frame->getModuleID(); 
          }
        }
        
        // 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) {
            
          try {
            
            JTriggeredEvent object(*event, *timesliceRouter, *moduleRouter, parameters.TMaxLocal_ns, getTimeRange(parameters));
            
            datawriter->put(object);
            
            numberOfEvents += 1;
            numberOfBytes  += object.getSize();
          }
          catch(const JControlHostException& error) {
            JErrorStream(logger) << error;
          }
        }

        if (parameters.writeL1()) {
      
          try {

            JTimesliceL1<JDAQTimesliceL1> object(timesliceL1, *timesliceRouter, *moduleRouter, parameters.TMaxLocal_ns);

            datawriter->put(object);

            numberOfBytes  += object.getSize();
          }
          catch(const JControlHostException& error) {
            JErrorStream(logger) << error;
          }
        }

        if (parameters.writeL2()) {
      
          try {

            JTimesliceL1<JDAQTimesliceL2> object(timesliceL2, *timesliceRouter, *moduleRouter, parameters.L2.TMaxLocal_ns);

            datawriter->put(object);

            numberOfBytes  += object.getSize();
          }
          catch(const JControlHostException& error) {
            JErrorStream(logger) << error;
          }
        }

        if (parameters.writeSN()) {
      
          try {

            JTimesliceL1<JDAQTimesliceSN> object(timesliceSN, *timesliceRouter, *moduleRouter, parameters.SN.TMaxLocal_ns);

            datawriter->put(object);

            numberOfBytes  += object.getSize();
          }
          catch(const JControlHostException& error) {
            JErrorStream(logger) << error;
          }
        }
      }

      if (parameters.writeTimeslices() || parameters.writeL0()) { // backward compatible, remove later
        
        try {

          datawriter->put(pending_slice);         

          numberOfBytes  += pending_slice.getSize();
        }
        catch(const JControlHostException& error) {
          JErrorStream(logger)<<error;    
        }
      }

    } catch(const JTriggerException& error) {
      JErrorStream(logger) << error.what();
      JErrorStream(logger) << "Run: " << pending_slice.getRunNumber() << " frame index: " <<  pending_slice.getFrameIndex() << " not processed!";
    }
  }
    

  void typeout()
  { 
    timer.stop();

    const double T_us = (double) timer.usec_wall;
    
    JNoticeStream(logger) << "Reporting statistics for this datafilter.";
    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;
    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 " << timeslices.size();
    JNoticeStream(logger) << "Current queue size "  << queueSize;
    JNoticeStream(logger) << "Current number of frames per slice expected: " << frames_per_slice;
    
    //if (T_us > 0) { JNoticeStream(logger) << "Packet rate  [kHz] "      << number_of_packets_received * 1000 / T_us; }
    //if (T_us > 0) { JNoticeStream(logger) << "Data   rate [MB/s] "      << number_of_bytes_received          / T_us; }    
    //if (T_us > 0) { JNoticeStream(logger) << "Event rate  [Hz] "        << numberOfEvents * 1e6 / T_us; }
    //if (T_us > 0) { JNoticeStream(logger) << "Output data rate [MB/s] " << numberOfBytes        / T_us; }
    
    JNoticeStream(logger) << "Number of timeslices processed total/incomplete " << numberOfTimeslicesProcessed << "/" << numberOfIncompleteTimeslicesProcessed;
    JNoticeStream(logger) << "Number of frames with invalid PMTs dropped "      << numberOfFramesWithInvalidPMTs;
    
    if (numberOfTimeslicesProcessed > 0) { JNoticeStream(logger) << "Real time per timeslice [ms] "       << timer.usec_wall / 1000 / numberOfTimeslicesProcessed; }
    if (numberOfTimeslicesProcessed > 0) { JNoticeStream(logger) << "User CPU  time per timeslice [ms] "  << timer.usec_ucpu / 1000 / numberOfTimeslicesProcessed; }
    if (numberOfTimeslicesProcessed > 0) { JNoticeStream(logger) << "System CPU time per timeslice [ms] " << timer.usec_scpu / 1000 / numberOfTimeslicesProcessed; }
    
    const double processedSlicesTimeMus   =  numberOfTimeslicesProcessed      * getFrameTime() / 1000;
    const double processedDetectorTimeMus = (maxFrameNumber - minFrameNumber) * getFrameTime() / 1000;
    
    if (processedSlicesTimeMus   > 0) { JNoticeStream(logger) << "Performance factor (inaccurate estimate): "                << T_us / processedSlicesTimeMus;   }
    if (processedDetectorTimeMus > 0) { JNoticeStream(logger) << "Performance factor whole detector (inaccurate estimate): " << T_us / processedDetectorTimeMus; }

    timer.start();
  }

    
private:
  
  typedef JControlHostObjectOutput<JDAQTypes_t>  JControlHost_t;
  JSinglePointer<JControlHost_t> datawriter;     /// controlhost to send data to the datawriter
  
  std::string       hostname;         //!< host name of data server
  int    port;                                   /// serversocket port
  int    backlog;
  int    buffer_size;

  JSinglePointer<JServerSocket> serversocket;    // input
  JChannelList_t                channelList;
  
  JTimer timer;  
  
  list<JDAQTimesliceL0>  timeslices;
  int                    current_slice_index;
    
  unsigned int         frames_per_slice;
  unsigned int         maximum_frames_per_slice;
  
  // trigger 
 
  JTriggerParameters parameters;
  string             interfaceInput;
  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;
  JMessageScheduler logErrorInvalidPMTs;

  // process management
  vector<JDAQProcess> dataFilters;
  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;
  long long int numberOfFramesWithInvalidPMTs;
  
  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;
  
  bool checkForInvalidPMTs;
  int maxHitsFrame;
};


/**
 * \file
 *
 * Application for real-time filtering of data.
 * \author rbruijn
 */
int main(int argc, char* argv[])
{
  using namespace std;
  
  string  server;
  string  logger;
  string  hostname;
  string  client_name;
  int     port;
  int     backlog;
  int     buffer_size;
  bool    use_cout;
  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)     = 8*1024*1024; // TCP buffer of 8 MB
    zap['c'] = make_field(use_cout);
    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);

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