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 () const
State	state () const
void	update_state ()
	~DQMFileIterator ()=default

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

Private Member Functions
void	collect (bool ignoreTimers)
void	monUpdateLumi (const LumiEntry &lumi)
unsigned	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_
unsigned	runPathMTime_
State	state_
std::string	streamLabel_

Detailed Description

Definition at line 21 of file DQMFileIterator.h.

Member Enumeration Documentation

State

enum dqmservices::DQMFileIterator::State

Enumerator
OPEN
EOR_CLOSING
EOR

Definition at line 53 of file DQMFileIterator.h.

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

Constructor & Destructor Documentation

DQMFileIterator()

dqmservices::DQMFileIterator::DQMFileIterator

(

edm::ParameterSet const &

pset

)

Definition at line 75 of file DQMFileIterator.cc.

References datafnPosition_, delayMillis_, flagScanOnce_, forceFileCheckTimeoutMillis_, nextLumiTimeoutMillis_, muonDTDigis_cfi::pset, 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();
  }

~DQMFileIterator()

dqmservices::DQMFileIterator::~DQMFileIterator ( )

default

Member Function Documentation

advanceToLumi()

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

Definition at line 142 of file DQMFileIterator.cc.

References boostedTaus_cff::doc, lastLumiLoad_, BXlumiParameters_cfi::lumi, lumiSeen_, mon_, monUpdateLumi(), nextLumiNumber_, submitPVValidationJobs::now, and PixelMapPlotter::reason.

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

                                                                         {
    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
      boost::property_tree::ptree doc;
      doc.put("next_lumi", nextLumiNumber_);
      mon_->outputUpdate(doc);
    }
  }

collect()

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

private

Definition at line 189 of file DQMFileIterator.cc.

References submitPVResolutionJobs::count, datafnPosition_, MillePedeFileConverter_cfg::e, eor_, cppFunctionSkipper::exception, corrVsCorr::filename, filesSeen_, flagScanOnce_, personalPlayback::fn, forceFileCheckTimeoutMillis_, label, dqmservices::DQMFileIterator::LumiEntry::load_json(), dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), lumiSeen_, monUpdateLumi(), mps_check::msg, mtimeHash(), dqmservices::DQMFileIterator::EorEntry::n_lumi, submitPVValidationJobs::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
    auto 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 std::filesystem::directory_entry;
    using std::filesystem::directory_iterator;

    std::string fn_eor;

    for (const auto& runPath : runPath_) {
      if (!std::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;
      }
    }
  }

delay()

void dqmservices::DQMFileIterator::delay

(

)

Definition at line 382 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);
  }

fillDescription()

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

static

Definition at line 389 of file DQMFileIterator.cc.

References submitPVResolutionJobs::desc, 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.");
  }

initialise()

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

lastLumiFound()

unsigned int dqmservices::DQMFileIterator::lastLumiFound

(

)

Definition at line 134 of file DQMFileIterator.cc.

References lumiSeen_.

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

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

    return 1;
  }

logFileAction()

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

Definition at line 367 of file DQMFileIterator.cc.

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

Referenced by dqmservices::DQMProtobufReader::beginLuminosityBlock(), dqmservices::DQMStreamerReader::checkNext(), 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();
  }

logLumiState()

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

Definition at line 372 of file DQMFileIterator.cc.

References 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.");
    }
  }

lumiReady()

bool dqmservices::DQMFileIterator::lumiReady

(

)

Definition at line 126 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;
  }

monUpdateLumi()

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

private

Definition at line 165 of file DQMFileIterator.cc.

References boostedTaus_cff::doc, and mon_.

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

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

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

mtimeHash()

unsigned dqmservices::DQMFileIterator::mtimeHash ( ) const

private

Definition at line 174 of file DQMFileIterator.cc.

References submitPVResolutionJobs::count, castor_dqm_sourceclient_file_cfg::path, and runPath_.

Referenced by collect().

                                            {
    unsigned mtime_now = 0;

    for (const auto& path : runPath_) {
      if (!std::filesystem::exists(path))
        continue;

      auto write_time = std::filesystem::last_write_time(path);
      mtime_now =
          mtime_now ^ std::chrono::duration_cast<std::chrono::microseconds>(write_time.time_since_epoch()).count();
    }

    return mtime_now;
  }

open()

DQMFileIterator::LumiEntry dqmservices::DQMFileIterator::open

(

)

Definition at line 120 of file DQMFileIterator.cc.

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

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

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

pop()

void dqmservices::DQMFileIterator::pop

(

)

Referenced by SequenceTypes.Schedule::_replaceIfHeldDirectly(), and esMonitoring.LineHistoryEnd::write().

reset()

void dqmservices::DQMFileIterator::reset

(

void

)

Definition at line 90 of file DQMFileIterator.cc.

References collect(), boostedTaus_cff::doc, eor_, filesSeen_, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, lumiSeen_, mon_, nextLumiNumber_, submitPVValidationJobs::now, runInputDir_, runNumber_, runPath_, submitPVValidationJobs::split(), state_, cond::impl::to_string(), unpackBuffers-CaloStage2::token, and update_state().

Referenced by DQMFileIterator().

                              {
    runPath_.clear();

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

    for (const auto& token : tokens) {
      runPath_.push_back(fmt::sprintf("%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()) {
      boost::property_tree::ptree doc;
      doc.put("run", runNumber_);
      doc.put("next_lumi", nextLumiNumber_);
      doc.put("fi_state", std::to_string(state_));
      mon_->outputUpdate(doc);
    }
  }

runNumber()

unsigned int dqmservices::DQMFileIterator::runNumber ( ) const

inline

Definition at line 89 of file DQMFileIterator.h.

References runNumber_.

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

{ return runNumber_; }

state()

State dqmservices::DQMFileIterator::state ( ) const

inline

Definition at line 63 of file DQMFileIterator.h.

References state_.

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

{ return state_; }

update_state()

void dqmservices::DQMFileIterator::update_state

(

)

Definition at line 308 of file DQMFileIterator.cc.

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

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

                                     {
    using std::chrono::duration_cast;
    using std::chrono::high_resolution_clock;
    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_) {
        boost::property_tree::ptree doc;
        doc.put("fi_state", std::to_string(state_));
        mon_->outputUpdate(doc);
      }
    }
  }

Member Data Documentation

datafnPosition_

unsigned int dqmservices::DQMFileIterator::datafnPosition_

private

Definition at line 105 of file DQMFileIterator.h.

Referenced by collect(), and DQMFileIterator().

delayMillis_

unsigned long dqmservices::DQMFileIterator::delayMillis_

private

Definition at line 99 of file DQMFileIterator.h.

Referenced by delay(), and DQMFileIterator().

eor_

EorEntry dqmservices::DQMFileIterator::eor_

private

Definition at line 108 of file DQMFileIterator.h.

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

filesSeen_

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

private

Definition at line 113 of file DQMFileIterator.h.

Referenced by collect(), and reset().

flagScanOnce_

bool dqmservices::DQMFileIterator::flagScanOnce_

private

Definition at line 102 of file DQMFileIterator.h.

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

forceFileCheckTimeoutMillis_

long dqmservices::DQMFileIterator::forceFileCheckTimeoutMillis_

private

Definition at line 101 of file DQMFileIterator.h.

Referenced by collect(), and DQMFileIterator().

lastLumiLoad_

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

private

Definition at line 121 of file DQMFileIterator.h.

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

lumiSeen_

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

private

Definition at line 112 of file DQMFileIterator.h.

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

mon_

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

private

Definition at line 128 of file DQMFileIterator.h.

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

nextLumiNumber_

unsigned int dqmservices::DQMFileIterator::nextLumiNumber_

private

Definition at line 111 of file DQMFileIterator.h.

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

nextLumiTimeoutMillis_

long dqmservices::DQMFileIterator::nextLumiTimeoutMillis_

private

Definition at line 100 of file DQMFileIterator.h.

Referenced by DQMFileIterator(), and update_state().

runInputDir_

std::string dqmservices::DQMFileIterator::runInputDir_

private

Definition at line 97 of file DQMFileIterator.h.

Referenced by DQMFileIterator(), and reset().

runNumber_

unsigned int dqmservices::DQMFileIterator::runNumber_

private

Definition at line 96 of file DQMFileIterator.h.

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

runPath_

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

private

Definition at line 106 of file DQMFileIterator.h.

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

runPathLastCollect_

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

private

Definition at line 118 of file DQMFileIterator.h.

Referenced by collect().

runPathMTime_

unsigned dqmservices::DQMFileIterator::runPathMTime_

private

Definition at line 117 of file DQMFileIterator.h.

Referenced by collect().

state_

State dqmservices::DQMFileIterator::state_

private

Definition at line 109 of file DQMFileIterator.h.

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

streamLabel_

std::string dqmservices::DQMFileIterator::streamLabel_

private

Definition at line 98 of file DQMFileIterator.h.

Referenced by collect(), and DQMFileIterator().