#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

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_

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. (At the present time concurrent runs are not possible, but there plans to implement that in the future). 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 will be 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 52 of file MergeableRunProductMetadata.h.

Member Enumeration Documentation

◆ MergeDecision

enum edm::MergeableRunProductMetadata::MergeDecision

Enumerator
MERGE
REPLACE
IGNORE

Definition at line 54 of file MergeableRunProductMetadata.h.

54 { MERGE, REPLACE, IGNORE };

Constructor & Destructor Documentation

◆ MergeableRunProductMetadata()

edm::MergeableRunProductMetadata::MergeableRunProductMetadata ( MergeableRunProductProcesses const & mergeableRunProductProcesses )

Definition at line 14 of file MergeableRunProductMetadata.cc.

16 : mergeableRunProductProcesses_(&mergeableRunProductProcesses),

17 metadataForProcesses_(mergeableRunProductProcesses.size()) {}

◆ ~MergeableRunProductMetadata()

edm::MergeableRunProductMetadata::~MergeableRunProductMetadata ( )

override

Definition at line 19 of file MergeableRunProductMetadata.cc.

19 {}

Member Function Documentation

◆ addEntryToStoredMetadata()

void edm::MergeableRunProductMetadata::addEntryToStoredMetadata ( StoredMergeableRunProductMetadata & storedMetadata ) const

Definition at line 162 of file MergeableRunProductMetadata.cc.

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

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

◆ 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.

                                                                                     {
     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;
   }

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().

◆ getMergeDecision()

MergeableRunProductMetadata::MergeDecision edm::MergeableRunProductMetadata::getMergeDecision ( std::string const & processThatCreatedProduct ) const

Definition at line 260 of file MergeableRunProductMetadata.cc.

                                                         {
     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;
   }

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

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

◆ getProcessName()

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

inline

Definition at line 87 of file MergeableRunProductMetadata.h.

                                                               {
       return mergeableRunProductProcesses_->getProcessName(index);
     }

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

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

◆ gotLumisFromIndexIntoFile()

bool edm::MergeableRunProductMetadata::gotLumisFromIndexIntoFile ( ) const

inline

Definition at line 127 of file MergeableRunProductMetadata.h.

127 { return gotLumisFromIndexIntoFile_; }

References gotLumisFromIndexIntoFile_.

◆ knownImproperlyMerged()

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

overridevirtual

Implements edm::MergeableRunProductMetadataBase.

Definition at line 272 of file MergeableRunProductMetadata.cc.

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

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

◆ lumisFromIndexIntoFile()

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

inline

Definition at line 125 of file MergeableRunProductMetadata.h.

125 { return lumisFromIndexIntoFile_; }

References lumisFromIndexIntoFile_.

◆ lumisProcessed()

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

inline

Definition at line 129 of file MergeableRunProductMetadata.h.

129 { return lumisProcessed_; }

References lumisProcessed_.

Referenced by preWriteRun().

◆ mergeLumisFromIndexIntoFile()

void edm::MergeableRunProductMetadata::mergeLumisFromIndexIntoFile ( )

private

Definition at line 302 of file MergeableRunProductMetadata.cc.

                                                                 {
     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;
   }

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

Referenced by preReadFile(), and preWriteRun().

◆ metadataForOneProcess()

MergeableRunProductMetadata::MetadataForProcess const * edm::MergeableRunProductMetadata::metadataForOneProcess ( std::string const & processName ) const

Definition at line 288 of file MergeableRunProductMetadata.cc.

                                           {
     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;
   }

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

Referenced by getMergeDecision(), and knownImproperlyMerged().

◆ metadataForProcesses()

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

inline

Definition at line 123 of file MergeableRunProductMetadata.h.

123 { return metadataForProcesses_; }

References metadataForProcesses_.

◆ postWriteRun()

void edm::MergeableRunProductMetadata::postWriteRun ( )

Definition at line 155 of file MergeableRunProductMetadata.cc.

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

References lumisProcessed_, and metadataForProcesses_.

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

◆ preReadFile()

void edm::MergeableRunProductMetadata::preReadFile ( )

Definition at line 21 of file MergeableRunProductMetadata.cc.

21 { mergeLumisFromIndexIntoFile(); }

References mergeLumisFromIndexIntoFile().

◆ preWriteRun()

void edm::MergeableRunProductMetadata::preWriteRun ( )

Definition at line 125 of file MergeableRunProductMetadata.cc.

                                                 {
     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()));
     }
   }

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

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

◆ readRun()

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

Definition at line 23 of file MergeableRunProductMetadata.cc.

                                                            {
     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

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_().

◆ writeLumi()

void edm::MergeableRunProductMetadata::writeLumi ( LuminosityBlockNumber_t lumi )

Definition at line 118 of file MergeableRunProductMetadata.cc.

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