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

void	delay ()

	DQMFileIterator (edm::ParameterSet const &pset)

const LumiEntry &	front ()

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

unsigned int	lastLumiFound ()

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

bool	lumiReady ()

std::string	make_path_data (const LumiEntry &lumi)

void	pop ()

void	reset ()

unsigned int	runNumber ()

State	state ()

void	update_state ()

void	updateWatchdog ()

	~DQMFileIterator ()

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

Private Member Functions
void	collect (bool ignoreTimers)

Private Attributes
unsigned int	currentLumi_

unsigned int	datafnPosition_

unsigned long	delayMillis_

EorEntry	eor_

long	forceFileCheckTimeoutMillis_

std::chrono::high_resolution_clock::time_point	lastLumiLoad_

std::map< unsigned int, LumiEntry >	lumiSeen_

edm::Service < DQMMonitoringService >	mon_

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 61 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 76 of file DQMFileIterator.cc.

76 {}

Member Function Documentation

void dqmservices::DQMFileIterator::advanceToLumi ( unsigned int lumi )

Definition at line 120 of file DQMFileIterator.cc.

References currentLumi_, getDQMSummary::iter, lastLumiLoad_, fjr2json::lumi, lumiSeen_, mon_, fileCollector::now, runNumber_, and cond::to_string().

Referenced by pop(), and dqmservices::DQMStreamerReader::reset_().

                                                      {
   using boost::property_tree::ptree;
   using boost::str;
 
   unsigned int prev_lumi = currentLumi_;
 
   currentLumi_ = lumi;
   lastLumiLoad_ = std::chrono::high_resolution_clock::now();
 
   // report the successful lumi file open
   if (mon_.isAvailable()) {
     ptree children;
 
     auto iter = lumiSeen_.begin();
     for (; iter != lumiSeen_.end(); ++iter) {
       children.put(std::to_string(iter->first), iter->second.filename);
     }
 
     mon_->registerExtra("lumiSeen", children);
     mon_->reportLumiSection(runNumber_, prev_lumi);
   }
 }

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

private

Definition at line 151 of file DQMFileIterator.cc.

References prof2calltree::count, datafnPosition_, eor_, lut2db_cfg::filename, forceFileCheckTimeoutMillis_, diffTwoXMLs::label, dqmservices::DQMFileIterator::LumiEntry::load_json(), dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), fjr2json::lumi, lumiSeen_, dqmservices::DQMFileIterator::EorEntry::n_lumi, fileCollector::now, query::result, DTTTrigCorrFirst::run, runNumber_, runPath_, runPathLastCollect_, runPathMTime_, streamLabel_, AlCaHLTBitMon_QueryRunRegistry::string, and edmStreamStallGrapher::t.

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 < 100)) {
     return;
   }
 
   // check if directory changed
   std::time_t t = boost::filesystem::last_write_time(runPath_);
 
   if ((!ignoreTimers) && (last_ms < forceFileCheckTimeoutMillis_) && (t == runPathMTime_)) {
     //logFileAction("Directory hasn't changed.");
     return;
   } else {
     //logFileAction("Directory changed, updating.");
   }
 
   runPathMTime_ = t;
   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+)(_.*).jsn");
 
     const std::string filename = di->path().filename().string();
     const std::string fn = di->path().string();
 
     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];
 
       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_) {
         logFileAction("Found and skipped json file (stream label mismatch): ",
                       fn);
         continue;
       }
 
       LumiEntry lumi_jsn = LumiEntry::load_json(fn, lumi, datafnPosition_);
       lumiSeen_.emplace(lumi, lumi_jsn);
       logFileAction("Found and loaded json file: ", 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 309 of file DQMFileIterator.cc.

References delayMillis_, and updateWatchdog().

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

                             {
   //logFileAction("Streamer waiting for the next LS.");
 
   updateWatchdog();
   usleep(delayMillis_ * 1000);
   updateWatchdog();
 }

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

static

Definition at line 317 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.");
 }

const DQMFileIterator::LumiEntry & dqmservices::DQMFileIterator::front ( )

Definition at line 94 of file DQMFileIterator.cc.

References currentLumi_, and lumiSeen_.

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

                                                        {
   return lumiSeen_[currentLumi_];
 }

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

unsigned int dqmservices::DQMFileIterator::lastLumiFound ( )

Definition at line 112 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 294 of file DQMFileIterator.cc.

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

Referenced by collect(), dqmservices::DQMProtobufReader::getNextItemType(), dqmservices::DQMStreamerReader::openNextFile_(), dqmservices::DQMProtobufReader::readLuminosityBlock_(), dqmservices::DQMProtobufReader::readLuminosityBlockAuxiliary_(), dqmservices::DQMProtobufReader::readRun_(), dqmservices::DQMProtobufReader::readRunAuxiliary_(), dqmservices::DQMStreamerReader::reset_(), and update_state().

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

bool dqmservices::DQMFileIterator::lumiReady ( )

Definition at line 102 of file DQMFileIterator.cc.

References currentLumi_, and lumiSeen_.

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

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

std::string dqmservices::DQMFileIterator::make_path_data ( const LumiEntry & lumi )

Definition at line 143 of file DQMFileIterator.cc.

References dqmservices::DQMFileIterator::LumiEntry::datafilename, AlCaHLTBitMon_ParallelJobs::p, cmsHarvester::path, and runPath_.

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

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

void dqmservices::DQMFileIterator::pop ( )

Definition at line 98 of file DQMFileIterator.cc.

References advanceToLumi(), and currentLumi_.

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

                           {
   advanceToLumi(currentLumi_ + 1);
 }

void dqmservices::DQMFileIterator::reset ( void )

Definition at line 78 of file DQMFileIterator.cc.

References collect(), currentLumi_, eor_, cmsPerfStripChart::format, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, lumiSeen_, fileCollector::now, runInputDir_, runNumber_, runPath_, state_, and update_state().

Referenced by DQMFileIterator().

                             {
   runPath_ = str(boost::format("%s/run%06d") % runInputDir_ % runNumber_);
 
   eor_.loaded = false;
   state_ = State::OPEN;
   currentLumi_ = 1;
   lumiSeen_.clear();
 
   lastLumiLoad_ = std::chrono::high_resolution_clock::now();
 
   collect(true);
   update_state();
 }

unsigned int dqmservices::DQMFileIterator::runNumber ( )

Definition at line 110 of file DQMFileIterator.cc.

References runNumber_.

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

110 { return runNumber_; }

dqmservices::DQMFileIterator::runNumber_

unsigned int runNumber_

Definition: DQMFileIterator.h:84

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

Definition at line 92 of file DQMFileIterator.cc.

References state_.

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

92 { return state_; }

dqmservices::DQMFileIterator::state_

State state_

Definition: DQMFileIterator.h:97

void dqmservices::DQMFileIterator::update_state ( )

Definition at line 240 of file DQMFileIterator.cc.

References collect(), prof2calltree::count, currentLumi_, eor_, getDQMSummary::iter, lastLumiLoad_, dqmservices::DQMFileIterator::EorEntry::loaded, logFileAction(), lumiSeen_, producerFileCleanner::msg, dqmservices::DQMFileIterator::EorEntry::n_lumi, 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(currentLumi_);
     if ((iter != lumiSeen_.end()) && iter->first != currentLumi_) {
 
       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(currentLumi_) + " .. " +
                std::to_string(iter->first - 1);
         msg += ", currentLumi_ is now " + std::to_string(iter->first);
 
         logFileAction(msg);
 
         currentLumi_ = iter->first;
       }
     }
   }
 
   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 (currentLumi_ > eor_.n_lumi) {
       state_ = State::EOR;
     }
   }
 
   if (state_ != old_state) {
     logFileAction("Streamer state changed: ",
                   std::to_string(old_state) + "->" + std::to_string(state_));
   }
 }

void dqmservices::DQMFileIterator::updateWatchdog ( )

Definition at line 301 of file DQMFileIterator.cc.

References TablePrint::write, and x.

Referenced by delay(), and dqmservices::DQMStreamerReader::prepareNextEvent().

                                      {
   const char* x = getenv("WATCHDOG_FD");
   if (x) {
     int fd = atoi(x);
     write(fd, ".\n", 2);
   }
 }