#include <StreamSchedule.h>

Classes
struct	AliasInfo

Public Types
typedef std::vector< Worker * >	AllWorkers

typedef std::vector< Path >	NonTrigPaths

typedef std::vector< WorkerInPath >	PathWorkers

typedef std::vector< Path >	TrigPaths

typedef std::shared_ptr< HLTGlobalStatus const >	TrigResConstPtr

typedef std::shared_ptr< HLTGlobalStatus >	TrigResPtr

typedef std::vector< std::string >	vstring

typedef std::shared_ptr< Worker >	WorkerPtr

typedef std::vector< Worker * >	Workers

Public Member Functions
AllWorkers const &	allWorkersBeginEnd () const
	returns the collection of pointers to workers More...

AllWorkers const &	allWorkersLumisAndEvents () const

AllWorkers const &	allWorkersRuns () const

void	availablePaths (std::vector< std::string > &oLabelsToFill) const
	adds to oLabelsToFill the labels for all paths in the process More...

void	beginStream ()

void	clearCounters ()
	Clear all the counters in the trigger report. More...

StreamContext const &	context () const

void	deleteModule (std::string const &iLabel)
	Delete the module with label iLabel. More...

void	endStream ()

std::vector< ModuleDescription const * >	getAllModuleDescriptions () const

void	getTriggerReport (TriggerReport &rep) const

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

void	moduleDescriptionsInEndPath (std::string const &iEndPathLabel, std::vector< ModuleDescription const *> &descriptions, unsigned int hint) const

void	moduleDescriptionsInPath (std::string const &iPathLabel, std::vector< ModuleDescription const *> &descriptions, unsigned int hint) const

void	modulesInPath (std::string const &iPathLabel, std::vector< std::string > &oLabelsToFill) const
	adds to oLabelsToFill in execution order the labels of all modules in path iPathLabel More...

unsigned int	numberOfUnscheduledModules () const

void	processOneEventAsync (WaitingTaskHolder iTask, EventTransitionInfo &, ServiceToken const &token, std::vector< edm::propagate_const< std::shared_ptr< PathStatusInserter >>> &pathStatusInserters)

template<typename T >
void	processOneStreamAsync (WaitingTaskHolder iTask, typename T::TransitionInfoType &transitionInfo, ServiceToken const &token, bool cleaningUpAfterException=false)

void	replaceModule (maker::ModuleHolder *iMod, std::string const &iLabel)
	clone the type of module with label iLabel but configure with iPSet. More...

StreamID	streamID () const

	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 >, ParameterSet &proc_pset, service::TriggerNamesService const &tns, PreallocationConfiguration const &prealloc, ProductRegistry &pregistry, ExceptionToActionTable const &actions, std::shared_ptr< ActivityRegistry > areg, std::shared_ptr< ProcessConfiguration const > processConfiguration, StreamID streamID, ProcessContext const *processContext)

	StreamSchedule (StreamSchedule const &)=delete

int	totalEvents () const

int	totalEventsFailed () const

int	totalEventsPassed () const

AllWorkers const &	unscheduledWorkersLumisAndEvents () const

Private Member Functions
ExceptionToActionTable const &	actionTable () const
	returns the action table More...

void	addToAllWorkers (Worker *w)

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

void	fillTrigPath (ParameterSet &proc_pset, ProductRegistry &preg, PreallocationConfiguration const *prealloc, std::shared_ptr< ProcessConfiguration const > processConfiguration, int bitpos, std::string const &name, TrigResPtr, std::vector< std::string > const &endPathNames, ConditionalTaskHelper const &conditionalTaskHelper, std::unordered_set< std::string > &allConditionalModules)

void	fillWorkers (ParameterSet &proc_pset, ProductRegistry &preg, PreallocationConfiguration const *prealloc, std::shared_ptr< ProcessConfiguration const > processConfiguration, std::string const &name, bool ignoreFilters, PathWorkers &out, std::vector< std::string > const &endPathNames, ConditionalTaskHelper const &conditionalTaskHelper, std::unordered_set< std::string > &allConditionalModules)

void	finishedPaths (std::atomic< std::exception_ptr *> &, WaitingTaskHolder, EventTransitionInfo &)

std::exception_ptr	finishProcessOneEvent (std::exception_ptr)

void	handleException (StreamContext const &, ServiceWeakToken const &, bool cleaningUpAfterException, std::exception_ptr &) const noexcept

void	makePathStatusInserters (std::vector< edm::propagate_const< std::shared_ptr< PathStatusInserter >>> &pathStatusInserters, std::vector< edm::propagate_const< std::shared_ptr< EndPathStatusInserter >>> &endPathStatusInserters, ExceptionToActionTable const &actions)

template<typename T >
void	postScheduleSignal (StreamContext const *, ServiceWeakToken const &, std::exception_ptr &) const noexcept

template<typename T >
void	preScheduleSignal (StreamContext const *) const

void	reportSkipped (EventPrincipal const &ep) const

void	resetAll ()

void	resetEarlyDelete ()

TrigResConstPtr	results () const

TrigResPtr &	results ()

std::vector< Worker * >	tryToPlaceConditionalModules (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)

Private Attributes
std::shared_ptr< ActivityRegistry >	actReg_

std::vector< BranchToCount >	earlyDeleteBranchToCount_

std::vector< EarlyDeleteHelper >	earlyDeleteHelpers_

std::vector< unsigned int >	earlyDeleteHelperToBranchIndicies_

std::vector< int >	empty_end_paths_

std::vector< int >	empty_trig_paths_

TrigPaths	end_paths_

std::vector< edm::propagate_const< WorkerPtr > >	endPathStatusInserterWorkers_

unsigned int	number_of_unscheduled_modules_

std::vector< edm::propagate_const< WorkerPtr > >	pathStatusInserterWorkers_

edm::propagate_const< TrigResPtr >	results_

edm::propagate_const< WorkerPtr >	results_inserter_

StreamContext	streamContext_

StreamID	streamID_

int	total_events_

int	total_passed_

TrigPaths	trig_paths_

WorkerManager	workerManagerBeginEnd_

WorkerManager	workerManagerLumisAndEvents_

WorkerManager	workerManagerRuns_

Detailed Description

Definition at line 122 of file StreamSchedule.h.

Member Typedef Documentation

◆ AllWorkers

typedef std::vector<Worker*> edm::StreamSchedule::AllWorkers

Definition at line 130 of file StreamSchedule.h.

◆ NonTrigPaths

typedef std::vector<Path> edm::StreamSchedule::NonTrigPaths

Definition at line 126 of file StreamSchedule.h.

◆ PathWorkers

typedef std::vector<WorkerInPath> edm::StreamSchedule::PathWorkers

Definition at line 134 of file StreamSchedule.h.

◆ TrigPaths

typedef std::vector<Path> edm::StreamSchedule::TrigPaths

Definition at line 125 of file StreamSchedule.h.

◆ TrigResConstPtr

typedef std::shared_ptr<HLTGlobalStatus const> edm::StreamSchedule::TrigResConstPtr

Definition at line 128 of file StreamSchedule.h.

◆ TrigResPtr

typedef std::shared_ptr<HLTGlobalStatus> edm::StreamSchedule::TrigResPtr

Definition at line 127 of file StreamSchedule.h.

◆ vstring

typedef std::vector<std::string> edm::StreamSchedule::vstring

Definition at line 124 of file StreamSchedule.h.

◆ WorkerPtr

typedef std::shared_ptr<Worker> edm::StreamSchedule::WorkerPtr

Definition at line 129 of file StreamSchedule.h.

◆ Workers

typedef std::vector<Worker*> edm::StreamSchedule::Workers

Definition at line 132 of file StreamSchedule.h.

Constructor & Destructor Documentation

◆ StreamSchedule() [1/2]

edm::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 &	pregistry,
		ExceptionToActionTable const &	actions,
		std::shared_ptr< ActivityRegistry >	areg,
		std::shared_ptr< ProcessConfiguration const >	processConfiguration,
		StreamID	streamID,
		ProcessContext const *	processContext
	)

Definition at line 349 of file StreamSchedule.cc.

References actions, actReg_, edm::WorkerManager::addToAllWorkers(), addToAllWorkers(), edm::WorkerManager::addToUnscheduledWorkers(), allWorkersLumisAndEvents(), cms::cuda::assert(), end_paths_, fillEndPath(), fillTrigPath(), edm::propagate_const< T >::get(), edm::service::TriggerNamesService::getEndPaths(), edm::ParameterSet::getParameter(), edm::ParameterSet::getPSetForUpdate(), edm::service::TriggerNamesService::getTrigPaths(), label, dqm-mbProfile::log, makePathStatusInserters(), HerwigMaxPtPartonFilter_cfi::moduleLabel, number_of_unscheduled_modules_, results(), results_inserter_, DBoxMetadataHelper::set_difference(), streamID(), AlCaHLTBitMon_QueryRunRegistry::string, trig_paths_, edm::StreamID::value(), workerManagerBeginEnd_, workerManagerLumisAndEvents_, and workerManagerRuns_.

       : 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

◆ StreamSchedule() [2/2]

edm::StreamSchedule::StreamSchedule ( StreamSchedule const & )

delete

Member Function Documentation

◆ actionTable()

ExceptionToActionTable const& edm::StreamSchedule::actionTable ( ) const

inlineprivate

returns the action table

Definition at line 244 of file StreamSchedule.h.

References edm::WorkerManager::actionTable(), and workerManagerLumisAndEvents_.

Referenced by fillEndPath(), fillTrigPath(), and finishedPaths().

244 { return workerManagerLumisAndEvents_.actionTable(); }

edm::WorkerManager::actionTable

ExceptionToActionTable const & actionTable() const

Definition: WorkerManager.h:85

edm::StreamSchedule::workerManagerLumisAndEvents_

WorkerManager workerManagerLumisAndEvents_

Definition: StreamSchedule.h:318

◆ addToAllWorkers()

void edm::StreamSchedule::addToAllWorkers ( Worker * w )

private

Definition at line 1370 of file StreamSchedule.cc.

References edm::WorkerManager::addToAllWorkers(), w(), and workerManagerLumisAndEvents_.

Referenced by fillEndPath(), fillTrigPath(), makePathStatusInserters(), and StreamSchedule().

1370 { workerManagerLumisAndEvents_.addToAllWorkers(w); }

w

T w() const

Definition: extBasic3DVector.h:229

edm::StreamSchedule::workerManagerLumisAndEvents_

WorkerManager workerManagerLumisAndEvents_

Definition: StreamSchedule.h:318

edm::WorkerManager::addToAllWorkers

void addToAllWorkers(Worker *w)

Definition: WorkerManager.cc:141

◆ allWorkersBeginEnd()

AllWorkers const& edm::StreamSchedule::allWorkersBeginEnd ( ) const

inline

returns the collection of pointers to workers

Definition at line 224 of file StreamSchedule.h.

References edm::WorkerManager::allWorkers(), and workerManagerBeginEnd_.

Referenced by replaceModule().

224 { return workerManagerBeginEnd_.allWorkers(); }

edm::StreamSchedule::workerManagerBeginEnd_

WorkerManager workerManagerBeginEnd_

Definition: StreamSchedule.h:316

edm::WorkerManager::allWorkers

AllWorkers const & allWorkers() const

Definition: WorkerManager.h:80

◆ allWorkersLumisAndEvents()

AllWorkers const& edm::StreamSchedule::allWorkersLumisAndEvents ( ) const

inline

Definition at line 226 of file StreamSchedule.h.

References edm::WorkerManager::allWorkers(), and workerManagerLumisAndEvents_.

Referenced by clearCounters(), getAllModuleDescriptions(), getTriggerReport(), initializeEarlyDelete(), replaceModule(), and StreamSchedule().

226 { return workerManagerLumisAndEvents_.allWorkers(); }

edm::WorkerManager::allWorkers

AllWorkers const & allWorkers() const

Definition: WorkerManager.h:80

edm::StreamSchedule::workerManagerLumisAndEvents_

WorkerManager workerManagerLumisAndEvents_

Definition: StreamSchedule.h:318

◆ allWorkersRuns()

AllWorkers const& edm::StreamSchedule::allWorkersRuns ( ) const

inline

Definition at line 225 of file StreamSchedule.h.

References edm::WorkerManager::allWorkers(), and workerManagerRuns_.

Referenced by replaceModule().

225 { return workerManagerRuns_.allWorkers(); }

edm::StreamSchedule::workerManagerRuns_

WorkerManager workerManagerRuns_

Definition: StreamSchedule.h:317

edm::WorkerManager::allWorkers

AllWorkers const & allWorkers() const

Definition: WorkerManager.h:80

◆ availablePaths()

void edm::StreamSchedule::availablePaths ( std::vector< std::string > & oLabelsToFill ) const

adds to oLabelsToFill the labels for all paths in the process

Definition at line 1228 of file StreamSchedule.cc.

References edm::Path::name(), HcalDetIdTransform::transform(), and trig_paths_.

                                                                                {
     oLabelsToFill.reserve(trig_paths_.size());
     std::transform(trig_paths_.begin(),
                    trig_paths_.end(),
                    std::back_inserter(oLabelsToFill),
                    std::bind(&Path::name, std::placeholders::_1));
   }

◆ beginStream()

void edm::StreamSchedule::beginStream ( )

Definition at line 985 of file StreamSchedule.cc.

References edm::WorkerManager::beginStream(), streamContext_, streamID_, and workerManagerBeginEnd_.

985 { workerManagerBeginEnd_.beginStream(streamID_, streamContext_); }

edm::StreamSchedule::workerManagerBeginEnd_

WorkerManager workerManagerBeginEnd_

Definition: StreamSchedule.h:316

edm::StreamSchedule::streamContext_

StreamContext streamContext_

Definition: StreamSchedule.h:352

edm::WorkerManager::beginStream

void beginStream(StreamID iID, StreamContext &streamContext)

Definition: WorkerManager.cc:127

edm::StreamSchedule::streamID_

StreamID streamID_

Definition: StreamSchedule.h:351

◆ clearCounters()

void edm::StreamSchedule::clearCounters ( )

Clear all the counters in the trigger report.

Definition at line 1360 of file StreamSchedule.cc.

References allWorkersLumisAndEvents(), edm::Path::clearCounters(), edm::Worker::clearCounters(), end_paths_, edm::for_all(), total_events_, total_passed_, and trig_paths_.

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

◆ context()

StreamContext const& edm::StreamSchedule::context ( ) const

inline

Definition at line 233 of file StreamSchedule.h.

References streamContext_.

233 { return streamContext_; }

edm::StreamSchedule::streamContext_

StreamContext streamContext_

Definition: StreamSchedule.h:352

◆ deleteModule()

void edm::StreamSchedule::deleteModule ( std::string const & iLabel )

Delete the module with label iLabel.

Definition at line 1013 of file StreamSchedule.cc.

References edm::WorkerManager::deleteModuleIfExists(), workerManagerBeginEnd_, workerManagerLumisAndEvents_, and workerManagerRuns_.

                                                            {
     workerManagerBeginEnd_.deleteModuleIfExists(iLabel);
     workerManagerRuns_.deleteModuleIfExists(iLabel);
     workerManagerLumisAndEvents_.deleteModuleIfExists(iLabel);
   }

◆ endStream()

void edm::StreamSchedule::endStream ( )

Definition at line 987 of file StreamSchedule.cc.

References edm::WorkerManager::endStream(), streamContext_, streamID_, and workerManagerBeginEnd_.

987 { workerManagerBeginEnd_.endStream(streamID_, streamContext_); }

edm::WorkerManager::endStream

void endStream(StreamID iID, StreamContext &streamContext)

Definition: WorkerManager.cc:133

edm::StreamSchedule::workerManagerBeginEnd_

WorkerManager workerManagerBeginEnd_

Definition: StreamSchedule.h:316

edm::StreamSchedule::streamContext_

StreamContext streamContext_

Definition: StreamSchedule.h:352

edm::StreamSchedule::streamID_

StreamID streamID_

Definition: StreamSchedule.h:351

◆ fillEndPath()

void edm::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
	)

private

Definition at line 947 of file StreamSchedule.cc.

References actionTable(), actReg_, addToAllWorkers(), empty_end_paths_, end_paths_, fillWorkers(), edm::PathContext::kEndPath, Skims_PA_cff::name, and streamContext_.

Referenced by StreamSchedule().

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

◆ fillTrigPath()

void edm::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
	)

private

Definition at line 913 of file StreamSchedule.cc.

References actionTable(), actReg_, addToAllWorkers(), empty_trig_paths_, fillWorkers(), edm::PathContext::kPath, Skims_PA_cff::name, streamContext_, and trig_paths_.

Referenced by StreamSchedule().

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

◆ fillWorkers()

void edm::StreamSchedule::fillWorkers	(	ParameterSet &	proc_pset,
		ProductRegistry &	preg,
		PreallocationConfiguration const *	prealloc,
		std::shared_ptr< ProcessConfiguration const >	processConfiguration,
		std::string const &	name,
		bool	ignoreFilters,
		PathWorkers &	out,
		std::vector< std::string > const &	endPathNames,
		ConditionalTaskHelper const &	conditionalTaskHelper,
		std::unordered_set< std::string > &	allConditionalModules
	)

private

Definition at line 805 of file StreamSchedule.cc.

References edm::ConditionalTaskHelper::aliasMap(), edm::ConditionalTaskHelper::conditionalModuleBranches(), edm::errors::Configuration, edm::Worker::description(), Exception, edm::ParameterSet::getParameter(), edm::ParameterSet::getUntrackedParameter(), edm::WorkerInPath::Ignore, ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, edm::Worker::kFilter, HerwigMaxPtPartonFilter_cfi::moduleLabel, edm::ModuleDescription::moduleName(), edm::Worker::moduleType(), Skims_PA_cff::name, edm::WorkerInPath::Normal, or, MillePedeFileConverter_cfg::out, hltMonBTagIPClient_cfi::pathName, edm::search_all(), AlCaHLTBitMon_QueryRunRegistry::string, tryToPlaceConditionalModules(), edm::WorkerInPath::Veto, and workerManagerLumisAndEvents_.

Referenced by fillEndPath(), and fillTrigPath().

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

◆ finishedPaths()

void edm::StreamSchedule::finishedPaths	(	std::atomic< std::exception_ptr *> &	iExcept,
		WaitingTaskHolder	iWait,
		EventTransitionInfo &	info
	)

private

Definition at line 1139 of file StreamSchedule.cc.

References writedatasetfile::action, actionTable(), cms::Exception::addContext(), cms::cuda::assert(), CMS_SA_ALLOW, cms::Exception::context(), edm::WaitingTaskHolder::doneWaiting(), MillePedeFileConverter_cfg::e, edm::ExceptionToActionTable::find(), edm::propagate_const< T >::get(), edm::exception_actions::IgnoreCompletely, info(), edm::printCmsExceptionWarning(), results_, results_inserter_, streamContext_, streamID_, total_passed_, and edm::exception_actions::TryToContinue.

Referenced by processOneEventAsync().

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

◆ finishProcessOneEvent()

std::exception_ptr edm::StreamSchedule::finishProcessOneEvent ( std::exception_ptr iExcept )

private

Definition at line 1196 of file StreamSchedule.cc.

References actReg_, edm::addContextAndPrintException(), CMS_SA_ALLOW, cms::Exception::context(), edm::ExceptionFromThisContext, resetEarlyDelete(), streamContext_, and edm::convertException::wrap().

Referenced by processOneEventAsync().

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

◆ getAllModuleDescriptions()

std::vector< ModuleDescription const * > edm::StreamSchedule::getAllModuleDescriptions ( ) const

Return a vector allowing const access to all the ModuleDescriptions for this StreamSchedule. *** N.B. *** Ownership of the ModuleDescriptions is not *** passed to the caller. Do not call delete on these *** pointers!

Definition at line 1019 of file StreamSchedule.cc.

References allWorkersLumisAndEvents(), AlCaHLTBitMon_ParallelJobs::p, and mps_fire::result.

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

◆ getTriggerReport()

void edm::StreamSchedule::getTriggerReport ( TriggerReport & rep ) const

Return the trigger report information on paths, modules-in-path, modules-in-endpath, and modules.

Definition at line 1350 of file StreamSchedule.cc.

References allWorkersLumisAndEvents(), end_paths_, edm::fillPathSummary(), edm::fillWorkerSummary(), cuy::rep, totalEvents(), totalEventsFailed(), totalEventsPassed(), and trig_paths_.

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

◆ handleException()

void edm::StreamSchedule::handleException	(	StreamContext const &	streamContext,
		ServiceWeakToken const &	weakToken,
		bool	cleaningUpAfterException,
		std::exception_ptr &	excpt
	)		const

privatenoexcept

Definition at line 1437 of file StreamSchedule.cc.

References actReg_, edm::addContextAndPrintException(), CMS_SA_ALLOW, cms::Exception::context(), edm::exceptionContext(), edm::ExceptionFromThisContext, findAndChange::op, and edm::convertException::wrap().

Referenced by processOneStreamAsync().

                                                                                {
     //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);
       }
       ServiceRegistry::Operate op(weakToken.lock());
       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 {
       ServiceRegistry::Operate op(weakToken.lock());
       actReg_->preStreamEarlyTerminationSignal_(streamContext, TerminationOrigin::ExceptionFromThisContext);
     } catch (...) {
     }
   }

◆ initializeEarlyDelete()

void edm::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
	)

Definition at line 516 of file StreamSchedule.cc.

References allWorkersLumisAndEvents(), cms::cuda::assert(), MicroEventContent_cff::branch, edm::maker::ModuleHolder::createOutputModuleCommunicator(), dumpMFGeometry_cfg::delta, submitPVResolutionJobs::desc, earlyDeleteBranchToCount_, earlyDeleteHelpers_, earlyDeleteHelperToBranchIndicies_, end_paths_, edm::ModuleRegistry::forAllModuleHolders(), dqm-mbProfile::format, newFWLiteAna::found, edm::InEvent, ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, B2GTnPMonitor_cfi::item, ALPAKA_ACCELERATOR_NAMESPACE::pixelClustering::pixelStatus::kEmpty, MainPageGenerator::l, dqmiodumpmetadata::n, AlCaHLTBitMon_ParallelJobs::p, resetEarlyDelete(), AlCaHLTBitMon_QueryRunRegistry::string, groupFilesInBlocks::temp, trig_paths_, mitigatedMETSequence_cff::U, and w().

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

◆ makePathStatusInserters()

void edm::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
	)

private

Definition at line 1386 of file StreamSchedule.cc.

References actions, actReg_, addToAllWorkers(), empty_end_paths_, empty_trig_paths_, end_paths_, endPathStatusInserterWorkers_, edm::get_underlying(), pathStatusInserterWorkers_, and trig_paths_.

Referenced by StreamSchedule().

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

◆ moduleDescriptionsInEndPath()

void edm::StreamSchedule::moduleDescriptionsInEndPath	(	std::string const &	iEndPathLabel,
		std::vector< ModuleDescription const *> &	descriptions,
		unsigned int	hint
	)		const

Definition at line 1278 of file StreamSchedule.cc.

References end_paths_, newFWLiteAna::found, mps_fire::i, and edm::Path::name().

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

◆ moduleDescriptionsInPath()

void edm::StreamSchedule::moduleDescriptionsInPath	(	std::string const &	iPathLabel,
		std::vector< ModuleDescription const *> &	descriptions,
		unsigned int	hint
	)		const

Definition at line 1249 of file StreamSchedule.cc.

References newFWLiteAna::found, mps_fire::i, edm::Path::name(), and trig_paths_.

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

◆ modulesInPath()

void edm::StreamSchedule::modulesInPath	(	std::string const &	iPathLabel,
		std::vector< std::string > &	oLabelsToFill
	)		const

adds to oLabelsToFill in execution order the labels of all modules in path iPathLabel

Definition at line 1236 of file StreamSchedule.cc.

References mps_fire::i, edm::Path::name(), and trig_paths_.

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

◆ numberOfUnscheduledModules()

unsigned int edm::StreamSchedule::numberOfUnscheduledModules ( ) const

inline

Definition at line 231 of file StreamSchedule.h.

References number_of_unscheduled_modules_.

231 { return number_of_unscheduled_modules_; }

edm::StreamSchedule::number_of_unscheduled_modules_

unsigned int number_of_unscheduled_modules_

Definition: StreamSchedule.h:349

◆ postScheduleSignal()

template<typename T >

void edm::StreamSchedule::postScheduleSignal	(	StreamContext const *	streamContext,
		ServiceWeakToken const &	weakToken,
		std::exception_ptr &	excpt
	)		const

privatenoexcept

Definition at line 432 of file StreamSchedule.h.

References cms::Exception::addContext(), edm::exceptionContext(), findAndChange::op, and edm::convertException::wrap().

                                                                                   {
     try {
       convertException::wrap([this, &weakToken, streamContext]() {
         ServiceRegistry::Operate op(weakToken.lock());
         T::postScheduleSignal(actReg_.get(), streamContext);
       });
     } catch (cms::Exception& ex) {
       if (not excpt) {
         std::ostringstream ost;
         ex.addContext("Handling post signal, likely in a service function");
         exceptionContext(ost, *streamContext);
         ex.addContext(ost.str());
         excpt = std::current_exception();
       }
     }
   }

◆ preScheduleSignal()

template<typename T >

void edm::StreamSchedule::preScheduleSignal ( StreamContext const * streamContext ) const

private

Definition at line 419 of file StreamSchedule.h.

References actReg_, cms::Exception::addContext(), edm::exceptionContext(), and edm::convertException::wrap().

                                                                                  {
     try {
       convertException::wrap([this, streamContext]() { T::preScheduleSignal(actReg_.get(), streamContext); });
     } catch (cms::Exception& ex) {
       std::ostringstream ost;
       ex.addContext("Handling pre signal, likely in a service function");
       exceptionContext(ost, *streamContext);
       ex.addContext(ost.str());
       throw;
     }
   }

◆ processOneEventAsync()

void edm::StreamSchedule::processOneEventAsync	(	WaitingTaskHolder	iTask,
		EventTransitionInfo &	info,
		ServiceToken const &	token,
		std::vector< edm::propagate_const< std::shared_ptr< PathStatusInserter >>> &	pathStatusInserters
	)

Definition at line 1030 of file StreamSchedule.cc.

References actReg_, CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), empty_end_paths_, empty_trig_paths_, end_paths_, endPathStatusInserterWorkers_, SiStripBadComponentsDQMServiceTemplate_cfg::ep, finishedPaths(), finishProcessOneEvent(), edm::WaitingTaskHolder::group(), info(), ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, edm::ServiceWeakToken::lock(), edm::make_waiting_task(), eostools::move(), edm::hlt::Pass, pathStatusInserterWorkers_, edm::WorkerManager::processAccumulatorsAsync(), resetAll(), results_, edm::WorkerManager::setupOnDemandSystem(), edm::WorkerManager::setupResolvers(), streamContext_, streamID_, total_events_, trig_paths_, and workerManagerLumisAndEvents_.

                                                                                              {
     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) {
           //this is used to prioritize this error over one
           // that happens in EndPath or Accumulate
           pathErrorPtr->store(new std::exception_ptr(*iPtr));
         }
         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());
     }
   }

◆ processOneStreamAsync()

template<typename T >

void edm::StreamSchedule::processOneStreamAsync	(	WaitingTaskHolder	iTask,
		typename T::TransitionInfoType &	transitionInfo,
		ServiceToken const &	token,
		bool	cleaningUpAfterException = `false`
	)

Definition at line 361 of file StreamSchedule.h.

References CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), edm::WaitingTaskHolder::group(), watchdog::group, h, handleException(), info(), edm::InLumi, edm::ServiceWeakToken::lock(), edm::make_functor_task(), edm::make_waiting_task(), eostools::move(), findAndChange::op, edm::WorkerManager::processOneOccurrenceAsync(), edm::WorkerManager::resetAll(), alignCSCRings::s, streamContext_, streamID_, TrackValidation_cff::task, unpackBuffers-CaloStage2::token, edm::StreamID::value(), workerManagerLumisAndEvents_, and workerManagerRuns_.

                                                                             {
     auto group = iHolder.group();
     auto const& principal = transitionInfo.principal();
     T::setStreamContext(streamContext_, principal);
 
     ServiceWeakToken weakToken = token;
     auto doneTask = make_waiting_task([this, iHolder = std::move(iHolder), cleaningUpAfterException, weakToken](
                                           std::exception_ptr const* iPtr) mutable {
       std::exception_ptr excpt;
       if (iPtr) {
         excpt = *iPtr;
         handleException(streamContext_, weakToken, cleaningUpAfterException, excpt);
       }
       postScheduleSignal<T>(&streamContext_, weakToken, excpt);
       iHolder.doneWaiting(excpt);
     });
 
     auto task =
         make_functor_task([this, h = WaitingTaskHolder(*group, doneTask), info = transitionInfo, weakToken]() mutable {
           auto token = weakToken.lock();
           ServiceRegistry::Operate op(token);
           // Caught exception is propagated via WaitingTaskHolder
           WorkerManager* workerManager = &workerManagerRuns_;
           if (T::branchType_ == InLumi) {
             workerManager = &workerManagerLumisAndEvents_;
           }
           CMS_SA_ALLOW try {
             preScheduleSignal<T>(&streamContext_);
             workerManager->resetAll();
           } catch (...) {
             h.doneWaiting(std::current_exception());
             return;
           }
 
           workerManager->processOneOccurrenceAsync<T>(h, info, token, streamID_, &streamContext_, &streamContext_);
         });
 
     if (streamID_.value() == 0) {
       //Enqueueing will start another thread if there is only
       // one thread in the job. Having stream == 0 use spawn
       // avoids starting up another thread when there is only one stream.
       group->run([task]() {
         TaskSentry s{task};
         task->execute();
       });
     } else {
       oneapi::tbb::task_arena arena{oneapi::tbb::task_arena::attach()};
       arena.enqueue([task]() {
         TaskSentry s{task};
         task->execute();
       });
     }
   }

◆ replaceModule()

void edm::StreamSchedule::replaceModule	(	maker::ModuleHolder *	iMod,
		std::string const &	iLabel
	)

clone the type of module with label iLabel but configure with iPSet.

Definition at line 989 of file StreamSchedule.cc.

References allWorkersBeginEnd(), allWorkersLumisAndEvents(), allWorkersRuns(), edm::maker::ModuleHolder::replaceModuleFor(), streamContext_, and streamID_.

                                                                                      {
     for (auto const& worker : allWorkersBeginEnd()) {
       if (worker->description()->moduleLabel() == iLabel) {
         iMod->replaceModuleFor(worker);
         worker->beginStream(streamID_, streamContext_);
         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;
       }
     }
   }

◆ reportSkipped()

void edm::StreamSchedule::reportSkipped ( EventPrincipal const & ep ) const

inlineprivate

Definition at line 355 of file StreamSchedule.h.

References SiStripBadComponentsDQMServiceTemplate_cfg::ep.

                                                                           {
     Service<JobReport> reportSvc;
     reportSvc->reportSkippedEvent(ep.id().run(), ep.id().event());
   }

◆ resetAll()

void edm::StreamSchedule::resetAll ( )

private

Definition at line 1368 of file StreamSchedule.cc.

References results_.

Referenced by processOneEventAsync().

1368 { results_->reset(); }

edm::StreamSchedule::results_

edm::propagate_const< TrigResPtr > results_

Definition: StreamSchedule.h:321

◆ resetEarlyDelete()

void edm::StreamSchedule::resetEarlyDelete ( )

private

Definition at line 1372 of file StreamSchedule.cc.

References submitPVResolutionJobs::count, earlyDeleteBranchToCount_, earlyDeleteHelpers_, and earlyDeleteHelperToBranchIndicies_.

Referenced by finishProcessOneEvent(), and initializeEarlyDelete().

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

◆ results() [1/2]

TrigResConstPtr edm::StreamSchedule::results ( ) const

inlineprivate

Definition at line 297 of file StreamSchedule.h.

References edm::get_underlying_safe(), and results_.

Referenced by StreamSchedule().

297 { return get_underlying_safe(results_); }

edm::get_underlying_safe

constexpr std::shared_ptr< T > & get_underlying_safe(propagate_const< std::shared_ptr< T >> &iP)

Definition: get_underlying_safe.h:41

edm::StreamSchedule::results_

edm::propagate_const< TrigResPtr > results_

Definition: StreamSchedule.h:321

◆ results() [2/2]

TrigResPtr& edm::StreamSchedule::results ( )

inlineprivate

Definition at line 298 of file StreamSchedule.h.

References edm::get_underlying_safe(), and results_.

298 { return get_underlying_safe(results_); }

edm::get_underlying_safe

constexpr std::shared_ptr< T > & get_underlying_safe(propagate_const< std::shared_ptr< T >> &iP)

Definition: get_underlying_safe.h:41

edm::StreamSchedule::results_

edm::propagate_const< TrigResPtr > results_

Definition: StreamSchedule.h:321

◆ streamID()

StreamID edm::StreamSchedule::streamID ( ) const

inline

Definition at line 167 of file StreamSchedule.h.

References streamID_.

Referenced by StreamSchedule().

167 { return streamID_; }

edm::StreamSchedule::streamID_

StreamID streamID_

Definition: StreamSchedule.h:351

◆ totalEvents()

int edm::StreamSchedule::totalEvents ( ) const

inline

Return the number of events this StreamSchedule has tried to process (inclues both successes and failures, including failures due to exceptions during processing).

Definition at line 194 of file StreamSchedule.h.

References total_events_.

Referenced by getTriggerReport(), and totalEventsFailed().

194 { return total_events_; }

edm::StreamSchedule::total_events_

int total_events_

Definition: StreamSchedule.h:347

◆ totalEventsFailed()

int edm::StreamSchedule::totalEventsFailed ( ) const

inline

Return the number of events that have not passed any trigger. (N.B. totalEventsFailed() + totalEventsPassed() == totalEvents()

Definition at line 202 of file StreamSchedule.h.

References totalEvents(), and totalEventsPassed().

Referenced by getTriggerReport().

202 { return totalEvents() - totalEventsPassed(); }

edm::StreamSchedule::totalEventsPassed

int totalEventsPassed() const

Definition: StreamSchedule.h:198

edm::StreamSchedule::totalEvents

int totalEvents() const

Definition: StreamSchedule.h:194

◆ totalEventsPassed()

int edm::StreamSchedule::totalEventsPassed ( ) const

inline

Return the number of events which have been passed by one or more trigger paths.

Definition at line 198 of file StreamSchedule.h.

References total_passed_.

Referenced by getTriggerReport(), and totalEventsFailed().

198 { return total_passed_; }

edm::StreamSchedule::total_passed_

int total_passed_

Definition: StreamSchedule.h:348

◆ tryToPlaceConditionalModules()

std::vector< Worker * > edm::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
	)

private

Definition at line 729 of file StreamSchedule.cc.

References cms::cuda::assert(), edm::TypeID::className(), edm::Worker::consumesInfo(), edm::Worker::description(), HerwigMaxPtPartonFilter_cfi::moduleLabel, edm::ModuleDescription::moduleLabel(), or, edm::PRODUCT_TYPE, AlCaHLTBitMon_QueryRunRegistry::string, edm::productholderindexhelper::typeIsViewCompatible(), and workerManagerLumisAndEvents_.

Referenced by fillWorkers().

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

◆ unscheduledWorkersLumisAndEvents()

AllWorkers const& edm::StreamSchedule::unscheduledWorkersLumisAndEvents ( ) const

inline

Definition at line 228 of file StreamSchedule.h.

References edm::WorkerManager::unscheduledWorkers(), and workerManagerLumisAndEvents_.

                                                                {
       return workerManagerLumisAndEvents_.unscheduledWorkers();
     }

Member Data Documentation

◆ actReg_

std::shared_ptr<ActivityRegistry> edm::StreamSchedule::actReg_

private

Definition at line 319 of file StreamSchedule.h.

Referenced by fillEndPath(), fillTrigPath(), finishProcessOneEvent(), handleException(), makePathStatusInserters(), preScheduleSignal(), processOneEventAsync(), and StreamSchedule().

◆ earlyDeleteBranchToCount_

std::vector<BranchToCount> edm::StreamSchedule::earlyDeleteBranchToCount_

private

Definition at line 335 of file StreamSchedule.h.

Referenced by initializeEarlyDelete(), and resetEarlyDelete().

◆ earlyDeleteHelpers_

std::vector<EarlyDeleteHelper> edm::StreamSchedule::earlyDeleteHelpers_

private

Definition at line 345 of file StreamSchedule.h.

Referenced by initializeEarlyDelete(), and resetEarlyDelete().

◆ earlyDeleteHelperToBranchIndicies_

std::vector<unsigned int> edm::StreamSchedule::earlyDeleteHelperToBranchIndicies_

private

Definition at line 342 of file StreamSchedule.h.

Referenced by initializeEarlyDelete(), and resetEarlyDelete().

◆ empty_end_paths_

std::vector<int> edm::StreamSchedule::empty_end_paths_

private

Definition at line 330 of file StreamSchedule.h.

Referenced by fillEndPath(), makePathStatusInserters(), and processOneEventAsync().

◆ empty_trig_paths_

std::vector<int> edm::StreamSchedule::empty_trig_paths_

private

Definition at line 329 of file StreamSchedule.h.

Referenced by fillTrigPath(), makePathStatusInserters(), and processOneEventAsync().

◆ end_paths_

TrigPaths edm::StreamSchedule::end_paths_

private

Definition at line 328 of file StreamSchedule.h.

Referenced by clearCounters(), fillEndPath(), getTriggerReport(), initializeEarlyDelete(), makePathStatusInserters(), moduleDescriptionsInEndPath(), processOneEventAsync(), and StreamSchedule().

◆ endPathStatusInserterWorkers_

std::vector<edm::propagate_const<WorkerPtr> > edm::StreamSchedule::endPathStatusInserterWorkers_

private

Definition at line 325 of file StreamSchedule.h.

Referenced by makePathStatusInserters(), and processOneEventAsync().

◆ number_of_unscheduled_modules_

unsigned int edm::StreamSchedule::number_of_unscheduled_modules_

private

Definition at line 349 of file StreamSchedule.h.

Referenced by numberOfUnscheduledModules(), and StreamSchedule().

◆ pathStatusInserterWorkers_

std::vector<edm::propagate_const<WorkerPtr> > edm::StreamSchedule::pathStatusInserterWorkers_

private

Definition at line 324 of file StreamSchedule.h.

Referenced by makePathStatusInserters(), and processOneEventAsync().

◆ results_

edm::propagate_const<TrigResPtr> edm::StreamSchedule::results_

private

Definition at line 321 of file StreamSchedule.h.

Referenced by finishedPaths(), processOneEventAsync(), resetAll(), and results().

◆ results_inserter_

edm::propagate_const<WorkerPtr> edm::StreamSchedule::results_inserter_

private

Definition at line 323 of file StreamSchedule.h.

Referenced by finishedPaths(), and StreamSchedule().

◆ streamContext_

StreamContext edm::StreamSchedule::streamContext_

private

Definition at line 352 of file StreamSchedule.h.

Referenced by beginStream(), context(), endStream(), fillEndPath(), fillTrigPath(), finishedPaths(), finishProcessOneEvent(), processOneEventAsync(), processOneStreamAsync(), and replaceModule().

◆ streamID_

StreamID edm::StreamSchedule::streamID_

private

Definition at line 351 of file StreamSchedule.h.

Referenced by beginStream(), endStream(), finishedPaths(), processOneEventAsync(), processOneStreamAsync(), replaceModule(), and streamID().

◆ total_events_

int edm::StreamSchedule::total_events_

private

Definition at line 347 of file StreamSchedule.h.

Referenced by clearCounters(), processOneEventAsync(), and totalEvents().

◆ total_passed_

int edm::StreamSchedule::total_passed_

private

Definition at line 348 of file StreamSchedule.h.

Referenced by clearCounters(), finishedPaths(), and totalEventsPassed().

◆ trig_paths_

TrigPaths edm::StreamSchedule::trig_paths_

private

Definition at line 327 of file StreamSchedule.h.

Referenced by availablePaths(), clearCounters(), fillTrigPath(), getTriggerReport(), initializeEarlyDelete(), makePathStatusInserters(), moduleDescriptionsInPath(), modulesInPath(), processOneEventAsync(), and StreamSchedule().

◆ workerManagerBeginEnd_

WorkerManager edm::StreamSchedule::workerManagerBeginEnd_

private

Definition at line 316 of file StreamSchedule.h.

Referenced by allWorkersBeginEnd(), beginStream(), deleteModule(), endStream(), and StreamSchedule().

◆ workerManagerLumisAndEvents_

WorkerManager edm::StreamSchedule::workerManagerLumisAndEvents_

private

Definition at line 318 of file StreamSchedule.h.

Referenced by actionTable(), addToAllWorkers(), allWorkersLumisAndEvents(), deleteModule(), fillWorkers(), processOneEventAsync(), processOneStreamAsync(), StreamSchedule(), tryToPlaceConditionalModules(), and unscheduledWorkersLumisAndEvents().

◆ workerManagerRuns_

WorkerManager edm::StreamSchedule::workerManagerRuns_

private

Definition at line 317 of file StreamSchedule.h.

Referenced by allWorkersRuns(), deleteModule(), processOneStreamAsync(), and StreamSchedule().

Classes

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ AllWorkers

◆ NonTrigPaths

◆ PathWorkers

◆ TrigPaths

◆ TrigResConstPtr

◆ TrigResPtr

◆ vstring

◆ WorkerPtr

◆ Workers

Constructor & Destructor Documentation

◆ StreamSchedule() [1/2]

◆ StreamSchedule() [2/2]

Member Function Documentation

◆ actionTable()

◆ addToAllWorkers()

◆ allWorkersBeginEnd()

◆ allWorkersLumisAndEvents()

◆ allWorkersRuns()

◆ availablePaths()

◆ beginStream()

◆ clearCounters()

◆ context()

◆ deleteModule()

◆ endStream()

◆ fillEndPath()

◆ fillTrigPath()

◆ fillWorkers()

◆ finishedPaths()

◆ finishProcessOneEvent()

◆ getAllModuleDescriptions()

◆ getTriggerReport()

◆ handleException()

◆ initializeEarlyDelete()

◆ makePathStatusInserters()

◆ moduleDescriptionsInEndPath()

◆ moduleDescriptionsInPath()

◆ modulesInPath()

◆ numberOfUnscheduledModules()

◆ postScheduleSignal()

◆ preScheduleSignal()

◆ processOneEventAsync()

◆ processOneStreamAsync()

◆ replaceModule()

◆ reportSkipped()

◆ resetAll()

◆ resetEarlyDelete()

◆ results() [1/2]

◆ results() [2/2]

◆ streamID()

◆ totalEvents()

◆ totalEventsFailed()

◆ totalEventsPassed()

◆ tryToPlaceConditionalModules()

◆ unscheduledWorkersLumisAndEvents()

Member Data Documentation

◆ actReg_

◆ earlyDeleteBranchToCount_

◆ earlyDeleteHelpers_

◆ earlyDeleteHelperToBranchIndicies_

◆ empty_end_paths_

◆ empty_trig_paths_

◆ end_paths_

◆ endPathStatusInserterWorkers_

◆ number_of_unscheduled_modules_

◆ pathStatusInserterWorkers_

◆ results_

◆ results_inserter_

◆ streamContext_

◆ streamID_

◆ total_events_

◆ total_passed_

◆ trig_paths_

◆ workerManagerBeginEnd_