edm::OutputProcessBlockHelper Class Reference

#include <OutputProcessBlockHelper.h>

Public Member Functions
void	fillCacheIndices (StoredProcessBlockHelper &) const
ProcessBlockHelperBase const *	processBlockHelper () const
std::vector< std::string > const &	processesWithProcessBlockProducts () const
bool	productsFromInputKept () const
void	updateAfterProductSelection (std::set< std::string > const &processesWithKeptProcessBlockProducts, ProcessBlockHelperBase const &)

Private Member Functions
unsigned int	nAddedProcesses () const
void	setProcessOffset (unsigned int iFile, unsigned int nInputProcesses, std::vector< std::vector< unsigned int >> const &nEntries, std::vector< unsigned int > &processOffset) const
void	setStoredFileInProcessOffset (unsigned int iFile, unsigned int nInputProcesses, std::vector< std::vector< unsigned int >> const &nEntries, std::vector< unsigned int > &storedFileInProcessOffset) const
void	setStoredProcessOffset (unsigned int nInputProcesses, std::vector< std::vector< unsigned int >> const &nEntries, std::vector< unsigned int > &storedProcessOffset) const
std::vector< unsigned int > const &	translateFromStoredIndex () const

Private Attributes
unsigned int	nAddedProcesses_ = 0
ProcessBlockHelperBase const *	processBlockHelper_ = nullptr
std::vector< std::string >	processesWithProcessBlockProducts_
bool	productsFromInputKept_ = false
std::vector< unsigned int >	translateFromStoredIndex_

Friends
class	TestOneOutput

Detailed Description

Author

W. David Dagenhart, created 28 December, 2020

Definition at line 20 of file OutputProcessBlockHelper.h.

Member Function Documentation

fillCacheIndices()

void edm::OutputProcessBlockHelper::fillCacheIndices

(

StoredProcessBlockHelper &

storedProcessBlockHelper

)

const

Definition at line 54 of file OutputProcessBlockHelper.cc.

References cms::cuda::assert(), edm::ProcessBlockHelperBase::cacheEntriesPerFile(), edm::ProcessBlockHelperBase::cacheIndexVectorsPerFile(), mps_fire::i, edm::ProcessBlockHelperBase::invalidCacheIndex(), eostools::move(), nAddedProcesses_, edm::ProcessBlockHelperBase::nEntries(), edm::ProcessBlockHelperBase::processBlockCacheIndices(), processBlockHelper_, edm::StoredProcessBlockHelper::processesWithProcessBlockProducts(), productsFromInputKept_, edm::StoredProcessBlockHelper::setProcessBlockCacheIndices(), setProcessOffset(), setStoredFileInProcessOffset(), setStoredProcessOffset(), findQualityFiles::size, and translateFromStoredIndex_.

                                                                                                          {
    // The stored cache indices are the ones we want to fill.
    // This will get written to the output file.
    // Note for output the vector of vectors is flattened into a single vector
    std::vector<unsigned int> storedCacheIndices;

    // Number of processes in StoredProcessBlockHelper.
    unsigned int nStoredProcesses = storedProcessBlockHelper.processesWithProcessBlockProducts().size();

    if (!productsFromInputKept_) {
      // This is really simple if we are not keeping any ProcessBlock products
      // from the input file. Deal with that special case first.
      // Except for the special case of a SubProcess, nStoredProcesses will be 1.
      assert(nAddedProcesses_ == nStoredProcesses);
      storedCacheIndices.reserve(nStoredProcesses);
      for (unsigned int i = 0; i < nStoredProcesses; ++i) {
        storedCacheIndices.push_back(i);
      }
      storedProcessBlockHelper.setProcessBlockCacheIndices(std::move(storedCacheIndices));
      return;
    }

    // Cache indices of the main ProcessBlockHelper we use as input. This
    // ProcessBlockHelper is owned by the EventProcessor.
    std::vector<std::vector<unsigned int>> const& cacheIndices = processBlockHelper_->processBlockCacheIndices();

    // We need to convert the cache indices in the ProcessBlockHelper to have different values when
    // put in the StoredProcessBlockHelper. The values are not the same because the ProcessBlocks are
    // not read in the same order in this process as compared to the next process which will read
    // the output file that is being written (the read order is the same as the order the cache
    // objects are placed in the cache vectors). In this process, they are ordered first by input file,
    // second by process and last by TTree entry. In the next process, this output file will be read
    // as a single input file. The ProcessBlocks are read in process order (this will be a subset
    // of the process list in ProcessBlockHelper, maybe smaller or maybe the same), and finally in
    // order of entry in the TTree. This conversion is done by subtracting and adding some
    // offsets and a lot of the following code involves calculating these offsets to do the conversion.

    // We will need the info in these to calculate the offsets
    std::vector<unsigned int> const& cacheIndexVectorsPerFile = processBlockHelper_->cacheIndexVectorsPerFile();
    std::vector<unsigned int> const& cacheEntriesPerFile = processBlockHelper_->cacheEntriesPerFile();
    std::vector<std::vector<unsigned int>> const& nEntries = processBlockHelper_->nEntries();

    assert(!cacheIndices.empty());
    // Count processes in the input file with saved ProcessBlock products in the output
    unsigned int nInputProcesses = 0;
    for (unsigned int iStoredProcess = 0; iStoredProcess < nStoredProcesses; ++iStoredProcess) {
      // The existing cache indices in processBlockHelper include only indices
      // corresponding to the processes in the input files. If there are more, then
      // they correspond to current process (and there only will be more if some
      // of the newly produced ProcessBlock products are going to be kept).
      // There will be at most 1 added process except in the case of SubProcesses.
      if (translateFromStoredIndex_[iStoredProcess] < cacheIndices[0].size()) {
        ++nInputProcesses;
      }
    }

    // The following are the 4 offsets. The first two are defined relative to the
    // cache sequence in this process. The second two offsets are defined relative
    // to the cache sequence when the output file we are writing is read.

    // 1. Total number of cache entries in all input files prior to the current input file
    unsigned int fileOffset = 0;

    // 2. For each process, the total number of cache entries in processes in the current
    // input file and before the process
    std::vector<unsigned int> processOffset(nInputProcesses, 0);

    // 3. For each input process with ProcessBlock products stored by this
    // output module, the total number of cache entries in earlier input processes
    // that have ProcessBlock products stored by this output module.
    // Summed over all input files and including only processes in StoredProcessBlockHelper.
    // Plus an extra element at the end that includes all entries in all such processes.
    assert(!nEntries.empty());
    std::vector<unsigned int> storedProcessOffset(nInputProcesses + 1, 0);

    // 4. For each process with ProcessBlock products stored by this output module,
    // the total number of cache entries in that process in all input files before
    // the current input file.
    std::vector<unsigned int> storedFileInProcessOffset(nInputProcesses, 0);

    setStoredProcessOffset(nInputProcesses, nEntries, storedProcessOffset);

    storedCacheIndices.reserve(cacheIndices.size() * nStoredProcesses);

    unsigned int iFile = 0;
    unsigned int innerVectorsCurrentFile = 0;

    // In ProcessBlockHelper, there is a vector which contains vectors
    // of cache indices. Iterate over the inner vectors.
    for (auto const& innerVectorOfCacheIndices : cacheIndices) {
      // The inner vectors are associated with input files. Several contiguous
      // inner vectors can be associated with the same input file. Check to
      // see if we have crossed an input file boundary and set the file
      // index, iFile, at the next file associated with at least
      // one inner vector if necessary.
      while (innerVectorsCurrentFile == cacheIndexVectorsPerFile[iFile]) {
        // Sum cache entries for all files before the current file in
        // ProcessBlockHelper
        fileOffset += cacheEntriesPerFile[iFile];
        ++iFile;
        innerVectorsCurrentFile = 0;
      }
      if (innerVectorsCurrentFile == 0) {
        // Call these when the input file changes
        setProcessOffset(iFile, nInputProcesses, nEntries, processOffset);
        setStoredFileInProcessOffset(iFile, nInputProcesses, nEntries, storedFileInProcessOffset);
      }
      ++innerVectorsCurrentFile;

      assert(nInputProcesses + nAddedProcesses_ == nStoredProcesses);

      // Really fill the cache indices that will be stored in the output file in this loop
      for (unsigned int iStoredProcess = 0; iStoredProcess < nStoredProcesses; ++iStoredProcess) {
        unsigned int storedCacheIndex = ProcessBlockHelperBase::invalidCacheIndex();
        if (iStoredProcess < nInputProcesses) {
          unsigned int cacheIndex = innerVectorOfCacheIndices[translateFromStoredIndex_[iStoredProcess]];
          if (cacheIndex != ProcessBlockHelperBase::invalidCacheIndex()) {
            // The offsets count in the cache sequence to the first entry in
            // the current input file and process
            unsigned int inputOffset = fileOffset + processOffset[iStoredProcess];
            unsigned int storedOffset = storedProcessOffset[iStoredProcess] + storedFileInProcessOffset[iStoredProcess];
            storedCacheIndex = cacheIndex - inputOffset + storedOffset;
          }
        } else {
          // This corresponds to the current process if it has newly produced
          // ProcessBlock products (plus possibly SubProcesses).
          storedCacheIndex = storedProcessOffset[nInputProcesses] + iStoredProcess - nInputProcesses;
        }
        storedCacheIndices.push_back(storedCacheIndex);
      }
    }
    storedProcessBlockHelper.setProcessBlockCacheIndices(std::move(storedCacheIndices));
  }

nAddedProcesses()

unsigned int edm::OutputProcessBlockHelper::nAddedProcesses ( ) const

inlineprivate

Definition at line 39 of file OutputProcessBlockHelper.h.

References nAddedProcesses_.

{ return nAddedProcesses_; }

processBlockHelper()

ProcessBlockHelperBase const* edm::OutputProcessBlockHelper::processBlockHelper ( ) const

inline

Definition at line 33 of file OutputProcessBlockHelper.h.

References processBlockHelper_.

Referenced by updateAfterProductSelection().

{ return processBlockHelper_; }

processesWithProcessBlockProducts()

std::vector<std::string> const& edm::OutputProcessBlockHelper::processesWithProcessBlockProducts ( ) const

inline

Definition at line 22 of file OutputProcessBlockHelper.h.

References processesWithProcessBlockProducts_.

Referenced by edm::PoolOutputModule::respondToOpenInputFile().

                                                                          {
      return processesWithProcessBlockProducts_;
    }

productsFromInputKept()

bool edm::OutputProcessBlockHelper::productsFromInputKept ( ) const

inline

Definition at line 31 of file OutputProcessBlockHelper.h.

References productsFromInputKept_.

{ return productsFromInputKept_; }

setProcessOffset()

void edm::OutputProcessBlockHelper::setProcessOffset ( unsigned int  iFile, unsigned int  nInputProcesses, std::vector< std::vector< unsigned int >> const &  nEntries, std::vector< unsigned int > &  processOffset  ) const

private

Definition at line 207 of file OutputProcessBlockHelper.cc.

References hltrates_dqm_sourceclient-live_cfg::offset, and translateFromStoredIndex_.

Referenced by fillCacheIndices().

                                                                                                {
    unsigned int iStored = 0;
    for (auto& offset : processOffset) {
      offset = 0;
      unsigned int iProcess = translateFromStoredIndex_[iStored];
      for (unsigned int jProcess = 0; jProcess < iProcess; ++jProcess) {
        offset += nEntries[iFile][jProcess];
      }
      ++iStored;
    }
  }

setStoredFileInProcessOffset()

void edm::OutputProcessBlockHelper::setStoredFileInProcessOffset ( unsigned int  iFile, unsigned int  nInputProcesses, std::vector< std::vector< unsigned int >> const &  nEntries, std::vector< unsigned int > &  storedFileInProcessOffset  ) const

private

Definition at line 222 of file OutputProcessBlockHelper.cc.

References cms::cuda::assert(), hltrates_dqm_sourceclient-live_cfg::offset, findQualityFiles::size, and translateFromStoredIndex_.

Referenced by fillCacheIndices().

                                                                {
    unsigned int iStored = 0;
    for (auto& offset : storedFileInProcessOffset) {
      offset = 0;
      unsigned int indexInEventProcessor = translateFromStoredIndex_[iStored];
      // loop over input files before current input file
      for (unsigned int jFile = 0; jFile < iFile; ++jFile) {
        assert(indexInEventProcessor < nEntries[jFile].size());
        offset += nEntries[jFile][indexInEventProcessor];
      }
      ++iStored;
    }
  }

setStoredProcessOffset()

void edm::OutputProcessBlockHelper::setStoredProcessOffset ( unsigned int  nInputProcesses, std::vector< std::vector< unsigned int >> const &  nEntries, std::vector< unsigned int > &  storedProcessOffset  ) const

private

Definition at line 188 of file OutputProcessBlockHelper.cc.

References cms::cuda::assert(), hltrates_dqm_sourceclient-live_cfg::offset, and translateFromStoredIndex_.

Referenced by fillCacheIndices().

                                                                                                            {
    unsigned int iStored = 0;
    for (auto& offset : storedProcessOffset) {
      offset = 0;
      // loop over earlier processes
      for (unsigned int jStored = 0; jStored < iStored; ++jStored) {
        unsigned int indexInEventProcessor = translateFromStoredIndex_[jStored];
        // loop over input files
        for (auto const& entries : nEntries) {
          assert(indexInEventProcessor < entries.size());
          offset += entries[indexInEventProcessor];
        }
      }
      ++iStored;
    }
  }

translateFromStoredIndex()

std::vector<unsigned int> const& edm::OutputProcessBlockHelper::translateFromStoredIndex ( ) const

inlineprivate

Definition at line 38 of file OutputProcessBlockHelper.h.

References translateFromStoredIndex_.

{ return translateFromStoredIndex_; }

updateAfterProductSelection()

void edm::OutputProcessBlockHelper::updateAfterProductSelection	(	std::set< std::string > const &	processesWithKeptProcessBlockProducts,
		ProcessBlockHelperBase const &	processBlockHelper
	)

Definition at line 12 of file OutputProcessBlockHelper.cc.

References edm::ProcessBlockHelperBase::addedProcesses(), cms::cuda::assert(), spr::find(), nAddedProcesses_, edm::ProcessBlockHelperBase::nProcessesInFirstFile(), LaserDQM_cfg::process, processBlockHelper(), processBlockHelper_, edm::ProcessBlockHelperBase::processesWithProcessBlockProducts(), processesWithProcessBlockProducts_, productsFromInputKept_, edm::ProcessBlockHelperBase::topProcessesWithProcessBlockProducts(), and translateFromStoredIndex_.

Referenced by edm::core::OutputModuleCore::selectProducts().

                                                        {
    processBlockHelper_ = &processBlockHelper;

    // Copy the list of processes with ProcessBlock products from the EventProcessor or SubProcess,
    // except remove any processes where the output module calling this has dropped all ProcessBlock
    // products. We want to maintain the same order and only remove elements. Fill a vector that can
    // translate from one process list to the other.
    assert(processesWithProcessBlockProducts_.empty());
    unsigned int iProcess = 0;
    for (auto const& process : processBlockHelper.processesWithProcessBlockProducts()) {
      if (processesWithKeptProcessBlockProducts.find(process) != processesWithKeptProcessBlockProducts.end()) {
        processesWithProcessBlockProducts_.emplace_back(process);
        translateFromStoredIndex_.emplace_back(iProcess);
      }
      ++iProcess;
    }

    for (auto const& addedProcess : processBlockHelper.addedProcesses()) {
      // count new processes producing ProcessBlock products that are
      // kept by the OutputModule. There can be at most 1 except
      // in the case of SubProcesses.
      if (std::find(processesWithProcessBlockProducts_.begin(),
                    processesWithProcessBlockProducts_.end(),
                    addedProcess) != processesWithProcessBlockProducts_.end()) {
        ++nAddedProcesses_;
      }
    }

    // Determine if any ProcessBlock product from the input file is kept by the output module.
    // Do this by looking for a process name on both the list of processes with ProcessBlock
    // products kept by the output module and process names from the input with ProcessBlock
    // products.
    productsFromInputKept_ =
        std::find_first_of(processesWithProcessBlockProducts_.begin(),
                           processesWithProcessBlockProducts_.end(),
                           processBlockHelper.topProcessesWithProcessBlockProducts().begin(),
                           processBlockHelper.topProcessesWithProcessBlockProducts().begin() +
                               processBlockHelper.nProcessesInFirstFile()) != processesWithProcessBlockProducts_.end();
  }

Friends And Related Function Documentation

TestOneOutput

friend class TestOneOutput

friend

Definition at line 37 of file OutputProcessBlockHelper.h.

Member Data Documentation

nAddedProcesses_

unsigned int edm::OutputProcessBlockHelper::nAddedProcesses_ = 0

private

Definition at line 62 of file OutputProcessBlockHelper.h.

Referenced by fillCacheIndices(), nAddedProcesses(), and updateAfterProductSelection().

processBlockHelper_

ProcessBlockHelperBase const* edm::OutputProcessBlockHelper::processBlockHelper_ = nullptr

private

Definition at line 70 of file OutputProcessBlockHelper.h.

Referenced by fillCacheIndices(), processBlockHelper(), and updateAfterProductSelection().

processesWithProcessBlockProducts_

std::vector<std::string> edm::OutputProcessBlockHelper::processesWithProcessBlockProducts_

private

Definition at line 57 of file OutputProcessBlockHelper.h.

Referenced by processesWithProcessBlockProducts(), and updateAfterProductSelection().

productsFromInputKept_

bool edm::OutputProcessBlockHelper::productsFromInputKept_ = false

private

Definition at line 72 of file OutputProcessBlockHelper.h.

Referenced by fillCacheIndices(), productsFromInputKept(), and updateAfterProductSelection().

translateFromStoredIndex_

std::vector<unsigned int> edm::OutputProcessBlockHelper::translateFromStoredIndex_

private

Definition at line 66 of file OutputProcessBlockHelper.h.

Referenced by fillCacheIndices(), setProcessOffset(), setStoredFileInProcessOffset(), setStoredProcessOffset(), translateFromStoredIndex(), and updateAfterProductSelection().