File indexing completed on 2022-04-28 22:23:51
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;
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 }
0086
0087 }
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 & ) {
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
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
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
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
0226
0227
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
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
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
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
0330
0331
0332
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
0373
0374
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);