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

96 {}

Member Function Documentation

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

Definition at line 154 of file DQMFileIterator.cc.

References lastLumiLoad_, fjr2json::lumi, lumiSeen_, mon_, monUpdateLumi(), nextLumiNumber_, and fileCollector::now.

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_, alignCSCRings::e, eor_, cppFunctionSkipper::exception, lut2db_cfg::filename, filesSeen_, flagScanOnce_, personalPlayback::fn, forceFileCheckTimeoutMillis_, diffTwoXMLs::label, dqmservices::DQMFileIterator::LumiEntry::load_json(), dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), fjr2json::lumi, lumiSeen_, monUpdateLumi(), visualization-live-secondInstance_cfg::msg, mtimeHash(), dqmservices::DQMFileIterator::EorEntry::n_lumi, fileCollector::now, or, mps_fire::result, DTTTrigCorrFirst::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(fn);
 
         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(), logLumiState(), dqmservices::DQMStreamerReader::openNextFile_(), 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 visualization-live-secondInstance_cfg::msg.

Referenced by dqmservices::DQMProtobufReader::beginLuminosityBlock(), dqmservices::DQMStreamerReader::closeFile_(), and dqmservices::DQMStreamerReader::openNextFile_().

                                                          {
   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 dqmservices::DQMFileIterator::LumiEntry::file_ls, cmsPerfStripChart::format, mon_, and dqmservices::DQMFileIterator::LumiEntry::state.

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 fed_dqm_sourceclient-live_cfg::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(), fjr2json::lumi, lumiSeen_, and nextLumiNumber_.

Referenced by Vispa.Plugins.ConfigEditor.ConfigEditorTabController.ConfigEditorTabController::importConfig(), dqmservices::DQMStreamerReader::openNextFile_(), 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(), eor_, filesSeen_, cmsPerfStripChart::format, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, lumiSeen_, mon_, nextLumiNumber_, fileCollector::now, runInputDir_, runNumber_, runPath_, 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 (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_().

144 { return runNumber_; }

dqmservices::DQMFileIterator::runNumber_

unsigned int runNumber_

Definition: DQMFileIterator.h:92

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

Definition at line 128 of file DQMFileIterator.cc.

References state_.

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

128 { return state_; }

dqmservices::DQMFileIterator::state_

State state_

Definition: DQMFileIterator.h:105

void dqmservices::DQMFileIterator::update_state ( )

Definition at line 322 of file DQMFileIterator.cc.

References advanceToLumi(), collect(), KineDebug3::count(), eor_, flagScanOnce_, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), lumiSeen_, mon_, visualization-live-secondInstance_cfg::msg, dqmservices::DQMFileIterator::EorEntry::n_lumi, nextLumiNumber_, nextLumiTimeoutMillis_, fileCollector::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);
     }
   }
 }