JDataWriter.cc

Go to the documentation of this file.

#include <string>
#include <iostream>
#include <sstream>
#include <iomanip>
#include <map>

#include "Jeep/JParser.hh"
#include "Jeep/JProperties.hh"
#include "Jeep/JTimer.hh"
#include "Jeep/JTimekeeper.hh"
#include "Jeep/JPrint.hh"
#include "Jeep/JeepToolkit.hh"
#include "JNet/JControlHost.hh"
#include "JLang/JException.hh"
#include "JLang/JSharedPointer.hh"
#include "JLogger/JMessageScheduler.hh"
#include "km3net-dataformat/online/JDAQ.hh"
#include "JDAQ/JDAQTags.hh"
#include "JDAQ/JDAQEventIO.hh"
#include "JDAQ/JDAQTimesliceIO.hh"
#include "JDAQ/JDAQSummarysliceIO.hh"
#include "JTrigger/JTriggerParameters.hh"
#include "JRuncontrol/JDAQClient.hh"
#include "JRuncontrol/JRuncontrolToolkit.hh"
#include "JIO/JByteArrayIO.hh"
#include "JTools/JAutoMap.hh"
#include "JSupport/JMeta.hh"
#include "JSupport/JSupport.hh"
#include "JSupport/JAutoTreeWriter.hh"


/**
 * Type definition of auto map.
 */
typedef JSUPPORT::JAutoTreeWriter<JNET::JTag> JTreeWriter_t;


namespace JSUPPORT {

  /**
   * Get key for given DAQ data type.
   *
   * \param  type             data type
   * \return                  map element
   */
  template<>
  template<class T>
  JNET::JTag JAutoTreeWriter<JNET::JTag>::getKey(JLANG::JType<T> type)
  {
    return getTag<T>();
  }
}


namespace KM3NETDAQ {

  /**
   * Runcontrol client to write data to disk.
   * In state running, this application will write ROOT formatted data from the data filters to disk.
   */
  class JDataWriter :
    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  path                 default path
     */
    JDataWriter(const std::string& name,
                const std::string& server,
                const std::string& hostname,
                JLogger*           logger,
                const int          level,
                const std::string& path) :
      JDAQClient(name, server, logger, level),
      datawriter(),
      path      (path),
      hostname  (hostname)
    {
      replaceEvent(RC_CMD, RC_DWRITER, ev_configure);

      JControlHost::Throw(true);

      // map ControlHost tag to TTree writer.

      writer.insert<JDAQTypes_t>();
    }


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

      // start server

      try {

        datawriter.reset(new JControlHost(hostname));

        datawriter->setReceiveBufferSize(DWRITER_RECEIVE_BUFFER_SIZE);

        datawriter->MyId(getFullName());

        JSubscriptionList buffer;

        for (JTreeWriter_t::iterator i = writer.begin(); i != writer.end(); ++i) {
          buffer.add(JSubscriptionAll(i->first));
        }
        
        buffer.add(JSubscriptionAll(IO_TRIGGER_PARAMETERS));

        datawriter->Subscribe(buffer);
        datawriter->SendMeAlways();

        JNoticeStream(logger) << "Established connection to " << hostname;
      }
      catch(const std::exception& exception) {
        JErrorStream(logger) << exception.what();
      }
    }


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

      long long int update_s = 10;
      long long int logger_s =  5;

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

      properties["path"]     = path;
      properties["update_s"] = update_s;
      properties["logger_s"] = logger_s;
      
      properties.read(string(buffer, length));
      
      if (update_s <=  0) { update_s = 1; }
      if (logger_s <=  0) { logger_s = 1; }

      setClockInterval(update_s * 1000000LL);

      JDebugStream(logger) << "Path <" << path << ">";
      JDebugStream(logger) << "Update period [s] " << update_s;

      logErrorRun   = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));
      logErrorFile  = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));
      logErrorTag   = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));
      logErrorState = JMessageScheduler(logger, JTimekeeper(logger_s * 1000000LL));

      numberOfEvents = 0;
      numberOfBytes  = 0;
    }


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


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


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

      JStatusStream(logger) << "Start run " << getDetectorID() << ' ' << getRunNumber();

      if (writer.is_open()) {

        JErrorStream (logger) << "Previous file still open -> close";

        writer.close();
      }

      ostringstream os;

      for (int i = 0; !writer.is_open() && i != MAXIMUM_FILE_NUMBER; ++i) {

        os.str("");

        os << getFullPath(path) 
           << "KM3NeT" 
           << "_"  << FILL(8,'0') << getDetectorID() 
           << "_"  << FILL(8,'0') << getRunNumber();

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

        os << ".root";

        try {
          writer.open(os.str().c_str());
        }
        catch(const std::exception& exception) {
          JErrorStream(logger) << exception.what();
        }
      }

      if (writer.is_open())
        JNoticeStream(logger) << "Output file "     << os.str();
      else
        JErrorStream (logger) << "File not opened " << os.str();

      numberOfEvents = 0;
      numberOfBytes  = 0;

      putObject(writer.getFile(), meta);

      timer.reset();

      logErrorRun  .reset();
      logErrorFile .reset();
      logErrorTag  .reset();
      logErrorState.reset();

      run_db.reset(getRunNumber());
    }


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

      if (timer.usec_wall > 0) {
        JStatusStream(logger) << "I/O "  << (int) (numberOfBytes / timer.usec_wall) << " MB/s";
      }

      if (!run_db.is_written(getRunNumber())) {
        JErrorStream(logger) << "No trigger parameters written for run " << getRunNumber();
      }
        
      writer.close();

      // Release resources.
      
      std::vector<char> null;

      this->buffer.swap(null);
    }


    virtual void setSelect(JFileDescriptorMask& mask) const override
    {
      if (datawriter.is_valid()) {
        mask.set(*datawriter);
      }
    }


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

      if (datawriter.is_valid() && mask.has(*datawriter)) {
              
        try {
          
          JPrefix prefix;
          
          datawriter->WaitHead(prefix);
          
          timer.start();
          
          buffer.resize(prefix.getSize());
          
          datawriter->GetFullData(buffer.data(), buffer.size());

          
          if (prefix.getTag() == IO_TRIGGER_PARAMETERS) {

            try {
              run_db.read(buffer.data(), buffer.size());
            }
            catch(const std::exception& error) {

              JErrorStream(logger) << "Fatal error reading trigger parameters \"" << error.what() << "\"; trigger ev_error.";

              ev_error();
            }
          }

          
          if (isRunning()) {

            // Write trigger parameters for current run if not yet done
            
            run_db.write(getRunNumber(), writer.getFile());

            JTreeWriter_t::iterator i = writer.find(prefix.toString());
            
            if (i != writer.end()) {
              
              TFile* out = i->second->GetCurrentFile();
              
              if (out != NULL && out->IsOpen()) {
                
                JDAQPreamble preamble;
                Version_t    version;
                JDAQHeader   header;
              
                JByteArrayReader in(buffer.data(), buffer.size());
                
                in >> preamble >> version >> header;
                
                in.seekg(0);   // rewind
                
                if (header.getRunNumber() == getRunNumber()) {
                  
                  const Int_t nb = i->second->copy(in);

                  if (nb < (int) buffer.size() || in.tellg() != (int) buffer.size()) {
                    JWarningStream(logger) << "Inconsistency at copy of " 
                                           << prefix.toString() << ' ' 
                                           << buffer.size()     << ' ' 
                                           << in.tellg()        << ' ' 
                                           << nb;
                  }

                  if (prefix.getTag() == IO_EVENT)
                    numberOfEvents += 1;
                  numberOfBytes    += buffer.size();
                
                  if (prefix.getTag() == IO_EVENT && numberOfEvents == 1) {
                    typeout();
                  }
                  
                } else {
                  JErrorStream(logErrorRun) << "Inconsistent run number " 
                                            << header.getRunNumber() 
                                            << " != " 
                                            << getRunNumber();
                }
              } else {
                JErrorStream(logErrorFile) << "Output file not open";
              }
            } else {
              if (prefix.getTag() != IO_TRIGGER_PARAMETERS) {
                JErrorStream(logErrorTag) << "Unknown tag <" << prefix.toString() << ">, no data written";
              }
            }       
          } else {
            JWarningStream(logErrorState) << "Not in running state <" << prefix.toString() << ">, no data written";
          }

          timer.stop();
        }
        catch(const std::exception& error) {

          JErrorStream(logger) << "Fatal error \"" << error.what() << "\"; trigger ev_error.";

          ev_error();
        }
      }
    }


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


    /**
     * Report status of data writing.
     */
    void typeout()
    {
      std::ostringstream message;

      message << getFullName()  << KM3NETDAQ::getTokenDelimeter() << numberOfBytes  << ' ' << numberOfEvents;
      
      logger.typeout(RC_LOG, message.str());
      logger.status(message.str());
    }

    JMeta meta;                                 //!< meta data

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

  private:

    JLANG::JSharedPointer<JControlHost> datawriter;
    std::string   path;                 // directory for output file

    JEEP::JTimer  timer;                // timer for I/O measurement
    Long64_t      numberOfEvents;       // total number of events
    long long int numberOfBytes;        // total number of bytes

    JMessageScheduler logErrorRun;
    JMessageScheduler logErrorFile;
    JMessageScheduler logErrorTag;
    JMessageScheduler logErrorState;

    std::string       hostname;         //!< host name of data server
    JTreeWriter_t     writer;           //!< TTree writer
    std::vector<char> buffer;           //!< internal buffer for incoming data

    
    /**
     * Auxiliary data structure for I/O of trigger parameters.
     */
    struct JValue_t {
      /**
       * Default constructor.
       */
      JValue_t() :
        count(0),
        is_written(false)
      {}

      JTriggerParameters parameters;    //!< trigger parameters
      int                count;         //!< reader count
      bool               is_written;    //!< writer status
    };

    
    /**
     * Map run number to trigger parameters.
     */
    struct JRunDB :
      public std::map<int, JValue_t>
    {
      /**
       * Remove all entries before given run.
       *
       * \param  run       run number
       */
      inline void reset(const int run)
      {
        while (!this->empty() && this->begin()->first < run) {
          this->erase(this->begin());
        }
      }

      /**
       * Check if trigger parameters have been written for given run.
       *
       * \param  run       run number
       * \return           true if written; else false.
       */
      inline bool is_written(const int run) const
      {
        const_iterator p = this->find(run);

        return p != this->end() && p->second.is_written;
      }

      /**
       * Read trigger parameters.
       *
       * \param  data      data
       * \param  size      size
       */
      void read(const char* const data, const size_t size)
      {
        using namespace std;
        using namespace JPP;


        const string buffer(data, size);

        istringstream in(buffer);

        int                run = -1;
        JTriggerParameters parameters;
        
        in >> run;

        if (!in) {
          THROW(JIOException, "Error reading run number for trigger parameters " << run << endl << in.rdbuf());
        }
        
        in >> parameters; 

        in.clear(std::ios::eofbit);

        if (!in) {
          THROW(JIOException, "Error reading trigger parameters " << in.rdbuf());
        }

        JValue_t& value = (*this)[run];
          
        if (value.count == 0) {
          value.parameters = parameters;
        }
          
        value.count += 1;
            
        if (!parameters.equals(value.parameters)) {
          THROW(JException, "Inconsistent trigger parameters " << endl << value.parameters << " != " << endl << parameters);
        }
      }

      /**
       * Write trigger parameters for given run if not yet done.
       *
       * \param  run       run number
       * \param  file      pointer to ROOT file
       */
      inline void write(const int run, TFile* file)
      {
        if (file != NULL) {
          
          iterator p = this->find(run);

          if (p != this->end() && p->second.count != 0 && !p->second.is_written) {

            file->WriteTObject(&p->second.parameters);

            p->second.is_written = true;
          }
        }
      }
    };

    JRunDB run_db;
  };
}


/**
 * \file
 *
 * Application for writing real-time data to disk.
 * \author 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;
  bool    use_cout;
  string  path;
  int     debug;

  try {

    JParser<> zap("Application for writing real-time data to disk.");

    zap['H'] = make_field(server,      "host name of server for command messages")               = "localhost";
    zap['M'] = make_field(logger,      "host name of server for logger messages")                = "localhost";
    zap['D'] = make_field(hostname,    "host name of server for incoming data from data filter") = "localhost";
    zap['u'] = make_field(client_name, "client name")                                            = "%";
    zap['c'] = make_field(use_cout,    "print to terminal");
    zap['p'] = make_field(path,        "directory for permanent archival of data")               = "";
    zap['d'] = make_field(debug,       "debug level")                                            =      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);

  JDataWriter dwriter(getProcessName(client_name, argv[0]), server, hostname, out, debug, path);

  dwriter.meta = JMeta(argc, argv);

  dwriter.enter();
  dwriter.run();
}