Public Member Functions
	CondorStatusService (ParameterSet const &pset, edm::ActivityRegistry &ar)

	CondorStatusService (const CondorStatusService &)=delete

CondorStatusService &	operator= (const CondorStatusService &)=delete

	~CondorStatusService ()

Static Public Member Functions
static void	fillDescriptions (ConfigurationDescriptions &descriptions)

Private Member Functions
void	beginPost ()

void	beginPre (PathsAndConsumesOfModulesBase const &, ProcessContext const &processContext)

void	endPost ()

void	eventPost (StreamContext const &iContext)

void	filePost (std::string const &, bool)

void	firstUpdate ()

bool	isChirpSupported ()

void	lastUpdate ()

void	lumiPost (GlobalContext const &)

void	preSourceConstruction (ModuleDescription const &md, int maxEvents, int maxLumis, int maxSecondsUntilRampdown)

void	runPost (GlobalContext const &)

void	update ()

template<typename T >
bool	updateChirp (const std::string &key_suffix, const T &value)

bool	updateChirpImpl (std::string const &key, std::string const &value)

bool	updateChirpQuoted (const std::string &key_suffix, const std::string &value)

void	updateImpl (time_t secsSinceLastUpdate)

Private Attributes
time_t	m_beginJob = 0

bool	m_debug

float	m_emaInterval = m_defaultEmaInterval

std::atomic< std::uint_least64_t >	m_events

std::atomic< std::uint_least64_t >	m_files

std::uint_least64_t	m_lastEventCount = 0

std::atomic< time_t >	m_lastUpdate

std::atomic< std::uint_least64_t >	m_lumis

edm::ParameterSetID	m_processParameterSetID

float	m_rate = 0

std::atomic< std::uint_least64_t >	m_runs

std::atomic_flag	m_shouldUpdate

std::string	m_tag

time_t	m_updateInterval = m_defaultUpdateInterval

Static Private Attributes
static float	m_defaultEmaInterval = 15 * 60

static unsigned int	m_defaultUpdateInterval = 3 * 60

Detailed Description

Definition at line 35 of file CondorStatusUpdater.cc.

Constructor & Destructor Documentation

CondorStatusService::CondorStatusService	(	ParameterSet const &	pset,
		edm::ActivityRegistry &	ar
	)

explicit

Definition at line 92 of file CondorStatusUpdater.cc.

References beginPost(), beginPre(), endPost(), eventPost(), edm::ParameterSet::exists(), filePost(), firstUpdate(), edm::ParameterSet::getUntrackedParameter(), isChirpSupported(), lumiPost(), m_debug, m_emaInterval, m_shouldUpdate, m_tag, m_updateInterval, runPost(), AlCaHLTBitMon_QueryRunRegistry::string, edm::ActivityRegistry::watchPostBeginJob(), edm::ActivityRegistry::watchPostCloseFile(), edm::ActivityRegistry::watchPostEndJob(), edm::ActivityRegistry::watchPostEvent(), edm::ActivityRegistry::watchPostGlobalEndLumi(), edm::ActivityRegistry::watchPostGlobalEndRun(), and edm::ActivityRegistry::watchPreBeginJob().

Referenced by ~CondorStatusService().

     : m_debug(false), m_lastUpdate(0), m_events(0), m_lumis(0), m_runs(0), m_files(0) {
   m_shouldUpdate.clear();
   if (pset.exists("debug")) {
     m_debug = true;
   }
   if (!isChirpSupported()) {
     return;
   }
 
   firstUpdate();
 
   ar.watchPostCloseFile(this, &CondorStatusService::filePost);
   ar.watchPostEvent(this, &CondorStatusService::eventPost);
   ar.watchPostGlobalEndLumi(this, &CondorStatusService::lumiPost);
   ar.watchPostGlobalEndRun(this, &CondorStatusService::runPost);
   ar.watchPreBeginJob(this, &CondorStatusService::beginPre);
   ar.watchPostBeginJob(this, &CondorStatusService::beginPost);
   ar.watchPostEndJob(this, &CondorStatusService::endPost);
 
   if (pset.exists("updateIntervalSeconds")) {
     m_updateInterval = pset.getUntrackedParameter<unsigned int>("updateIntervalSeconds");
   }
   if (pset.exists("EMAInterval")) {
     m_emaInterval = pset.getUntrackedParameter<double>("EMAInterval");
   }
   if (pset.exists("tag")) {
     m_tag = pset.getUntrackedParameter<std::string>("tag");
   }
 }

edm::service::CondorStatusService::~CondorStatusService ( )

inline

edm::service::CondorStatusService::CondorStatusService ( const CondorStatusService & )

delete

Member Function Documentation

void CondorStatusService::beginPost ( )

private

Definition at line 149 of file CondorStatusUpdater.cc.

References LaserTracksInput_cfi::fileNames, edm::ParameterSet::getParameterSet(), edm::getParameterSet(), edm::ParameterSet::getUntrackedParameter(), edm::ParameterSet::getUntrackedParameterSet(), m_beginJob, m_processParameterSetID, Data_TkAlMinBias_Run2018C_PromptReco_v3_cff::maxEvents, edm::LuminosityBlockID::maxLuminosityBlockNumber(), cmsPerfSuiteHarvest::now, muonDTDigis_cfi::pset, edm::sortAndRemoveOverlaps(), update(), and updateChirp().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                     {
   ParameterSet const &processParameterSet = edm::getParameterSet(m_processParameterSetID);
   const edm::ParameterSet &pset = processParameterSet.getParameterSet("@main_input");
   // PSet info from edm::ScheduleItems
   int maxEvents =
       processParameterSet.getUntrackedParameterSet("maxEvents", ParameterSet()).getUntrackedParameter<int>("input", -1);
   int maxLumis = processParameterSet.getUntrackedParameterSet("maxLuminosityBlocks", ParameterSet())
                      .getUntrackedParameter<int>("input", -1);
 
   // lumisToProcess from EventSkipperByID (PoolSource and similar)
   std::vector<edm::LuminosityBlockRange> toProcess = pset.getUntrackedParameter<std::vector<LuminosityBlockRange>>(
       "lumisToProcess", std::vector<LuminosityBlockRange>());
   edm::sortAndRemoveOverlaps(toProcess);
   uint64_t lumiCount = 0;
   for (auto const &range : toProcess) {
     if (range.startRun() != range.endRun()) {
       break;
     }
     if (range.endLumi() >= edm::LuminosityBlockID::maxLuminosityBlockNumber()) {
       break;
     }
     lumiCount += (range.endLumi() - range.startLumi());
   }
   // Handle sources deriving from ProducerSourceBase
   unsigned int eventsPerLumi = pset.getUntrackedParameter<unsigned int>("numberEventsInLuminosityBlock", 0);
   if ((lumiCount == 0) && (maxEvents > 0) && (eventsPerLumi > 0)) {
     lumiCount = static_cast<unsigned int>(std::ceil(static_cast<float>(maxEvents) / static_cast<float>(eventsPerLumi)));
   }
 
   std::vector<std::string> fileNames =
       pset.getUntrackedParameter<std::vector<std::string>>("fileNames", std::vector<std::string>());
   std::stringstream ss_max_files;
   ss_max_files << fileNames.size();
   updateChirp("MaxFiles", ss_max_files.str());
 
   if (lumiCount > 0) {
     if (maxLumis < 0) {
       maxLumis = lumiCount;
     }
     if (maxLumis > static_cast<int>(lumiCount)) {
       maxLumis = lumiCount;
     }
   }
   if (maxEvents > 0) {
     std::stringstream ss_max_events;
     ss_max_events << maxEvents;
     updateChirp("MaxEvents", ss_max_events.str());
   }
   if (maxLumis > 0) {
     std::stringstream ss_max_lumis;
     ss_max_lumis << maxLumis;
     updateChirp("MaxLumis", ss_max_lumis.str());
   }
 
   m_beginJob = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
   update();
 }

void CondorStatusService::beginPre	(	PathsAndConsumesOfModulesBase const &	,
		ProcessContext const &	processContext
	)

private

Definition at line 143 of file CondorStatusUpdater.cc.

References edm::Hash< I >::isValid(), m_processParameterSetID, and edm::ProcessContext::parameterSetID().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                                                                                               {
   if (!m_processParameterSetID.isValid()) {
     m_processParameterSetID = processContext.parameterSetID();
   }
 }

void CondorStatusService::endPost ( )

private

Definition at line 207 of file CondorStatusUpdater.cc.

References lastUpdate().

Referenced by CondorStatusService(), and ~CondorStatusService().

207 { lastUpdate(); }

edm::service::CondorStatusService::lastUpdate

void lastUpdate()

Definition: CondorStatusUpdater.cc:236

void CondorStatusService::eventPost ( StreamContext const & iContext )

private

Definition at line 123 of file CondorStatusUpdater.cc.

References m_events, and update().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                                                  {
   m_events++;
   update();
 }

void CondorStatusService::filePost	(	std::string const &	,
		bool
	)

private

Definition at line 138 of file CondorStatusUpdater.cc.

References m_files, and update().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                                               {
   m_files++;
   update();
 }

void CondorStatusService::fillDescriptions ( ConfigurationDescriptions & descriptions )

static

Definition at line 407 of file CondorStatusUpdater.cc.

References edm::ConfigurationDescriptions::add(), edm::ParameterSetDescription::addOptionalUntracked(), m_defaultEmaInterval, m_defaultUpdateInterval, edm::ParameterSetDescription::setComment(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by ~CondorStatusService().

                                                                                   {
   ParameterSetDescription desc;
   desc.setComment("Service to update HTCondor with the current CMSSW status.");
   desc.addOptionalUntracked<unsigned int>("updateIntervalSeconds", m_defaultUpdateInterval)
       ->setComment("Interval, in seconds, for HTCondor updates");
   desc.addOptionalUntracked<bool>("debug", false)->setComment("Enable debugging of this service");
   desc.addOptionalUntracked<double>("EMAInterval", m_defaultEmaInterval)
       ->setComment("Interval, in seconds, to calculate event rate over (using EMA)");
   desc.addOptionalUntracked<std::string>("tag")->setComment(
       "Identifier tag for this process (a value of 'Foo' results in ClassAd attributes of the form 'ChirpCMSSWFoo*')");
   descriptions.add("CondorStatusService", desc);
 }

void CondorStatusService::firstUpdate ( )

private

Definition at line 217 of file CondorStatusUpdater.cc.

References edm::CPUServiceBase::cpuInfo(), edm::Service< T >::isAvailable(), AlCaHLTBitMon_QueryRunRegistry::string, updateChirp(), updateChirpQuoted(), and updateImpl().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                       {
   // Note we always update all our statistics to 0 / false / -1
   // This allows us to overwrite the activities of a previous cmsRun process
   // within this HTCondor job.
   updateImpl(0);
   updateChirp("MaxFiles", "-1");
   updateChirp("MaxEvents", "-1");
   updateChirp("MaxLumis", "-1");
   updateChirp("Done", "false");
 
   edm::Service<edm::CPUServiceBase> cpusvc;
   std::string models;
   double avgSpeed;
   if (cpusvc.isAvailable() && cpusvc->cpuInfo(models, avgSpeed)) {
     updateChirpQuoted("CPUModels", models);
     updateChirp("CPUSpeed", avgSpeed);
   }
 }

bool CondorStatusService::isChirpSupported ( )

private

Definition at line 209 of file CondorStatusUpdater.cc.

References m_debug, and updateChirp().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                            {
   if (m_debug) {
     return true;
   }
 
   return getenv("_CONDOR_CHIRP_CONFIG") && updateChirp("Elapsed", "0");
 }

void CondorStatusService::lastUpdate ( )

private

Definition at line 236 of file CondorStatusUpdater.cc.

References gather_cfg::cout, edm::Service< T >::isAvailable(), m_lastUpdate, cmsPerfSuiteHarvest::now, updateChirp(), and updateImpl().

Referenced by endPost(), and ~CondorStatusService().

                                      {
   time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
   updateImpl(now - m_lastUpdate);
   updateChirp("Done", "true");
   edm::Service<edm::CPUServiceBase> cpusvc;
   if (!cpusvc.isAvailable()) {
     std::cout << "At post, CPU service is NOT available.\n";
   }
 }

void CondorStatusService::lumiPost ( GlobalContext const & )

private

Definition at line 128 of file CondorStatusUpdater.cc.

References m_lumis, and update().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                                         {
   m_lumis++;
   update();
 }

CondorStatusService& edm::service::CondorStatusService::operator= ( const CondorStatusService & )

delete

Referenced by ~CondorStatusService().

void edm::service::CondorStatusService::preSourceConstruction	(	ModuleDescription const &	md,
		int	maxEvents,
		int	maxLumis,
		int	maxSecondsUntilRampdown
	)

private

Referenced by ~CondorStatusService().

void CondorStatusService::runPost ( GlobalContext const & )

private

Definition at line 133 of file CondorStatusUpdater.cc.

References m_runs, and update().

Referenced by CondorStatusService(), and ~CondorStatusService().

                                                        {
   m_runs++;
   update();
 }

void CondorStatusService::update ( )

inlineprivate

Definition at line 246 of file CondorStatusUpdater.cc.

References m_lastUpdate, m_shouldUpdate, m_updateInterval, cmsPerfSuiteHarvest::now, and updateImpl().

                                  {
   time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
   if ((now - m_lastUpdate.load(std::memory_order_relaxed)) > m_updateInterval) {
     if (!m_shouldUpdate.test_and_set(std::memory_order_acquire)) {
       try {
         time_t sinceLastUpdate = now - m_lastUpdate;
         m_lastUpdate = now;
         updateImpl(sinceLastUpdate);
         m_shouldUpdate.clear(std::memory_order_release);
       } catch (...) {
         m_shouldUpdate.clear(std::memory_order_release);
         throw;
       }
     }
   }
 }

template<typename T >

bool CondorStatusService::updateChirp	(	const std::string &	key_suffix,
		const T &	value
	)

private

Definition at line 354 of file CondorStatusUpdater.cc.

References updateChirpImpl(), and relativeConstraints::value.

Referenced by beginPost(), firstUpdate(), isChirpSupported(), lastUpdate(), updateImpl(), and ~CondorStatusService().

                                                                                  {
   std::stringstream ss;
   ss << value;
   return updateChirpImpl(key_suffix, ss.str());
 }

bool CondorStatusService::updateChirpImpl	(	std::string const &	key,
		std::string const &	value
	)

private

Definition at line 372 of file CondorStatusUpdater.cc.

References cmsBatch::argv, gather_cfg::cout, crabWrapper::key, m_debug, m_tag, sysUtil::pid, mps_update::status, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by updateChirp(), updateChirpQuoted(), and ~CondorStatusService().

                                                                                              {
   std::stringstream ss;
   ss << "ChirpCMSSW" << m_tag << key_suffix;
   std::string key = ss.str();
   if (m_debug) {
     std::cout << "condor_chirp set_job_attr_delayed " << key << " " << value << std::endl;
   }
   int pid = 0;
   posix_spawn_file_actions_t file_actions;
   int devnull_fd = open("/dev/null", O_RDWR);
   if (devnull_fd == -1) {
     return false;
   }
   posix_spawn_file_actions_init(&file_actions);
   posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 1);
   posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 2);
   const std::string chirp_name = "condor_chirp";
   const std::string set_job_attr = "set_job_attr_delayed";
   std::vector<const char *> argv;
   argv.push_back(chirp_name.c_str());
   argv.push_back(set_job_attr.c_str());
   argv.push_back(key.c_str());
   argv.push_back(value.c_str());
   argv.push_back(nullptr);
   int status = posix_spawnp(&pid, "condor_chirp", &file_actions, nullptr, const_cast<char *const *>(&argv[0]), environ);
   close(devnull_fd);
   posix_spawn_file_actions_destroy(&file_actions);
   if (status) {
     return false;
   }
   while ((waitpid(pid, &status, 0) == -1) && errno == -EINTR) {
   }
   return status == 0;
 }

bool CondorStatusService::updateChirpQuoted	(	const std::string &	key_suffix,
		const std::string &	value
	)

private

Definition at line 360 of file CondorStatusUpdater.cc.

References EnergyCorrector::c, AlCaHLTBitMon_QueryRunRegistry::string, updateChirpImpl(), and relativeConstraints::value.

Referenced by firstUpdate(), and ~CondorStatusService().

                                                                                                {
   std::string value_copy = value;
   // Remove double-quotes or the \ character (as it has special escaping semantics in ClassAds).
   // Make sure we have ASCII characters.
   // Otherwise, remainder is allowed (including tabs, newlines, single-quotes).
   value_copy.erase(
       remove_if(
           value_copy.begin(), value_copy.end(), [](const char &c) { return !isascii(c) || (c == '"') || (c == '\\'); }),
       value_copy.end());
   return updateChirpImpl(key_suffix, "\"" + value_copy + "\"");
 }

void CondorStatusService::updateImpl ( time_t secsSinceLastUpdate )

private

Definition at line 263 of file CondorStatusUpdater.cc.

References XrdAdaptor::XrdStatisticsService::condorUpdate(), JetChargeProducer_cfi::exp, f, edm::TimingServiceBase::getTotalCPU(), edm::Service< T >::isAvailable(), m_beginJob, m_emaInterval, m_events, m_files, m_lastEventCount, m_lumis, m_rate, m_runs, SiStripPI::max, min(), cmsPerfSuiteHarvest::now, StorageAccount::read, StorageAccount::readv, AlCaHLTBitMon_QueryRunRegistry::string, StorageAccount::summary(), StorageAccount::tokenForStorageClassName(), updateChirp(), StorageAccount::write, and StorageAccount::writev.

Referenced by firstUpdate(), lastUpdate(), update(), and ~CondorStatusService().

                                                            {
   time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
   time_t jobTime = now - m_beginJob;
 
   edm::Service<edm::TimingServiceBase> timingsvc;
   if (timingsvc.isAvailable()) {
     updateChirp("TotalCPU", timingsvc->getTotalCPU());
   }
 
   updateChirp("LastUpdate", now);
 
   if (!m_events || (m_events > m_lastEventCount)) {
     updateChirp("Events", m_events);
   }
 
   updateChirp("Lumis", m_lumis);
 
   updateChirp("Runs", m_runs);
 
   updateChirp("Files", m_files);
 
   float ema_coeff = 1 - std::exp(-static_cast<float>(sinceLastUpdate) /
                                  std::max(std::min(m_emaInterval, static_cast<float>(jobTime)), 1.0f));
   if (sinceLastUpdate > 0) {
     updateChirp("Elapsed", jobTime);
     m_rate = ema_coeff * static_cast<float>(m_events - m_lastEventCount) / static_cast<float>(sinceLastUpdate) +
              (1.0 - ema_coeff) * m_rate;
     m_lastEventCount = m_events;
     updateChirp("EventRate", m_rate);
   }
 
   // If Xrootd was used, pull the statistics from there.
   edm::Service<XrdAdaptor::XrdStatisticsService> xrdsvc;
   if (xrdsvc.isAvailable()) {
     for (auto const &iter : xrdsvc->condorUpdate()) {
       std::string site = iter.first;
       site.erase(std::remove_if(site.begin(), site.end(), [](char x) { return !isalnum(x) && (x != '_'); }),
                  site.end());
       auto &iostats = iter.second;
       updateChirp("IOSite_" + site + "_ReadBytes", iostats.bytesRead);
       updateChirp("IOSite_" + site + "_ReadTimeMS",
                   std::chrono::duration_cast<std::chrono::milliseconds>(iostats.transferTime).count());
     }
   }
 
   // Update storage account information
   auto const &stats = StorageAccount::summary();
   uint64_t readOps = 0;
   uint64_t readVOps = 0;
   uint64_t readSegs = 0;
   uint64_t readBytes = 0;
   uint64_t readTimeTotal = 0;
   uint64_t writeBytes = 0;
   uint64_t writeTimeTotal = 0;
   const auto token = StorageAccount::tokenForStorageClassName("tstoragefile");
   for (const auto &storage : stats) {
     // StorageAccount records statistics for both the TFile layer and the
     // StorageFactory layer.  However, the StorageFactory statistics tend to
     // be more accurate as various backends may alter the incoming read requests
     // (such as when lazy-download is used).
     if (storage.first == token.value()) {
       continue;
     }
     for (const auto &counter : storage.second) {
       if (counter.first == static_cast<int>(StorageAccount::Operation::read)) {
         readOps += counter.second.successes;
         readSegs++;
         readBytes += counter.second.amount;
         readTimeTotal += counter.second.timeTotal;
       } else if (counter.first == static_cast<int>(StorageAccount::Operation::readv)) {
         readVOps += counter.second.successes;
         readSegs += counter.second.vector_count;
         readBytes += counter.second.amount;
         readTimeTotal += counter.second.timeTotal;
       } else if ((counter.first == static_cast<int>(StorageAccount::Operation::write)) ||
                  (counter.first == static_cast<int>(StorageAccount::Operation::writev))) {
         writeBytes += counter.second.amount;
         writeTimeTotal += counter.second.timeTotal;
       }
     }
   }
   updateChirp("ReadOps", readOps);
   updateChirp("ReadVOps", readVOps);
   updateChirp("ReadSegments", readSegs);
   updateChirp("ReadBytes", readBytes);
   updateChirp("ReadTimeMsecs", readTimeTotal / (1000 * 1000));
   updateChirp("WriteBytes", writeBytes);
   updateChirp("WriteTimeMsecs", writeTimeTotal / (1000 * 1000));
 }