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 "tbb/concurrent_unordered_map.h"
#include "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;
      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_;
      tbb::concurrent_vector<InputFile> inputFilesSecSource_;
      tbb::concurrent_vector<OutputFile> outputFiles_;
      std::map<std::string, long long> readBranches_;
      std::map<std::string, long long> readBranchesSecFile_;
      tbb::concurrent_unordered_map<std::string, AtomicLongLong> readBranchesSecSource_;
      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 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:
    edm::propagate_const<std::unique_ptr<JobReportImpl>> impl_;
    std::mutex write_mutex;
  };
 
  std::ostream& operator<<(std::ostream& os, JobReport::InputFile const& f);
  std::ostream& operator<<(std::ostream& os, JobReport::OutputFile const& f);
}  // namespace edm
#endif