SubProcess.cc

Go to the documentation of this file.

#include "FWCore/Framework/interface/SubProcess.h"

#include "DataFormats/Common/interface/ProductData.h"
#include "DataFormats/Common/interface/ThinnedAssociation.h"
#include "DataFormats/Provenance/interface/BranchIDListHelper.h"
#include "DataFormats/Provenance/interface/EventSelectionID.h"
#include "DataFormats/Provenance/interface/ProcessHistoryID.h"
#include "DataFormats/Provenance/interface/ProcessHistoryRegistry.h"
#include "DataFormats/Provenance/interface/ProductRegistry.h"
#include "DataFormats/Provenance/interface/LuminosityBlockAuxiliary.h"
#include "DataFormats/Provenance/interface/RunAuxiliary.h"
#include "DataFormats/Provenance/interface/ThinnedAssociationsHelper.h"
#include "FWCore/Framework/interface/EventPrincipal.h"
#include "FWCore/Framework/interface/FileBlock.h"
#include "FWCore/Framework/interface/ProductHolder.h"
#include "FWCore/Framework/interface/HistoryAppender.h"
#include "FWCore/Framework/interface/LuminosityBlockPrincipal.h"
#include "FWCore/Framework/interface/OccurrenceTraits.h"
#include "FWCore/Framework/interface/OutputModuleDescription.h"
#include "FWCore/Framework/interface/RunPrincipal.h"
#include "FWCore/Framework/interface/getAllTriggerNames.h"
#include "FWCore/Framework/interface/TriggerNamesService.h"
#include "FWCore/Framework/src/EventSetupsController.h"
#include "FWCore/Framework/src/SignallingProductRegistry.h"
#include "FWCore/Framework/src/PreallocationConfiguration.h"
#include "FWCore/ParameterSet/interface/IllegalParameters.h"
#include "FWCore/ParameterSet/interface/ParameterSet.h"
#include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
#include "FWCore/Utilities/interface/ExceptionCollector.h"

#include <cassert>
#include <string>
#include <vector>

namespace edm {

  SubProcess::SubProcess(ParameterSet& parameterSet,
                         ParameterSet const& topLevelParameterSet,
                         std::shared_ptr<ProductRegistry const> parentProductRegistry,
                         std::shared_ptr<BranchIDListHelper const> parentBranchIDListHelper,
                         ThinnedAssociationsHelper const& parentThinnedAssociationsHelper,
                         eventsetup::EventSetupsController& esController,
                         ActivityRegistry& parentActReg,
                         ServiceToken const& token,
                         serviceregistry::ServiceLegacy iLegacy,
                         PreallocationConfiguration const& preallocConfig,
                         ProcessContext const* parentProcessContext) :
      serviceToken_(),
      parentPreg_(parentProductRegistry),
      preg_(),
      branchIDListHelper_(),
      act_table_(),
      processConfiguration_(),
      historyLumiOffset_(preallocConfig.numberOfStreams()),
      historyRunOffset_(historyLumiOffset_+preallocConfig.numberOfLuminosityBlocks()),
      processHistoryRegistries_(historyRunOffset_+ preallocConfig.numberOfRuns()),
      historyAppenders_(historyRunOffset_+preallocConfig.numberOfRuns()),
      principalCache_(),
      esp_(),
      schedule_(),
      parentToChildPhID_(),
      subProcesses_(),
      processParameterSet_(),
      productSelectorRules_(parameterSet, "outputCommands", "OutputModule"),
      productSelector_(),
      wantAllEvents_(true) {
  
    //Setup the event selection
    Service<service::TriggerNamesService> tns;
    
    ParameterSet selectevents =
    parameterSet.getUntrackedParameterSet("SelectEvents", ParameterSet());
    
    selectevents.registerIt(); // Just in case this PSet is not registered
    wantAllEvents_ = detail::configureEventSelector(selectevents,
                                                    tns->getProcessName(),
                                                    getAllTriggerNames(),
                                                    selectors_);
    std::map<std::string, std::vector<std::pair<std::string, int> > > outputModulePathPositions;
    selector_config_id_ = detail::registerProperSelectionInfo(selectevents,
                                                              "",
                                                              outputModulePathPositions,
                                                              parentProductRegistry->anyProductProduced());

    std::map<BranchID, bool> keepAssociation;
    selectProducts(*parentProductRegistry, parentThinnedAssociationsHelper, keepAssociation);

    std::string const maxEvents("maxEvents");
    std::string const maxLumis("maxLuminosityBlocks");

    // propagate_const<T> has no reset() function
    processParameterSet_ = std::unique_ptr<ParameterSet>(parameterSet.popParameterSet(std::string("process")).release()); 

    // if this process has a maxEvents or maxLuminosityBlocks parameter set, remove them.
    if(processParameterSet_->exists(maxEvents)) {
      processParameterSet_->popParameterSet(maxEvents);
    }
    if(processParameterSet_->exists(maxLumis)) {
      processParameterSet_->popParameterSet(maxLumis);
    }

    // if the top level process has a maxEvents or maxLuminosityBlocks parameter set, add them to this process.
    if(topLevelParameterSet.exists(maxEvents)) {
      processParameterSet_->addUntrackedParameter<ParameterSet>(maxEvents, topLevelParameterSet.getUntrackedParameterSet(maxEvents));
    }
    if(topLevelParameterSet.exists(maxLumis)) {
      processParameterSet_->addUntrackedParameter<ParameterSet>(maxLumis, topLevelParameterSet.getUntrackedParameterSet(maxLumis));
    }

    // If there are subprocesses, pop the subprocess parameter sets out of the process parameter set
    std::unique_ptr<std::vector<ParameterSet> > subProcessVParameterSet(popSubProcessVParameterSet(*processParameterSet_));
    bool hasSubProcesses = subProcessVParameterSet.get() != nullptr;
  
    ScheduleItems items(*parentProductRegistry, *this);
    actReg_ = items.actReg_;

    ParameterSet const& optionsPset(processParameterSet_->getUntrackedParameterSet("options", ParameterSet()));
    IllegalParameters::setThrowAnException(optionsPset.getUntrackedParameter<bool>("throwIfIllegalParameter", true));

    //initialize the services
    ServiceToken iToken;

    // get any configured services.
    std::unique_ptr<std::vector<ParameterSet> > serviceSets = processParameterSet_->popVParameterSet(std::string("services")); 

    ServiceToken newToken = items.initServices(*serviceSets, *processParameterSet_, token, iLegacy, false);
    parentActReg.connectToSubProcess(*items.actReg_);
    serviceToken_ = items.addCPRandTNS(*processParameterSet_, newToken);


    //make the services available
    ServiceRegistry::Operate operate(serviceToken_);

    // intialize miscellaneous items
    items.initMisc(*processParameterSet_);

    // intialize the event setup provider
    esp_ = esController.makeProvider(*processParameterSet_);

    branchIDListHelper_ = items.branchIDListHelper();
    updateBranchIDListHelper(parentBranchIDListHelper->branchIDLists());

    thinnedAssociationsHelper_ = items.thinnedAssociationsHelper();
    thinnedAssociationsHelper_->updateFromParentProcess(parentThinnedAssociationsHelper, keepAssociation, droppedBranchIDToKeptBranchID_);

    // intialize the Schedule
    schedule_ = items.initSchedule(*processParameterSet_,hasSubProcesses,preallocConfig,&processContext_);

    // set the items
    act_table_ = std::move(items.act_table_);
    preg_ = items.preg();
    //CMS-THREADING this only works since Run/Lumis are synchronous so when principalCache asks for
    // the reducedProcessHistoryID from a full ProcessHistoryID that registry will not be in use by
    // another thread. We really need to change how this is done in the PrincipalCache.
    principalCache_.setProcessHistoryRegistry(processHistoryRegistries_[historyRunOffset_]);


    processConfiguration_ = items.processConfiguration();
    processContext_.setProcessConfiguration(processConfiguration_.get());
    processContext_.setParentProcessContext(parentProcessContext);

    principalCache_.setNumberOfConcurrentPrincipals(preallocConfig);
    for(unsigned int index = 0; index < preallocConfig.numberOfStreams(); ++index) {
      auto ep = std::make_shared<EventPrincipal>(preg_,
                                                 branchIDListHelper(),
                                                 thinnedAssociationsHelper(),
                                                 *processConfiguration_,
                                                 &(historyAppenders_[index]),
                                                 index);
      ep->preModuleDelayedGetSignal_.connect(std::cref(items.actReg_->preModuleEventDelayedGetSignal_));
      ep->postModuleDelayedGetSignal_.connect(std::cref(items.actReg_->postModuleEventDelayedGetSignal_));
      principalCache_.insert(ep);
    }
    if(hasSubProcesses) {
      if(subProcesses_ == nullptr) {
        subProcesses_ = std::make_unique<std::vector<SubProcess> >();
      }
      subProcesses_->reserve(subProcessVParameterSet->size());
      for(auto& subProcessPSet : *subProcessVParameterSet) {
        subProcesses_->emplace_back(subProcessPSet,
                                    topLevelParameterSet,
                                    preg_,
                                    branchIDListHelper(),
                                    *thinnedAssociationsHelper_,
                                    esController,
                                    *items.actReg_,
                                    newToken,
                                    iLegacy,
                                    preallocConfig,
                                    &processContext_);
       }
    }
  }

  SubProcess::~SubProcess() {}

  void
  SubProcess::doBeginJob() {
    this->beginJob();
  }
  
  void
  SubProcess::doEndJob() {
    endJob();
  }
  

  void
  SubProcess::beginJob() {
    if(!droppedBranchIDToKeptBranchID().empty()) {
      fixBranchIDListsForEDAliases(droppedBranchIDToKeptBranchID());
    }
    ServiceRegistry::Operate operate(serviceToken_);
    pathsAndConsumesOfModules_.initialize(schedule_.get(), preg_);
    actReg_->preBeginJobSignal_(pathsAndConsumesOfModules_, processContext_);
    schedule_->beginJob(*preg_);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doBeginJob();
      }
    }
  }

  void
  SubProcess::endJob() {
    ServiceRegistry::Operate operate(serviceToken_);
    ExceptionCollector c("Multiple exceptions were thrown while executing endJob. An exception message follows for each.");
    schedule_->endJob(c);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        c.call([&subProcess](){ subProcess.doEndJob();});
      }
    }
    if(c.hasThrown()) {
      c.rethrow();
    }
  }

  void
  SubProcess::selectProducts(ProductRegistry const& preg,
                             ThinnedAssociationsHelper const& parentThinnedAssociationsHelper,
                             std::map<BranchID, bool>& keepAssociation) {
    if(productSelector_.initialized()) return;
    productSelector_.initialize(productSelectorRules_, preg.allBranchDescriptions());
    
    // TODO: See if we can collapse keptProducts_ and productSelector_ into a
    // single object. See the notes in the header for ProductSelector
    // for more information.
    
    std::map<BranchID, BranchDescription const*> trueBranchIDToKeptBranchDesc;
    std::vector<BranchDescription const*> associationDescriptions;
    std::set<BranchID> keptProductsInEvent;
    
    for(auto const& it : preg.productList()) {
      BranchDescription const& desc = it.second;
      if(desc.transient()) {
        // if the class of the branch is marked transient, output nothing
      } else if(!desc.present() && !desc.produced()) {
        // else if the branch containing the product has been previously dropped,
        // output nothing
      } else if(desc.unwrappedType() == typeid(ThinnedAssociation)) {
        associationDescriptions.push_back(&desc);
      } else if(productSelector_.selected(desc)) {
        keepThisBranch(desc, trueBranchIDToKeptBranchDesc, keptProductsInEvent);
      }
    }

    parentThinnedAssociationsHelper.selectAssociationProducts(associationDescriptions,
                                                              keptProductsInEvent,
                                                              keepAssociation);

    for(auto association : associationDescriptions) {
      if(keepAssociation[association->branchID()]) {
        keepThisBranch(*association, trueBranchIDToKeptBranchDesc, keptProductsInEvent);
      }
    }

    // Now fill in a mapping needed in the case that a branch was dropped while its EDAlias was kept.
    ProductSelector::fillDroppedToKept(preg, trueBranchIDToKeptBranchDesc, droppedBranchIDToKeptBranchID_);
  }

  void SubProcess::keepThisBranch(BranchDescription const& desc,
                                  std::map<BranchID, BranchDescription const*>& trueBranchIDToKeptBranchDesc,
                                  std::set<BranchID>& keptProductsInEvent) {

    ProductSelector::checkForDuplicateKeptBranch(desc,
                                                 trueBranchIDToKeptBranchDesc);

    if(desc.branchType() == InEvent) {
      if(desc.produced()) {
        keptProductsInEvent.insert(desc.originalBranchID());
      } else {
        keptProductsInEvent.insert(desc.branchID());
      }
    }
    // Now put it in the list of selected branches.
    keptProducts_[desc.branchType()].push_back(&desc);
  }

  void
  SubProcess::fixBranchIDListsForEDAliases(std::map<BranchID::value_type, BranchID::value_type> const& droppedBranchIDToKeptBranchID) {
    // Check for branches dropped while an EDAlias was kept.
    // Replace BranchID of each dropped branch with that of the kept alias.
    for(BranchIDList& branchIDList : branchIDListHelper_->mutableBranchIDLists()) {
      for(BranchID::value_type& branchID : branchIDList) {
        std::map<BranchID::value_type, BranchID::value_type>::const_iterator iter = droppedBranchIDToKeptBranchID.find(branchID);
        if(iter != droppedBranchIDToKeptBranchID.end()) {
          branchID = iter->second;
        }
      }
    }
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.fixBranchIDListsForEDAliases(droppedBranchIDToKeptBranchID);
      }
    }
  }

  void
  SubProcess::doEvent(EventPrincipal const& ep) {
    ServiceRegistry::Operate operate(serviceToken_);
    /* BEGIN relevant bits from OutputModule::doEvent */
    if(!wantAllEvents_) {
      // use module description and const_cast unless interface to
      // event is changed to just take a const EventPrincipal
      if(!selectors_.wantEvent(ep, nullptr)) {
        return;
      }
    }
    process(ep);
    /* END relevant bits from OutputModule::doEvent */
  }

  void
  SubProcess::process(EventPrincipal const& principal) {
    EventAuxiliary aux(principal.aux());
    aux.setProcessHistoryID(principal.processHistoryID());

    EventSelectionIDVector esids{principal.eventSelectionIDs()};
    if (principal.productRegistry().anyProductProduced() || !wantAllEvents_) {
      esids.push_back(selector_config_id_);
    }

    EventPrincipal& ep = principalCache_.eventPrincipal(principal.streamID().value());
    auto & processHistoryRegistry = processHistoryRegistries_[principal.streamID().value()];
    processHistoryRegistry.registerProcessHistory(principal.processHistory());
    BranchListIndexes bli(principal.branchListIndexes());
    branchIDListHelper_->fixBranchListIndexes(bli);
    ep.fillEventPrincipal(aux,
                          processHistoryRegistry,
                          std::move(esids),
                          std::move(bli),
                          *(principal.productProvenanceRetrieverPtr()),//NOTE: this transfers the per product provenance
                          principal.reader());
    ep.setLuminosityBlockPrincipal(principalCache_.lumiPrincipalPtr());
    propagateProducts(InEvent, principal, ep);
    typedef OccurrenceTraits<EventPrincipal, BranchActionStreamBegin> Traits;
    schedule_->processOneEvent<Traits>(ep.streamID().value(),ep, esp_->eventSetup());
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doEvent(ep);
      }
    }
    ep.clearEventPrincipal();
  }

  void
  SubProcess::doBeginRun(RunPrincipal const& principal, IOVSyncValue const& ts) {
    ServiceRegistry::Operate operate(serviceToken_);
    beginRun(principal,ts);
  }

  void
  SubProcess::beginRun(RunPrincipal const& principal, IOVSyncValue const& ts) {
    auto aux = std::make_shared<RunAuxiliary>(principal.aux());
    aux->setProcessHistoryID(principal.processHistoryID());
    auto rpp = std::make_shared<RunPrincipal>(aux, preg_, *processConfiguration_, &(historyAppenders_[historyRunOffset_+principal.index()]),principal.index());
    auto & processHistoryRegistry = processHistoryRegistries_[historyRunOffset_+principal.index()];
    processHistoryRegistry.registerProcessHistory(principal.processHistory());
    rpp->fillRunPrincipal(processHistoryRegistry, principal.reader());
    principalCache_.insert(rpp);

    ProcessHistoryID const& parentInputReducedPHID = principal.reducedProcessHistoryID();
    ProcessHistoryID const& inputReducedPHID       = rpp->reducedProcessHistoryID();

    parentToChildPhID_.insert(std::make_pair(parentInputReducedPHID,inputReducedPHID));

    RunPrincipal& rp = *principalCache_.runPrincipalPtr();
    propagateProducts(InRun, principal, rp);
    typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin> Traits;
    schedule_->processOneGlobal<Traits>(rp, esp_->eventSetupForInstance(ts));
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doBeginRun(rp, ts);
      }
    }
  }

  void
  SubProcess::doEndRun(RunPrincipal const& principal, IOVSyncValue const& ts, bool cleaningUpAfterException) {
    ServiceRegistry::Operate operate(serviceToken_);
    endRun(principal,ts,cleaningUpAfterException);
  }

  void
  SubProcess::endRun(RunPrincipal const& principal, IOVSyncValue const& ts, bool cleaningUpAfterException) {
    RunPrincipal& rp = *principalCache_.runPrincipalPtr();
    propagateProducts(InRun, principal, rp);
    typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd> Traits;
    schedule_->processOneGlobal<Traits>(rp, esp_->eventSetupForInstance(ts), cleaningUpAfterException);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doEndRun(rp, ts, cleaningUpAfterException);
      }
    }
  }

  void
  SubProcess::writeRun(ProcessHistoryID const& parentPhID, int runNumber) {
    ServiceRegistry::Operate operate(serviceToken_);
    std::map<ProcessHistoryID, ProcessHistoryID>::const_iterator it = parentToChildPhID_.find(parentPhID);
    assert(it != parentToChildPhID_.end());
    schedule_->writeRun(principalCache_.runPrincipal(it->second, runNumber), &processContext_);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.writeRun(it->second, runNumber);
      }
    }
  }

  void
  SubProcess::deleteRunFromCache(ProcessHistoryID const& parentPhID, int runNumber) {
    std::map<ProcessHistoryID, ProcessHistoryID>::const_iterator it = parentToChildPhID_.find(parentPhID);
    assert(it != parentToChildPhID_.end());
    principalCache_.deleteRun(it->second, runNumber);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.deleteRunFromCache(it->second, runNumber);
      }
    }
  }

  void
  SubProcess::doBeginLuminosityBlock(LuminosityBlockPrincipal const& principal, IOVSyncValue const& ts) {
    ServiceRegistry::Operate operate(serviceToken_);
    beginLuminosityBlock(principal,ts);
  }

  void
  SubProcess::beginLuminosityBlock(LuminosityBlockPrincipal const& principal, IOVSyncValue const& ts) {
    auto aux = std::make_shared<LuminosityBlockAuxiliary>(principal.aux());
    aux->setProcessHistoryID(principal.processHistoryID());
    auto lbpp = std::make_shared<LuminosityBlockPrincipal>(aux, preg_, *processConfiguration_, &(historyAppenders_[historyLumiOffset_+principal.index()]),principal.index());
    auto & processHistoryRegistry = processHistoryRegistries_[historyLumiOffset_+principal.index()];
    processHistoryRegistry.registerProcessHistory(principal.processHistory());
    lbpp->fillLuminosityBlockPrincipal(processHistoryRegistry, principal.reader());
    lbpp->setRunPrincipal(principalCache_.runPrincipalPtr());
    principalCache_.insert(lbpp);
    LuminosityBlockPrincipal& lbp = *principalCache_.lumiPrincipalPtr();
    propagateProducts(InLumi, principal, lbp);
    typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin> Traits;
    schedule_->processOneGlobal<Traits>(lbp, esp_->eventSetupForInstance(ts));
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doBeginLuminosityBlock(lbp, ts);
      }
    }
  }

  void
  SubProcess::doEndLuminosityBlock(LuminosityBlockPrincipal const& principal, IOVSyncValue const& ts, bool cleaningUpAfterException) {
    ServiceRegistry::Operate operate(serviceToken_);
    endLuminosityBlock(principal,ts,cleaningUpAfterException);
  }

  void
  SubProcess::endLuminosityBlock(LuminosityBlockPrincipal const& principal, IOVSyncValue const& ts, bool cleaningUpAfterException) {
    LuminosityBlockPrincipal& lbp = *principalCache_.lumiPrincipalPtr();
    propagateProducts(InLumi, principal, lbp);
    typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd> Traits;
    schedule_->processOneGlobal<Traits>(lbp, esp_->eventSetupForInstance(ts), cleaningUpAfterException);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doEndLuminosityBlock(lbp, ts, cleaningUpAfterException);
      }
    }
  }

  void
  SubProcess::writeLumi(ProcessHistoryID const& parentPhID, int runNumber, int lumiNumber) {
    ServiceRegistry::Operate operate(serviceToken_);
    std::map<ProcessHistoryID, ProcessHistoryID>::const_iterator it = parentToChildPhID_.find(parentPhID);
    assert(it != parentToChildPhID_.end());
    schedule_->writeLumi(principalCache_.lumiPrincipal(it->second, runNumber, lumiNumber), &processContext_);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.writeLumi(it->second, runNumber, lumiNumber);
      }
    }
  }

  void
  SubProcess::deleteLumiFromCache(ProcessHistoryID const& parentPhID, int runNumber, int lumiNumber) {
    std::map<ProcessHistoryID, ProcessHistoryID>::const_iterator it = parentToChildPhID_.find(parentPhID);
    assert(it != parentToChildPhID_.end());
    principalCache_.deleteLumi(it->second, runNumber, lumiNumber);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.deleteLumiFromCache(it->second, runNumber, lumiNumber);
      }
    }
  }
  
  void
  SubProcess::doBeginStream(unsigned int iID) {
    ServiceRegistry::Operate operate(serviceToken_);
    schedule_->beginStream(iID);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doBeginStream(iID);
      }
    }
  }

  void
  SubProcess::doEndStream(unsigned int iID) {
    ServiceRegistry::Operate operate(serviceToken_);
    schedule_->endStream(iID);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.doEndStream(iID);
      }
    }
  }

  void
  SubProcess::doStreamBeginRun(unsigned int id, RunPrincipal const& principal, IOVSyncValue const& ts) {
    ServiceRegistry::Operate operate(serviceToken_);
    {
      RunPrincipal& rp = *principalCache_.runPrincipalPtr();
      typedef OccurrenceTraits<RunPrincipal, BranchActionStreamBegin> Traits;
      schedule_->processOneStream<Traits>(id,rp, esp_->eventSetupForInstance(ts));
      if(hasSubProcesses()) {
        for(auto& subProcess : *subProcesses_) {
          subProcess.doStreamBeginRun(id,rp, ts);
        }
      }
    }
  }
  
  void
  SubProcess::doStreamEndRun(unsigned int id, RunPrincipal const& principal, IOVSyncValue const& ts, bool cleaningUpAfterException) {
    ServiceRegistry::Operate operate(serviceToken_);
    {
      RunPrincipal& rp = *principalCache_.runPrincipalPtr();
      typedef OccurrenceTraits<RunPrincipal, BranchActionStreamEnd> Traits;
      schedule_->processOneStream<Traits>(id,rp, esp_->eventSetupForInstance(ts),cleaningUpAfterException);
      if(hasSubProcesses()) {
        for(auto& subProcess : *subProcesses_) {
          subProcess.doStreamEndRun(id,rp, ts,cleaningUpAfterException);
        }
      }
    }
  }
  
  void
  SubProcess::doStreamBeginLuminosityBlock(unsigned int id, LuminosityBlockPrincipal const& principal, IOVSyncValue const& ts) {
    ServiceRegistry::Operate operate(serviceToken_);
    {
      LuminosityBlockPrincipal& lbp = *principalCache_.lumiPrincipalPtr();
      typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin> Traits;
      schedule_->processOneStream<Traits>(id,lbp, esp_->eventSetupForInstance(ts));
      if(hasSubProcesses()) {
        for(auto& subProcess : *subProcesses_) {
          subProcess.doStreamBeginLuminosityBlock(id,lbp, ts);
        }
      }
    }
  }
  
  void
  SubProcess::doStreamEndLuminosityBlock(unsigned int id, LuminosityBlockPrincipal const& principal, IOVSyncValue const& ts, bool cleaningUpAfterException) {
    ServiceRegistry::Operate operate(serviceToken_);
    {
      LuminosityBlockPrincipal& lbp = *principalCache_.lumiPrincipalPtr();
      typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd> Traits;
      schedule_->processOneStream<Traits>(id,lbp, esp_->eventSetupForInstance(ts),cleaningUpAfterException);
      if(hasSubProcesses()) {
        for(auto& subProcess : *subProcesses_) {
          subProcess.doStreamEndLuminosityBlock(id,lbp, ts,cleaningUpAfterException);
        }
      }
    }
  }


  void
  SubProcess::propagateProducts(BranchType type, Principal const& parentPrincipal, Principal& principal) const {
    SelectedProducts const& keptVector = keptProducts()[type];
    for(auto const& item : keptVector) {
      ProductHolderBase const* parentProductHolder = parentPrincipal.getProductHolder(item->branchID());
      if(parentProductHolder != nullptr) {
        ProductData const& parentData = parentProductHolder->productData();
        ProductHolderBase* productHolder = principal.getModifiableProductHolder(item->branchID());
        if(productHolder != nullptr) {
          ProductData& thisData = productHolder->productData();
          //Propagate the per event(run)(lumi) data for this product to the subprocess.
          //First, the product itself.
          thisData.connectTo(parentData);
          // Sets unavailable flag, if known that product is not available
          (void)productHolder->productUnavailable();
        }
      }
    }
  }

  void SubProcess::updateBranchIDListHelper(BranchIDLists const& branchIDLists) {
    branchIDListHelper_->updateFromParent(branchIDLists);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.updateBranchIDListHelper(branchIDListHelper_->branchIDLists());
      }
    }
  }

  // Call respondToOpenInputFile() on all Modules
  void
  SubProcess::respondToOpenInputFile(FileBlock const& fb) {
    ServiceRegistry::Operate operate(serviceToken_);
    schedule_->respondToOpenInputFile(fb);
    if(hasSubProcesses()) {
      for(auto& subProcess : *subProcesses_) {
        subProcess.respondToOpenInputFile(fb);
      }
    }
  }

  // free function
  std::unique_ptr<std::vector<ParameterSet> >
  popSubProcessVParameterSet(ParameterSet& parameterSet) {
    std::vector<std::string> subProcesses = parameterSet.getUntrackedParameter<std::vector<std::string> >("@all_subprocesses");
    if(!subProcesses.empty()) {
      return parameterSet.popVParameterSet("subProcesses");
    }
    return std::unique_ptr<std::vector<ParameterSet> >(nullptr);
  }
}