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

std::string	make_path (const std::string &fn)

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)

Private Attributes
unsigned int	datafnPosition_

unsigned long	delayMillis_

EorEntry	eor_

std::unordered_set< std::string >	filesSeen_

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::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 43 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 58 of file DQMFileIterator.cc.

References datafnPosition_, delayMillis_, 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");
 
   forceFileCheckTimeoutMillis_ = 5015;
   reset();
 }

dqmservices::DQMFileIterator::~DQMFileIterator ( )

Definition at line 73 of file DQMFileIterator.cc.

73 {}

Member Function Documentation

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

Definition at line 125 of file DQMFileIterator.cc.

References asciidump::doc, getDQMSummary::iter, 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 168 of file DQMFileIterator.cc.

References prof2calltree::count, datafnPosition_, alignCSCRings::e, eor_, cppFunctionSkipper::exception, lut2db_cfg::filename, filesSeen_, forceFileCheckTimeoutMillis_, diffTwoXMLs::label, dqmservices::DQMFileIterator::LumiEntry::load_json(), dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), fjr2json::lumi, lumiSeen_, monUpdateLumi(), producerFileCleanner::msg, dqmservices::DQMFileIterator::EorEntry::n_lumi, fileCollector::now, query::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 = boost::filesystem::last_write_time(runPath_);
 
   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;
 
   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(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()) {
     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 353 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 360 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<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 117 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 336 of file DQMFileIterator.cc.

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

Referenced by collect(), logLumiState(), dqmservices::DQMStreamerReader::openNextFile_(), dqmservices::DQMProtobufReader::readLuminosityBlock_(), dqmservices::DQMStreamerReader::reset_(), 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 343 of file DQMFileIterator.cc.

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

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

                                                                               {
   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 107 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;
 }

std::string dqmservices::DQMFileIterator::make_path ( const std::string & fn )

Definition at line 160 of file DQMFileIterator.cc.

References AlCaHLTBitMon_ParallelJobs::p, cmsHarvester::path, and runPath_.

Referenced by dqmservices::DQMStreamerReader::openFile_(), dqmservices::DQMStreamerReader::openNextFile_(), and dqmservices::DQMProtobufReader::readLuminosityBlock_().

                                                         {
   if (boost::starts_with(fn, "/")) return fn;
 
   boost::filesystem::path p(runPath_);
   p /= fn;
   return p.string();
 }

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

private

Definition at line 151 of file DQMFileIterator.cc.

References asciidump::doc, 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);
 }

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

Definition at line 101 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.History::write().

void dqmservices::DQMFileIterator::reset ( void )

Definition at line 75 of file DQMFileIterator.cc.

References collect(), asciidump::doc, eor_, filesSeen_, cmsPerfStripChart::format, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, lumiSeen_, mon_, nextLumiNumber_, fileCollector::now, runInputDir_, runNumber_, runPath_, state_, cond::to_string(), and update_state().

Referenced by DQMFileIterator().

                             {
   runPath_ = str(boost::format("%s/run%06d") % runInputDir_ % 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 115 of file DQMFileIterator.cc.

References runNumber_.

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

115 { return runNumber_; }

dqmservices::DQMFileIterator::runNumber_

unsigned int runNumber_

Definition: DQMFileIterator.h:88

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

Definition at line 99 of file DQMFileIterator.cc.

References state_.

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

99 { return state_; }

dqmservices::DQMFileIterator::state_

State state_

Definition: DQMFileIterator.h:100

void dqmservices::DQMFileIterator::update_state ( )

Definition at line 277 of file DQMFileIterator.cc.

References advanceToLumi(), collect(), prof2calltree::count, asciidump::doc, eor_, getDQMSummary::iter, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), lumiSeen_, mon_, producerFileCleanner::msg, dqmservices::DQMFileIterator::EorEntry::n_lumi, nextLumiNumber_, nextLumiTimeoutMillis_, fileCollector::now, state_, AlCaHLTBitMon_QueryRunRegistry::string, and cond::to_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;
 
   collect(false);
 
   // now update the state
   State old_state = state_;
 
   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);
     }
   }
 }