DQMStreamerReader.cc

Go to the documentation of this file.

#include "FWCore/MessageLogger/interface/MessageLogger.h"
#include "FWCore/Utilities/interface/Exception.h"
#include "FWCore/Utilities/interface/EDMException.h"
#include "FWCore/ParameterSet/interface/ParameterSet.h"
#include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
#include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
#include "FWCore/Sources/interface/EventSkipperByID.h"
#include "FWCore/Utilities/interface/UnixSignalHandlers.h"

#include "FWCore/Utilities/interface/RegexMatch.h"
#include "DQMStreamerReader.h"

#include <fstream>
#include <queue>
#include <cstdlib>
#include <boost/regex.hpp>
#include <boost/format.hpp>
#include <boost/range.hpp>
#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include <IOPool/Streamer/interface/DumpTools.h>

namespace dqmservices {

  DQMStreamerReader::DQMStreamerReader(edm::ParameterSet const& pset, edm::InputSourceDescription const& desc)
      : StreamerInputSource(pset, desc), fiterator_(pset) {
    runNumber_ = pset.getUntrackedParameter<unsigned int>("runNumber");
    runInputDir_ = pset.getUntrackedParameter<std::string>("runInputDir");
    hltSel_ = pset.getUntrackedParameter<std::vector<std::string> >("SelectEvents");

    minEventsPerLs_ = pset.getUntrackedParameter<int>("minEventsPerLumi");
    flagSkipFirstLumis_ = pset.getUntrackedParameter<bool>("skipFirstLumis");
    flagEndOfRunKills_ = pset.getUntrackedParameter<bool>("endOfRunKills");
    flagDeleteDatFiles_ = pset.getUntrackedParameter<bool>("deleteDatFiles");

    triggerSel();

    reset_();
  }

  DQMStreamerReader::~DQMStreamerReader() {
    // Sometimes(?) the destructor called after service registry was already destructed
    // and closeFile_ throws away no ServiceRegistry found exception...
    //
    // Normally, this file should be closed before this destructor is called.
    //closeFileImp_("destructor");
  }

  void DQMStreamerReader::reset_() {
    // We have to load at least a single header,
    // so the ProductRegistry gets initialized.
    //
    // This must happen here (inside the constructor),
    // as ProductRegistry gets frozen after we initialize:
    // https://cmssdt.cern.ch/SDT/lxr/source/FWCore/Framework/src/Schedule.cc#441

    fiterator_.logFileAction("Waiting for the first lumi in order to initialize.");

    fiterator_.update_state();

    // Fast-forward to the last open file.
    if (flagSkipFirstLumis_) {
      unsigned int l = fiterator_.lastLumiFound();
      if (l > 1) {
        fiterator_.advanceToLumi(l, "skipped: fast-forward to the latest lumi");
      }
    }

    for (;;) {
      bool next = prepareNextFile();

      // check for end of run
      if (!next) {
        fiterator_.logFileAction("End of run reached before DQMStreamerReader was initialised.");
        return;
      }

      // check if we have a file openned
      if (file_.open()) {
        // we are now initialised
        break;
      }

      // wait
      fiterator_.delay();
    }

    fiterator_.logFileAction("DQMStreamerReader initialised.");
  }

  void DQMStreamerReader::openFileImp_(const DQMFileIterator::LumiEntry& entry) {
    processedEventPerLs_ = 0;
    edm::ParameterSet pset;

    std::string path = entry.get_data_path();

    file_.lumi_ = entry;
    file_.streamFile_.reset(new edm::StreamerInputFile(path));

    InitMsgView const* header = getHeaderMsg();
    deserializeAndMergeWithRegistry(*header, false);

    // dump the list of HLT trigger name from the header
    //  dumpInitHeader(header);

    // if specific trigger selection is requested, check if the requested triggers
    // match with trigger paths in the header file
    if (!acceptAllEvt_) {
      Strings tnames;
      header->hltTriggerNames(tnames);

      pset.addParameter<Strings>("SelectEvents", hltSel_);
      eventSelector_.reset(new TriggerSelector(pset, tnames));

      // check if any trigger path name requested matches with trigger name in the
      // header file
      matchTriggerSel(tnames);
    }

    // our initialization
    processedEventPerLs_ = 0;

    if (flagDeleteDatFiles_) {
      // unlink the file
      unlink(path.c_str());
    }
  }

  void DQMStreamerReader::closeFileImp_(const std::string& reason) {
    if (file_.open()) {
      file_.streamFile_->closeStreamerFile();
      file_.streamFile_ = nullptr;

      fiterator_.logLumiState(file_.lumi_, "close: " + reason);
    }
  }

  bool DQMStreamerReader::openNextFileImp_() {
    closeFileImp_("skipping to another file");

    DQMFileIterator::LumiEntry currentLumi = fiterator_.open();
    std::string p = currentLumi.get_data_path();

    if (boost::filesystem::exists(p)) {
      try {
        openFileImp_(currentLumi);
        return true;
      } catch (const cms::Exception& e) {
        fiterator_.logFileAction(std::string("Can't deserialize registry data (in open file): ") + e.what(), p);
        fiterator_.logLumiState(currentLumi, "error: data file corrupted");

        closeFileImp_("data file corrupted");
        return false;
      }
    } else {
      /* dat file missing */
      fiterator_.logFileAction("Data file (specified in json) is missing:", p);
      fiterator_.logLumiState(currentLumi, "error: data file missing");

      return false;
    }
  }

  InitMsgView const* DQMStreamerReader::getHeaderMsg() {
    InitMsgView const* header = file_.streamFile_->startMessage();

    if (header->code() != Header::INIT) {  // INIT Msg
      throw edm::Exception(edm::errors::FileReadError, "DQMStreamerReader::readHeader")
          << "received wrong message type: expected INIT, got " << header->code() << "\n";
    }

    return header;
  }

  EventMsgView const* DQMStreamerReader::getEventMsg() {
    if (!file_.streamFile_->next()) {
      return nullptr;
    }

    EventMsgView const* msg = file_.streamFile_->currentRecord();

    //  if (msg != nullptr) dumpEventView(msg);
    return msg;
  }

  bool DQMStreamerReader::prepareNextFile() {
    typedef DQMFileIterator::State State;

    for (;;) {
      fiterator_.update_state();

      if (edm::shutdown_flag.load()) {
        fiterator_.logFileAction("Shutdown flag was set, shutting down.");

        closeFileImp_("shutdown flag is set");
        return false;
      }

      // check for end of run file and force quit
      if (flagEndOfRunKills_ && (fiterator_.state() != State::OPEN)) {
        closeFileImp_("forced end-of-run");
        return false;
      }

      // check for end of run and quit if everything has been processed.
      // this clean exit
      if ((!file_.open()) && (!fiterator_.lumiReady()) && (fiterator_.state() == State::EOR)) {
        return false;
      }

      // if this is end of run and no more files to process
      // close it
      if ((processedEventPerLs_ >= minEventsPerLs_) && (!fiterator_.lumiReady()) &&
          (fiterator_.state() == State::EOR)) {
        closeFileImp_("graceful end-of-run");
        return false;
      }

      // skip to the next file if we have no files openned yet
      if (!file_.open()) {
        if (fiterator_.lumiReady()) {
          openNextFileImp_();
          // we might need to open once more (if .dat is missing)
          continue;
        }
      }

      // or if there is a next file and enough eventshas been processed.
      if (fiterator_.lumiReady() && (processedEventPerLs_ >= minEventsPerLs_)) {
        openNextFileImp_();
        // we might need to open once more (if .dat is missing)
        continue;
      }

      return true;
    }
  }

  EventMsgView const* DQMStreamerReader::prepareNextEvent() {
    EventMsgView const* eview = nullptr;
    typedef DQMFileIterator::State State;

    // wait for the next event
    for (;;) {
      // edm::LogAbsolute("DQMStreamerReader")
      //     << "State loop.";
      bool next = prepareNextFile();
      if (!next)
        return nullptr;

      // sleep
      if (!file_.open()) {
        // the reader does not exist
        fiterator_.delay();
      } else {
        // our reader exists, try to read out an event
        eview = getEventMsg();

        if (eview == nullptr) {
          // read unsuccessful
          // this means end of file, so close the file
          closeFileImp_("eof");
        } else {
          if (!acceptEvent(eview)) {
            continue;
          } else {
            return eview;
          }
        }
      }
    }
    return eview;
  }

  bool DQMStreamerReader::checkNextEvent() {
    try {
      EventMsgView const* eview = prepareNextEvent();
      if (eview == nullptr) {
        return false;
      }

      // this is reachable only if eview is set
      // and the file is openned
      if (file_.streamFile_->newHeader()) {
        // A new file has been opened and we must compare Headers here !!
        // Get header/init from reader

        InitMsgView const* header = getHeaderMsg();
        deserializeAndMergeWithRegistry(*header, true);
      }

      deserializeEvent(*eview);
    } catch (const cms::Exception& e) {
      // try to recover from corrupted files/events
      fiterator_.logFileAction(std::string("Can't deserialize event or registry data: ") + e.what());
      closeFileImp_("data file corrupted");

      // this is not optimal, but hopefully we won't catch this many times in a row
      return checkNextEvent();
    }

    processedEventPerLs_ += 1;

    return true;
  }

  bool DQMStreamerReader::triggerSel() {
    acceptAllEvt_ = false;
    for (Strings::const_iterator i(hltSel_.begin()), end(hltSel_.end()); i != end; ++i) {
      std::string hltPath(*i);
      boost::erase_all(hltPath, " \t");
      if (hltPath == "*")
        acceptAllEvt_ = true;
    }
    return acceptAllEvt_;
  }

  bool DQMStreamerReader::matchTriggerSel(Strings const& tnames) {
    matchTriggerSel_ = false;
    for (Strings::const_iterator i(hltSel_.begin()), end(hltSel_.end()); i != end; ++i) {
      std::string hltPath(*i);
      boost::erase_all(hltPath, " \t");
      std::vector<Strings::const_iterator> matches = edm::regexMatch(tnames, hltPath);
      if (!matches.empty()) {
        matchTriggerSel_ = true;
      }
    }

    if (!matchTriggerSel_) {
      edm::LogWarning("Trigger selection does not match any trigger path!!!") << std::endl;
    }

    return matchTriggerSel_;
  }

  bool DQMStreamerReader::acceptEvent(const EventMsgView* evtmsg) {
    if (acceptAllEvt_)
      return true;
    if (!matchTriggerSel_)
      return false;

    std::vector<unsigned char> hltTriggerBits_;
    int hltTriggerCount_ = evtmsg->hltCount();
    if (hltTriggerCount_ > 0) {
      hltTriggerBits_.resize(1 + (hltTriggerCount_ - 1) / 4);
    }
    evtmsg->hltTriggerBits(&hltTriggerBits_[0]);

    if (eventSelector_->wantAll() || eventSelector_->acceptEvent(&hltTriggerBits_[0], evtmsg->hltCount())) {
      return true;
    } else {
      return false;
    }
  }

  void DQMStreamerReader::skip(int toSkip) {
    try {
      for (int i = 0; i != toSkip; ++i) {
        EventMsgView const* evMsg = prepareNextEvent();

        if (evMsg == nullptr) {
          return;
        }
      }
    } catch (const cms::Exception& e) {
      // try to recover from corrupted files/events
      fiterator_.logFileAction(std::string("Can't deserialize event data: ") + e.what());
      closeFileImp_("data file corrupted");
    }
  }

  void DQMStreamerReader::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
    edm::ParameterSetDescription desc;
    desc.setComment("Reads events from streamer files.");

    desc.addUntracked<std::vector<std::string> >("SelectEvents")->setComment("HLT path to select events ");

    desc.addUntracked<int>("minEventsPerLumi", 1)
        ->setComment(
            "Minimum number of events to process per lumisection, "
            "before switching to a new input file. If the next file "
            "does not yet exist, "
            "the number of processed events will be bigger.");

    desc.addUntracked<bool>("skipFirstLumis", false)
        ->setComment(
            "Skip (and ignore the minEventsPerLumi parameter) for the files "
            "which have been available at the begining of the processing. "
            "If set to true, the reader will open last available file for "
            "processing.");

    desc.addUntracked<bool>("deleteDatFiles", false)
        ->setComment(
            "Delete data files after they have been closed, in order to "
            "save disk space.");

    desc.addUntracked<bool>("endOfRunKills", false)
        ->setComment(
            "Kill the processing as soon as the end-of-run file appears, even if "
            "there are/will be unprocessed lumisections.");

    // desc.addUntracked<unsigned int>("skipEvents", 0U)
    //    ->setComment("Skip the first 'skipEvents' events that otherwise would "
    //                 "have been processed.");

    // This next parameter is read in the base class, but its default value
    // depends on the derived class, so it is set here.
    desc.addUntracked<bool>("inputFileTransitionsEachEvent", false);

    DQMFileIterator::fillDescription(desc);
    edm::StreamerInputSource::fillDescription(desc);
    edm::EventSkipperByID::fillDescription(desc);

    descriptions.add("source", desc);
  }

}  // namespace dqmservices

#include "FWCore/Framework/interface/InputSourceMacros.h"
#include "FWCore/Framework/interface/MakerMacros.h"

typedef dqmservices::DQMStreamerReader DQMStreamerReader;
DEFINE_FWK_INPUT_SOURCE(DQMStreamerReader);