JobReport.h

Go to the documentation of this file.

#ifndef FWCore_MessageLogger_JobReport_h
#define FWCore_MessageLogger_JobReport_h
// -*- C++ -*-
//
// Package:     MessageLogger
// Class  :     JobReport
//
//
// Original Author:  Marc Paterno
//

/*
Changes Log 1: 2009/01/14 10:29:00, Natalia Garcia Nebot
        Modified and added some methods to report CPU and memory information from /proc/cpuinfo
        and /proc/meminfo files and Memory statistics
*/

#include "DataFormats/Provenance/interface/RunLumiEventNumber.h"
#include "FWCore/Utilities/interface/InputType.h"
#include "FWCore/Utilities/interface/get_underlying_safe.h"

#include <atomic>
#include <cstddef>
#include <iosfwd>
#include <map>
#include <memory>
#include <mutex>
#include <set>
#include <string>
#include <vector>

#include "oneapi/tbb/concurrent_unordered_map.h"
#include "oneapi/tbb/concurrent_vector.h"

namespace edm {

  class JobReport {
  public:
    typedef unsigned int RunNumber;
    typedef std::size_t Token;

    struct RunReport {
      RunNumber runNumber;
      std::map<unsigned int, unsigned long> lumiSectionsToNEvents;
    };

    struct InputFile {
      typedef std::vector<std::string> StringVector;

      std::string logicalFileName;
      std::string physicalFileName;
      std::string catalog;
      std::string inputType;             // primaryFiles, secondaryFiles, mixingFiles
      std::string inputSourceClassName;  // class which created the file
      std::string moduleLabel;           // name of class instance
      std::string guid;
      std::size_t numEventsRead;
      StringVector branchNames;
      std::map<RunNumber, RunReport> runReports;
      bool fileHasBeenClosed;
      std::set<std::string> fastClonedBranches;
    };

    struct OutputFile {
      typedef InputFile::StringVector StringVector;

      std::string logicalFileName;
      std::string physicalFileName;
      std::string catalog;
      std::string outputModuleClassName;
      std::string moduleLabel;  // name of class instance
      std::string guid;
      std::string dataType;
      std::string branchHash;
      std::size_t numEventsWritten;
      StringVector branchNames;
      std::vector<Token> contributingInputs;
      oneapi::tbb::concurrent_vector<Token> contributingInputsSecSource;
      std::map<std::string, bool> fastCopyingInputs;
      std::map<RunNumber, RunReport> runReports;
      bool fileHasBeenClosed;
    };

    class AtomicLongLong {
    public:
      AtomicLongLong() : value_(0) {}
      AtomicLongLong(AtomicLongLong const& r) : value_(r.value_.load()) {}
      std::atomic<long long>& value() { return value_; }
      std::atomic<long long> const& value() const { return value_; }

    private:
      std::atomic<long long> value_;
    };

    struct JobReportImpl {
      JobReportImpl& operator=(JobReportImpl const&) = delete;
      JobReportImpl(JobReportImpl const&) = delete;

      InputFile& getInputFileForToken(InputType inputType, Token t);
      OutputFile& getOutputFileForToken(Token t);

      /*
         * Add the input file token provided to every output
         * file currently available.
         * Used whenever a new input file is opened, it's token
         * is added to all open output files as a contributor
         */
      void insertInputForOutputs(InputType inputType, Token t);

      /*
         * Associate a Lumi Section to all open output files
         *
         */
      void associateLumiSection(JobReport::Token token,
                                unsigned int runNumber,
                                unsigned int lumiSection,
                                unsigned long nEvents);

      /*
         * Associate a Lumi Section to all open input files
         *
         */
      void associateInputLumiSection(unsigned int runNumber, unsigned int lumiSection);

      /*
         * Associate a run to all open output files
         */
      void associateRun(JobReport::Token token, unsigned int runNumber);

      /*
         * Associate a run to all open output files
         */
      void associateInputRun(unsigned int runNumber);

      /*
         * Write an InputFile object to the Logger
         * Generate XML string for InputFile instance and dispatch to
         * job report via MessageLogger
         */
      void writeInputFile(InputFile const& f);

      /*
         * Write an OutputFile object to the Logger
         * Generate an XML string for the OutputFile provided and
         * dispatch it to the logger
         * Contributing input tokens are resolved to the input LFN and PFN
         *
         * TODO: We have not yet addressed the issue where we cleanup not
         * contributing input files.
         * Also, it is possible to get fake input to output file mappings
         * if an input file is open already when a new output file is opened
         * but the input gets closed without contributing events to the
         * output file due to filtering etc.
         *
         */
      void writeOutputFile(OutputFile const& f);

      /*
         *  Flush all open files to logger in event of a problem.
         */
      void flushFiles(void);

      JobReportImpl(std::ostream* iOst) : printedReadBranches_(false), ost_(iOst) {}

      std::ostream const* ost() const { return get_underlying_safe(ost_); }
      std::ostream*& ost() { return get_underlying_safe(ost_); }

      std::vector<InputFile> inputFiles_;
      oneapi::tbb::concurrent_vector<InputFile> inputFilesSecSource_;
      oneapi::tbb::concurrent_vector<OutputFile> outputFiles_;
      std::map<std::string, long long> readBranches_;
      std::map<std::string, long long> readBranchesSecFile_;
      oneapi::tbb::concurrent_unordered_map<std::string, AtomicLongLong> readBranchesSecSource_;
      std::atomic<bool> printedReadBranches_;
      std::vector<InputFile>::size_type lastOpenedPrimaryInputFile_;
      edm::propagate_const<std::ostream*> ost_;
    };

    JobReport();
    //Does not take ownership of pointer
    JobReport(std::ostream* outputStream);

    JobReport& operator=(JobReport const&) = delete;
    JobReport(JobReport const&) = delete;

    ~JobReport();

    Token inputFileOpened(std::string const& physicalFileName,
                          std::string const& logicalFileName,
                          std::string const& catalog,
                          std::string const& inputType,
                          std::string const& inputSourceClassName,
                          std::string const& moduleLabel,
                          std::string const& guid,
                          std::vector<std::string> const& branchNames);

    void eventReadFromFile(InputType inputType, Token fileToken);

    // CMS-THREADING Current implementation requires an instance of an
    // OuputModule run on only one thread at a time.
    void reportDataType(Token fileToken, std::string const& dataType);

    void inputFileClosed(InputType inputType, Token fileToken);

    Token outputFileOpened(std::string const& physicalFileName,
                           std::string const& logicalFileName,
                           std::string const& catalog,
                           std::string const& outputModuleClassName,
                           std::string const& moduleLabel,
                           std::string const& guid,
                           std::string const& dataType,
                           std::string const& branchHash,
                           std::vector<std::string> const& branchNames);

    // CMS-THREADING Current implementation requires an instance of an
    // OuputModule run on only one thread at a time.
    void eventWrittenToFile(Token fileToken, RunNumber_t run, EventNumber_t event);

    void outputFileClosed(Token fileToken);

    void reportSkippedEvent(RunNumber_t run, EventNumber_t event);

    void reportRunNumber(JobReport::Token token, unsigned int run);

    void reportLumiSection(JobReport::Token token, unsigned int run, unsigned int lumiSectId, unsigned long nEvents = 0);

    void reportInputLumiSection(unsigned int run, unsigned int lumiSectId);

    void reportInputRunNumber(unsigned int run);

    void reportError(std::string const& shortDesc, std::string const& longDesc, int const& exitCode);

    void reportShutdownSignal();

    void reportSkippedFile(std::string const& pfn, std::string const& lfn);

    void reportFallbackAttempt(std::string const& pfn, std::string const& lfn, std::string const& err);

    void reportAnalysisFile(std::string const& fileName, std::map<std::string, std::string> const& fileData);

    void reportMemoryInfo(std::vector<std::string> const& memoryData);

    void reportMessageInfo(std::map<std::string, double> const& messageData);

    void reportReadBranches();

    void reportReadBranch(InputType inputType, std::string const& branchName);

    void reportFastClonedBranches(std::set<std::string> const& fastClonedBranches, long long nEvents);

    void reportRandomStateFile(std::string const& name);

    /*
       * Report information about fast copying. Called for each open output file
       * whenever an input file is opened.
       */
    void reportFastCopyingStatus(Token t, std::string const& inputFileName, bool fastCopying);

    void reportPerformanceSummary(std::string const& metricClass, std::map<std::string, std::string> const& metrics);

    void reportPerformanceForModule(std::string const& metricClass,
                                    std::string const& moduleName,
                                    std::map<std::string, std::string> const& metrics);

    std::string dumpFiles(void);

  protected:
    edm::propagate_const<std::unique_ptr<JobReportImpl>>& impl() { return impl_; }

  private:
    void temporarilyCloseXML();
    edm::propagate_const<std::unique_ptr<JobReportImpl>> impl_;
    std::mutex write_mutex;
    bool errorLogged_ = false;
  };

  std::ostream& operator<<(std::ostream& os, JobReport::InputFile const& f);
  std::ostream& operator<<(std::ostream& os, JobReport::OutputFile const& f);
}  // namespace edm
#endif