RamdiskMonitor.cc

Go to the documentation of this file.

#include <filesystem>
#include <map>
#include <vector>
#include <sys/stat.h>

#include <fmt/printf.h>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/range.hpp>
#include <boost/regex.hpp>

#include "DQMServices/Core/interface/DQMOneEDAnalyzer.h"
#include "DQMServices/Core/interface/DQMStore.h"
#include "DQMServices/StreamerIO/plugins/DQMFileIterator.h"
#include "FWCore/Framework/interface/Event.h"
#include "FWCore/Framework/interface/EventSetup.h"
#include "FWCore/MessageLogger/interface/MessageLogger.h"
#include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
#include "FWCore/ParameterSet/interface/ParameterSet.h"
#include "FWCore/ParameterSet/interface/ParameterSetDescription.h"

namespace dqm {
  namespace rdm {
    struct Empty {};
  }  // namespace rdm
  class RamdiskMonitor : public DQMOneEDAnalyzer<edm::LuminosityBlockCache<rdm::Empty>> {
  public:
    RamdiskMonitor(const edm::ParameterSet &ps);
    ~RamdiskMonitor() override;
    static void fillDescriptions(edm::ConfigurationDescriptions &descriptions);

  protected:
    void bookHistograms(DQMStore::IBooker &, edm::Run const &, edm::EventSetup const &) override;
    std::shared_ptr<rdm::Empty> globalBeginLuminosityBlock(edm::LuminosityBlock const &lumi,
                                                           edm::EventSetup const &eSetup) const override;
    void globalEndLuminosityBlock(edm::LuminosityBlock const &lumi, edm::EventSetup const &eSetup) final {}
    void analyze(edm::Event const &e, edm::EventSetup const &eSetup) override{};

    void analyzeFile(std::string fn, unsigned int run, unsigned int lumi, std::string label) const;
    double getRunTimestamp() const;

    const unsigned int runNumber_;
    const std::string runInputDir_;
    const std::vector<std::string> streamLabels_;
    const std::string runPath_;

    struct StreamME {
      MonitorElement *eventsAccepted;
      MonitorElement *eventsProcessed;

      MonitorElement *deliveryDelayMTime;
      MonitorElement *deliveryDelayCTime;
    };

    std::map<std::string, StreamME> streams_;
    mutable std::set<std::string> filesSeen_;
    mutable double global_start_ = 0.;

    static constexpr double LUMI = 23.310893056;
  };

  RamdiskMonitor::RamdiskMonitor(const edm::ParameterSet &ps)
      : runNumber_{ps.getUntrackedParameter<unsigned int>("runNumber")},
        runInputDir_{ps.getUntrackedParameter<std::string>("runInputDir")},
        streamLabels_{ps.getUntrackedParameter<std::vector<std::string>>("streamLabels")},
        runPath_{fmt::sprintf("%s/run%06d", runInputDir_, runNumber_)}

  {}

  RamdiskMonitor::~RamdiskMonitor(){};

  void RamdiskMonitor::bookHistograms(DQMStore::IBooker &ib, edm::Run const &, edm::EventSetup const &) {
    for (const auto &stream : streamLabels_) {
      edm::LogInfo("RamdiskMonitor") << "Booking: " << stream;

      ib.cd();
      ib.setCurrentFolder(std::string("Info/RamdiskMonitor/") + stream + "/");

      StreamME m;

      m.eventsAccepted = ib.book1D("EventAccepted", "# of accepted events per lumi", 4, 0., 4.);
      m.eventsProcessed = ib.book1D("EventProcessed", "# of processed events per lumi", 4, 0., 4.);
      m.deliveryDelayMTime =
          ib.book1D("DeliveryDelayMTime", "Observed delivery delay for the data file (mtime).", 4, 0., 4.);
      m.deliveryDelayCTime =
          ib.book1D("DeliveryDelayCTime", "Observed delivery delay for the data file (ctime).", 4, 0., 4.);

      m.eventsAccepted->getTH1F()->SetCanExtend(TH1::kXaxis);
      m.eventsProcessed->getTH1F()->SetCanExtend(TH1::kXaxis);
      m.deliveryDelayMTime->getTH1F()->SetCanExtend(TH1::kXaxis);
      m.deliveryDelayCTime->getTH1F()->SetCanExtend(TH1::kXaxis);

      m.eventsAccepted->setAxisTitle("Luminosity Section", 1);
      m.eventsProcessed->setAxisTitle("Luminosity Section", 1);
      m.deliveryDelayMTime->setAxisTitle("Luminosity Section", 1);
      m.deliveryDelayCTime->setAxisTitle("Luminosity Section", 1);

      m.eventsAccepted->setAxisTitle("Number of events", 2);
      m.eventsProcessed->setAxisTitle("Number of events", 2);
      m.deliveryDelayMTime->setAxisTitle("Delay (s.)", 2);
      m.deliveryDelayCTime->setAxisTitle("Delay (s.)", 2);

      streams_[stream] = m;
    }
  };

  double RamdiskMonitor::getRunTimestamp() const {
    if (global_start_ != 0)
      return global_start_;

    std::string run_global = fmt::sprintf("%s/.run%06d.global", runInputDir_, runNumber_);
    struct stat st;
    if (::stat(run_global.c_str(), &st) != 0) {
      edm::LogWarning("RamdiskMonitor") << "Stat failed: " << run_global;
      return 0.;
    }

    global_start_ = st.st_mtime;
    edm::LogPrint("RamdiskMonitor") << "Run start timestamp: " << global_start_;
    return global_start_;
  };

  void RamdiskMonitor::analyzeFile(std::string fn, unsigned int run, unsigned int lumi, std::string label) const {
    using LumiEntry = dqmservices::DQMFileIterator::LumiEntry;

    // we are disabled, at least for this stream
    if (streams_.empty())
      return;

    auto itStream = streams_.find(label);
    if (itStream == streams_.end()) {
      edm::LogPrint("RamdiskMonitor") << "Stream not monitored [" << label << "]: " << fn;
      return;
    }

    StreamME m = itStream->second;

    // decode json and fill in some histograms
    LumiEntry lumi_jsn = LumiEntry::load_json(runPath_, fn, lumi, -1);
    m.eventsAccepted->setBinContent(lumi, lumi_jsn.n_events_accepted);
    m.eventsProcessed->setBinContent(lumi, lumi_jsn.n_events_processed);

    // collect stat struct and calculate mtimes
    struct stat st;
    if (::stat(fn.c_str(), &st) != 0) {
      edm::LogWarning("RamdiskMonitor") << "Stat failed: " << fn;
      return;
    }

    // get start offset (from .global)
    // abort the calculation if it does not exist
    double start_offset = getRunTimestamp();
    if (start_offset <= 0)
      return;

    // check fff_dqmtools (separate repository)
    // for calculation details
    double mtime = st.st_mtime;
    double ctime = st.st_ctime;

    // timeout from the begging of the run
    double start_offset_mtime = mtime - start_offset - LUMI;
    double start_offset_ctime = ctime - start_offset - LUMI;
    double lumi_offset = (lumi - 1) * LUMI;

    // timeout from the time we think this lumi happenned
    double delay_mtime = start_offset_mtime - lumi_offset;
    double delay_ctime = start_offset_ctime - lumi_offset;

    m.deliveryDelayMTime->setBinContent(lumi, delay_mtime);
    m.deliveryDelayCTime->setBinContent(lumi, delay_ctime);
  };

  std::shared_ptr<dqm::rdm::Empty> RamdiskMonitor::globalBeginLuminosityBlock(edm::LuminosityBlock const &,
                                                                              edm::EventSetup const &eSetup) const {
    // search filesystem to find available lumi section files
    using std::filesystem::directory_entry;
    using std::filesystem::directory_iterator;

    directory_iterator dend;
    for (directory_iterator di(runPath_); di != dend; ++di) {
      const boost::regex fn_re("run(\\d+)_ls(\\d+)_([a-zA-Z0-9]+)(_.*)?\\.jsn");

      const std::string filename = di->path().filename().string();
      const std::string fn = di->path().string();

      if (filesSeen_.find(filename) != filesSeen_.end()) {
        continue;
      }

      boost::smatch result;
      if (boost::regex_match(filename, result, fn_re)) {
        unsigned int run = std::stoi(result[1]);
        unsigned int lumi = std::stoi(result[2]);
        std::string label = result[3];

        filesSeen_.insert(filename);

        if (run != runNumber_)
          continue;

        // check if this is EoR
        if ((lumi == 0) && (label == "EoR")) {
          // do not handle
          continue;
        }

        try {
          this->analyzeFile(fn, run, lumi, label);
        } catch (const std::exception &e) {
          // it's likely we have read it too soon
          filesSeen_.erase(filename);

          std::string msg("Found, tried to load the json, but failed (");
          msg += e.what();
          msg += "): ";
          edm::LogWarning("RamdiskMonitor") << msg;
        }
      }
    }

    // @TODO lookup info for the current lumi
    return std::shared_ptr<dqm::rdm::Empty>();
  }

  void RamdiskMonitor::fillDescriptions(edm::ConfigurationDescriptions &d) {
    edm::ParameterSetDescription desc;

    desc.setComment(
        "Analyses file timestams in the /fff/ramdisk and creates monitor "
        "elements.");

    desc.addUntracked<std::vector<std::string>>("streamLabels")->setComment("List of streams to monitor.");

    desc.addUntracked<unsigned int>("runNumber")->setComment("Run number passed via configuration file.");

    desc.addUntracked<std::string>("runInputDir")->setComment("Directory where the DQM files will appear.");

    d.add("RamdiskMonitor", desc);
  }

}  // namespace dqm

#include "FWCore/Framework/interface/MakerMacros.h"

typedef dqm::RamdiskMonitor RamdiskMonitor;
DEFINE_FWK_MODULE(RamdiskMonitor);