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 "FWCore/Utilities/interface/InputType.h"

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

#include "boost/scoped_ptr.hpp"
#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::set<unsigned int> lumiSections;
      };

      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);

        /*
         * 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::vector<InputFile> inputFiles_;
        tbb::concurrent_vector<InputFile> inputFilesSecSource_;
        std::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_;
        std::ostream* ost_;
      };

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

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

      ~JobReport();

      void childAfterFork(std::string const& jobReportFile, unsigned int childIndex, unsigned int numberOfChildren);

      void parentBeforeFork(std::string const& jobReportFile, unsigned int numberOfChildren);

      void parentAfterFork(std::string const& jobReportFile);

      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, unsigned int run, unsigned int event);

      void outputFileClosed(Token fileToken);

      void reportSkippedEvent(unsigned int run, unsigned int event);

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

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

      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:
      boost::scoped_ptr<JobReportImpl>& impl() {return impl_;}

   private:
      boost::scoped_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);
}
#endif