#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)

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 18 of file DQMFileIterator.h.

Member Enumeration Documentation

enum dqmservices::DQMFileIterator::State

Enumerator
OPEN
EOR_CLOSING
EOR

Definition at line 50 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 74 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 89 of file DQMFileIterator.cc.

89 {}

Member Function Documentation

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

Definition at line 147 of file DQMFileIterator.cc.

References lastLumiLoad_, dqmdumpme::lumi, lumiSeen_, mon_, monUpdateLumi(), nextLumiNumber_, submitPVValidationJobs::now, PixelMapPlotter::reason, and str.

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

                                                                          {
     using boost::str;
     using boost::property_tree::ptree;
 
     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 197 of file DQMFileIterator.cc.

References submitPVResolutionJobs::count, datafnPosition_, alignCSCRings::e, eor_, cppFunctionSkipper::exception, lut2db_cfg::filename, filesSeen_, flagScanOnce_, personalPlayback::fn, forceFileCheckTimeoutMillis_, label, dqmservices::DQMFileIterator::LumiEntry::load_json(), dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), dqmdumpme::lumi, lumiSeen_, monUpdateLumi(), mps_check::msg, mtimeHash(), dqmservices::DQMFileIterator::EorEntry::n_lumi, submitPVValidationJobs::now, or, mps_fire::result, submitPVValidationJobs::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;
       }
     }
   }

void dqmservices::DQMFileIterator::delay ( )

Definition at line 390 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 397 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 139 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 375 of file DQMFileIterator.cc.

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

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

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

Definition at line 380 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 129 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 173 of file DQMFileIterator.cc.

References dqmservices::DQMFileIterator::LumiEntry::file_ls, mon_, and dqmservices::DQMFileIterator::LumiEntry::state.

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

                                                            {
     if (!mon_.isAvailable())
       return;
 
     ptree doc;
     doc.put(fmt::sprintf("extra.lumi_seen.lumi%06d", lumi.file_ls), lumi.state);
     mon_->outputUpdate(doc);
   }

unsigned dqmservices::DQMFileIterator::mtimeHash ( ) const

private

Definition at line 182 of file DQMFileIterator.cc.

References submitPVResolutionJobs::count, fed_dqm_sourceclient-live_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;
   }

DQMFileIterator::LumiEntry dqmservices::DQMFileIterator::open ( )

Definition at line 123 of file DQMFileIterator.cc.

References advanceToLumi(), dqmdumpme::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;
   }

void dqmservices::DQMFileIterator::pop ( )

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

void dqmservices::DQMFileIterator::reset ( void )

Definition at line 91 of file DQMFileIterator.cc.

References collect(), eor_, filesSeen_, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, lumiSeen_, mon_, nextLumiNumber_, submitPVValidationJobs::now, runInputDir_, runNumber_, runPath_, submitPVValidationJobs::split(), state_, 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()) {
       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 137 of file DQMFileIterator.cc.

References runNumber_.

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

137 { return runNumber_; }

dqmservices::DQMFileIterator::runNumber_

unsigned int runNumber_

Definition: DQMFileIterator.h:93

DQMFileIterator::State dqmservices::DQMFileIterator::state ( )

Definition at line 121 of file DQMFileIterator.cc.

References state_.

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

121 { return state_; }

dqmservices::DQMFileIterator::state_

State state_

Definition: DQMFileIterator.h:106

void dqmservices::DQMFileIterator::update_state ( )

Definition at line 316 of file DQMFileIterator.cc.

References advanceToLumi(), collect(), submitPVResolutionJobs::count, eor_, flagScanOnce_, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), lumiSeen_, mon_, mps_check::msg, dqmservices::DQMFileIterator::EorEntry::n_lumi, nextLumiNumber_, nextLumiTimeoutMillis_, submitPVValidationJobs::now, state_, and AlCaHLTBitMon_QueryRunRegistry::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_) {
         ptree doc;
         doc.put("fi_state", std::to_string(state_));
         mon_->outputUpdate(doc);
       }
     }
   }