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

enum edm::MergeableRunProductMetadata::MergeDecision

Enumerator
MERGE
REPLACE
IGNORE

Definition at line 55 of file MergeableRunProductMetadata.h.

{ MERGE, REPLACE, IGNORE };

Constructor & Destructor Documentation

edm::MergeableRunProductMetadata::MergeableRunProductMetadata

(

MergeableRunProductProcesses const &

mergeableRunProductProcesses

)

Definition at line 15 of file MergeableRunProductMetadata.cc.

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

edm::MergeableRunProductMetadata::~MergeableRunProductMetadata ( )

override

Definition at line 20 of file MergeableRunProductMetadata.cc.

                                                            {
  }

Member Function Documentation

void edm::MergeableRunProductMetadata::addEntryToStoredMetadata

(

StoredMergeableRunProductMetadata &

storedMetadata

)

const

Definition at line 180 of file MergeableRunProductMetadata.cc.

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

Referenced by postWriteRun(), and edm::RootOutputFile::writeRun().

                                                                                    {

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

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 235 of file MergeableRunProductMetadata.cc.

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

Referenced by addEntryToStoredMetadata(), and lumisProcessed().

                                                  {

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

MergeableRunProductMetadata::MergeDecision edm::MergeableRunProductMetadata::getMergeDecision

(

std::string const &

processThatCreatedProduct

)

const

Definition at line 297 of file MergeableRunProductMetadata.cc.

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

Referenced by edm::DataManagingProductResolver::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;
  }

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

inline

Definition at line 89 of file MergeableRunProductMetadata.h.

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

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

                                                              {
      return mergeableRunProductProcesses_->getProcessName(index);
    }

bool edm::MergeableRunProductMetadata::gotLumisFromIndexIntoFile ( ) const

inline

Definition at line 133 of file MergeableRunProductMetadata.h.

References gotLumisFromIndexIntoFile_.

{ return gotLumisFromIndexIntoFile_; }

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

overridevirtual

Implements edm::MergeableRunProductMetadataBase.

Definition at line 311 of file MergeableRunProductMetadata.cc.

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

                                                                                                     {

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

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

inline

Definition at line 131 of file MergeableRunProductMetadata.h.

References lumisFromIndexIntoFile_.

{ return lumisFromIndexIntoFile_; }

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

inline

Definition at line 136 of file MergeableRunProductMetadata.h.

References addProcess(), lumisProcessed_, and mergeLumisFromIndexIntoFile().

Referenced by preWriteRun().

{ return lumisProcessed_; }

void edm::MergeableRunProductMetadata::mergeLumisFromIndexIntoFile ( )

private

Definition at line 342 of file MergeableRunProductMetadata.cc.

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

Referenced by lumisProcessed(), 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;
  }

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

(

std::string const &

processName

)

const

Definition at line 329 of file MergeableRunProductMetadata.cc.

References getProcessName(), and metadataForProcesses_.

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

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

inline

Definition at line 128 of file MergeableRunProductMetadata.h.

References metadataForProcesses_.

{ return metadataForProcesses_; }

void edm::MergeableRunProductMetadata::postWriteRun

(

)

Definition at line 171 of file MergeableRunProductMetadata.cc.

References addEntryToStoredMetadata(), lumisProcessed_, and metadataForProcesses_.

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

                                                 {

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

void edm::MergeableRunProductMetadata::preReadFile

(

)

Definition at line 23 of file MergeableRunProductMetadata.cc.

References mergeLumisFromIndexIntoFile().

                                                {
    mergeLumisFromIndexIntoFile();
  }

void edm::MergeableRunProductMetadata::preWriteRun

(

)

Definition at line 137 of file MergeableRunProductMetadata.cc.

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

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

                                                {

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

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

Definition at line 27 of file MergeableRunProductMetadata.cc.

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

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

void edm::MergeableRunProductMetadata::writeLumi

(

LuminosityBlockNumber_t

lumi

)

Definition at line 129 of file MergeableRunProductMetadata.cc.

References lumisProcessed_, and metadataForProcesses_.

                                                                          {

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

Member Data Documentation

bool edm::MergeableRunProductMetadata::gotLumisFromIndexIntoFile_ = false

private

Definition at line 157 of file MergeableRunProductMetadata.h.

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

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

private

Definition at line 156 of file MergeableRunProductMetadata.h.

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

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

private

Definition at line 159 of file MergeableRunProductMetadata.h.

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

MergeableRunProductProcesses const* edm::MergeableRunProductMetadata::mergeableRunProductProcesses_

private

Definition at line 148 of file MergeableRunProductMetadata.h.

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

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

private

Definition at line 154 of file MergeableRunProductMetadata.h.

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