Project CMSSW displayed by LXR CMSSW_15_0_X_2024-12-09-2300/​FWCore/​Framework/​src/​StreamSchedule.cc	Source navigation Diff markup Identifier search General search
	Version: CMSSW_15_0_X_2024-12-09-2300 CMSSW_15_0_X_2024-12-08-2300 CMSSW_15_0_X_2024-12-06-2300 CMSSW_15_0_X_2024-12-05-2300 CMSSW_15_0_X_2024-12-04-2300 CMSSW_15_0_X_2024-12-03-2300 Revert   Change

File indexing completed on 2024-10-29 06:08:21

 #include "FWCore/Framework/interface/StreamSchedule.h"
 
 #include "DataFormats/Provenance/interface/BranchIDListHelper.h"
 #include "DataFormats/Provenance/interface/EventID.h"
 #include "DataFormats/Provenance/interface/ProcessConfiguration.h"
 #include "DataFormats/Provenance/interface/ProductRegistry.h"
 #include "DataFormats/Provenance/interface/ProductResolverIndexHelper.h"
 #include "DataFormats/Provenance/interface/Timestamp.h"
 #include "FWCore/Framework/src/OutputModuleDescription.h"
 #include "FWCore/Framework/interface/TriggerNamesService.h"
 #include "FWCore/Framework/src/TriggerReport.h"
 #include "FWCore/Framework/src/TriggerTimingReport.h"
 #include "FWCore/Framework/src/Factory.h"
 #include "FWCore/Framework/interface/OutputModuleCommunicator.h"
 #include "FWCore/Framework/src/TriggerResultInserter.h"
 #include "FWCore/Framework/src/PathStatusInserter.h"
 #include "FWCore/Framework/src/EndPathStatusInserter.h"
 #include "FWCore/Framework/interface/WorkerInPath.h"
 #include "FWCore/Framework/interface/maker/ModuleHolder.h"
 #include "FWCore/Framework/interface/maker/WorkerT.h"
 #include "FWCore/Framework/interface/ModuleRegistry.h"
 #include "FWCore/MessageLogger/interface/MessageLogger.h"
 #include "FWCore/ParameterSet/interface/ParameterSet.h"
 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
 #include "FWCore/ParameterSet/interface/Registry.h"
 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
 #include "FWCore/ServiceRegistry/interface/PathContext.h"
 #include "FWCore/Utilities/interface/Algorithms.h"
 #include "FWCore/Utilities/interface/ExceptionCollector.h"
 #include "FWCore/Utilities/interface/LuminosityBlockIndex.h"
 #include "FWCore/Utilities/interface/RunIndex.h"
 
 #include "LuminosityBlockProcessingStatus.h"
 #include "processEDAliases.h"
 
 #include <algorithm>
 #include <cassert>
 #include <cstdlib>
 #include <functional>
 #include <iomanip>
 #include <limits>
 #include <list>
 #include <map>
 #include <fmt/format.h>
 
 namespace edm {
 
   namespace {
 
     // Function template to transform each element in the input range to
     // a value placed into the output range. The supplied function
     // should take a const_reference to the 'input', and write to a
     // reference to the 'output'.
     template <typename InputIterator, typename ForwardIterator, typename Func>
     void transform_into(InputIterator begin, InputIterator end, ForwardIterator out, Func func) {
       for (; begin != end; ++begin, ++out)
         func(*begin, *out);
     }
 
     // Function template that takes a sequence 'from', a sequence
     // 'to', and a callable object 'func'. It and applies
     // transform_into to fill the 'to' sequence with the values
     // calcuated by the callable object, taking care to fill the
     // outupt only if all calls succeed.
     template <typename FROM, typename TO, typename FUNC>
     void fill_summary(FROM const& from, TO& to, FUNC func) {
       if (to.size() != from.size()) {
         TO temp(from.size());
         transform_into(from.begin(), from.end(), temp.begin(), func);
         to.swap(temp);
       } else {
         transform_into(from.begin(), from.end(), to.begin(), func);
       }
     }
 
     class BeginStreamTraits {
     public:
       static void preScheduleSignal(ActivityRegistry* activityRegistry, StreamContext const* streamContext) {
         activityRegistry->preBeginStreamSignal_(*streamContext);
       }
       static void postScheduleSignal(ActivityRegistry* activityRegistry, StreamContext const* streamContext) {
         activityRegistry->postBeginStreamSignal_(*streamContext);
       }
     };
 
     class EndStreamTraits {
     public:
       static void preScheduleSignal(ActivityRegistry* activityRegistry, StreamContext const* streamContext) {
         activityRegistry->preEndStreamSignal_(*streamContext);
       }
       static void postScheduleSignal(ActivityRegistry* activityRegistry, StreamContext const* streamContext) {
         activityRegistry->postEndStreamSignal_(*streamContext);
       }
     };
 
     // -----------------------------
 
     // Here we make the trigger results inserter directly.  This should
     // probably be a utility in the WorkerRegistry or elsewhere.
 
     StreamSchedule::WorkerPtr makeInserter(ExceptionToActionTable const& actions,
                                            std::shared_ptr<ActivityRegistry> areg,
                                            std::shared_ptr<TriggerResultInserter> inserter) {
       StreamSchedule::WorkerPtr ptr(
           new edm::WorkerT<TriggerResultInserter::ModuleType>(inserter, inserter->moduleDescription(), &actions));
       ptr->setActivityRegistry(areg);
       return ptr;
     }
 
     void initializeBranchToReadingWorker(std::vector<std::string> const& branchesToDeleteEarly,
                                          ProductRegistry const& preg,
                                          std::multimap<std::string, Worker*>& branchToReadingWorker) {
       auto vBranchesToDeleteEarly = branchesToDeleteEarly;
       // Remove any duplicates
       std::sort(vBranchesToDeleteEarly.begin(), vBranchesToDeleteEarly.end(), std::less<std::string>());
       vBranchesToDeleteEarly.erase(std::unique(vBranchesToDeleteEarly.begin(), vBranchesToDeleteEarly.end()),
                                    vBranchesToDeleteEarly.end());
 
       // Are the requested items in the product registry?
       auto allBranchNames = preg.allBranchNames();
       //the branch names all end with a period, which we do not want to compare with
       for (auto& b : allBranchNames) {
         b.resize(b.size() - 1);
       }
       std::sort(allBranchNames.begin(), allBranchNames.end(), std::less<std::string>());
       std::vector<std::string> temp;
       temp.reserve(vBranchesToDeleteEarly.size());
 
       std::set_intersection(vBranchesToDeleteEarly.begin(),
                             vBranchesToDeleteEarly.end(),
                             allBranchNames.begin(),
                             allBranchNames.end(),
                             std::back_inserter(temp));
       vBranchesToDeleteEarly.swap(temp);
       if (temp.size() != vBranchesToDeleteEarly.size()) {
         std::vector<std::string> missingProducts;
         std::set_difference(temp.begin(),
                             temp.end(),
                             vBranchesToDeleteEarly.begin(),
                             vBranchesToDeleteEarly.end(),
                             std::back_inserter(missingProducts));
         LogInfo l("MissingProductsForCanDeleteEarly");
         l << "The following products in the 'canDeleteEarly' list are not available in this job and will be ignored.";
         for (auto const& n : missingProducts) {
           l << "\n " << n;
         }
       }
       //set placeholder for the branch, we will remove the nullptr if a
       // module actually wants the branch.
       for (auto const& branch : vBranchesToDeleteEarly) {
         branchToReadingWorker.insert(std::make_pair(branch, static_cast<Worker*>(nullptr)));
       }
     }
 
     Worker* getWorker(std::string const& moduleLabel,
                       ParameterSet& proc_pset,
                       WorkerManager& workerManager,
                       ProductRegistry& preg,
                       PreallocationConfiguration const* prealloc,
                       std::shared_ptr<ProcessConfiguration const> processConfiguration) {
       bool isTracked;
       ParameterSet* modpset = proc_pset.getPSetForUpdate(moduleLabel, isTracked);
       if (modpset == nullptr) {
         return nullptr;
       }
       assert(isTracked);
 
       return workerManager.getWorker(*modpset, preg, prealloc, processConfiguration, moduleLabel);
     }
 
     // If ConditionalTask modules exist in the container of module
     // names, returns the range (std::pair) for the modules. The range
     // excludes the special markers '#' (right before the
     // ConditionalTask modules) and '@' (last element).
     // If the module name container does not contain ConditionalTask
     // modules, returns std::pair of end iterators.
     template <typename T>
     auto findConditionalTaskModulesRange(T& modnames) {
       auto beg = std::find(modnames.begin(), modnames.end(), "#");
       if (beg == modnames.end()) {
         return std::pair(modnames.end(), modnames.end());
       }
       return std::pair(beg + 1, std::prev(modnames.end()));
     }
 
     std::optional<std::string> findBestMatchingAlias(
         std::unordered_multimap<std::string, edm::BranchDescription const*> const& conditionalModuleBranches,
         std::unordered_multimap<std::string, StreamSchedule::AliasInfo> const& aliasMap,
         std::string const& productModuleLabel,
         ConsumesInfo const& consumesInfo) {
       std::optional<std::string> best;
       int wildcardsInBest = std::numeric_limits<int>::max();
       bool bestIsAmbiguous = false;
 
       auto updateBest = [&best, &wildcardsInBest, &bestIsAmbiguous](
                             std::string const& label, bool instanceIsWildcard, bool typeIsWildcard) {
         int const wildcards = static_cast<int>(instanceIsWildcard) + static_cast<int>(typeIsWildcard);
         if (wildcards == 0) {
           bestIsAmbiguous = false;
           return true;
         }
         if (not best or wildcards < wildcardsInBest) {
           best = label;
           wildcardsInBest = wildcards;
           bestIsAmbiguous = false;
         } else if (best and *best != label and wildcardsInBest == wildcards) {
           bestIsAmbiguous = true;
         }
         return false;
       };
 
       auto findAlias = aliasMap.equal_range(productModuleLabel);
       for (auto it = findAlias.first; it != findAlias.second; ++it) {
         std::string const& aliasInstanceLabel =
             it->second.instanceLabel != "*" ? it->second.instanceLabel : it->second.originalInstanceLabel;
         bool const instanceIsWildcard = (aliasInstanceLabel == "*");
         if (instanceIsWildcard or consumesInfo.instance() == aliasInstanceLabel) {
           bool const typeIsWildcard = it->second.friendlyClassName == "*";
           if (typeIsWildcard or (consumesInfo.type().friendlyClassName() == it->second.friendlyClassName)) {
             if (updateBest(it->second.originalModuleLabel, instanceIsWildcard, typeIsWildcard)) {
               return it->second.originalModuleLabel;
             }
           } else if (consumesInfo.kindOfType() == ELEMENT_TYPE) {
             //consume is a View so need to do more intrusive search
             //find matching branches in module
             auto branches = conditionalModuleBranches.equal_range(productModuleLabel);
             for (auto itBranch = branches.first; itBranch != branches.second; ++it) {
               if (typeIsWildcard or itBranch->second->productInstanceName() == it->second.originalInstanceLabel) {
                 if (productholderindexhelper::typeIsViewCompatible(consumesInfo.type(),
                                                                    TypeID(itBranch->second->wrappedType().typeInfo()),
                                                                    itBranch->second->className())) {
                   if (updateBest(it->second.originalModuleLabel, instanceIsWildcard, typeIsWildcard)) {
                     return it->second.originalModuleLabel;
                   }
                 }
               }
             }
           }
         }
       }
       if (bestIsAmbiguous) {
         throw Exception(errors::UnimplementedFeature)
             << "Encountered ambiguity when trying to find a best-matching alias for\n"
             << " friendly class name " << consumesInfo.type().friendlyClassName() << "\n"
             << " module label " << productModuleLabel << "\n"
             << " product instance name " << consumesInfo.instance() << "\n"
             << "when processing EDAliases for modules in ConditionalTasks. Two aliases have the same number of "
                "wildcards ("
             << wildcardsInBest << ")";
       }
       return best;
     }
   }  // namespace
 
   // -----------------------------
 
   typedef std::vector<std::string> vstring;
 
   // -----------------------------
 
   class ConditionalTaskHelper {
   public:
     using AliasInfo = StreamSchedule::AliasInfo;
 
     ConditionalTaskHelper(ParameterSet& proc_pset,
                           ProductRegistry& preg,
                           PreallocationConfiguration const* prealloc,
                           std::shared_ptr<ProcessConfiguration const> processConfiguration,
                           WorkerManager& workerManagerLumisAndEvents,
                           std::vector<std::string> const& trigPathNames) {
       std::unordered_set<std::string> allConditionalMods;
       for (auto const& pathName : trigPathNames) {
         auto const modnames = proc_pset.getParameter<vstring>(pathName);
 
         //Pull out ConditionalTask modules
         auto condRange = findConditionalTaskModulesRange(modnames);
         if (condRange.first == condRange.second)
           continue;
 
         //the last entry should be ignored since it is required to be "@"
         allConditionalMods.insert(condRange.first, condRange.second);
       }
 
       for (auto const& cond : allConditionalMods) {
         //force the creation of the conditional modules so alias check can work
         (void)getWorker(cond, proc_pset, workerManagerLumisAndEvents, preg, prealloc, processConfiguration);
       }
 
       fillAliasMap(proc_pset, allConditionalMods);
       processSwitchEDAliases(proc_pset, preg, *processConfiguration, allConditionalMods);
 
       //find branches created by the conditional modules
       for (auto const& prod : preg.productList()) {
         if (allConditionalMods.find(prod.first.moduleLabel()) != allConditionalMods.end()) {
           conditionalModsBranches_.emplace(prod.first.moduleLabel(), &prod.second);
         }
       }
     }
 
     std::unordered_multimap<std::string, AliasInfo> const& aliasMap() const { return aliasMap_; }
 
     std::unordered_multimap<std::string, edm::BranchDescription const*> conditionalModuleBranches(
         std::unordered_set<std::string> const& conditionalmods) const {
       std::unordered_multimap<std::string, edm::BranchDescription const*> ret;
       for (auto const& mod : conditionalmods) {
         auto range = conditionalModsBranches_.equal_range(mod);
         ret.insert(range.first, range.second);
       }
       return ret;
     }
 
   private:
     void fillAliasMap(ParameterSet const& proc_pset, std::unordered_set<std::string> const& allConditionalMods) {
       auto aliases = proc_pset.getParameter<std::vector<std::string>>("@all_aliases");
       std::string const star("*");
       for (auto const& alias : aliases) {
         auto info = proc_pset.getParameter<edm::ParameterSet>(alias);
         auto aliasedToModuleLabels = info.getParameterNames();
         for (auto const& mod : aliasedToModuleLabels) {
           if (not mod.empty() and mod[0] != '@' and allConditionalMods.find(mod) != allConditionalMods.end()) {
             auto aliasVPSet = info.getParameter<std::vector<edm::ParameterSet>>(mod);
             for (auto const& aliasPSet : aliasVPSet) {
               std::string type = star;
               std::string instance = star;
               std::string originalInstance = star;
               if (aliasPSet.exists("type")) {
                 type = aliasPSet.getParameter<std::string>("type");
               }
               if (aliasPSet.exists("toProductInstance")) {
                 instance = aliasPSet.getParameter<std::string>("toProductInstance");
               }
               if (aliasPSet.exists("fromProductInstance")) {
                 originalInstance = aliasPSet.getParameter<std::string>("fromProductInstance");
               }
 
               aliasMap_.emplace(alias, AliasInfo{type, instance, originalInstance, mod});
             }
           }
         }
       }
     }
 
     void processSwitchEDAliases(ParameterSet const& proc_pset,
                                 ProductRegistry& preg,
                                 ProcessConfiguration const& processConfiguration,
                                 std::unordered_set<std::string> const& allConditionalMods) {
       auto const& all_modules = proc_pset.getParameter<std::vector<std::string>>("@all_modules");
       std::vector<std::string> switchEDAliases;
       for (auto const& module : all_modules) {
         auto const& mod_pset = proc_pset.getParameter<edm::ParameterSet>(module);
         if (mod_pset.getParameter<std::string>("@module_type") == "SwitchProducer") {
           auto const& all_cases = mod_pset.getParameter<std::vector<std::string>>("@all_cases");
           for (auto const& case_label : all_cases) {
             auto range = aliasMap_.equal_range(case_label);
             if (range.first != range.second) {
               switchEDAliases.push_back(case_label);
             }
           }
         }
       }
       detail::processEDAliases(
           switchEDAliases, allConditionalMods, proc_pset, processConfiguration.processName(), preg);
     }
 
     std::unordered_multimap<std::string, AliasInfo> aliasMap_;
     std::unordered_multimap<std::string, edm::BranchDescription const*> conditionalModsBranches_;
   };
 
   // -----------------------------
 
   StreamSchedule::StreamSchedule(
       std::shared_ptr<TriggerResultInserter> inserter,
       std::vector<edm::propagate_const<std::shared_ptr<PathStatusInserter>>>& pathStatusInserters,
       std::vector<edm::propagate_const<std::shared_ptr<EndPathStatusInserter>>>& endPathStatusInserters,
       std::shared_ptr<ModuleRegistry> modReg,
       ParameterSet& proc_pset,
       service::TriggerNamesService const& tns,
       PreallocationConfiguration const& prealloc,
       ProductRegistry& preg,
       ExceptionToActionTable const& actions,
       std::shared_ptr<ActivityRegistry> areg,
       std::shared_ptr<ProcessConfiguration const> processConfiguration,
       StreamID streamID,
       ProcessContext const* processContext)
       : workerManagerBeginEnd_(modReg, areg, actions),
         workerManagerRuns_(modReg, areg, actions),
         workerManagerLumisAndEvents_(modReg, areg, actions),
         actReg_(areg),
         results_(new HLTGlobalStatus(tns.getTrigPaths().size())),
         results_inserter_(),
         trig_paths_(),
         end_paths_(),
         total_events_(),
         total_passed_(),
         number_of_unscheduled_modules_(0),
         streamID_(streamID),
         streamContext_(streamID_, processContext) {
     bool hasPath = false;
     std::vector<std::string> const& pathNames = tns.getTrigPaths();
     std::vector<std::string> const& endPathNames = tns.getEndPaths();
 
     ConditionalTaskHelper conditionalTaskHelper(
         proc_pset, preg, &prealloc, processConfiguration, workerManagerLumisAndEvents_, pathNames);
     std::unordered_set<std::string> conditionalModules;
 
     int trig_bitpos = 0;
     trig_paths_.reserve(pathNames.size());
     for (auto const& trig_name : pathNames) {
       fillTrigPath(proc_pset,
                    preg,
                    &prealloc,
                    processConfiguration,
                    trig_bitpos,
                    trig_name,
                    results(),
                    endPathNames,
                    conditionalTaskHelper,
                    conditionalModules);
       ++trig_bitpos;
       hasPath = true;
     }
 
     if (hasPath) {
       // the results inserter stands alone
       inserter->setTrigResultForStream(streamID.value(), results());
 
       results_inserter_ = makeInserter(actions, actReg_, inserter);
       addToAllWorkers(results_inserter_.get());
     }
 
     // fill normal endpaths
     int bitpos = 0;
     end_paths_.reserve(endPathNames.size());
     for (auto const& end_path_name : endPathNames) {
       fillEndPath(proc_pset,
                   preg,
                   &prealloc,
                   processConfiguration,
                   bitpos,
                   end_path_name,
                   endPathNames,
                   conditionalTaskHelper,
                   conditionalModules);
       ++bitpos;
     }
 
     makePathStatusInserters(pathStatusInserters, endPathStatusInserters, actions);
 
     //See if all modules were used
     std::set<std::string> usedWorkerLabels;
     for (auto const& worker : allWorkersLumisAndEvents()) {
       usedWorkerLabels.insert(worker->description()->moduleLabel());
     }
     std::vector<std::string> modulesInConfig(proc_pset.getParameter<std::vector<std::string>>("@all_modules"));
     std::set<std::string> modulesInConfigSet(modulesInConfig.begin(), modulesInConfig.end());
     std::vector<std::string> unusedLabels;
     set_difference(modulesInConfigSet.begin(),
                    modulesInConfigSet.end(),
                    usedWorkerLabels.begin(),
                    usedWorkerLabels.end(),
                    back_inserter(unusedLabels));
     std::set<std::string> unscheduledLabels;
     std::vector<std::string> shouldBeUsedLabels;
     if (!unusedLabels.empty()) {
       //Need to
       // 1) create worker
       // 2) if it is a WorkerT<EDProducer>, add it to our list
       // 3) hand list to our delayed reader
       for (auto const& label : unusedLabels) {
         bool isTracked;
         ParameterSet* modulePSet(proc_pset.getPSetForUpdate(label, isTracked));
         assert(isTracked);
         assert(modulePSet != nullptr);
         workerManagerLumisAndEvents_.addToUnscheduledWorkers(
             *modulePSet, preg, &prealloc, processConfiguration, label, unscheduledLabels, shouldBeUsedLabels);
       }
       if (!shouldBeUsedLabels.empty()) {
         std::ostringstream unusedStream;
         unusedStream << "'" << shouldBeUsedLabels.front() << "'";
         for (std::vector<std::string>::iterator itLabel = shouldBeUsedLabels.begin() + 1,
                                                 itLabelEnd = shouldBeUsedLabels.end();
              itLabel != itLabelEnd;
              ++itLabel) {
           unusedStream << ",'" << *itLabel << "'";
         }
         LogInfo("path") << "The following module labels are not assigned to any path:\n" << unusedStream.str() << "\n";
       }
     }
     number_of_unscheduled_modules_ = unscheduledLabels.size();
 
     // Print conditional modules that were not consumed in any of their associated Paths
     if (streamID.value() == 0 and not conditionalModules.empty()) {
       // Intersection of unscheduled and ConditionalTask modules gives
       // directly the set of conditional modules that were not
       // consumed by anything in the Paths associated to the
       // corresponding ConditionalTask.
       std::vector<std::string_view> labelsToPrint;
       std::copy_if(
           unscheduledLabels.begin(),
           unscheduledLabels.end(),
           std::back_inserter(labelsToPrint),
           [&conditionalModules](auto const& lab) { return conditionalModules.find(lab) != conditionalModules.end(); });
 
       if (not labelsToPrint.empty()) {
         edm::LogWarning log("NonConsumedConditionalModules");
         log << "The following modules were part of some ConditionalTask, but were not\n"
             << "consumed by any other module in any of the Paths to which the ConditionalTask\n"
             << "was associated. Perhaps they should be either removed from the\n"
             << "job, or moved to a Task to make it explicit they are unscheduled.\n";
         for (auto const& modLabel : labelsToPrint) {
           log.format("\n {}", modLabel);
         }
       }
     }
 
     for (auto const& worker : allWorkersLumisAndEvents()) {
       std::string const& moduleLabel = worker->description()->moduleLabel();
 
       // The new worker pointers will be null for the TriggerResultsInserter, PathStatusInserter, and
       // EndPathStatusInserter because there are no ParameterSets for those in the configuration.
       // We could add special code to create workers for those, but instead we skip them because they
       // do not have beginStream, endStream, or run/lumi begin/end stream transition functions.
 
       Worker* workerBeginEnd =
           getWorker(moduleLabel, proc_pset, workerManagerBeginEnd_, preg, &prealloc, processConfiguration);
       if (workerBeginEnd) {
         workerManagerBeginEnd_.addToAllWorkers(workerBeginEnd);
       }
 
       Worker* workerRuns = getWorker(moduleLabel, proc_pset, workerManagerRuns_, preg, &prealloc, processConfiguration);
       if (workerRuns) {
         workerManagerRuns_.addToAllWorkers(workerRuns);
       }
     }
 
   }  // StreamSchedule::StreamSchedule
 
   void StreamSchedule::initializeEarlyDelete(ModuleRegistry& modReg,
                                              std::vector<std::string> const& branchesToDeleteEarly,
                                              std::multimap<std::string, std::string> const& referencesToBranches,
                                              std::vector<std::string> const& modulesToSkip,
                                              edm::ProductRegistry const& preg) {
     // setup the list with those products actually registered for this job
     std::multimap<std::string, Worker*> branchToReadingWorker;
     initializeBranchToReadingWorker(branchesToDeleteEarly, preg, branchToReadingWorker);
 
     const std::vector<std::string> kEmpty;
     std::map<Worker*, unsigned int> reserveSizeForWorker;
     unsigned int upperLimitOnReadingWorker = 0;
     unsigned int upperLimitOnIndicies = 0;
     unsigned int nUniqueBranchesToDelete = branchToReadingWorker.size();
 
     //talk with output modules first
     modReg.forAllModuleHolders([&branchToReadingWorker, &nUniqueBranchesToDelete](maker::ModuleHolder* iHolder) {
       auto comm = iHolder->createOutputModuleCommunicator();
       if (comm) {
         if (!branchToReadingWorker.empty()) {
           //If an OutputModule needs a product, we can't delete it early
           // so we should remove it from our list
           SelectedProductsForBranchType const& kept = comm->keptProducts();
           for (auto const& item : kept[InEvent]) {
             BranchDescription const& desc = *item.first;
             auto found = branchToReadingWorker.equal_range(desc.branchName());
             if (found.first != found.second) {
               --nUniqueBranchesToDelete;
               branchToReadingWorker.erase(found.first, found.second);
             }
           }
         }
       }
     });
 
     if (branchToReadingWorker.empty()) {
       return;
     }
 
     std::unordered_set<std::string> modulesToExclude(modulesToSkip.begin(), modulesToSkip.end());
     for (auto w : allWorkersLumisAndEvents()) {
       if (modulesToExclude.end() != modulesToExclude.find(w->description()->moduleLabel())) {
         continue;
       }
       //determine if this module could read a branch we want to delete early
       auto consumes = w->consumesInfo();
       if (not consumes.empty()) {
         bool foundAtLeastOneMatchingBranch = false;
         for (auto const& product : consumes) {
           std::string branch = fmt::format("{}_{}_{}_{}",
                                            product.type().friendlyClassName(),
                                            product.label().data(),
                                            product.instance().data(),
                                            product.process().data());
           {
             //Handle case where worker directly consumes product
             auto found = branchToReadingWorker.end();
             if (product.process().empty()) {
               auto startFound = branchToReadingWorker.lower_bound(branch);
               if (startFound != branchToReadingWorker.end()) {
                 if (startFound->first.substr(0, branch.size()) == branch) {
                   //match all processNames here, even if it means multiple matches will happen
                   found = startFound;
                 }
               }
             } else {
               auto exactFound = branchToReadingWorker.equal_range(branch);
               if (exactFound.first != exactFound.second) {
                 found = exactFound.first;
               }
             }
             if (found != branchToReadingWorker.end()) {
               if (not foundAtLeastOneMatchingBranch) {
                 ++upperLimitOnReadingWorker;
                 foundAtLeastOneMatchingBranch = true;
               }
               ++upperLimitOnIndicies;
               ++reserveSizeForWorker[w];
               if (nullptr == found->second) {
                 found->second = w;
               } else {
                 branchToReadingWorker.insert(make_pair(found->first, w));
               }
             }
           }
           {
             //Handle case where indirectly consumes product
             auto found = referencesToBranches.end();
             if (product.process().empty()) {
               auto startFound = referencesToBranches.lower_bound(branch);
               if (startFound != referencesToBranches.end()) {
                 if (startFound->first.substr(0, branch.size()) == branch) {
                   //match all processNames here, even if it means multiple matches will happen
                   found = startFound;
                 }
               }
             } else {
               //can match exactly
               auto exactFound = referencesToBranches.equal_range(branch);
               if (exactFound.first != exactFound.second) {
                 found = exactFound.first;
               }
             }
             if (found != referencesToBranches.end()) {
               for (auto itr = found; (itr != referencesToBranches.end()) and (itr->first == found->first); ++itr) {
                 auto foundInBranchToReadingWorker = branchToReadingWorker.find(itr->second);
                 if (foundInBranchToReadingWorker == branchToReadingWorker.end()) {
                   continue;
                 }
                 if (not foundAtLeastOneMatchingBranch) {
                   ++upperLimitOnReadingWorker;
                   foundAtLeastOneMatchingBranch = true;
                 }
                 ++upperLimitOnIndicies;
                 ++reserveSizeForWorker[w];
                 if (nullptr == foundInBranchToReadingWorker->second) {
                   foundInBranchToReadingWorker->second = w;
                 } else {
                   branchToReadingWorker.insert(make_pair(itr->second, w));
                 }
               }
             }
           }
         }
       }
     }
     {
       auto it = branchToReadingWorker.begin();
       std::vector<std::string> unusedBranches;
       while (it != branchToReadingWorker.end()) {
         if (it->second == nullptr) {
           unusedBranches.push_back(it->first);
           //erasing the object invalidates the iterator so must advance it first
           auto temp = it;
           ++it;
           branchToReadingWorker.erase(temp);
         } else {
           ++it;
         }
       }
       if (not unusedBranches.empty()) {
         LogWarning l("UnusedProductsForCanDeleteEarly");
         l << "The following products in the 'canDeleteEarly' list are not used in this job and will be ignored.\n"
              " If possible, remove the producer from the job.";
         for (auto const& n : unusedBranches) {
           l << "\n " << n;
         }
       }
     }
     if (!branchToReadingWorker.empty()) {
       earlyDeleteHelpers_.reserve(upperLimitOnReadingWorker);
       earlyDeleteHelperToBranchIndicies_.resize(upperLimitOnIndicies, 0);
       earlyDeleteBranchToCount_.reserve(nUniqueBranchesToDelete);
       std::map<const Worker*, EarlyDeleteHelper*> alreadySeenWorkers;
       std::string lastBranchName;
       size_t nextOpenIndex = 0;
       unsigned int* beginAddress = &(earlyDeleteHelperToBranchIndicies_.front());
       for (auto& branchAndWorker : branchToReadingWorker) {
         if (lastBranchName != branchAndWorker.first) {
           //have to put back the period we removed earlier in order to get the proper name
           BranchID bid(branchAndWorker.first + ".");
           earlyDeleteBranchToCount_.emplace_back(bid, 0U);
           lastBranchName = branchAndWorker.first;
         }
         auto found = alreadySeenWorkers.find(branchAndWorker.second);
         if (alreadySeenWorkers.end() == found) {
           //NOTE: we will set aside enough space in earlyDeleteHelperToBranchIndicies_ to accommodate
           // all the branches that might be read by this worker. However, initially we will only tell the
           // EarlyDeleteHelper about the first one. As additional branches are added via 'appendIndex' the
           // EarlyDeleteHelper will automatically advance its internal end pointer.
           size_t index = nextOpenIndex;
           size_t nIndices = reserveSizeForWorker[branchAndWorker.second];
           assert(index < earlyDeleteHelperToBranchIndicies_.size());
           earlyDeleteHelperToBranchIndicies_[index] = earlyDeleteBranchToCount_.size() - 1;
           earlyDeleteHelpers_.emplace_back(beginAddress + index, beginAddress + index + 1, &earlyDeleteBranchToCount_);
           branchAndWorker.second->setEarlyDeleteHelper(&(earlyDeleteHelpers_.back()));
           alreadySeenWorkers.insert(std::make_pair(branchAndWorker.second, &(earlyDeleteHelpers_.back())));
           nextOpenIndex += nIndices;
         } else {
           found->second->appendIndex(earlyDeleteBranchToCount_.size() - 1);
         }
       }
 
       //Now we can compactify the earlyDeleteHelperToBranchIndicies_ since we may have over estimated the
       // space needed for each module
       auto itLast = earlyDeleteHelpers_.begin();
       for (auto it = earlyDeleteHelpers_.begin() + 1; it != earlyDeleteHelpers_.end(); ++it) {
         if (itLast->end() != it->begin()) {
           //figure the offset for next Worker since it hasn't been moved yet so it has the original address
           unsigned int delta = it->begin() - itLast->end();
           it->shiftIndexPointers(delta);
 
           earlyDeleteHelperToBranchIndicies_.erase(
               earlyDeleteHelperToBranchIndicies_.begin() + (itLast->end() - beginAddress),
               earlyDeleteHelperToBranchIndicies_.begin() + (it->begin() - beginAddress));
         }
         itLast = it;
       }
       earlyDeleteHelperToBranchIndicies_.erase(
           earlyDeleteHelperToBranchIndicies_.begin() + (itLast->end() - beginAddress),
           earlyDeleteHelperToBranchIndicies_.end());
 
       //now tell the paths about the deleters
       for (auto& p : trig_paths_) {
         p.setEarlyDeleteHelpers(alreadySeenWorkers);
       }
       for (auto& p : end_paths_) {
         p.setEarlyDeleteHelpers(alreadySeenWorkers);
       }
       resetEarlyDelete();
     }
   }
 
   std::vector<Worker*> StreamSchedule::tryToPlaceConditionalModules(
       Worker* worker,
       std::unordered_set<std::string>& conditionalModules,
       std::unordered_multimap<std::string, edm::BranchDescription const*> const& conditionalModuleBranches,
       std::unordered_multimap<std::string, AliasInfo> const& aliasMap,
       ParameterSet& proc_pset,
       ProductRegistry& preg,
       PreallocationConfiguration const* prealloc,
       std::shared_ptr<ProcessConfiguration const> processConfiguration) {
     std::vector<Worker*> returnValue;
     auto const& consumesInfo = worker->consumesInfo();
     auto moduleLabel = worker->description()->moduleLabel();
     using namespace productholderindexhelper;
     for (auto const& ci : consumesInfo) {
       if (not ci.skipCurrentProcess() and
           (ci.process().empty() or ci.process() == processConfiguration->processName())) {
         auto productModuleLabel = std::string(ci.label());
         bool productFromConditionalModule = false;
         auto itFound = conditionalModules.find(productModuleLabel);
         if (itFound == conditionalModules.end()) {
           //Check to see if this was an alias
           //note that aliasMap was previously filtered so only the conditional modules remain there
           auto foundAlias = findBestMatchingAlias(conditionalModuleBranches, aliasMap, productModuleLabel, ci);
           if (foundAlias) {
             productModuleLabel = *foundAlias;
             productFromConditionalModule = true;
             itFound = conditionalModules.find(productModuleLabel);
             //check that the alias-for conditional module has not been used
             if (itFound == conditionalModules.end()) {
               continue;
             }
           }
         } else {
           //need to check the rest of the data product info
           auto findBranches = conditionalModuleBranches.equal_range(productModuleLabel);
           for (auto itBranch = findBranches.first; itBranch != findBranches.second; ++itBranch) {
             if (itBranch->second->productInstanceName() == ci.instance()) {
               if (ci.kindOfType() == PRODUCT_TYPE) {
                 if (ci.type() == itBranch->second->unwrappedTypeID()) {
                   productFromConditionalModule = true;
                   break;
                 }
               } else {
                 //this is a view
                 if (typeIsViewCompatible(
                         ci.type(), TypeID(itBranch->second->wrappedType().typeInfo()), itBranch->second->className())) {
                   productFromConditionalModule = true;
                   break;
                 }
               }
             }
           }
         }
         if (productFromConditionalModule) {
           auto condWorker = getWorker(
               productModuleLabel, proc_pset, workerManagerLumisAndEvents_, preg, prealloc, processConfiguration);
           assert(condWorker);
 
           conditionalModules.erase(itFound);
 
           auto dependents = tryToPlaceConditionalModules(condWorker,
                                                          conditionalModules,
                                                          conditionalModuleBranches,
                                                          aliasMap,
                                                          proc_pset,
                                                          preg,
                                                          prealloc,
                                                          processConfiguration);
           returnValue.insert(returnValue.end(), dependents.begin(), dependents.end());
           returnValue.push_back(condWorker);
         }
       }
     }
     return returnValue;
   }
 
   void StreamSchedule::fillWorkers(ParameterSet& proc_pset,
                                    ProductRegistry& preg,
                                    PreallocationConfiguration const* prealloc,
                                    std::shared_ptr<ProcessConfiguration const> processConfiguration,
                                    std::string const& pathName,
                                    bool ignoreFilters,
                                    PathWorkers& out,
                                    std::vector<std::string> const& endPathNames,
                                    ConditionalTaskHelper const& conditionalTaskHelper,
                                    std::unordered_set<std::string>& allConditionalModules) {
     vstring modnames = proc_pset.getParameter<vstring>(pathName);
     PathWorkers tmpworkers;
 
     //Pull out ConditionalTask modules
     auto condRange = findConditionalTaskModulesRange(modnames);
 
     std::unordered_set<std::string> conditionalmods;
     //An EDAlias may be redirecting to a module on a ConditionalTask
     std::unordered_multimap<std::string, edm::BranchDescription const*> conditionalModsBranches;
     std::unordered_map<std::string, unsigned int> conditionalModOrder;
     if (condRange.first != condRange.second) {
       for (auto it = condRange.first; it != condRange.second; ++it) {
         // ordering needs to skip the # token in the path list
         conditionalModOrder.emplace(*it, it - modnames.begin() - 1);
       }
       //the last entry should be ignored since it is required to be "@"
       conditionalmods = std::unordered_set<std::string>(std::make_move_iterator(condRange.first),
                                                         std::make_move_iterator(condRange.second));
 
       conditionalModsBranches = conditionalTaskHelper.conditionalModuleBranches(conditionalmods);
       modnames.erase(std::prev(condRange.first), modnames.end());
 
       // Make a union of all conditional modules from all Paths
       allConditionalModules.insert(conditionalmods.begin(), conditionalmods.end());
     }
 
     unsigned int placeInPath = 0;
     for (auto const& name : modnames) {
       //Modules except EDFilters are set to run concurrently by default
       bool doNotRunConcurrently = false;
       WorkerInPath::FilterAction filterAction = WorkerInPath::Normal;
       if (name[0] == '!') {
         filterAction = WorkerInPath::Veto;
       } else if (name[0] == '-' or name[0] == '+') {
         filterAction = WorkerInPath::Ignore;
       }
       if (name[0] == '|' or name[0] == '+') {
         //cms.wait was specified so do not run concurrently
         doNotRunConcurrently = true;
       }
 
       std::string moduleLabel = name;
       if (filterAction != WorkerInPath::Normal or name[0] == '|') {
         moduleLabel.erase(0, 1);
       }
 
       Worker* worker =
           getWorker(moduleLabel, proc_pset, workerManagerLumisAndEvents_, preg, prealloc, processConfiguration);
       if (worker == nullptr) {
         std::string pathType("endpath");
         if (!search_all(endPathNames, pathName)) {
           pathType = std::string("path");
         }
         throw Exception(errors::Configuration)
             << "The unknown module label \"" << moduleLabel << "\" appears in " << pathType << " \"" << pathName
             << "\"\n please check spelling or remove that label from the path.";
       }
 
       if (ignoreFilters && filterAction != WorkerInPath::Ignore && worker->moduleType() == Worker::kFilter) {
         // We have a filter on an end path, and the filter is not explicitly ignored.
         // See if the filter is allowed.
         std::vector<std::string> allowed_filters = proc_pset.getUntrackedParameter<vstring>("@filters_on_endpaths");
         if (!search_all(allowed_filters, worker->description()->moduleName())) {
           // Filter is not allowed. Ignore the result, and issue a warning.
           filterAction = WorkerInPath::Ignore;
           LogWarning("FilterOnEndPath") << "The EDFilter '" << worker->description()->moduleName()
                                         << "' with module label '" << moduleLabel << "' appears on EndPath '"
                                         << pathName << "'.\n"
                                         << "The return value of the filter will be ignored.\n"
                                         << "To suppress this warning, either remove the filter from the endpath,\n"
                                         << "or explicitly ignore it in the configuration by using cms.ignore().\n";
         }
       }
       bool runConcurrently = not doNotRunConcurrently;
       if (runConcurrently && worker->moduleType() == Worker::kFilter and filterAction != WorkerInPath::Ignore) {
         runConcurrently = false;
       }
 
       auto condModules = tryToPlaceConditionalModules(worker,
                                                       conditionalmods,
                                                       conditionalModsBranches,
                                                       conditionalTaskHelper.aliasMap(),
                                                       proc_pset,
                                                       preg,
                                                       prealloc,
                                                       processConfiguration);
       for (auto condMod : condModules) {
         tmpworkers.emplace_back(
             condMod, WorkerInPath::Ignore, conditionalModOrder[condMod->description()->moduleLabel()], true);
       }
 
       tmpworkers.emplace_back(worker, filterAction, placeInPath, runConcurrently);
       ++placeInPath;
     }
 
     out.swap(tmpworkers);
   }
 
   void StreamSchedule::fillTrigPath(ParameterSet& proc_pset,
                                     ProductRegistry& preg,
                                     PreallocationConfiguration const* prealloc,
                                     std::shared_ptr<ProcessConfiguration const> processConfiguration,
                                     int bitpos,
                                     std::string const& name,
                                     TrigResPtr trptr,
                                     std::vector<std::string> const& endPathNames,
                                     ConditionalTaskHelper const& conditionalTaskHelper,
                                     std::unordered_set<std::string>& allConditionalModules) {
     PathWorkers tmpworkers;
     fillWorkers(proc_pset,
                 preg,
                 prealloc,
                 processConfiguration,
                 name,
                 false,
                 tmpworkers,
                 endPathNames,
                 conditionalTaskHelper,
                 allConditionalModules);
 
     // an empty path will cause an extra bit that is not used
     if (!tmpworkers.empty()) {
       trig_paths_.emplace_back(
           bitpos, name, tmpworkers, trptr, actionTable(), actReg_, &streamContext_, PathContext::PathType::kPath);
     } else {
       empty_trig_paths_.push_back(bitpos);
     }
     for (WorkerInPath const& workerInPath : tmpworkers) {
       addToAllWorkers(workerInPath.getWorker());
     }
   }
 
   void StreamSchedule::fillEndPath(ParameterSet& proc_pset,
                                    ProductRegistry& preg,
                                    PreallocationConfiguration const* prealloc,
                                    std::shared_ptr<ProcessConfiguration const> processConfiguration,
                                    int bitpos,
                                    std::string const& name,
                                    std::vector<std::string> const& endPathNames,
                                    ConditionalTaskHelper const& conditionalTaskHelper,
                                    std::unordered_set<std::string>& allConditionalModules) {
     PathWorkers tmpworkers;
     fillWorkers(proc_pset,
                 preg,
                 prealloc,
                 processConfiguration,
                 name,
                 true,
                 tmpworkers,
                 endPathNames,
                 conditionalTaskHelper,
                 allConditionalModules);
 
     if (!tmpworkers.empty()) {
       end_paths_.emplace_back(bitpos,
                               name,
                               tmpworkers,
                               TrigResPtr(),
                               actionTable(),
                               actReg_,
                               &streamContext_,
                               PathContext::PathType::kEndPath);
     } else {
       empty_end_paths_.push_back(bitpos);
     }
     for (WorkerInPath const& workerInPath : tmpworkers) {
       addToAllWorkers(workerInPath.getWorker());
     }
   }
 
   void StreamSchedule::beginStream() {
     streamContext_.setTransition(StreamContext::Transition::kBeginStream);
     streamContext_.setEventID(EventID(0, 0, 0));
     streamContext_.setRunIndex(RunIndex::invalidRunIndex());
     streamContext_.setLuminosityBlockIndex(LuminosityBlockIndex::invalidLuminosityBlockIndex());
     streamContext_.setTimestamp(Timestamp());
 
     std::exception_ptr exceptionInStream;
     CMS_SA_ALLOW try {
       preScheduleSignal<BeginStreamTraits>(&streamContext_);
       workerManagerBeginEnd_.beginStream(streamID_, streamContext_);
     } catch (...) {
       exceptionInStream = std::current_exception();
     }
 
     postScheduleSignal<BeginStreamTraits>(&streamContext_, exceptionInStream);
 
     if (exceptionInStream) {
       bool cleaningUpAfterException = false;
       handleException(streamContext_, cleaningUpAfterException, exceptionInStream);
     }
     streamContext_.setTransition(StreamContext::Transition::kInvalid);
 
     if (exceptionInStream) {
       std::rethrow_exception(exceptionInStream);
     }
   }
 
   void StreamSchedule::endStream(ExceptionCollector& collector, std::mutex& collectorMutex) noexcept {
     streamContext_.setTransition(StreamContext::Transition::kEndStream);
     streamContext_.setEventID(EventID(0, 0, 0));
     streamContext_.setRunIndex(RunIndex::invalidRunIndex());
     streamContext_.setLuminosityBlockIndex(LuminosityBlockIndex::invalidLuminosityBlockIndex());
     streamContext_.setTimestamp(Timestamp());
 
     std::exception_ptr exceptionInStream;
     CMS_SA_ALLOW try {
       preScheduleSignal<EndStreamTraits>(&streamContext_);
       workerManagerBeginEnd_.endStream(streamID_, streamContext_, collector, collectorMutex);
     } catch (...) {
       exceptionInStream = std::current_exception();
     }
 
     postScheduleSignal<EndStreamTraits>(&streamContext_, exceptionInStream);
 
     if (exceptionInStream) {
       std::lock_guard<std::mutex> collectorLock(collectorMutex);
       collector.call([&exceptionInStream]() { std::rethrow_exception(exceptionInStream); });
     }
     streamContext_.setTransition(StreamContext::Transition::kInvalid);
   }
 
   void StreamSchedule::replaceModule(maker::ModuleHolder* iMod, std::string const& iLabel) {
     for (auto const& worker : allWorkersBeginEnd()) {
       if (worker->description()->moduleLabel() == iLabel) {
         iMod->replaceModuleFor(worker);
 
         streamContext_.setTransition(StreamContext::Transition::kBeginStream);
         streamContext_.setEventID(EventID(0, 0, 0));
         streamContext_.setRunIndex(RunIndex::invalidRunIndex());
         streamContext_.setLuminosityBlockIndex(LuminosityBlockIndex::invalidLuminosityBlockIndex());
         streamContext_.setTimestamp(Timestamp());
         try {
           worker->beginStream(streamID_, streamContext_);
         } catch (cms::Exception& ex) {
           streamContext_.setTransition(StreamContext::Transition::kInvalid);
           ex.addContext("Executing StreamSchedule::replaceModule");
           throw;
         }
         streamContext_.setTransition(StreamContext::Transition::kInvalid);
         break;
       }
     }
 
     for (auto const& worker : allWorkersRuns()) {
       if (worker->description()->moduleLabel() == iLabel) {
         iMod->replaceModuleFor(worker);
         break;
       }
     }
 
     for (auto const& worker : allWorkersLumisAndEvents()) {
       if (worker->description()->moduleLabel() == iLabel) {
         iMod->replaceModuleFor(worker);
         break;
       }
     }
   }
 
   void StreamSchedule::deleteModule(std::string const& iLabel) {
     workerManagerBeginEnd_.deleteModuleIfExists(iLabel);
     workerManagerRuns_.deleteModuleIfExists(iLabel);
     workerManagerLumisAndEvents_.deleteModuleIfExists(iLabel);
   }
 
   std::vector<ModuleDescription const*> StreamSchedule::getAllModuleDescriptions() const {
     std::vector<ModuleDescription const*> result;
     result.reserve(allWorkersLumisAndEvents().size());
 
     for (auto const& worker : allWorkersLumisAndEvents()) {
       ModuleDescription const* p = worker->description();
       result.push_back(p);
     }
     return result;
   }
 
   void StreamSchedule::processOneEventAsync(
       WaitingTaskHolder iTask,
       EventTransitionInfo& info,
       ServiceToken const& serviceToken,
       std::vector<edm::propagate_const<std::shared_ptr<PathStatusInserter>>>& pathStatusInserters) {
     EventPrincipal& ep = info.principal();
 
     // Caught exception is propagated via WaitingTaskHolder
     CMS_SA_ALLOW try {
       this->resetAll();
 
       using Traits = OccurrenceTraits<EventPrincipal, BranchActionStreamBegin>;
 
       Traits::setStreamContext(streamContext_, ep);
       //a service may want to communicate with another service
       ServiceRegistry::Operate guard(serviceToken);
       Traits::preScheduleSignal(actReg_.get(), &streamContext_);
 
       // Data dependencies need to be set up before marking empty
       // (End)Paths complete in case something consumes the status of
       // the empty (EndPath)
       workerManagerLumisAndEvents_.setupResolvers(ep);
       workerManagerLumisAndEvents_.setupOnDemandSystem(info);
 
       HLTPathStatus hltPathStatus(hlt::Pass, 0);
       for (int empty_trig_path : empty_trig_paths_) {
         results_->at(empty_trig_path) = hltPathStatus;
         pathStatusInserters[empty_trig_path]->setPathStatus(streamID_, hltPathStatus);
         std::exception_ptr except = pathStatusInserterWorkers_[empty_trig_path]
                                         ->runModuleDirectly<OccurrenceTraits<EventPrincipal, BranchActionStreamBegin>>(
                                             info, streamID_, ParentContext(&streamContext_), &streamContext_);
         if (except) {
           iTask.doneWaiting(except);
           return;
         }
       }
       if (not endPathStatusInserterWorkers_.empty()) {
         for (int empty_end_path : empty_end_paths_) {
           std::exception_ptr except =
               endPathStatusInserterWorkers_[empty_end_path]
                   ->runModuleDirectly<OccurrenceTraits<EventPrincipal, BranchActionStreamBegin>>(
                       info, streamID_, ParentContext(&streamContext_), &streamContext_);
           if (except) {
             iTask.doneWaiting(except);
             return;
           }
         }
       }
 
       ++total_events_;
 
       //use to give priorities on an error to ones from Paths
       auto pathErrorHolder = std::make_unique<std::atomic<std::exception_ptr*>>(nullptr);
       auto pathErrorPtr = pathErrorHolder.get();
       ServiceWeakToken weakToken = serviceToken;
       auto allPathsDone = make_waiting_task(
           [iTask, this, weakToken, pathError = std::move(pathErrorHolder)](std::exception_ptr const* iPtr) mutable {
             ServiceRegistry::Operate operate(weakToken.lock());
 
             std::exception_ptr ptr;
             if (pathError->load()) {
               ptr = *pathError->load();
               delete pathError->load();
             }
             if ((not ptr) and iPtr) {
               ptr = *iPtr;
             }
             iTask.doneWaiting(finishProcessOneEvent(ptr));
           });
       //The holder guarantees that if the paths finish before the loop ends
       // that we do not start too soon. It also guarantees that the task will
       // run under that condition.
       WaitingTaskHolder allPathsHolder(*iTask.group(), allPathsDone);
 
       auto pathsDone = make_waiting_task([allPathsHolder, pathErrorPtr, transitionInfo = info, this, weakToken](
                                              std::exception_ptr const* iPtr) mutable {
         ServiceRegistry::Operate operate(weakToken.lock());
 
         if (iPtr) {
           // free previous value of pathErrorPtr, if any;
           // prioritize this error over one that happens in EndPath or Accumulate
           auto currentPtr = pathErrorPtr->exchange(new std::exception_ptr(*iPtr));
           assert(currentPtr == nullptr);
         }
         finishedPaths(*pathErrorPtr, std::move(allPathsHolder), transitionInfo);
       });
 
       //The holder guarantees that if the paths finish before the loop ends
       // that we do not start too soon. It also guarantees that the task will
       // run under that condition.
       WaitingTaskHolder taskHolder(*iTask.group(), pathsDone);
 
       //start end paths first so on single threaded the paths will run first
       WaitingTaskHolder hAllPathsDone(*iTask.group(), allPathsDone);
       for (auto it = end_paths_.rbegin(), itEnd = end_paths_.rend(); it != itEnd; ++it) {
         it->processEventUsingPathAsync(hAllPathsDone, info, serviceToken, streamID_, &streamContext_);
       }
 
       for (auto it = trig_paths_.rbegin(), itEnd = trig_paths_.rend(); it != itEnd; ++it) {
         it->processEventUsingPathAsync(taskHolder, info, serviceToken, streamID_, &streamContext_);
       }
 
       ParentContext parentContext(&streamContext_);
       workerManagerLumisAndEvents_.processAccumulatorsAsync<OccurrenceTraits<EventPrincipal, BranchActionStreamBegin>>(
           hAllPathsDone, info, serviceToken, streamID_, parentContext, &streamContext_);
     } catch (...) {
       iTask.doneWaiting(std::current_exception());
     }
   }
 
   void StreamSchedule::finishedPaths(std::atomic<std::exception_ptr*>& iExcept,
                                      WaitingTaskHolder iWait,
                                      EventTransitionInfo& info) {
     if (iExcept) {
       // Caught exception is propagated via WaitingTaskHolder
       CMS_SA_ALLOW try { std::rethrow_exception(*(iExcept.load())); } catch (cms::Exception& e) {
         exception_actions::ActionCodes action = actionTable().find(e.category());
         assert(action != exception_actions::IgnoreCompletely);
         if (action == exception_actions::TryToContinue) {
           edm::printCmsExceptionWarning("TryToContinue", e);
           *(iExcept.load()) = std::exception_ptr();
         } else {
           *(iExcept.load()) = std::current_exception();
         }
       } catch (...) {
         *(iExcept.load()) = std::current_exception();
       }
     }
 
     if ((not iExcept) and results_->accept()) {
       ++total_passed_;
     }
 
     if (nullptr != results_inserter_.get()) {
       // Caught exception is propagated to the caller
       CMS_SA_ALLOW try {
         //Even if there was an exception, we need to allow results inserter
         // to run since some module may be waiting on its results.
         ParentContext parentContext(&streamContext_);
         using Traits = OccurrenceTraits<EventPrincipal, BranchActionStreamBegin>;
 
         auto expt = results_inserter_->runModuleDirectly<Traits>(info, streamID_, parentContext, &streamContext_);
         if (expt) {
           std::rethrow_exception(expt);
         }
       } catch (cms::Exception& ex) {
         if (not iExcept) {
           if (ex.context().empty()) {
             std::ostringstream ost;
             ost << "Processing Event " << info.principal().id();
             ex.addContext(ost.str());
           }
           iExcept.store(new std::exception_ptr(std::current_exception()));
         }
       } catch (...) {
         if (not iExcept) {
           iExcept.store(new std::exception_ptr(std::current_exception()));
         }
       }
     }
     std::exception_ptr ptr;
     if (iExcept) {
       ptr = *iExcept.load();
     }
     iWait.doneWaiting(ptr);
   }
 
   std::exception_ptr StreamSchedule::finishProcessOneEvent(std::exception_ptr iExcept) {
     using Traits = OccurrenceTraits<EventPrincipal, BranchActionStreamBegin>;
 
     if (iExcept) {
       //add context information to the exception and print message
       try {
         convertException::wrap([&]() { std::rethrow_exception(iExcept); });
       } catch (cms::Exception& ex) {
         bool const cleaningUpAfterException = false;
         if (ex.context().empty()) {
           addContextAndPrintException("Calling function StreamSchedule::processOneEvent", ex, cleaningUpAfterException);
         } else {
           addContextAndPrintException("", ex, cleaningUpAfterException);
         }
         iExcept = std::current_exception();
       }
 
       actReg_->preStreamEarlyTerminationSignal_(streamContext_, TerminationOrigin::ExceptionFromThisContext);
     }
     // Caught exception is propagated to the caller
     CMS_SA_ALLOW try { Traits::postScheduleSignal(actReg_.get(), &streamContext_); } catch (...) {
       if (not iExcept) {
         iExcept = std::current_exception();
       }
     }
     if (not iExcept) {
       resetEarlyDelete();
     }
 
     return iExcept;
   }
 
   void StreamSchedule::availablePaths(std::vector<std::string>& oLabelsToFill) const {
     oLabelsToFill.reserve(trig_paths_.size());
     std::transform(trig_paths_.begin(),
                    trig_paths_.end(),
                    std::back_inserter(oLabelsToFill),
                    std::bind(&Path::name, std::placeholders::_1));
   }
 
   void StreamSchedule::modulesInPath(std::string const& iPathLabel, std::vector<std::string>& oLabelsToFill) const {
     TrigPaths::const_iterator itFound = std::find_if(
         trig_paths_.begin(),
         trig_paths_.end(),
         std::bind(std::equal_to<std::string>(), iPathLabel, std::bind(&Path::name, std::placeholders::_1)));
     if (itFound != trig_paths_.end()) {
       oLabelsToFill.reserve(itFound->size());
       for (size_t i = 0; i < itFound->size(); ++i) {
         oLabelsToFill.push_back(itFound->getWorker(i)->description()->moduleLabel());
       }
     }
   }
 
   void StreamSchedule::moduleDescriptionsInPath(std::string const& iPathLabel,
                                                 std::vector<ModuleDescription const*>& descriptions,
                                                 unsigned int hint) const {
     descriptions.clear();
     bool found = false;
     TrigPaths::const_iterator itFound;
 
     if (hint < trig_paths_.size()) {
       itFound = trig_paths_.begin() + hint;
       if (itFound->name() == iPathLabel)
         found = true;
     }
     if (!found) {
       // if the hint did not work, do it the slow way
       itFound = std::find_if(
           trig_paths_.begin(),
           trig_paths_.end(),
           std::bind(std::equal_to<std::string>(), iPathLabel, std::bind(&Path::name, std::placeholders::_1)));
       if (itFound != trig_paths_.end())
         found = true;
     }
     if (found) {
       descriptions.reserve(itFound->size());
       for (size_t i = 0; i < itFound->size(); ++i) {
         descriptions.push_back(itFound->getWorker(i)->description());
       }
     }
   }
 
   void StreamSchedule::moduleDescriptionsInEndPath(std::string const& iEndPathLabel,
                                                    std::vector<ModuleDescription const*>& descriptions,
                                                    unsigned int hint) const {
     descriptions.clear();
     bool found = false;
     TrigPaths::const_iterator itFound;
 
     if (hint < end_paths_.size()) {
       itFound = end_paths_.begin() + hint;
       if (itFound->name() == iEndPathLabel)
         found = true;
     }
     if (!found) {
       // if the hint did not work, do it the slow way
       itFound = std::find_if(
           end_paths_.begin(),
           end_paths_.end(),
           std::bind(std::equal_to<std::string>(), iEndPathLabel, std::bind(&Path::name, std::placeholders::_1)));
       if (itFound != end_paths_.end())
         found = true;
     }
     if (found) {
       descriptions.reserve(itFound->size());
       for (size_t i = 0; i < itFound->size(); ++i) {
         descriptions.push_back(itFound->getWorker(i)->description());
       }
     }
   }
 
   static void fillModuleInPathSummary(Path const& path, size_t which, ModuleInPathSummary& sum) {
     sum.timesVisited += path.timesVisited(which);
     sum.timesPassed += path.timesPassed(which);
     sum.timesFailed += path.timesFailed(which);
     sum.timesExcept += path.timesExcept(which);
     sum.moduleLabel = path.getWorker(which)->description()->moduleLabel();
     sum.bitPosition = path.bitPosition(which);
   }
 
   static void fillPathSummary(Path const& path, PathSummary& sum) {
     sum.name = path.name();
     sum.bitPosition = path.bitPosition();
     sum.timesRun += path.timesRun();
     sum.timesPassed += path.timesPassed();
     sum.timesFailed += path.timesFailed();
     sum.timesExcept += path.timesExcept();
 
     Path::size_type sz = path.size();
     if (sum.moduleInPathSummaries.empty()) {
       std::vector<ModuleInPathSummary> temp(sz);
       for (size_t i = 0; i != sz; ++i) {
         fillModuleInPathSummary(path, i, temp[i]);
       }
       sum.moduleInPathSummaries.swap(temp);
     } else {
       assert(sz == sum.moduleInPathSummaries.size());
       for (size_t i = 0; i != sz; ++i) {
         fillModuleInPathSummary(path, i, sum.moduleInPathSummaries[i]);
       }
     }
   }
 
   static void fillWorkerSummaryAux(Worker const& w, WorkerSummary& sum) {
     sum.timesVisited += w.timesVisited();
     sum.timesRun += w.timesRun();
     sum.timesPassed += w.timesPassed();
     sum.timesFailed += w.timesFailed();
     sum.timesExcept += w.timesExcept();
     sum.moduleLabel = w.description()->moduleLabel();
   }
 
   static void fillWorkerSummary(Worker const* pw, WorkerSummary& sum) { fillWorkerSummaryAux(*pw, sum); }
 
   void StreamSchedule::getTriggerReport(TriggerReport& rep) const {
     rep.eventSummary.totalEvents += totalEvents();
     rep.eventSummary.totalEventsPassed += totalEventsPassed();
     rep.eventSummary.totalEventsFailed += totalEventsFailed();
 
     fill_summary(trig_paths_, rep.trigPathSummaries, &fillPathSummary);
     fill_summary(end_paths_, rep.endPathSummaries, &fillPathSummary);
     fill_summary(allWorkersLumisAndEvents(), rep.workerSummaries, &fillWorkerSummary);
   }
 
   void StreamSchedule::clearCounters() {
     using std::placeholders::_1;
     total_events_ = total_passed_ = 0;
     for_all(trig_paths_, std::bind(&Path::clearCounters, _1));
     for_all(end_paths_, std::bind(&Path::clearCounters, _1));
     for_all(allWorkersLumisAndEvents(), std::bind(&Worker::clearCounters, _1));
   }
 
   void StreamSchedule::resetAll() { results_->reset(); }
 
   void StreamSchedule::addToAllWorkers(Worker* w) { workerManagerLumisAndEvents_.addToAllWorkers(w); }
 
   void StreamSchedule::resetEarlyDelete() {
     //must be sure we have cleared the count first
     for (auto& count : earlyDeleteBranchToCount_) {
       count.count = 0;
     }
     //now reset based on how many helpers use that branch
     for (auto& index : earlyDeleteHelperToBranchIndicies_) {
       ++(earlyDeleteBranchToCount_[index].count);
     }
     for (auto& helper : earlyDeleteHelpers_) {
       helper.reset();
     }
   }
 
   void StreamSchedule::makePathStatusInserters(
       std::vector<edm::propagate_const<std::shared_ptr<PathStatusInserter>>>& pathStatusInserters,
       std::vector<edm::propagate_const<std::shared_ptr<EndPathStatusInserter>>>& endPathStatusInserters,
       ExceptionToActionTable const& actions) {
     int bitpos = 0;
     unsigned int indexEmpty = 0;
     unsigned int indexOfPath = 0;
     for (auto& pathStatusInserter : pathStatusInserters) {
       std::shared_ptr<PathStatusInserter> inserterPtr = get_underlying(pathStatusInserter);
       WorkerPtr workerPtr(
           new edm::WorkerT<PathStatusInserter::ModuleType>(inserterPtr, inserterPtr->moduleDescription(), &actions));
       pathStatusInserterWorkers_.emplace_back(workerPtr);
       workerPtr->setActivityRegistry(actReg_);
       addToAllWorkers(workerPtr.get());
 
       // A little complexity here because a C++ Path object is not
       // instantiated and put into end_paths if there are no modules
       // on the configured path.
       if (indexEmpty < empty_trig_paths_.size() && bitpos == empty_trig_paths_.at(indexEmpty)) {
         ++indexEmpty;
       } else {
         trig_paths_.at(indexOfPath).setPathStatusInserter(inserterPtr.get(), workerPtr.get());
         ++indexOfPath;
       }
       ++bitpos;
     }
 
     bitpos = 0;
     indexEmpty = 0;
     indexOfPath = 0;
     for (auto& endPathStatusInserter : endPathStatusInserters) {
       std::shared_ptr<EndPathStatusInserter> inserterPtr = get_underlying(endPathStatusInserter);
       WorkerPtr workerPtr(
           new edm::WorkerT<EndPathStatusInserter::ModuleType>(inserterPtr, inserterPtr->moduleDescription(), &actions));
       endPathStatusInserterWorkers_.emplace_back(workerPtr);
       workerPtr->setActivityRegistry(actReg_);
       addToAllWorkers(workerPtr.get());
 
       // A little complexity here because a C++ Path object is not
       // instantiated and put into end_paths if there are no modules
       // on the configured path.
       if (indexEmpty < empty_end_paths_.size() && bitpos == empty_end_paths_.at(indexEmpty)) {
         ++indexEmpty;
       } else {
         end_paths_.at(indexOfPath).setPathStatusInserter(nullptr, workerPtr.get());
         ++indexOfPath;
       }
       ++bitpos;
     }
   }
 
   void StreamSchedule::handleException(StreamContext const& streamContext,
                                        bool cleaningUpAfterException,
                                        std::exception_ptr& excpt) const noexcept {
     //add context information to the exception and print message
     try {
       convertException::wrap([&excpt]() { std::rethrow_exception(excpt); });
     } catch (cms::Exception& ex) {
       std::ostringstream ost;
       // In most cases the exception will already have context at this point,
       // but add some context here in those rare cases where it does not.
       if (ex.context().empty()) {
         exceptionContext(ost, streamContext);
       }
       addContextAndPrintException(ost.str().c_str(), ex, cleaningUpAfterException);
       excpt = std::current_exception();
     }
     // We are already handling an earlier exception, so ignore it
     // if this signal results in another exception being thrown.
     CMS_SA_ALLOW try {
       actReg_->preStreamEarlyTerminationSignal_(streamContext, TerminationOrigin::ExceptionFromThisContext);
     } catch (...) {
     }
   }
 }  // namespace edm

This page was automatically generated by the 2.2.1 LXR engine. The LXR team