StreamerIO/plugins/DQMMonitoringService.cc

0001 #include "DQMMonitoringService.h"
0002
0003 #include "DataFormats/Provenance/interface/Timestamp.h"
0004 #include "FWCore/MessageLogger/interface/MessageLogger.h"
0005 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0006 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
0007 #include "FWCore/ServiceRegistry/interface/GlobalContext.h"
0008 #include "FWCore/Utilities/interface/StreamID.h"
0009
0010 #include <cstdlib>
0011 #include <ctime>
0012 #include <exception>
0013 #include <iostream>
0014
0015 #include <boost/property_tree/json_parser.hpp>
0016
0017 #include <fmt/printf.h>
0018
0019 /*
0020  * This service is very similar to the FastMonitoringService in the HLT,
0021  * except that it is used for monitoring online DQM applications
0022  */
0023
0024 namespace dqmservices {
0025
0026   DQMMonitoringService::DQMMonitoringService(const edm::ParameterSet& pset, edm::ActivityRegistry& ar) {
0027     const char* x = std::getenv("DQM2_SOCKET");
0028     if (x) {
0029       std::cerr << "Monitoring pipe: " << x << std::endl;
0030       mstream_.connect(boost::asio::local::stream_protocol::endpoint(x));
0031     } else {
0032       std::cerr << "Monitoring file not found, disabling." << std::endl;
0033     }
0034
0035     // init counters
0036     nevents_ = 0;
0037
0038     last_lumi_ = 0;
0039     last_lumi_nevents_ = 0;
0040     last_lumi_time_ = std::chrono::high_resolution_clock::now();
0041
0042     run_ = 0;
0043     lumi_ = 0;
0044
0045     ar.watchPreGlobalBeginLumi(this, &DQMMonitoringService::evLumi);
0046     ar.watchPreSourceEvent(this, &DQMMonitoringService::evEvent);
0047   }
0048
0049   void DQMMonitoringService::outputLumiUpdate() {
0050     auto now = std::chrono::high_resolution_clock::now();
0051
0052     boost::property_tree::ptree doc;
0053
0054     // these might be different than the numbers we want to report
0055     // rate/stats per lumi are calculated from last_*_ fields
0056     doc.put("cmssw_run", run_);
0057     doc.put("cmssw_lumi", lumi_);
0058     doc.put("events_total", nevents_);
0059
0060     // do statistics for the last (elapsed) ls
0061     auto lumi_millis = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_lumi_time_).count();
0062     auto lumi_events = nevents_ - last_lumi_nevents_;
0063     auto lumi_last = last_lumi_;
0064
0065     if ((lumi_last > 0) && (lumi_millis > 0)) {
0066       doc.put("lumi_number", lumi_last);
0067       doc.put("lumi_events", lumi_events);
0068       doc.put("lumi_duration_ms", lumi_millis);
0069
0070       float rate = ((float)lumi_events * 1000) / lumi_millis;
0071       doc.put("events_rate", rate);
0072
0073       // also save the history entry
0074       boost::property_tree::ptree plumi;
0075       plumi.put("n", lumi_last);
0076       plumi.put("nevents", lumi_events);
0077       plumi.put("nmillis", lumi_millis);
0078       plumi.put("rate", rate);
0079
0080       std::time_t hkey = std::time(nullptr);
0081       doc.add_child(fmt::sprintf("extra.lumi_stats.%d", hkey), plumi);
0082     }
0083
0084     outputUpdate(doc);
0085   }
0086
0087   void DQMMonitoringService::evLumi(edm::GlobalContext const& iContext) {
0088     // these might be different than the numbers we want to report
0089     // rate/stats per lumi are calculated from last_*_ fields
0090     run_ = iContext.luminosityBlockID().run();
0091     lumi_ = iContext.luminosityBlockID().luminosityBlock();
0092
0093     outputLumiUpdate();
0094
0095     last_lumi_time_ = std::chrono::high_resolution_clock::now();
0096     last_lumi_nevents_ = nevents_;
0097     last_lumi_ = lumi_;
0098   }
0099
0100   void DQMMonitoringService::evEvent(edm::StreamID const& iContext) {
0101     nevents_ += 1;
0102     tryUpdate();
0103   }
0104
0105   void DQMMonitoringService::outputUpdate(boost::property_tree::ptree& doc) {
0106     if (!mstream_)
0107       return;
0108
0109     try {
0110       last_update_time_ = std::chrono::high_resolution_clock::now();
0111       doc.put("update_timestamp", std::time(nullptr));
0112
0113       write_json(mstream_, doc, false);
0114       mstream_.flush();
0115     } catch (std::exception const& exc) {
0116       LogDebug("DQMMonitoringService") << "Exception thrown in outputUpdate method: " << exc.what();
0117     }
0118   }
0119
0120   void DQMMonitoringService::keepAlive() {
0121     if (!mstream_)
0122       return;
0123
0124     mstream_ << "\n";
0125     mstream_.flush();
0126
0127     tryUpdate();
0128   }
0129
0130   void DQMMonitoringService::tryUpdate() {
0131     if (!mstream_)
0132       return;
0133
0134     // sometimes we don't see any transition for a very long time
0135     // but we still want updates
0136     // luckily, keepAlive is called rather often by the input source
0137     auto const now = std::chrono::high_resolution_clock::now();
0138     auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_update_time_).count();
0139     if (millis >= (25 * 1000)) {
0140       outputLumiUpdate();
0141     }
0142   }
0143
0144 }  // namespace dqmservices
0145
0146 #include "FWCore/ServiceRegistry/interface/ServiceMaker.h"
0147
0148 using dqmservices::DQMMonitoringService;
0149 DEFINE_FWK_SERVICE(DQMMonitoringService);