test
CMS 3D CMS Logo

All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Pages
CondorStatusUpdater.cc
Go to the documentation of this file.
1 
15 
16 #include <fcntl.h>
17 #include <unistd.h>
18 #include <sys/wait.h>
19 #include <spawn.h>
20 #include <iostream>
21 #include <fstream>
22 #include <sstream>
23 #include <cmath>
24 #include <chrono>
25 #include <sstream>
26 #include <atomic>
27 #include <string>
28 #include <set>
29 
30 namespace edm {
31 
32  namespace service {
33 
35  {
36 
37  public:
38 
43 
44  static void fillDescriptions(ConfigurationDescriptions &descriptions);
45 
46  private:
47 
48  bool isChirpSupported();
49  template<typename T> bool updateChirp(const std::string &key_suffix, const T &value);
50  bool updateChirp(std::string const &key, std::string const &value);
51  inline void update();
52  void firstUpdate();
53  void lastUpdate();
54  void updateImpl(time_t secsSinceLastUpdate);
55 
56  void preSourceConstruction(ModuleDescription const &md, int maxEvents, int maxLumis, int maxSecondsUntilRampdown);
57  void eventPost(StreamContext const& iContext);
58  void lumiPost(GlobalContext const&);
59  void runPost(GlobalContext const&);
60  void beginPre(PathsAndConsumesOfModulesBase const&, ProcessContext const& processContext);
61  void beginPost();
62  void endPost();
63  void filePost(std::string const &, bool);
64 
65  bool m_debug;
66  std::atomic_flag m_shouldUpdate;
67  time_t m_beginJob = 0;
70  float m_rate = 0;
71  static constexpr float m_defaultEmaInterval = 15*60; // Time in seconds to average EMA over for event rate.
72  static constexpr unsigned int m_defaultUpdateInterval = 3*60;
73  std::atomic<time_t> m_lastUpdate;
74  std::atomic<std::uint_least64_t> m_events;
75  std::atomic<std::uint_least64_t> m_lumis;
76  std::atomic<std::uint_least64_t> m_runs;
77  std::atomic<std::uint_least64_t> m_files;
80 
81  std::uint_least64_t m_lastEventCount = 0;
82  };
83 
84  }
85 
86 }
87 
88 using namespace edm::service;
89 
92 
94  :
95  m_debug(false),
96  m_lastUpdate(0),
97  m_events(0),
98  m_lumis(0),
99  m_runs(0),
100  m_files(0)
101 {
102  m_shouldUpdate.clear();
103  if (pset.exists("debug"))
104  {
105  m_debug = true;
106  }
107  if (!isChirpSupported()) {return;}
108 
109  firstUpdate();
110 
118 
119  if (pset.exists("updateIntervalSeconds"))
120  {
121  m_updateInterval = pset.getUntrackedParameter<unsigned int>("updateIntervalSeconds");
122  }
123  if (pset.exists("EMAInterval"))
124  {
125  m_emaInterval = pset.getUntrackedParameter<double>("EMAInterval");
126  }
127  if (pset.exists("tag"))
128  {
129  m_tag = pset.getUntrackedParameter<std::string>("tag");
130  }
131 }
132 
133 
134 void
136 {
137  m_events++;
138  update();
139 }
140 
141 
142 void
144 {
145  m_lumis++;
146  update();
147 }
148 
149 
150 void
152 {
153  m_runs++;
154  update();
155 }
156 
157 
158 void
159 CondorStatusService::filePost(std::string const & /*lfn*/, bool /*usedFallback*/)
160 {
161  m_files++;
162  update();
163 }
164 
165 
166 void
168 {
170  {
171  m_processParameterSetID = processContext.parameterSetID();
172  }
173 }
174 
175 
176 void
178 {
179  ParameterSet const& processParameterSet = edm::getParameterSet(m_processParameterSetID);
180  const edm::ParameterSet &pset = processParameterSet.getParameterSet("@main_input");
181  // PSet info from edm::ScheduleItems
182  int maxEvents = processParameterSet.getUntrackedParameterSet("maxEvents", ParameterSet()).getUntrackedParameter<int>("input", -1);
183  int maxLumis = processParameterSet.getUntrackedParameterSet("maxLuminosityBlocks", ParameterSet()).getUntrackedParameter<int>("input", -1);
184 
185  // lumisToProcess from EventSkipperByID (PoolSource and similar)
186  std::vector<edm::LuminosityBlockRange> toProcess = pset.getUntrackedParameter<std::vector<LuminosityBlockRange> >("lumisToProcess", std::vector<LuminosityBlockRange>());
187  edm::sortAndRemoveOverlaps(toProcess);
188  uint64_t lumiCount = 0;
189  for (auto const &range : toProcess)
190  {
191  if (range.startRun() != range.endRun()) {break;}
192  if (range.endLumi() >= edm::LuminosityBlockID::maxLuminosityBlockNumber()) {break;}
193  lumiCount += (range.endLumi()-range.startLumi());
194  }
195  // Handle sources deriving from ProducerSourceBase
196  unsigned int eventsPerLumi = pset.getUntrackedParameter<unsigned int>("numberEventsInLuminosityBlock", 0);
197  if ((lumiCount == 0) && (maxEvents > 0) && (eventsPerLumi > 0))
198  {
199  lumiCount = static_cast<unsigned int>(std::ceil(static_cast<float>(maxEvents) / static_cast<float>(eventsPerLumi)));
200  }
201 
202  std::vector<std::string> fileNames = pset.getUntrackedParameter<std::vector<std::string>>("fileNames", std::vector<std::string>());
203  std::stringstream ss_max_files; ss_max_files << fileNames.size();
204  updateChirp("MaxFiles", ss_max_files.str());
205 
206  if (lumiCount > 0)
207  {
208  if (maxLumis < 0) {maxLumis = lumiCount;}
209  if (maxLumis > static_cast<int>(lumiCount))
210  {
211  maxLumis = lumiCount;
212  }
213  }
214  if (maxEvents > 0)
215  {
216  std::stringstream ss_max_events; ss_max_events << maxEvents;
217  updateChirp("MaxEvents", ss_max_events.str());
218  }
219  if (maxLumis > 0)
220  {
221  std::stringstream ss_max_lumis; ss_max_lumis << maxLumis;
222  updateChirp("MaxLumis", ss_max_lumis.str());
223  }
224 
225  m_beginJob = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
226  update();
227 }
228 
229 
230 void
232 {
233  lastUpdate();
234 }
235 
236 
237 bool
239 {
240  if (m_debug) {return true;}
241 
242  return getenv("_CONDOR_CHIRP_CONFIG") && updateChirp("Elapsed", "0");
243 }
244 
245 
246 void
248 {
249  // Note we always update all our statistics to 0 / false / -1
250  // This allows us to overwrite the activities of a previous cmsRun process
251  // within this HTCondor job.
252  updateImpl(0);
253  updateChirp("MaxFiles", "-1");
254  updateChirp("MaxEvents", "-1");
255  updateChirp("MaxLumis", "-1");
256  updateChirp("Done", "false");
257 
259  std::string models;
260  double avgSpeed;
261  if (cpusvc.isAvailable() && cpusvc->cpuInfo(models, avgSpeed)) {
262  updateChirp("CPUModels", models);
263  updateChirp("CPUSpeed", avgSpeed);
264  }
265 }
266 
267 
268 void
270 {
271  time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
272  updateImpl(now - m_lastUpdate);
273  updateChirp("Done", "true");
275  if (!cpusvc.isAvailable()) {
276  std::cout << "At post, CPU service is NOT available.\n";
277  }
278 }
279 
280 
281 void
283 {
284  time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
285  if ((now - m_lastUpdate.load(std::memory_order_relaxed)) > m_updateInterval)
286  {
287  if (!m_shouldUpdate.test_and_set(std::memory_order_acquire))
288  {
289  try
290  {
291  time_t sinceLastUpdate = now - m_lastUpdate;
292  m_lastUpdate = now;
293  updateImpl(sinceLastUpdate);
294  m_shouldUpdate.clear(std::memory_order_release);
295  }
296  catch (...)
297  {
298  m_shouldUpdate.clear(std::memory_order_release);
299  throw;
300  }
301  }
302  }
303 }
304 
305 
306 void
307 CondorStatusService::updateImpl(time_t sinceLastUpdate)
308 {
309  time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
310  time_t jobTime = now-m_beginJob;
311 
313  if (timingsvc.isAvailable()) {
314  updateChirp("TotalCPU", timingsvc->getTotalCPU());
315  }
316 
317  updateChirp("LastUpdate", now);
318 
319  if (!m_events || (m_events > m_lastEventCount))
320  {
321  updateChirp("Events", m_events);
322  }
323 
324  updateChirp("Lumis", m_lumis);
325 
326  updateChirp("Runs", m_runs);
327 
328  updateChirp("Files", m_files);
329 
330  float ema_coeff = 1 - std::exp(-static_cast<float>(sinceLastUpdate)/std::max(std::min(m_emaInterval, static_cast<float>(jobTime)), 1.0f));
331  if (sinceLastUpdate > 0)
332  {
333  updateChirp("Elapsed", jobTime);
334  m_rate = ema_coeff*static_cast<float>(m_events-m_lastEventCount)/static_cast<float>(sinceLastUpdate) + (1.0-ema_coeff)*m_rate;
336  updateChirp("EventRate", m_rate);
337  }
338 
339  // Update storage account information
340  auto const& stats = StorageAccount::summary();
341  uint64_t readOps = 0;
342  uint64_t readVOps = 0;
343  uint64_t readSegs = 0;
344  uint64_t readBytes = 0;
345  uint64_t readTimeTotal = 0;
346  uint64_t writeBytes = 0;
347  uint64_t writeTimeTotal = 0;
348  const auto token = StorageAccount::tokenForStorageClassName("tstoragefile");
349  for (const auto & storage : stats)
350  {
351  // StorageAccount records statistics for both the TFile layer and the
352  // StorageFactory layer. However, the StorageFactory statistics tend to
353  // be more accurate as various backends may alter the incoming read requests
354  // (such as when lazy-download is used).
355  if (storage.first == token.value()) {continue;}
356  for (const auto & counter : storage.second)
357  {
358  if (counter.first == static_cast<int>(StorageAccount::Operation::read))
359  {
360  readOps += counter.second.successes;
361  readSegs++;
362  readBytes += counter.second.amount;
363  readTimeTotal += counter.second.timeTotal;
364  }
365  else if (counter.first == static_cast<int>(StorageAccount::Operation::readv))
366  {
367  readVOps += counter.second.successes;
368  readSegs += counter.second.vector_count;
369  readBytes += counter.second.amount;
370  readTimeTotal += counter.second.timeTotal;
371  }
372  else if ((counter.first == static_cast<int>(StorageAccount::Operation::write)) || (counter.first == static_cast<int>(StorageAccount::Operation::writev)))
373  {
374  writeBytes += counter.second.amount;
375  writeTimeTotal += counter.second.timeTotal;
376  }
377  }
378  }
379  updateChirp("ReadOps", readOps);
380  updateChirp("ReadVOps", readVOps);
381  updateChirp("ReadSegments", readSegs);
382  updateChirp("ReadBytes", readBytes);
383  updateChirp("ReadTimeMsecs", readTimeTotal/(1000*1000));
384  updateChirp("WriteBytes", writeBytes);
385  updateChirp("WriteTimeMsecs", writeTimeTotal/(1000*1000));
386 }
387 
388 
389 template<typename T> bool
391 {
392  std::stringstream ss; ss << value;
393  return updateChirp(key_suffix, ss.str());
394 }
395 
396 
397 bool
399 {
400  std::stringstream ss; ss << "ChirpCMSSW" << m_tag << key_suffix;
401  std::string key = ss.str();
402  if (m_debug)
403  {
404  std::cout << "condor_chirp set_job_attr_delayed " << key << " " << value << std::endl;
405  }
406  int pid = 0;
407  posix_spawn_file_actions_t file_actions;
408  int devnull_fd = open("/dev/null", O_RDWR);
409  if (devnull_fd == -1) {return false;}
410  posix_spawn_file_actions_init(&file_actions);
411  posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 1);
412  posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 2);
413  const std::string chirp_name = "condor_chirp";
414  const std::string set_job_attr = "set_job_attr_delayed";
415  std::vector<const char *> argv;
416  argv.push_back(chirp_name.c_str());
417  argv.push_back(set_job_attr.c_str());
418  argv.push_back(key.c_str());
419  argv.push_back(value.c_str());
420  argv.push_back(NULL);
421  int status = posix_spawnp(&pid, "condor_chirp", &file_actions, NULL, const_cast<char* const*>(&argv[0]), environ);
422  close(devnull_fd);
423  posix_spawn_file_actions_destroy(&file_actions);
424  if (status)
425  {
426  return false;
427  }
428  while ((waitpid(pid, &status, 0) == -1) && errno == -EINTR) {}
429  return status == 0;
430 }
431 
432 
433 void
435 {
437  desc.setComment("Service to update HTCondor with the current CMSSW status.");
438  desc.addOptionalUntracked<unsigned int>("updateIntervalSeconds", m_defaultUpdateInterval)
439  ->setComment("Interval, in seconds, for HTCondor updates");
440  desc.addOptionalUntracked<bool>("debug", false)
441  ->setComment("Enable debugging of this service");
442  desc.addOptionalUntracked<double>("EMAInterval", m_defaultEmaInterval)
443  ->setComment("Interval, in seconds, to calculate event rate over (using EMA)");
444  desc.addOptionalUntracked<std::string>("tag")
445  ->setComment("Identifier tag for this process (a value of 'Foo' results in ClassAd attributes of the form 'ChirpCMSSWFoo*')");
446  descriptions.add("CondorStatusService", desc);
447 }
448 
449 
452 
T getUntrackedParameter(std::string const &, T const &) const
CondorStatusService(ParameterSet const &pset, edm::ActivityRegistry &ar)
void preSourceConstruction(ModuleDescription const &md, int maxEvents, int maxLumis, int maxSecondsUntilRampdown)
#define DEFINE_FWK_SERVICE_MAKER(concrete, maker)
Definition: ServiceMaker.h:117
void runPost(GlobalContext const &)
void watchPostEndJob(PostEndJob::slot_type const &iSlot)
void beginPre(PathsAndConsumesOfModulesBase const &, ProcessContext const &processContext)
virtual bool cpuInfo(std::string &models, double &avgSpeed)=0
CPU information - the models present and average speed.
void updateImpl(time_t secsSinceLastUpdate)
std::atomic< std::uint_least64_t > m_events
edm::serviceregistry::AllArgsMaker< edm::service::CondorStatusService > CondorStatusServiceMaker
ParameterSet const & getParameterSet(ParameterSetID const &id)
void watchPostEvent(PostEvent::slot_type const &iSlot)
static LuminosityBlockNumber_t maxLuminosityBlockNumber()
static const StorageStats & summary(void)
void lumiPost(GlobalContext const &)
bool exists(std::string const &parameterName) const
checks if a parameter exists
#define NULL
Definition: scimark2.h:8
ParameterSet getUntrackedParameterSet(std::string const &name, ParameterSet const &defaultValue) const
std::atomic< std::uint_least64_t > m_lumis
#define constexpr
ParameterSetID const & parameterSetID() const
void setComment(std::string const &value)
void watchPostCloseFile(PostCloseFile::slot_type const &iSlot)
static StorageClassToken tokenForStorageClassName(std::string const &iName)
CondorStatusService & operator=(const CondorStatusService &)=delete
std::atomic< std::uint_least64_t > m_runs
bool isAvailable() const
Definition: Service.h:46
double f[11][100]
void watchPostGlobalEndLumi(PostGlobalEndLumi::slot_type const &iSlot)
T min(T a, T b)
Definition: MathUtil.h:58
string key
FastSim: produces sample of signal events, overlayed with premixed minbias events.
void eventPost(StreamContext const &iContext)
unsigned long long uint64_t
Definition: Time.h:15
ParameterSet const & getParameterSet(std::string const &) const
tuple pid
Definition: sysUtil.py:22
void watchPostGlobalEndRun(PostGlobalEndRun::slot_type const &iSlot)
void add(std::string const &label, ParameterSetDescription const &psetDescription)
void watchPreBeginJob(PreBeginJob::slot_type const &iSlot)
convenience function for attaching to signal
virtual double getTotalCPU() const =0
bool updateChirp(const std::string &key_suffix, const T &value)
bool isValid() const
Definition: Hash.h:150
static std::atomic< unsigned int > counter
std::vector< EventRange > & sortAndRemoveOverlaps(std::vector< EventRange > &eventRange)
Definition: EventRange.cc:102
tuple cout
Definition: gather_cfg.py:145
ParameterDescriptionBase * addOptionalUntracked(U const &iLabel, T const &value)
tuple fileNames
Definition: LaserDQM_cfg.py:34
volatile std::atomic< bool > shutdown_flag false
long double T
static void fillDescriptions(ConfigurationDescriptions &descriptions)
tuple status
Definition: mps_update.py:57
void watchPostBeginJob(PostBeginJob::slot_type const &iSlot)
convenience function for attaching to signal
std::atomic< std::uint_least64_t > m_files
void filePost(std::string const &, bool)