Services/plugins/CondorStatusUpdater.cc

0001
0002 #include "DataFormats/Provenance/interface/ModuleDescription.h"
0003 #include "DataFormats/Provenance/interface/ParameterSetID.h"
0004 #include "FWCore/Utilities/interface/TimingServiceBase.h"
0005 #include "FWCore/Utilities/interface/CPUServiceBase.h"
0006 #include "FWCore/ServiceRegistry/interface/Service.h"
0007 #include "FWCore/ServiceRegistry/interface/ServiceMaker.h"
0008 #include "FWCore/ServiceRegistry/interface/ProcessContext.h"
0009 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
0010 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
0011 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
0012 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0013 #include "FWCore/ParameterSet/interface/Registry.h"
0014 #include "Utilities/StorageFactory/interface/StorageAccount.h"
0015 #include "Utilities/XrdAdaptor/interface/XrdStatistics.h"
0016 #include "FWCore/Utilities/interface/thread_safety_macros.h"
0017 #include "FWCore/Utilities/interface/processGUID.h"
0018
0019 #include <fcntl.h>
0020 #include <unistd.h>
0021 #include <sys/wait.h>
0022 #include <spawn.h>
0023 #include <iostream>
0024 #include <fstream>
0025 #include <sstream>
0026 #include <cmath>
0027 #include <chrono>
0028 #include <sstream>
0029 #include <atomic>
0030 #include <string>
0031 #include <set>
0032
0033 namespace edm {
0034
0035   namespace service {
0036
0037     class CondorStatusService {
0038     public:
0039       explicit CondorStatusService(ParameterSet const &pset, edm::ActivityRegistry &ar);
0040       ~CondorStatusService() {}
0041       CondorStatusService(const CondorStatusService &) = delete;
0042       CondorStatusService &operator=(const CondorStatusService &) = delete;
0043
0044       static void fillDescriptions(ConfigurationDescriptions &descriptions);
0045
0046     private:
0047       bool isChirpSupported();
0048       template <typename T>
0049       bool updateChirp(const std::string &key_suffix, const T &value);
0050       bool updateChirpQuoted(const std::string &key_suffix, const std::string &value);
0051       bool updateChirpImpl(std::string const &key, std::string const &value);
0052       inline void update();
0053       void firstUpdate();
0054       void lastUpdate();
0055       void updateImpl(time_t secsSinceLastUpdate);
0056
0057       void preSourceConstruction(ModuleDescription const &md, int maxEvents, int maxLumis, int maxSecondsUntilRampdown);
0058       void eventPost(StreamContext const &iContext);
0059       void lumiPost(GlobalContext const &);
0060       void runPost(GlobalContext const &);
0061       void beginPre(PathsAndConsumesOfModulesBase const &, ProcessContext const &processContext);
0062       void beginPost();
0063       void endPost();
0064       void filePost(std::string const &);
0065
0066       bool m_debug;
0067       std::atomic_flag m_shouldUpdate;
0068       time_t m_beginJob = 0;
0069       time_t m_updateInterval = m_defaultUpdateInterval;
0070       float m_emaInterval = m_defaultEmaInterval;
0071       float m_rate = 0;
0072       static constexpr float m_defaultEmaInterval = 15 * 60;  // Time in seconds to average EMA over for event rate.
0073       static constexpr unsigned int m_defaultUpdateInterval = 3 * 60;
0074       std::atomic<time_t> m_lastUpdate;
0075       std::atomic<std::uint_least64_t> m_events;
0076       std::atomic<std::uint_least64_t> m_lumis;
0077       std::atomic<std::uint_least64_t> m_runs;
0078       std::atomic<std::uint_least64_t> m_files;
0079       std::string m_tag;
0080       edm::ParameterSetID m_processParameterSetID;
0081
0082       std::uint_least64_t m_lastEventCount = 0;
0083     };
0084
0085   }  // namespace service
0086
0087 }  // namespace edm
0088
0089 using namespace edm::service;
0090
0091 const unsigned int CondorStatusService::m_defaultUpdateInterval;
0092 constexpr float CondorStatusService::m_defaultEmaInterval;
0093
0094 CondorStatusService::CondorStatusService(ParameterSet const &pset, edm::ActivityRegistry &ar)
0095     : m_debug(pset.getUntrackedParameter("debug", false)),
0096       m_lastUpdate(0),
0097       m_events(0),
0098       m_lumis(0),
0099       m_runs(0),
0100       m_files(0) {
0101   m_shouldUpdate.clear();
0102   if (not pset.getUntrackedParameter("enable", true)) {
0103     return;
0104   }
0105   if (!isChirpSupported()) {
0106     return;
0107   }
0108
0109   firstUpdate();
0110
0111   ar.watchPostCloseFile(this, &CondorStatusService::filePost);
0112   ar.watchPostEvent(this, &CondorStatusService::eventPost);
0113   ar.watchPostGlobalEndLumi(this, &CondorStatusService::lumiPost);
0114   ar.watchPostGlobalEndRun(this, &CondorStatusService::runPost);
0115   ar.watchPreBeginJob(this, &CondorStatusService::beginPre);
0116   ar.watchPostBeginJob(this, &CondorStatusService::beginPost);
0117   ar.watchPostEndJob(this, &CondorStatusService::endPost);
0118
0119   if (pset.exists("updateIntervalSeconds")) {
0120     m_updateInterval = pset.getUntrackedParameter<unsigned int>("updateIntervalSeconds");
0121   }
0122   if (pset.exists("EMAInterval")) {
0123     m_emaInterval = pset.getUntrackedParameter<double>("EMAInterval");
0124   }
0125   if (pset.exists("tag")) {
0126     m_tag = pset.getUntrackedParameter<std::string>("tag");
0127   }
0128 }
0129
0130 void CondorStatusService::eventPost(StreamContext const &iContext) {
0131   m_events++;
0132   update();
0133 }
0134
0135 void CondorStatusService::lumiPost(GlobalContext const &) {
0136   m_lumis++;
0137   update();
0138 }
0139
0140 void CondorStatusService::runPost(GlobalContext const &) {
0141   m_runs++;
0142   update();
0143 }
0144
0145 void CondorStatusService::filePost(std::string const & /*lfn*/) {
0146   m_files++;
0147   update();
0148 }
0149
0150 void CondorStatusService::beginPre(PathsAndConsumesOfModulesBase const &, ProcessContext const &processContext) {
0151   if (!m_processParameterSetID.isValid()) {
0152     m_processParameterSetID = processContext.parameterSetID();
0153   }
0154 }
0155
0156 void CondorStatusService::beginPost() {
0157   ParameterSet const &processParameterSet = edm::getParameterSet(m_processParameterSetID);
0158   const edm::ParameterSet &pset = processParameterSet.getParameterSet("@main_input");
0159   // PSet info from edm::ScheduleItems
0160   int maxEvents =
0161       processParameterSet.getUntrackedParameterSet("maxEvents", ParameterSet()).getUntrackedParameter<int>("input", -1);
0162   int maxLumis = processParameterSet.getUntrackedParameterSet("maxLuminosityBlocks", ParameterSet())
0163                      .getUntrackedParameter<int>("input", -1);
0164
0165   // lumisToProcess from EventSkipperByID (PoolSource and similar)
0166   std::vector<edm::LuminosityBlockRange> toProcess = pset.getUntrackedParameter<std::vector<LuminosityBlockRange>>(
0167       "lumisToProcess", std::vector<LuminosityBlockRange>());
0168   edm::sortAndRemoveOverlaps(toProcess);
0169   uint64_t lumiCount = 0;
0170   for (auto const &range : toProcess) {
0171     if (range.startRun() != range.endRun()) {
0172       break;
0173     }
0174     if (range.endLumi() >= edm::LuminosityBlockID::maxLuminosityBlockNumber()) {
0175       break;
0176     }
0177     lumiCount += (range.endLumi() - range.startLumi());
0178   }
0179   // Handle sources deriving from ProducerSourceBase
0180   unsigned int eventsPerLumi = pset.getUntrackedParameter<unsigned int>("numberEventsInLuminosityBlock", 0);
0181   if ((lumiCount == 0) && (maxEvents > 0) && (eventsPerLumi > 0)) {
0182     lumiCount = static_cast<unsigned int>(std::ceil(static_cast<float>(maxEvents) / static_cast<float>(eventsPerLumi)));
0183   }
0184
0185   std::vector<std::string> fileNames =
0186       pset.getUntrackedParameter<std::vector<std::string>>("fileNames", std::vector<std::string>());
0187   std::stringstream ss_max_files;
0188   ss_max_files << fileNames.size();
0189   updateChirp("MaxFiles", ss_max_files.str());
0190
0191   if (lumiCount > 0) {
0192     if (maxLumis < 0) {
0193       maxLumis = lumiCount;
0194     }
0195     if (maxLumis > static_cast<int>(lumiCount)) {
0196       maxLumis = lumiCount;
0197     }
0198   }
0199   if (maxEvents > 0) {
0200     std::stringstream ss_max_events;
0201     ss_max_events << maxEvents;
0202     updateChirp("MaxEvents", ss_max_events.str());
0203   }
0204   if (maxLumis > 0) {
0205     std::stringstream ss_max_lumis;
0206     ss_max_lumis << maxLumis;
0207     updateChirp("MaxLumis", ss_max_lumis.str());
0208   }
0209
0210   m_beginJob = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
0211   update();
0212 }
0213
0214 void CondorStatusService::endPost() { lastUpdate(); }
0215
0216 bool CondorStatusService::isChirpSupported() {
0217   if (m_debug) {
0218     return true;
0219   }
0220
0221   return std::getenv("_CONDOR_CHIRP_CONFIG") && updateChirp("Elapsed", "0");
0222 }
0223
0224 void CondorStatusService::firstUpdate() {
0225   // Note we always update all our statistics to 0 / false / -1
0226   // This allows us to overwrite the activities of a previous cmsRun process
0227   // within this HTCondor job.
0228   updateImpl(0);
0229   updateChirp("MaxFiles", "-1");
0230   updateChirp("MaxEvents", "-1");
0231   updateChirp("MaxLumis", "-1");
0232   updateChirp("Done", "false");
0233   updateChirpQuoted("Guid", edm::processGUID().toString());
0234
0235   edm::Service<edm::CPUServiceBase> cpusvc;
0236   std::string models;
0237   double avgSpeed;
0238   if (cpusvc.isAvailable() && cpusvc->cpuInfo(models, avgSpeed)) {
0239     updateChirpQuoted("CPUModels", models);
0240     updateChirp("CPUSpeed", avgSpeed);
0241   }
0242 }
0243
0244 void CondorStatusService::lastUpdate() {
0245   time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
0246   updateImpl(now - m_lastUpdate);
0247   updateChirp("Done", "true");
0248   edm::Service<edm::CPUServiceBase> cpusvc;
0249   if (!cpusvc.isAvailable()) {
0250     std::cout << "At post, CPU service is NOT available.\n";
0251   }
0252 }
0253
0254 void CondorStatusService::update() {
0255   time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
0256   if ((now - m_lastUpdate.load(std::memory_order_relaxed)) > m_updateInterval) {
0257     if (!m_shouldUpdate.test_and_set(std::memory_order_acquire)) {
0258       // Caught exception is rethrown
0259       CMS_SA_ALLOW try {
0260         time_t sinceLastUpdate = now - m_lastUpdate;
0261         m_lastUpdate = now;
0262         updateImpl(sinceLastUpdate);
0263         m_shouldUpdate.clear(std::memory_order_release);
0264       } catch (...) {
0265         m_shouldUpdate.clear(std::memory_order_release);
0266         throw;
0267       }
0268     }
0269   }
0270 }
0271
0272 void CondorStatusService::updateImpl(time_t sinceLastUpdate) {
0273   time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
0274   time_t jobTime = now - m_beginJob;
0275
0276   edm::Service<edm::TimingServiceBase> timingsvc;
0277   if (timingsvc.isAvailable()) {
0278     updateChirp("TotalCPU", timingsvc->getTotalCPU());
0279   }
0280
0281   updateChirp("LastUpdate", now);
0282
0283   if (!m_events || (m_events > m_lastEventCount)) {
0284     updateChirp("Events", m_events);
0285   }
0286
0287   updateChirp("Lumis", m_lumis);
0288
0289   updateChirp("Runs", m_runs);
0290
0291   updateChirp("Files", m_files);
0292
0293   float ema_coeff = 1 - std::exp(-static_cast<float>(sinceLastUpdate) /
0294                                  std::max(std::min(m_emaInterval, static_cast<float>(jobTime)), 1.0f));
0295   if (sinceLastUpdate > 0) {
0296     updateChirp("Elapsed", jobTime);
0297     m_rate = ema_coeff * static_cast<float>(m_events - m_lastEventCount) / static_cast<float>(sinceLastUpdate) +
0298              (1.0 - ema_coeff) * m_rate;
0299     m_lastEventCount = m_events;
0300     updateChirp("EventRate", m_rate);
0301   }
0302
0303   // If Xrootd was used, pull the statistics from there.
0304   edm::Service<xrd_adaptor::XrdStatistics> xrdsvc;
0305   if (xrdsvc.isAvailable()) {
0306     for (auto const &iter : xrdsvc->condorUpdate()) {
0307       std::string site = iter.first;
0308       site.erase(std::remove_if(site.begin(), site.end(), [](char x) { return !isalnum(x) && (x != '_'); }),
0309                  site.end());
0310       auto &iostats = iter.second;
0311       updateChirp("IOSite_" + site + "_ReadBytes", iostats.bytesRead);
0312       updateChirp("IOSite_" + site + "_ReadTimeMS",
0313                   std::chrono::duration_cast<std::chrono::milliseconds>(iostats.transferTime).count());
0314     }
0315   }
0316
0317   using namespace edm::storage;
0318   // Update storage account information
0319   auto const &stats = StorageAccount::summary();
0320   uint64_t readOps = 0;
0321   uint64_t readVOps = 0;
0322   uint64_t readSegs = 0;
0323   uint64_t readBytes = 0;
0324   uint64_t readTimeTotal = 0;
0325   uint64_t writeBytes = 0;
0326   uint64_t writeTimeTotal = 0;
0327   const auto token = StorageAccount::tokenForStorageClassName("tstoragefile");
0328   for (const auto &storage : stats) {
0329     // StorageAccount records statistics for both the TFile layer and the
0330     // StorageFactory layer.  However, the StorageFactory statistics tend to
0331     // be more accurate as various backends may alter the incoming read requests
0332     // (such as when lazy-download is used).
0333     if (storage.first == token.value()) {
0334       continue;
0335     }
0336     for (const auto &counter : storage.second) {
0337       if (counter.first == static_cast<int>(StorageAccount::Operation::read)) {
0338         readOps += counter.second.successes;
0339         readSegs++;
0340         readBytes += counter.second.amount;
0341         readTimeTotal += counter.second.timeTotal;
0342       } else if (counter.first == static_cast<int>(StorageAccount::Operation::readv)) {
0343         readVOps += counter.second.successes;
0344         readSegs += counter.second.vector_count;
0345         readBytes += counter.second.amount;
0346         readTimeTotal += counter.second.timeTotal;
0347       } else if ((counter.first == static_cast<int>(StorageAccount::Operation::write)) ||
0348                  (counter.first == static_cast<int>(StorageAccount::Operation::writev))) {
0349         writeBytes += counter.second.amount;
0350         writeTimeTotal += counter.second.timeTotal;
0351       }
0352     }
0353   }
0354   updateChirp("ReadOps", readOps);
0355   updateChirp("ReadVOps", readVOps);
0356   updateChirp("ReadSegments", readSegs);
0357   updateChirp("ReadBytes", readBytes);
0358   updateChirp("ReadTimeMsecs", readTimeTotal / (1000 * 1000));
0359   updateChirp("WriteBytes", writeBytes);
0360   updateChirp("WriteTimeMsecs", writeTimeTotal / (1000 * 1000));
0361 }
0362
0363 template <typename T>
0364 bool CondorStatusService::updateChirp(const std::string &key_suffix, const T &value) {
0365   std::stringstream ss;
0366   ss << value;
0367   return updateChirpImpl(key_suffix, ss.str());
0368 }
0369
0370 bool CondorStatusService::updateChirpQuoted(const std::string &key_suffix, const std::string &value) {
0371   std::string value_copy = value;
0372   // Remove double-quotes or the \ character (as it has special escaping semantics in ClassAds).
0373   // Make sure we have ASCII characters.
0374   // Otherwise, remainder is allowed (including tabs, newlines, single-quotes).
0375   value_copy.erase(
0376       remove_if(
0377           value_copy.begin(), value_copy.end(), [](const char &c) { return !isascii(c) || (c == '"') || (c == '\\'); }),
0378       value_copy.end());
0379   return updateChirpImpl(key_suffix, "\"" + value_copy + "\"");
0380 }
0381
0382 bool CondorStatusService::updateChirpImpl(const std::string &key_suffix, const std::string &value) {
0383   std::stringstream ss;
0384   ss << "ChirpCMSSW" << m_tag << key_suffix;
0385   std::string key = ss.str();
0386   if (m_debug) {
0387     std::cout << "condor_chirp set_job_attr_delayed " << key << " " << value << std::endl;
0388   }
0389   int pid = 0;
0390   posix_spawn_file_actions_t file_actions;
0391   int devnull_fd = open("/dev/null", O_RDWR);
0392   if (devnull_fd == -1) {
0393     return false;
0394   }
0395   posix_spawn_file_actions_init(&file_actions);
0396   posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 1);
0397   posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 2);
0398   const std::string chirp_name = "condor_chirp";
0399   const std::string set_job_attr = "set_job_attr_delayed";
0400   std::vector<const char *> argv;
0401   argv.push_back(chirp_name.c_str());
0402   argv.push_back(set_job_attr.c_str());
0403   argv.push_back(key.c_str());
0404   argv.push_back(value.c_str());
0405   argv.push_back(nullptr);
0406   int status = posix_spawnp(&pid, "condor_chirp", &file_actions, nullptr, const_cast<char *const *>(&argv[0]), environ);
0407   close(devnull_fd);
0408   posix_spawn_file_actions_destroy(&file_actions);
0409   if (status) {
0410     return false;
0411   }
0412   while ((waitpid(pid, &status, 0) == -1) && errno == -EINTR) {
0413   }
0414   return status == 0;
0415 }
0416
0417 void CondorStatusService::fillDescriptions(ConfigurationDescriptions &descriptions) {
0418   ParameterSetDescription desc;
0419   desc.setComment("Service to update HTCondor with the current CMSSW status.");
0420   desc.addOptionalUntracked<unsigned int>("updateIntervalSeconds", m_defaultUpdateInterval)
0421       ->setComment("Interval, in seconds, for HTCondor updates");
0422   desc.addOptionalUntracked<bool>("debug", false)->setComment("Enable debugging of this service");
0423   desc.addOptionalUntracked<double>("EMAInterval", m_defaultEmaInterval)
0424       ->setComment("Interval, in seconds, to calculate event rate over (using EMA)");
0425   desc.addOptionalUntracked<std::string>("tag")->setComment(
0426       "Identifier tag for this process (a value of 'Foo' results in ClassAd attributes of the form 'ChirpCMSSWFoo*')");
0427   desc.addOptionalUntracked<bool>("enable", true)->setComment("Enable this service");
0428   descriptions.add("CondorStatusService", desc);
0429 }
0430
0431 typedef edm::serviceregistry::AllArgsMaker<edm::service::CondorStatusService> CondorStatusServiceMaker;
0432 DEFINE_FWK_SERVICE_MAKER(CondorStatusService, CondorStatusServiceMaker);