edm::MergeableRunProductMetadata Class Reference

#include <MergeableRunProductMetadata.h>

Inheritance diagram for edm::MergeableRunProductMetadata:

Classes
class	MetadataForProcess

Public Types
enum	MergeDecision { MERGE, REPLACE, IGNORE }

Public Member Functions
void	addEntryToStoredMetadata (StoredMergeableRunProductMetadata &) const
MergeDecision	getMergeDecision (std::string const &processThatCreatedProduct) const
std::string const &	getProcessName (unsigned int index) const
bool	gotLumisFromIndexIntoFile () const
bool	knownImproperlyMerged (std::string const &processThatCreatedProduct) const override
std::vector< LuminosityBlockNumber_t > const &	lumisFromIndexIntoFile () const
oneapi::tbb::concurrent_vector< LuminosityBlockNumber_t > const &	lumisProcessed () const
	MergeableRunProductMetadata (MergeableRunProductProcesses const &)
MetadataForProcess const *	metadataForOneProcess (std::string const &processName) const
std::vector< MetadataForProcess > const &	metadataForProcesses () const
void	postWriteRun ()
void	preReadFile ()
void	preWriteRun ()
void	readRun (long long inputRunEntry, StoredMergeableRunProductMetadata const &inputStoredMergeableRunProductMetadata, IndexIntoFileItrHolder const &inputIndexIntoFileItr)
void	writeLumi (LuminosityBlockNumber_t lumi)
	~MergeableRunProductMetadata () override
Public Member Functions inherited from edm::MergeableRunProductMetadataBase
virtual	~MergeableRunProductMetadataBase ()

Private Member Functions
bool	addProcess (StoredMergeableRunProductMetadata &storedMetadata, MetadataForProcess const &metadataForProcess, unsigned int storedProcessIndex, unsigned long long beginProcess, unsigned long long endProcess) const
void	mergeLumisFromIndexIntoFile ()

Private Attributes
bool	gotLumisFromIndexIntoFile_ = false
std::vector< LuminosityBlockNumber_t >	lumisFromIndexIntoFile_
oneapi::tbb::concurrent_vector< LuminosityBlockNumber_t >	lumisProcessed_
MergeableRunProductProcesses const *	mergeableRunProductProcesses_
std::vector< MetadataForProcess >	metadataForProcesses_

Detailed Description

This class holds information used to decide how to merge together mergeable run products when multiple run entries with the same run number and ProcessHistoryID are read from input files contiguously.

Most of the information here is associated with the current run being processed. Most of it is cleared when a new run is started. If multiple runs are being processed concurrently, then there will be an object instantiated for each concurrent run. The primary RunPrincipal for the current run owns the object.

This class gets information from the input file from the StoredMergeableRunProductMetadata object and IndexIntoFile object. It uses that information to make the decision how to merge run products read from data, and puts information into the StoredMergeableRunProductMetadata written into the output file to use in later processing steps.

If there are SubProcesses, they use the same object as the top level process because they share the same input.

There is a TWIKI page on the Framework page of the Software Guide which explains the details about how this works. There are significant limitations related to what the Framework does and does not do managing mergeable run products.

Author

W. David Dagenhart, created 9 July, 2018

Definition at line 51 of file MergeableRunProductMetadata.h.

Member Enumeration Documentation

MergeDecision

enum edm::MergeableRunProductMetadata::MergeDecision

Enumerator
MERGE
REPLACE
IGNORE

Definition at line 53 of file MergeableRunProductMetadata.h.

{ MERGE, REPLACE, IGNORE };

Constructor & Destructor Documentation

MergeableRunProductMetadata()

edm::MergeableRunProductMetadata::MergeableRunProductMetadata

(

MergeableRunProductProcesses const &

mergeableRunProductProcesses

)

Definition at line 14 of file MergeableRunProductMetadata.cc.

      : mergeableRunProductProcesses_(&mergeableRunProductProcesses),
        metadataForProcesses_(mergeableRunProductProcesses.size()) {}

~MergeableRunProductMetadata()

edm::MergeableRunProductMetadata::~MergeableRunProductMetadata ( )

override

Definition at line 19 of file MergeableRunProductMetadata.cc.

{}

Member Function Documentation

addEntryToStoredMetadata()

void edm::MergeableRunProductMetadata::addEntryToStoredMetadata

(

StoredMergeableRunProductMetadata &

storedMetadata

)

const

Definition at line 162 of file MergeableRunProductMetadata.cc.

References addProcess(), mergeableRunProductProcesses_, metadataForProcesses_, edm::MergeableRunProductProcesses::processesWithMergeableRunProducts(), edm::StoredMergeableRunProductMetadata::processesWithMergeableRunProducts(), edm::StoredMergeableRunProductMetadata::singleRunEntries(), and edm::StoredMergeableRunProductMetadata::singleRunEntryAndProcesses().

                                                                                                                    {
    if (metadataForProcesses_.empty()) {
      return;
    }

    std::vector<std::string> const& storedProcesses = storedMetadata.processesWithMergeableRunProducts();
    if (storedProcesses.empty()) {
      return;
    }

    unsigned long long beginProcess = storedMetadata.singleRunEntryAndProcesses().size();
    unsigned long long endProcess = beginProcess;

    std::vector<std::string> const& processesWithMergeableRunProducts =
        mergeableRunProductProcesses_->processesWithMergeableRunProducts();

    for (unsigned int storedProcessIndex = 0; storedProcessIndex < storedProcesses.size(); ++storedProcessIndex) {
      // Look for a matching process. It is intentional that no process
      // is added when there is no match. storedProcesses only contains
      // processes which created mergeable run products selected by the
      // output module to be written out. processesWithMergeableRunProducts_
      // only has processes which created mergeable run products that were
      // read from the input data files. Note storedProcesses may be
      // missing processes because the output module dropped products.
      // The other vector may be missing processes because of SubProcesses.
      for (unsigned int transientProcessIndex = 0; transientProcessIndex < processesWithMergeableRunProducts.size();
           ++transientProcessIndex) {
        // This string comparison could be optimized away by storing an index mapping in
        // OutputModuleBase calculated once early in a job. (? Given how rare
        // mergeable run products are this optimization may not be worth doing)
        if (processesWithMergeableRunProducts[transientProcessIndex] == storedProcesses[storedProcessIndex]) {
          if (addProcess(storedMetadata,
                         metadataForProcesses_.at(transientProcessIndex),
                         storedProcessIndex,
                         beginProcess,
                         endProcess)) {
            ++endProcess;
          }
          break;
        }
      }
    }
    storedMetadata.singleRunEntries().emplace_back(beginProcess, endProcess);
  }

addProcess()

bool edm::MergeableRunProductMetadata::addProcess ( StoredMergeableRunProductMetadata &  storedMetadata, MetadataForProcess const &  metadataForProcess, unsigned int  storedProcessIndex, unsigned long long  beginProcess, unsigned long long  endProcess  ) const

private

Definition at line 207 of file MergeableRunProductMetadata.cc.

References edm::MergeableRunProductMetadata::MetadataForProcess::allLumisProcessed(), edm::StoredMergeableRunProductMetadata::allValidAndUseIndexIntoFile(), edm::StoredMergeableRunProductMetadata::SingleRunEntryAndProcess::beginLumi(), edm::StoredMergeableRunProductMetadata::SingleRunEntryAndProcess::endLumi(), newFWLiteAna::found, edm::MergeableRunProductMetadata::MetadataForProcess::lumis(), edm::StoredMergeableRunProductMetadata::lumis(), edm::StoredMergeableRunProductMetadata::singleRunEntryAndProcesses(), and edm::MergeableRunProductMetadata::MetadataForProcess::valid().

Referenced by addEntryToStoredMetadata().

                                                                                    {
    if (metadataForProcess.valid() && metadataForProcess.allLumisProcessed()) {
      return false;
    }

    storedMetadata.allValidAndUseIndexIntoFile() = false;

    unsigned long long iBeginLumi = 0;
    unsigned long long iEndLumi = 0;

    // See if we need to store the set of lumi numbers corresponding
    // to this process and run entry. If they were all processed then
    // we can just get the lumi numbers out of IndexIntoFile and do
    // not need to store them here
    if (!metadataForProcess.allLumisProcessed()) {
      // If we need to store the numbers, then we can check to
      // make sure this does not duplicate the lumi numbers we
      // stored for another process. If we did then we can just
      // just reference same indices and avoid copying a duplicate
      // sequence of lumi numbers. It is sufficient to check the
      // size only. As you go back in the time sequence of processes
      // the only thing that can happen is more lumi numbers appear
      // at steps where a run was only partially processed.
      bool found = false;
      for (unsigned long long kProcess = beginProcess; kProcess < endProcess; ++kProcess) {
        StoredMergeableRunProductMetadata::SingleRunEntryAndProcess const& storedSingleRunEntryAndProcess =
            storedMetadata.singleRunEntryAndProcesses().at(kProcess);

        if (metadataForProcess.lumis().size() ==
            (storedSingleRunEntryAndProcess.endLumi() - storedSingleRunEntryAndProcess.beginLumi())) {
          iBeginLumi = storedSingleRunEntryAndProcess.beginLumi();
          iEndLumi = storedSingleRunEntryAndProcess.endLumi();
          found = true;
          break;
        }
      }
      if (!found) {
        std::vector<LuminosityBlockNumber_t>& storedLumis = storedMetadata.lumis();
        std::vector<LuminosityBlockNumber_t> const& metdataLumis = metadataForProcess.lumis();
        iBeginLumi = storedLumis.size();
        storedLumis.insert(storedLumis.end(), metdataLumis.begin(), metdataLumis.end());
        iEndLumi = storedLumis.size();
      }
    }
    storedMetadata.singleRunEntryAndProcesses().emplace_back(
        iBeginLumi, iEndLumi, storedProcessIndex, metadataForProcess.valid(), metadataForProcess.allLumisProcessed());
    return true;
  }

getMergeDecision()

MergeableRunProductMetadata::MergeDecision edm::MergeableRunProductMetadata::getMergeDecision

(

std::string const &

processThatCreatedProduct

)

const

Definition at line 260 of file MergeableRunProductMetadata.cc.

References Exception, edm::errors::LogicError, MERGE, edm::MergeableRunProductMetadata::MetadataForProcess::mergeDecision(), and metadataForOneProcess().

Referenced by edm::MergeableInputProductResolver::mergeProduct().

                                                        {
    MetadataForProcess const* metadataForProcess = metadataForOneProcess(processThatCreatedProduct);
    if (metadataForProcess) {
      return metadataForProcess->mergeDecision();
    }
    throw Exception(errors::LogicError) << "MergeableRunProductMetadata::getMergeDecision could not find process.\n"
                                        << "It should not be possible for this error to occur.\n"
                                        << "Contact a Framework developer\n";
    return MERGE;
  }

getProcessName()

std::string const& edm::MergeableRunProductMetadata::getProcessName ( unsigned int  index ) const

inline

Definition at line 86 of file MergeableRunProductMetadata.h.

References edm::MergeableRunProductProcesses::getProcessName(), and mergeableRunProductProcesses_.

Referenced by Types.InputTag::cppTag(), and metadataForOneProcess().

                                                              {
      return mergeableRunProductProcesses_->getProcessName(index);
    }

gotLumisFromIndexIntoFile()

bool edm::MergeableRunProductMetadata::gotLumisFromIndexIntoFile ( ) const

inline

Definition at line 126 of file MergeableRunProductMetadata.h.

References gotLumisFromIndexIntoFile_.

{ return gotLumisFromIndexIntoFile_; }

knownImproperlyMerged()

bool edm::MergeableRunProductMetadata::knownImproperlyMerged ( std::string const &  processThatCreatedProduct ) const

overridevirtual

Implements edm::MergeableRunProductMetadataBase.

Definition at line 272 of file MergeableRunProductMetadata.cc.

References metadataForOneProcess(), and edm::MergeableRunProductMetadata::MetadataForProcess::valid().

                                                                                                          {
    MetadataForProcess const* metadataForProcess = metadataForOneProcess(processThatCreatedProduct);
    if (metadataForProcess) {
      return !metadataForProcess->valid();
    }
    return false;
  }

lumisFromIndexIntoFile()

std::vector<LuminosityBlockNumber_t> const& edm::MergeableRunProductMetadata::lumisFromIndexIntoFile ( ) const

inline

Definition at line 124 of file MergeableRunProductMetadata.h.

References lumisFromIndexIntoFile_.

{ return lumisFromIndexIntoFile_; }

lumisProcessed()

oneapi::tbb::concurrent_vector<LuminosityBlockNumber_t> const& edm::MergeableRunProductMetadata::lumisProcessed ( ) const

inline

Definition at line 128 of file MergeableRunProductMetadata.h.

References lumisProcessed_.

Referenced by preWriteRun().

{ return lumisProcessed_; }

mergeLumisFromIndexIntoFile()

void edm::MergeableRunProductMetadata::mergeLumisFromIndexIntoFile ( )

private

Definition at line 302 of file MergeableRunProductMetadata.cc.

References gotLumisFromIndexIntoFile_, lumisFromIndexIntoFile_, metadataForProcesses_, and groupFilesInBlocks::temp.

Referenced by preReadFile(), and preWriteRun().

                                                                {
    for (auto& metadataForProcess : metadataForProcesses_) {
      if (metadataForProcess.useIndexIntoFile()) {
        metadataForProcess.setUseIndexIntoFile(false);

        std::vector<LuminosityBlockNumber_t> temp;
        temp.reserve(metadataForProcess.lumis().size() + lumisFromIndexIntoFile_.size());
        std::vector<LuminosityBlockNumber_t>::const_iterator end1 = metadataForProcess.lumis().end();
        std::vector<LuminosityBlockNumber_t>::const_iterator end2 = lumisFromIndexIntoFile_.end();
        for (std::vector<LuminosityBlockNumber_t>::const_iterator iter1 = metadataForProcess.lumis().begin(),
                                                                  iter2 = lumisFromIndexIntoFile_.begin();
             iter1 != end1 || iter2 != end2;) {
          if (iter1 == end1) {
            temp.push_back(*iter2);
            ++iter2;
            continue;
          } else if (iter2 == end2) {
            temp.push_back(*iter1);
            ++iter1;
            continue;
          } else if (*iter1 < *iter2) {
            temp.push_back(*iter1);
            ++iter1;
          } else if (*iter1 > *iter2) {
            temp.push_back(*iter2);
            ++iter2;
          } else {
            // they must be equal
            temp.push_back(*iter1);
            ++iter1;
            ++iter2;
          }
        }
        metadataForProcess.lumis().swap(temp);
      }
    }
    lumisFromIndexIntoFile_.clear();
    gotLumisFromIndexIntoFile_ = false;
  }

metadataForOneProcess()

MergeableRunProductMetadata::MetadataForProcess const * edm::MergeableRunProductMetadata::metadataForOneProcess

(

std::string const &

processName

)

const

Definition at line 288 of file MergeableRunProductMetadata.cc.

References getProcessName(), metadataForProcesses_, and SimL1EmulatorRepack_CalouGT_cff::processName.

Referenced by getMergeDecision(), and knownImproperlyMerged().

                                          {
    unsigned int processIndex = 0;
    for (auto const& metadataForProcess : metadataForProcesses_) {
      // This string comparison could be optimized away by storing an index in
      // BranchDescription as a transient calculated once early in a job.
      if (getProcessName(processIndex) == processName) {
        return &metadataForProcess;
      }
      ++processIndex;
    }
    return nullptr;
  }

metadataForProcesses()

std::vector<MetadataForProcess> const& edm::MergeableRunProductMetadata::metadataForProcesses ( ) const

inline

Definition at line 122 of file MergeableRunProductMetadata.h.

References metadataForProcesses_.

{ return metadataForProcesses_; }

postWriteRun()

void edm::MergeableRunProductMetadata::postWriteRun

(

)

Definition at line 155 of file MergeableRunProductMetadata.cc.

References lumisProcessed_, and metadataForProcesses_.

                                                 {
    lumisProcessed_.clear();
    for (auto& metadataForProcess : metadataForProcesses_) {
      metadataForProcess.reset();
    }
  }

preReadFile()

void edm::MergeableRunProductMetadata::preReadFile

(

)

Definition at line 21 of file MergeableRunProductMetadata.cc.

References mergeLumisFromIndexIntoFile().

{ mergeLumisFromIndexIntoFile(); }

preWriteRun()

void edm::MergeableRunProductMetadata::preWriteRun

(

)

Definition at line 125 of file MergeableRunProductMetadata.cc.

References includes, lumisProcessed(), lumisProcessed_, mergeLumisFromIndexIntoFile(), metadataForProcesses_, jetsAK4_CHS_cff::sort, and tier0::unique().

Referenced by edm::EventProcessor::globalEndRunAsync().

                                                {
    if (metadataForProcesses_.empty()) {
      return;
    }

    mergeLumisFromIndexIntoFile();

    // Sort the lumiProcessed vector and ignore the duplicate
    // entries

    // Not sure if this copy is necessary. I'm copying because
    // I am not sure the standard algorithms work on TBB containers.
    // I couldn't find anything saying they did when I searched ...
    std::vector<LuminosityBlockNumber_t> lumisProcessed;
    lumisProcessed.reserve(lumisProcessed_.size());
    for (auto const& lumi : lumisProcessed_) {
      lumisProcessed.push_back(lumi);
    }

    std::sort(lumisProcessed.begin(), lumisProcessed.end());
    auto uniqueEnd = std::unique(lumisProcessed.begin(), lumisProcessed.end());

    for (auto& metadataForProcess : metadataForProcesses_) {
      // Did we process all the lumis in this process that were processed
      // in the process that created the mergeable run products.
      metadataForProcess.setAllLumisProcessed(std::includes(
          lumisProcessed.begin(), uniqueEnd, metadataForProcess.lumis().begin(), metadataForProcess.lumis().end()));
    }
  }

readRun()

void edm::MergeableRunProductMetadata::readRun	(	long long	inputRunEntry,
		StoredMergeableRunProductMetadata const &	inputStoredMergeableRunProductMetadata,
		IndexIntoFileItrHolder const &	inputIndexIntoFileItr
	)

Definition at line 23 of file MergeableRunProductMetadata.cc.

References edm::StoredMergeableRunProductMetadata::getLumiContent(), edm::IndexIntoFileItrHolder::getLumisInRun(), gotLumisFromIndexIntoFile_, IGNORE, edmLumisInFiles::lumis, lumisFromIndexIntoFile_, MERGE, mergeableRunProductProcesses_, metadataForProcesses_, edm::MergeableRunProductProcesses::processesWithMergeableRunProducts(), SimL1EmulatorRepack_CalouGT_cff::processName, REPLACE, AlCaHLTBitMon_QueryRunRegistry::string, groupFilesInBlocks::temp, and validateGeometry_cfg::valid.

Referenced by edm::RootFile::readRun_().

                                                           {
    unsigned int processIndex{0};
    for (auto& metadataForProcess : metadataForProcesses_) {
      bool valid = true;
      std::vector<LuminosityBlockNumber_t>::const_iterator lumisInRunBeingReadBegin;
      std::vector<LuminosityBlockNumber_t>::const_iterator lumisInRunBeingReadEnd;

      std::string const& processName = mergeableRunProductProcesses_->processesWithMergeableRunProducts()[processIndex];

      if (inputStoredMergeableRunProductMetadata.getLumiContent(
              inputRunEntry, processName, valid, lumisInRunBeingReadBegin, lumisInRunBeingReadEnd)) {
        // This is a reference to the container accumulating the luminosity
        // block numbers for the run entries read associated with the current
        // run being processed that correspond to the luminosity block content
        // for the mergeable run products created in the process.
        std::vector<LuminosityBlockNumber_t>& lumis = metadataForProcess.lumis();

        // In the following, iter1 refers to the lumis associated with run entries already read
        // and iter2 refers to the lumis associated with the current run entry being read.

        bool elementsIn2NotIn1 = false;
        bool elementsIn1NotIn2 = false;
        bool sharedElements = false;

        std::vector<LuminosityBlockNumber_t> temp;
        temp.reserve(lumis.size() + (lumisInRunBeingReadEnd - lumisInRunBeingReadBegin));
        std::vector<LuminosityBlockNumber_t>::const_iterator end1 = lumis.end();
        std::vector<LuminosityBlockNumber_t>::const_iterator end2 = lumisInRunBeingReadEnd;
        for (std::vector<LuminosityBlockNumber_t>::const_iterator iter1 = lumis.begin(),
                                                                  iter2 = lumisInRunBeingReadBegin;
             iter1 != end1 || iter2 != end2;) {
          if (iter1 == end1) {
            temp.push_back(*iter2);
            ++iter2;
            elementsIn2NotIn1 = true;
            continue;
          } else if (iter2 == end2) {
            temp.push_back(*iter1);
            ++iter1;
            elementsIn1NotIn2 = true;
            continue;
          } else if (*iter1 < *iter2) {
            temp.push_back(*iter1);
            ++iter1;
            elementsIn1NotIn2 = true;
          } else if (*iter1 > *iter2) {
            temp.push_back(*iter2);
            ++iter2;
            elementsIn2NotIn1 = true;
          } else {
            // they must be equal
            sharedElements = true;
            temp.push_back(*iter1);
            ++iter1;
            ++iter2;
          }
        }
        lumis.swap(temp);
        if (!sharedElements && elementsIn2NotIn1 && elementsIn1NotIn2) {
          metadataForProcess.setMergeDecision(MERGE);
          if (!valid) {
            metadataForProcess.setValid(false);
          }
        } else if (!elementsIn2NotIn1) {
          metadataForProcess.setMergeDecision(IGNORE);
        } else if (!elementsIn1NotIn2) {
          metadataForProcess.setMergeDecision(REPLACE);
          if (!valid) {
            metadataForProcess.setValid(false);
          }
        } else {
          // In this case there is no way to get the correct answer.
          // The result will always be invalid.
          metadataForProcess.setMergeDecision(MERGE);
          metadataForProcess.setValid(false);
        }

      } else {
        metadataForProcess.setMergeDecision(MERGE);
        if (!valid) {
          metadataForProcess.setValid(false);
        }
        metadataForProcess.setUseIndexIntoFile(true);
        if (!gotLumisFromIndexIntoFile_) {
          inputIndexIntoFileItr.getLumisInRun(lumisFromIndexIntoFile_);
          gotLumisFromIndexIntoFile_ = true;
        }
      }
      ++processIndex;
    }  // end of loop over processes
  }    // end of readRun function

writeLumi()

void edm::MergeableRunProductMetadata::writeLumi

(

LuminosityBlockNumber_t

lumi

)

Definition at line 118 of file MergeableRunProductMetadata.cc.

References lumisProcessed_, and metadataForProcesses_.

Referenced by edm::EventProcessor::writeLumiAsync().

                                                                          {
    if (metadataForProcesses_.empty()) {
      return;
    }
    lumisProcessed_.push_back(lumi);
  }

Member Data Documentation

gotLumisFromIndexIntoFile_

bool edm::MergeableRunProductMetadata::gotLumisFromIndexIntoFile_ = false

private

Definition at line 148 of file MergeableRunProductMetadata.h.

Referenced by gotLumisFromIndexIntoFile(), mergeLumisFromIndexIntoFile(), and readRun().

lumisFromIndexIntoFile_

std::vector<LuminosityBlockNumber_t> edm::MergeableRunProductMetadata::lumisFromIndexIntoFile_

private

Definition at line 147 of file MergeableRunProductMetadata.h.

Referenced by lumisFromIndexIntoFile(), mergeLumisFromIndexIntoFile(), and readRun().

lumisProcessed_

oneapi::tbb::concurrent_vector<LuminosityBlockNumber_t> edm::MergeableRunProductMetadata::lumisProcessed_

private

Definition at line 150 of file MergeableRunProductMetadata.h.

Referenced by lumisProcessed(), postWriteRun(), preWriteRun(), and writeLumi().

mergeableRunProductProcesses_

MergeableRunProductProcesses const* edm::MergeableRunProductMetadata::mergeableRunProductProcesses_

private

Definition at line 139 of file MergeableRunProductMetadata.h.

Referenced by addEntryToStoredMetadata(), getProcessName(), and readRun().

metadataForProcesses_

std::vector<MetadataForProcess> edm::MergeableRunProductMetadata::metadataForProcesses_

private

Definition at line 145 of file MergeableRunProductMetadata.h.

Referenced by addEntryToStoredMetadata(), mergeLumisFromIndexIntoFile(), metadataForOneProcess(), metadataForProcesses(), postWriteRun(), preWriteRun(), readRun(), and writeLumi().