dqmservices::DQMFileIterator Class Reference

#include <DQMFileIterator.h>

Classes
struct	EorEntry
struct	LumiEntry

Public Types
enum	State { OPEN = 0, EOR_CLOSING = 1, EOR = 2 }

Public Member Functions
void	advanceToLumi (unsigned int lumi, std::string reason)
void	delay ()
	DQMFileIterator (edm::ParameterSet const &pset)
void	initialise (int run, const std::string &, const std::string &)
unsigned int	lastLumiFound ()
void	logFileAction (const std::string &msg, const std::string &fileName="") const
void	logLumiState (const LumiEntry &lumi, const std::string &msg)
bool	lumiReady ()
LumiEntry	open ()
void	pop ()
void	reset ()
unsigned int	runNumber ()
State	state ()
void	update_state ()
	~DQMFileIterator ()

Static Public Member Functions
static void	fillDescription (edm::ParameterSetDescription &d)

Private Member Functions
void	collect (bool ignoreTimers)
void	monUpdateLumi (const LumiEntry &lumi)
std::time_t	mtimeHash () const

Private Attributes
unsigned int	datafnPosition_
unsigned long	delayMillis_
EorEntry	eor_
std::unordered_set< std::string >	filesSeen_
bool	flagScanOnce_
long	forceFileCheckTimeoutMillis_
std::chrono::high_resolution_clock::time_point	lastLumiLoad_
std::map< unsigned int, LumiEntry >	lumiSeen_
edm::Service< DQMMonitoringService >	mon_
unsigned int	nextLumiNumber_
long	nextLumiTimeoutMillis_
std::string	runInputDir_
unsigned int	runNumber_
std::vector< std::string >	runPath_
std::chrono::high_resolution_clock::time_point	runPathLastCollect_
std::time_t	runPathMTime_
State	state_
std::string	streamLabel_

Detailed Description

Definition at line 18 of file DQMFileIterator.h.

Member Enumeration Documentation

enum dqmservices::DQMFileIterator::State

Enumerator
OPEN
EOR_CLOSING
EOR

Definition at line 48 of file DQMFileIterator.h.

             {
    OPEN = 0,
    EOR_CLOSING = 1,  // EoR file found, but lumis are still pending
    EOR = 2,
  };

Constructor & Destructor Documentation

dqmservices::DQMFileIterator::DQMFileIterator

(

edm::ParameterSet const &

pset

)

Definition at line 80 of file DQMFileIterator.cc.

References datafnPosition_, delayMillis_, flagScanOnce_, forceFileCheckTimeoutMillis_, edm::ParameterSet::getUntrackedParameter(), nextLumiTimeoutMillis_, reset(), runInputDir_, runNumber_, streamLabel_, and AlCaHLTBitMon_QueryRunRegistry::string.

                                                            : state_(EOR) {
  runNumber_ = pset.getUntrackedParameter<unsigned int>("runNumber");
  datafnPosition_ = pset.getUntrackedParameter<unsigned int>("datafnPosition");
  runInputDir_ = pset.getUntrackedParameter<std::string>("runInputDir");
  streamLabel_ = pset.getUntrackedParameter<std::string>("streamLabel");
  delayMillis_ = pset.getUntrackedParameter<uint32_t>("delayMillis");
  nextLumiTimeoutMillis_ =
      pset.getUntrackedParameter<int32_t>("nextLumiTimeoutMillis");

  // scan one mode
  flagScanOnce_ = pset.getUntrackedParameter<bool>("scanOnce");

  forceFileCheckTimeoutMillis_ = 5015;
  reset();
}

dqmservices::DQMFileIterator::~DQMFileIterator

(

)

Definition at line 96 of file DQMFileIterator.cc.

{}

Member Function Documentation

void dqmservices::DQMFileIterator::advanceToLumi	(	unsigned int	lumi,
		std::string	reason
	)

Definition at line 154 of file DQMFileIterator.cc.

References boostedTaus_cff::doc, lastLumiLoad_, csvLumiCalc::lumi, lumiSeen_, mon_, monUpdateLumi(), nextLumiNumber_, cmsPerfSuiteHarvest::now, and str.

Referenced by open(), dqmservices::DQMStreamerReader::reset_(), and update_state().

                                                                       {
  using boost::property_tree::ptree;
  using boost::str;

  unsigned int currentLumi = nextLumiNumber_;

  nextLumiNumber_ = lumi;
  lastLumiLoad_ = std::chrono::high_resolution_clock::now();

  auto iter = lumiSeen_.lower_bound(currentLumi);

  while ((iter != lumiSeen_.end()) && ((iter->first) < nextLumiNumber_)) {
    iter->second.state = reason;
    monUpdateLumi(iter->second);

    ++iter;
  }

  if (mon_.isAvailable()) {
    // report the successful lumi file open
    ptree doc;
    doc.put("next_lumi", nextLumiNumber_);
    mon_->outputUpdate(doc);
  }
}

void dqmservices::DQMFileIterator::collect ( bool  ignoreTimers )

private

Definition at line 202 of file DQMFileIterator.cc.

References KineDebug3::count(), datafnPosition_, MillePedeFileConverter_cfg::e, eor_, cppFunctionSkipper::exception, corrVsCorr::filename, filesSeen_, flagScanOnce_, forceFileCheckTimeoutMillis_, label, dqmservices::DQMFileIterator::LumiEntry::load_json(), dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), lumiSeen_, monUpdateLumi(), mps_check::msg, mtimeHash(), dqmservices::DQMFileIterator::EorEntry::n_lumi, cmsPerfSuiteHarvest::now, or, mps_fire::result, writedatasetfile::run, runNumber_, runPath_, runPathLastCollect_, runPathMTime_, streamLabel_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by reset(), and update_state().

                                               {
  // search filesystem to find available lumi section files
  // or the end of run files

  auto now = std::chrono::high_resolution_clock::now();
  auto last_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                     now - runPathLastCollect_).count();

  // don't refresh if it's too soon
  if ((!ignoreTimers) && (last_ms >= 0) && (last_ms < 100)) {
    return;
  }

  // check if directory changed
  std::time_t mtime_now = mtimeHash();

  if ((!ignoreTimers) && (last_ms < forceFileCheckTimeoutMillis_) &&
      (mtime_now == runPathMTime_)) {
    // logFileAction("Directory hasn't changed.");
    return;
  } else {
    // logFileAction("Directory changed, updating.");
  }

  runPathMTime_ = mtime_now;
  runPathLastCollect_ = now;

  using boost::filesystem::directory_iterator;
  using boost::filesystem::directory_entry;

  std::string fn_eor;

  for (auto runPath : runPath_) {
    if (!boost::filesystem::exists(runPath)) {
      logFileAction("Directory does not exist: ", runPath);

      continue;
    }

    directory_iterator dend;
    for (directory_iterator di(runPath); di != dend; ++di) {
      const boost::regex fn_re("run(\\d+)_ls(\\d+)_([a-zA-Z0-9]+)(_.*)?\\.jsn");

      const std::string filename = di->path().filename().string();
      const std::string fn = di->path().string();

      if (filesSeen_.find(filename) != filesSeen_.end()) {
        continue;
      }

      boost::smatch result;
      if (boost::regex_match(filename, result, fn_re)) {
        unsigned int run = std::stoi(result[1]);
        unsigned int lumi = std::stoi(result[2]);
        std::string label = result[3];

        filesSeen_.insert(filename);

        if (run != runNumber_) continue;

        // check if this is EoR
        // for various reasons we have to load it after all other files
        if ((lumi == 0) && (label == "EoR") && (!eor_.loaded)) {
          fn_eor = fn;
          continue;
        }

        // check if lumi is loaded
        if (lumiSeen_.find(lumi) != lumiSeen_.end()) {
          continue;  // already loaded
        }

        // check if this belongs to us
        if (label != streamLabel_) {
          std::string msg("Found and skipped json file (stream label mismatch, ");
          msg += label + " [files] != " + streamLabel_ + " [config]";
          msg += "): ";
          logFileAction(msg, fn);
          continue;
        }

        try {
          LumiEntry lumi_jsn = LumiEntry::load_json(runPath, fn, lumi, datafnPosition_);
          lumiSeen_.emplace(lumi, lumi_jsn);
          logFileAction("Found and loaded json file: ", fn);

          monUpdateLumi(lumi_jsn);
        } catch (const std::exception& e) {
          // don't reset the mtime, keep it waiting
          filesSeen_.erase(filename);

          std::string msg("Found, tried to load the json, but failed (");
          msg += e.what();
          msg += "): ";
          logFileAction(msg, fn);
        }
      }
    }
  }

  if ((!fn_eor.empty()) or flagScanOnce_) {
    if (!fn_eor.empty()) {
        logFileAction("EoR file found: ", fn_eor);
    }

    // @TODO load EoR files correctly
    // eor_ = EorEntry::load_json(fn_eor);
    // logFileAction("Loaded eor file: ", fn_eor);

    // for now , set n_lumi to the highest _found_ lumi
    eor_.loaded = true;

    if (lumiSeen_.empty()) {
      eor_.n_lumi = 0;
    } else {
      eor_.n_lumi = lumiSeen_.rbegin()->first;
    }
  }
}

void dqmservices::DQMFileIterator::delay

(

)

Definition at line 401 of file DQMFileIterator.cc.

References delayMillis_, and mon_.

Referenced by dqmservices::DQMProtobufReader::getNextItemType(), dqmservices::DQMStreamerReader::prepareNextEvent(), and dqmservices::DQMStreamerReader::reset_().

                            {
  if (mon_.isAvailable()) mon_->keepAlive();

  usleep(delayMillis_ * 1000);
}

void dqmservices::DQMFileIterator::fillDescription ( edm::ParameterSetDescription &  d )

static

Definition at line 407 of file DQMFileIterator.cc.

References edm::ParameterSetDescription::addUntracked(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by dqmservices::DQMProtobufReader::fillDescriptions(), and dqmservices::DQMStreamerReader::fillDescriptions().

                                                                      {
  desc.addUntracked<unsigned int>("runNumber")
      ->setComment("Run number passed via configuration file.");

  desc.addUntracked<unsigned int>("datafnPosition", 3)
      ->setComment(
          "Data filename position in the positional arguments array 'data' in "
          "json file.");

  desc.addUntracked<std::string>("streamLabel")
      ->setComment("Stream label used in json discovery.");

  desc.addUntracked<uint32_t>("delayMillis")
      ->setComment("Number of milliseconds to wait between file checks.");

  desc.addUntracked<int32_t>("nextLumiTimeoutMillis", -1)
      ->setComment(
          "Number of milliseconds to wait before switching to the next lumi "
          "section if the current is missing, -1 to disable.");

  desc.addUntracked<bool>("scanOnce", false)
      ->setComment(
          "Don't repeat file scans: use what was found during the initial scan. "
          "EOR file is ignored and the state is set to 'past end of run'.");

  desc.addUntracked<std::string>("runInputDir")
      ->setComment("Directory where the DQM files will appear.");
}

void dqmservices::DQMFileIterator::initialise	(	int	run,
		const std::string &	,
		const std::string &
	)

unsigned int dqmservices::DQMFileIterator::lastLumiFound

(

)

Definition at line 146 of file DQMFileIterator.cc.

References lumiSeen_.

Referenced by dqmservices::DQMStreamerReader::reset_().

                                            {
  if (!lumiSeen_.empty()) {
    return lumiSeen_.rbegin()->first;
  }

  return 1;
}

void dqmservices::DQMFileIterator::logFileAction	(	const std::string &	msg,
		const std::string &	fileName = ""
	)		const

Definition at line 383 of file DQMFileIterator.cc.

References MillePedeFileConverter_cfg::fileName, and edm::FlushMessageLog().

Referenced by dqmservices::DQMProtobufReader::beginLuminosityBlock(), dqmservices::DQMStreamerReader::checkNextEvent(), collect(), dqmservices::DQMProtobufReader::getNextItemType(), logLumiState(), dqmservices::DQMStreamerReader::openNextFileImp_(), dqmservices::DQMStreamerReader::prepareNextFile(), dqmservices::DQMStreamerReader::reset_(), dqmservices::DQMStreamerReader::skip(), and update_state().

                                                                     {
  edm::LogAbsolute("fileAction") << std::setprecision(0) << edm::TimeOfDay()
                                 << "  " << msg << fileName;
  edm::FlushMessageLog();
}

void dqmservices::DQMFileIterator::logLumiState	(	const LumiEntry &	lumi,
		const std::string &	msg
	)

Definition at line 390 of file DQMFileIterator.cc.

References dqmservices::DQMFileIterator::LumiEntry::file_ls, logFileAction(), lumiSeen_, monUpdateLumi(), and mps_check::msg.

Referenced by dqmservices::DQMProtobufReader::beginLuminosityBlock(), dqmservices::DQMStreamerReader::closeFileImp_(), and dqmservices::DQMStreamerReader::openNextFileImp_().

                                                         {
  if (lumiSeen_.find(lumi.file_ls) != lumiSeen_.end()) {
    lumiSeen_[lumi.file_ls].state = msg;

    monUpdateLumi(lumiSeen_[lumi.file_ls]);
  } else {
    logFileAction("Internal error: referenced lumi is not the map.");
  }
}

bool dqmservices::DQMFileIterator::lumiReady

(

)

Definition at line 136 of file DQMFileIterator.cc.

References lumiSeen_, and nextLumiNumber_.

Referenced by dqmservices::DQMProtobufReader::getNextItemType(), and dqmservices::DQMStreamerReader::prepareNextFile().

                                {
  if (lumiSeen_.find(nextLumiNumber_) != lumiSeen_.end()) {
    return true;
  }

  return false;
}

void dqmservices::DQMFileIterator::monUpdateLumi ( const LumiEntry &  lumi )

private

Definition at line 180 of file DQMFileIterator.cc.

References boostedTaus_cff::doc, dqmservices::DQMFileIterator::LumiEntry::file_ls, cmsPerfStripChart::format, mon_, dqmservices::DQMFileIterator::LumiEntry::state, and str.

Referenced by advanceToLumi(), collect(), and logLumiState().

                                                         {
  if (!mon_.isAvailable()) return;

  ptree doc;
  doc.put(str(boost::format("extra.lumi_seen.lumi%06d") % lumi.file_ls),
          lumi.state);
  mon_->outputUpdate(doc);
}

std::time_t dqmservices::DQMFileIterator::mtimeHash ( ) const

private

Definition at line 189 of file DQMFileIterator.cc.

References callgraph::path, and runPath_.

Referenced by collect().

                                           {
  std::time_t mtime_now = 0;
  
  for (auto path : runPath_) {
    if (!boost::filesystem::exists(path))
      continue;

    mtime_now = mtime_now ^ boost::filesystem::last_write_time(path);
  }

  return mtime_now;
}

DQMFileIterator::LumiEntry dqmservices::DQMFileIterator::open

(

)

Definition at line 130 of file DQMFileIterator.cc.

References advanceToLumi(), csvLumiCalc::lumi, lumiSeen_, and nextLumiNumber_.

Referenced by Vispa.Plugins.ConfigEditor.ConfigEditorTabController.ConfigEditorTabController::importConfig(), dqmservices::DQMStreamerReader::openNextFileImp_(), and dqmservices::DQMProtobufReader::readLuminosityBlockAuxiliary_().

                                               {
  LumiEntry& lumi = lumiSeen_[nextLumiNumber_];
  advanceToLumi(nextLumiNumber_ + 1, "open: file iterator");
  return lumi;
}

void dqmservices::DQMFileIterator::pop

(

)

Referenced by esMonitoring.LineHistoryEnd::write().

void dqmservices::DQMFileIterator::reset

(

void

)

Definition at line 98 of file DQMFileIterator.cc.

References collect(), boostedTaus_cff::doc, eor_, filesSeen_, cmsPerfStripChart::format, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, lumiSeen_, mon_, nextLumiNumber_, cmsPerfSuiteHarvest::now, runInputDir_, runNumber_, runPath_, split, state_, str, and update_state().

Referenced by DQMFileIterator().

                            {
  runPath_.clear();

  std::vector<std::string> tokens;
  boost::split(tokens, runInputDir_, boost::is_any_of(":"));

  for (auto token: tokens) {
    runPath_.push_back(boost::str(boost::format("%s/run%06d") % token % runNumber_));
  }

  eor_.loaded = false;
  state_ = State::OPEN;
  nextLumiNumber_ = 1;
  lumiSeen_.clear();
  filesSeen_.clear();

  lastLumiLoad_ = std::chrono::high_resolution_clock::now();

  collect(true);
  update_state();

  if (mon_.isAvailable()) {
    ptree doc;
    doc.put("run", runNumber_);
    doc.put("next_lumi", nextLumiNumber_);
    doc.put("fi_state", std::to_string(state_));
    mon_->outputUpdate(doc);
  }
}

unsigned int dqmservices::DQMFileIterator::runNumber

(

)

Definition at line 144 of file DQMFileIterator.cc.

References runNumber_.

Referenced by dqmservices::DQMProtobufReader::readLuminosityBlockAuxiliary_(), and dqmservices::DQMProtobufReader::readRunAuxiliary_().

{ return runNumber_; }

DQMFileIterator::State dqmservices::DQMFileIterator::state

(

)

Definition at line 128 of file DQMFileIterator.cc.

References state_.

Referenced by dqmservices::DQMProtobufReader::getNextItemType(), and dqmservices::DQMStreamerReader::prepareNextFile().

{ return state_; }

void dqmservices::DQMFileIterator::update_state

(

)

Definition at line 322 of file DQMFileIterator.cc.

References advanceToLumi(), collect(), KineDebug3::count(), boostedTaus_cff::doc, eor_, flagScanOnce_, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), lumiSeen_, mon_, mps_check::msg, dqmservices::DQMFileIterator::EorEntry::n_lumi, nextLumiNumber_, nextLumiTimeoutMillis_, cmsPerfSuiteHarvest::now, state_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by dqmservices::DQMProtobufReader::getNextItemType(), dqmservices::DQMStreamerReader::prepareNextFile(), reset(), and dqmservices::DQMStreamerReader::reset_().

                                   {
  using std::chrono::high_resolution_clock;
  using std::chrono::duration_cast;
  using std::chrono::milliseconds;

  State old_state = state_;

  // in scanOnce mode we don't do repeated scans
  // whatever found at reset() is be used
  if (!flagScanOnce_) {
    collect(false);
  }

  if ((state_ == State::OPEN) && (eor_.loaded)) {
    state_ = State::EOR_CLOSING;
  }

  // special case for missing lumi files
  // skip to the next available, but after the timeout
  if ((state_ != State::EOR) && (nextLumiTimeoutMillis_ >= 0)) {
    auto iter = lumiSeen_.lower_bound(nextLumiNumber_);
    if ((iter != lumiSeen_.end()) && iter->first != nextLumiNumber_) {
      auto elapsed = high_resolution_clock::now() - lastLumiLoad_;
      auto elapsed_ms = duration_cast<milliseconds>(elapsed).count();

      if (elapsed_ms >= nextLumiTimeoutMillis_) {
        std::string msg("Timeout reached, skipping lumisection(s) ");
        msg += std::to_string(nextLumiNumber_) + " .. " +
               std::to_string(iter->first - 1);
        msg += ", nextLumiNumber_ is now " + std::to_string(iter->first);
        logFileAction(msg);

        advanceToLumi(iter->first, "skipped: timeout");
      }
    }
  }

  if (state_ == State::EOR_CLOSING) {
    // check if we parsed all lumis
    // n_lumi is both last lumi and the number of lumi
    // since lumis are indexed from 1

    // after all lumi have been pop()'ed
    // current lumi will become larger than the last lumi
    if (nextLumiNumber_ > eor_.n_lumi) {
      state_ = State::EOR;
    }
  }

  if (state_ != old_state) {
    logFileAction("Streamer state changed: ",
                  std::to_string(old_state) + "->" + std::to_string(state_));

    if (mon_) {
      ptree doc;
      doc.put("fi_state", std::to_string(state_));
      mon_->outputUpdate(doc);
    }
  }
}

Member Data Documentation

unsigned int dqmservices::DQMFileIterator::datafnPosition_

private

Definition at line 101 of file DQMFileIterator.h.

Referenced by collect(), and DQMFileIterator().

unsigned long dqmservices::DQMFileIterator::delayMillis_

private

Definition at line 95 of file DQMFileIterator.h.

Referenced by delay(), and DQMFileIterator().

EorEntry dqmservices::DQMFileIterator::eor_

private

Definition at line 104 of file DQMFileIterator.h.

Referenced by collect(), reset(), and update_state().

std::unordered_set<std::string> dqmservices::DQMFileIterator::filesSeen_

private

Definition at line 109 of file DQMFileIterator.h.

Referenced by collect(), and reset().

bool dqmservices::DQMFileIterator::flagScanOnce_

private

Definition at line 98 of file DQMFileIterator.h.

Referenced by collect(), DQMFileIterator(), and update_state().

long dqmservices::DQMFileIterator::forceFileCheckTimeoutMillis_

private

Definition at line 97 of file DQMFileIterator.h.

Referenced by collect(), and DQMFileIterator().

std::chrono::high_resolution_clock::time_point dqmservices::DQMFileIterator::lastLumiLoad_

private

Definition at line 117 of file DQMFileIterator.h.

Referenced by advanceToLumi(), reset(), and update_state().

std::map<unsigned int, LumiEntry> dqmservices::DQMFileIterator::lumiSeen_

private

Definition at line 108 of file DQMFileIterator.h.

Referenced by advanceToLumi(), collect(), lastLumiFound(), logLumiState(), lumiReady(), open(), reset(), and update_state().

edm::Service<DQMMonitoringService> dqmservices::DQMFileIterator::mon_

private

Definition at line 124 of file DQMFileIterator.h.

Referenced by advanceToLumi(), delay(), monUpdateLumi(), reset(), and update_state().

unsigned int dqmservices::DQMFileIterator::nextLumiNumber_

private

Definition at line 107 of file DQMFileIterator.h.

Referenced by advanceToLumi(), lumiReady(), open(), reset(), and update_state().

long dqmservices::DQMFileIterator::nextLumiTimeoutMillis_

private

Definition at line 96 of file DQMFileIterator.h.

Referenced by DQMFileIterator(), and update_state().

std::string dqmservices::DQMFileIterator::runInputDir_

private

Definition at line 93 of file DQMFileIterator.h.

Referenced by DQMFileIterator(), and reset().

unsigned int dqmservices::DQMFileIterator::runNumber_

private

Definition at line 92 of file DQMFileIterator.h.

Referenced by collect(), DQMFileIterator(), reset(), and runNumber().

std::vector<std::string> dqmservices::DQMFileIterator::runPath_

private

Definition at line 102 of file DQMFileIterator.h.

Referenced by collect(), mtimeHash(), and reset().

std::chrono::high_resolution_clock::time_point dqmservices::DQMFileIterator::runPathLastCollect_

private

Definition at line 114 of file DQMFileIterator.h.

Referenced by collect().

std::time_t dqmservices::DQMFileIterator::runPathMTime_

private

Definition at line 113 of file DQMFileIterator.h.

Referenced by collect().

State dqmservices::DQMFileIterator::state_

private

Definition at line 105 of file DQMFileIterator.h.

Referenced by reset(), state(), and update_state().

std::string dqmservices::DQMFileIterator::streamLabel_

private

Definition at line 94 of file DQMFileIterator.h.

Referenced by collect(), and DQMFileIterator().