edm::StreamSchedule Class Reference

#include <StreamSchedule.h>

Classes
class	SendTerminationSignalIfException

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 &	allWorkers () const
	returns the collection of pointers to workers More...
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	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, BranchIDListHelper &branchIDListHelper, ExceptionToActionTable const &actions, std::shared_ptr< ActivityRegistry > areg, std::shared_ptr< ProcessConfiguration > processConfiguration, bool allowEarlyDelete, StreamID streamID, ProcessContext const *processContext)
	StreamSchedule (StreamSchedule const &)=delete
int	totalEvents () const
int	totalEventsFailed () const
int	totalEventsPassed () 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)
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)
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)
void	finishedPaths (std::atomic< std::exception_ptr * > &, WaitingTaskHolder, EventTransitionInfo &)
std::exception_ptr	finishProcessOneEvent (std::exception_ptr)
void	initializeEarlyDelete (ModuleRegistry &modReg, edm::ParameterSet const &opts, edm::ProductRegistry const &preg, bool allowEarlyDelete)
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)
void	reportSkipped (EventPrincipal const &ep) const
void	resetAll ()
void	resetEarlyDelete ()
TrigResConstPtr	results () const
TrigResPtr &	results ()

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_
std::atomic< bool >	skippingEvent_
StreamContext	streamContext_
StreamID	streamID_
int	total_events_
int	total_passed_
TrigPaths	trig_paths_
WorkerManager	workerManager_

Detailed Description

Definition at line 152 of file StreamSchedule.h.

Member Typedef Documentation

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

Definition at line 160 of file StreamSchedule.h.

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

Definition at line 156 of file StreamSchedule.h.

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

Definition at line 164 of file StreamSchedule.h.

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

Definition at line 155 of file StreamSchedule.h.

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

Definition at line 158 of file StreamSchedule.h.

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

Definition at line 157 of file StreamSchedule.h.

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

Definition at line 154 of file StreamSchedule.h.

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

Definition at line 159 of file StreamSchedule.h.

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

Definition at line 162 of file StreamSchedule.h.

Constructor & Destructor Documentation

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

Definition at line 138 of file StreamSchedule.cc.

References actReg_, addToAllWorkers(), edm::WorkerManager::addToUnscheduledWorkers(), allWorkers(), 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(), edm::ParameterSet::getUntrackedParameterSet(), initializeEarlyDelete(), label, makePathStatusInserters(), number_of_unscheduled_modules_, results(), results_inserter_, DBoxMetadataHelper::set_difference(), trig_paths_, edm::StreamID::value(), and workerManager_.

      : workerManager_(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),
        skippingEvent_(false) {
    ParameterSet const& opts = proc_pset.getUntrackedParameterSet("options", ParameterSet());
    bool hasPath = false;
    std::vector<std::string> const& pathNames = tns.getTrigPaths();
    std::vector<std::string> const& endPathNames = tns.getEndPaths();

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

    makePathStatusInserters(pathStatusInserters, endPathStatusInserters, actions);

    //See if all modules were used
    std::set<std::string> usedWorkerLabels;
    for (auto const& worker : allWorkers()) {
      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);
        workerManager_.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();

    initializeEarlyDelete(*modReg, opts, preg, allowEarlyDelete);

  }  // StreamSchedule::StreamSchedule

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

delete

Member Function Documentation

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

inlineprivate

returns the action table

Definition at line 278 of file StreamSchedule.h.

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

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

{ return workerManager_.actionTable(); }

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

private

Definition at line 899 of file StreamSchedule.cc.

References edm::WorkerManager::addToAllWorkers(), and workerManager_.

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

{ workerManager_.addToAllWorkers(w); }

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

inline

returns the collection of pointers to workers

Definition at line 250 of file StreamSchedule.h.

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

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

{ return workerManager_.allWorkers(); }

void edm::StreamSchedule::availablePaths

(

std::vector< std::string > &

oLabelsToFill

)

const

adds to oLabelsToFill the labels for all paths in the process

Definition at line 755 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));
  }

void edm::StreamSchedule::beginStream

(

)

Definition at line 529 of file StreamSchedule.cc.

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

{ workerManager_.beginStream(streamID_, streamContext_); }

void edm::StreamSchedule::clearCounters

(

)

Clear all the counters in the trigger report.

Definition at line 886 of file StreamSchedule.cc.

References allWorkers(), 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(allWorkers(), std::bind(&Worker::clearCounters, _1));
  }

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

inline

Definition at line 254 of file StreamSchedule.h.

References streamContext_.

{ return streamContext_; }

void edm::StreamSchedule::deleteModule

(

std::string const &

iLabel

)

Delete the module with label iLabel.

Definition at line 549 of file StreamSchedule.cc.

References edm::WorkerManager::deleteModuleIfExists(), and workerManager_.

{ workerManager_.deleteModuleIfExists(iLabel); }

void edm::StreamSchedule::endStream

(

)

Definition at line 531 of file StreamSchedule.cc.

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

{ workerManager_.endStream(streamID_, streamContext_); }

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  )

private

Definition at line 500 of file StreamSchedule.cc.

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

Referenced by StreamSchedule().

                                                                             {
    PathWorkers tmpworkers;
    fillWorkers(proc_pset, preg, prealloc, processConfiguration, name, true, tmpworkers, endPathNames);

    if (!tmpworkers.empty()) {
      //EndPaths are not supposed to stop if SkipEvent type exception happens
      end_paths_.emplace_back(bitpos,
                              name,
                              tmpworkers,
                              TrigResPtr(),
                              actionTable(),
                              actReg_,
                              &streamContext_,
                              nullptr,
                              PathContext::PathType::kEndPath);
    } else {
      empty_end_paths_.push_back(bitpos);
    }
    for (WorkerInPath const& workerInPath : tmpworkers) {
      addToAllWorkers(workerInPath.getWorker());
    }
  }

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  )

private

Definition at line 470 of file StreamSchedule.cc.

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

Referenced by StreamSchedule().

                                                                              {
    PathWorkers tmpworkers;
    fillWorkers(proc_pset, preg, prealloc, processConfiguration, name, false, tmpworkers, endPathNames);

    // 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_,
                               &skippingEvent_,
                               PathContext::PathType::kPath);
    } else {
      empty_trig_paths_.push_back(bitpos);
    }
    for (WorkerInPath const& workerInPath : tmpworkers) {
      addToAllWorkers(workerInPath.getWorker());
    }
  }

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  )

private

Definition at line 399 of file StreamSchedule.cc.

References cms::cuda::assert(), edm::errors::Configuration, edm::Worker::description(), Exception, edm::ParameterSet::getParameter(), edm::ParameterSet::getPSetForUpdate(), edm::ParameterSet::getUntrackedParameter(), edm::WorkerManager::getWorker(), edm::WorkerInPath::Ignore, edm::Worker::kFilter, edm::ModuleDescription::moduleName(), edm::Worker::moduleType(), mergeVDriftHistosByStation::name, edm::WorkerInPath::Normal, or, EgammaValidation_cff::pathName, edm::search_all(), AlCaHLTBitMon_QueryRunRegistry::string, edm::WorkerInPath::Veto, and workerManager_.

Referenced by fillEndPath(), and fillTrigPath().

                                                                             {
    vstring modnames = proc_pset.getParameter<vstring>(pathName);
    PathWorkers tmpworkers;

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

      bool isTracked;
      ParameterSet* modpset = proc_pset.getPSetForUpdate(moduleLabel, isTracked);
      if (modpset == 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.";
      }
      assert(isTracked);

      Worker* worker = workerManager_.getWorker(*modpset, preg, prealloc, processConfiguration, moduleLabel);
      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;
      }
      tmpworkers.emplace_back(worker, filterAction, placeInPath, runConcurrently);
      ++placeInPath;
    }

    out.swap(tmpworkers);
  }

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

private

Definition at line 665 of file StreamSchedule.cc.

References mps_fire::action, actionTable(), cms::Exception::addContext(), cms::cuda::assert(), cms::Exception::category(), CMS_SA_ALLOW, cms::Exception::context(), edm::WaitingTaskHolder::doneWaiting(), alignCSCRings::e, edm::exception_actions::FailPath, edm::ExceptionToActionTable::find(), edm::propagate_const< T >::get(), edm::EventPrincipal::id(), edm::exception_actions::IgnoreCompletely, info(), edm::EventTransitionInfo::principal(), edm::printCmsExceptionWarning(), results_, results_inserter_, edm::exception_actions::SkipEvent, streamContext_, streamID_, and total_passed_.

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);
        assert(action != exception_actions::FailPath);
        if (action == exception_actions::SkipEvent) {
          edm::printCmsExceptionWarning("SkipEvent", 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 edm::StreamSchedule::finishProcessOneEvent ( std::exception_ptr  iExcept )

private

Definition at line 723 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;
  }

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 551 of file StreamSchedule.cc.

References allWorkers(), AlCaHLTBitMon_ParallelJobs::p, mps_fire::result, and findQualityFiles::size.

                                                                                     {
    std::vector<ModuleDescription const*> result;
    result.reserve(allWorkers().size());

    for (auto const& worker : allWorkers()) {
      ModuleDescription const* p = worker->description();
      result.push_back(p);
    }
    return result;
  }

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 876 of file StreamSchedule.cc.

References allWorkers(), end_paths_, edm::TriggerReport::endPathSummaries, edm::TriggerReport::eventSummary, edm::fillPathSummary(), edm::fillWorkerSummary(), edm::EventSummary::totalEvents, totalEvents(), edm::EventSummary::totalEventsFailed, totalEventsFailed(), edm::EventSummary::totalEventsPassed, totalEventsPassed(), trig_paths_, edm::TriggerReport::trigPathSummaries, and edm::TriggerReport::workerSummaries.

                                                                {
    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(allWorkers(), rep.workerSummaries, &fillWorkerSummary);
  }

void edm::StreamSchedule::initializeEarlyDelete ( ModuleRegistry &  modReg, edm::ParameterSet const &  opts, edm::ProductRegistry const &  preg, bool  allowEarlyDelete  )

private

Definition at line 243 of file StreamSchedule.cc.

References allWorkers(), edm::BranchDescription::branchName(), edm::maker::ModuleHolder::createOutputModuleCommunicator(), CommonMethods::delta(), submitPVResolutionJobs::desc, earlyDeleteBranchToCount_, earlyDeleteHelpers_, earlyDeleteHelperToBranchIndicies_, end_paths_, edm::ModuleRegistry::forAllModuleHolders(), newFWLiteAna::found, edm::pset::Registry::getMapped(), edm::InEvent, edm::pset::Registry::instance(), B2GTnPMonitor_cfi::item, cmsLHEtoEOSManager::l, dqmiodumpmetadata::n, AlCaHLTBitMon_ParallelJobs::p, TrackValidation_cff::pset, resetEarlyDelete(), AlCaHLTBitMon_QueryRunRegistry::string, trig_paths_, and w.

Referenced by StreamSchedule().

                                                                    {
    //for now, if have a subProcess, don't allow early delete
    // In the future we should use the SubProcess's 'keep list' to decide what can be kept
    if (not allowEarlyDelete)
      return;

    //see if 'canDeleteEarly' was set and if so setup the list with those products actually
    // registered for this job
    std::multimap<std::string, Worker*> branchToReadingWorker;
    initializeBranchToReadingWorker(opts, preg, branchToReadingWorker);

    //If no delete early items have been specified we don't have to do anything
    if (branchToReadingWorker.empty()) {
      return;
    }
    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;
    }

    for (auto w : allWorkers()) {
      //determine if this module could read a branch we want to delete early
      auto pset = pset::Registry::instance()->getMapped(w->description()->parameterSetID());
      if (nullptr != pset) {
        auto branches = pset->getUntrackedParameter<std::vector<std::string>>("mightGet", kEmpty);
        if (not branches.empty()) {
          ++upperLimitOnReadingWorker;
        }
        for (auto const& branch : branches) {
          auto found = branchToReadingWorker.equal_range(branch);
          if (found.first != found.second) {
            ++upperLimitOnIndicies;
            ++reserveSizeForWorker[w];
            if (nullptr == found.first->second) {
              found.first->second = w;
            } else {
              branchToReadingWorker.insert(make_pair(found.first->first, 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 or add the product to the producer's own 'mightGet' list.";
        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];
          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();
    }
  }

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 915 of file StreamSchedule.cc.

References edm::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;
    }
  }

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

Definition at line 805 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());
      }
    }
  }

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

Definition at line 776 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());
      }
    }
  }

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

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

inline

Definition at line 252 of file StreamSchedule.h.

References number_of_unscheduled_modules_.

{ return number_of_unscheduled_modules_; }

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

Definition at line 562 of file StreamSchedule.cc.

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

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

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

      workerManager_.setupResolvers(ep);
      workerManager_.setupOnDemandSystem(info);

      ++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->processOneOccurrenceAsync(hAllPathsDone, info, serviceToken, streamID_, &streamContext_);
      }

      for (auto it = trig_paths_.rbegin(), itEnd = trig_paths_.rend(); it != itEnd; ++it) {
        it->processOneOccurrenceAsync(taskHolder, info, serviceToken, streamID_, &streamContext_);
      }

      ParentContext parentContext(&streamContext_);
      workerManager_.processAccumulatorsAsync<OccurrenceTraits<EventPrincipal, BranchActionStreamBegin>>(
          hAllPathsDone, info, serviceToken, streamID_, parentContext, &streamContext_);
    } catch (...) {
      iTask.doneWaiting(std::current_exception());
    }
  }

template<typename T >

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

Definition at line 372 of file StreamSchedule.h.

References actReg_, edm::addContextAndPrintException(), CMS_SA_ALLOW, cms::Exception::context(), edm::WaitingTaskHolder::doneWaiting(), end_paths_, edm::ExceptionFromThisContext, edm::WaitingTaskHolder::group(), h, gpuClustering::id, info(), edm::ServiceWeakToken::lock(), edm::make_functor_task(), edm::make_waiting_task(), AlCaHLTBitMon_ParallelJobs::p, edm::WorkerManager::processOneOccurrenceAsync(), edm::WorkerManager::resetAll(), alignCSCRings::s, streamContext_, streamID_, unpackBuffers-CaloStage2::token, trig_paths_, edm::StreamID::value(), workerManager_, and edm::convertException::wrap().

                                                                            {
    auto const& principal = transitionInfo.principal();
    T::setStreamContext(streamContext_, principal);

    auto id = principal.id();
    ServiceWeakToken weakToken = token;
    auto doneTask = make_waiting_task(
        [this, iHolder, id, cleaningUpAfterException, weakToken](std::exception_ptr const* iPtr) mutable {
          std::exception_ptr excpt;
          if (iPtr) {
            excpt = *iPtr;
            //add context information to the exception and print message
            try {
              convertException::wrap([&]() { std::rethrow_exception(excpt); });
            } catch (cms::Exception& ex) {
              //TODO: should add the transition type info
              std::ostringstream ost;
              if (ex.context().empty()) {
                ost << "Processing " << T::transitionName() << " " << id;
              }
              ServiceRegistry::Operate op(weakToken.lock());
              addContextAndPrintException(ost.str().c_str(), ex, cleaningUpAfterException);
              excpt = std::current_exception();
            }

            ServiceRegistry::Operate op(weakToken.lock());
            actReg_->preStreamEarlyTerminationSignal_(streamContext_, TerminationOrigin::ExceptionFromThisContext);
          }
          // Caught exception is propagated via WaitingTaskHolder
          CMS_SA_ALLOW try {
            ServiceRegistry::Operate op(weakToken.lock());
            T::postScheduleSignal(actReg_.get(), &streamContext_);
          } catch (...) {
            if (not excpt) {
              excpt = std::current_exception();
            }
          }
          iHolder.doneWaiting(excpt);
        });

    auto task = make_functor_task(
        [this, h = WaitingTaskHolder(*iHolder.group(), doneTask), info = transitionInfo, weakToken]() mutable {
          auto token = weakToken.lock();
          ServiceRegistry::Operate op(token);
          // Caught exception is propagated via WaitingTaskHolder
          CMS_SA_ALLOW try {
            T::preScheduleSignal(actReg_.get(), &streamContext_);

            workerManager_.resetAll();
          } catch (...) {
            h.doneWaiting(std::current_exception());
            return;
          }

          for (auto& p : end_paths_) {
            p.runAllModulesAsync<T>(h, info, token, streamID_, &streamContext_);
          }

          for (auto& p : trig_paths_) {
            p.runAllModulesAsync<T>(h, info, token, streamID_, &streamContext_);
          }

          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.
      iHolder.group()->run([task]() {
        TaskSentry s{task};
        task->execute();
      });
    } else {
      tbb::task_arena arena{tbb::task_arena::attach()};
      arena.enqueue([task]() {
        TaskSentry s{task};
        task->execute();
      });
    }
  }

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 533 of file StreamSchedule.cc.

References allWorkers(), edm::Worker::beginStream(), newFWLiteAna::found, edm::maker::ModuleHolder::replaceModuleFor(), streamContext_, and streamID_.

                                                                                     {
    Worker* found = nullptr;
    for (auto const& worker : allWorkers()) {
      if (worker->description()->moduleLabel() == iLabel) {
        found = worker;
        break;
      }
    }
    if (nullptr == found) {
      return;
    }

    iMod->replaceModuleFor(found);
    found->beginStream(streamID_, streamContext_);
  }

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

inlineprivate

Definition at line 366 of file StreamSchedule.h.

References edm::EventID::event(), edm::EventPrincipal::id(), and edm::EventID::run().

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

void edm::StreamSchedule::resetAll ( )

private

Definition at line 894 of file StreamSchedule.cc.

References results_, and skippingEvent_.

Referenced by processOneEventAsync().

                                {
    skippingEvent_ = false;
    results_->reset();
  }

void edm::StreamSchedule::resetEarlyDelete ( )

private

Definition at line 901 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();
    }
  }

TrigResConstPtr edm::StreamSchedule::results ( ) const

inlineprivate

Definition at line 320 of file StreamSchedule.h.

References edm::get_underlying_safe(), and results_.

Referenced by StreamSchedule().

{ return get_underlying_safe(results_); }

TrigResPtr& edm::StreamSchedule::results ( )

inlineprivate

Definition at line 321 of file StreamSchedule.h.

References edm::get_underlying_safe(), and results_.

{ return get_underlying_safe(results_); }

StreamID edm::StreamSchedule::streamID ( ) const

inline

Definition at line 199 of file StreamSchedule.h.

References streamID_.

{ return streamID_; }

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 226 of file StreamSchedule.h.

References total_events_.

Referenced by getTriggerReport(), and totalEventsFailed().

{ return total_events_; }

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 234 of file StreamSchedule.h.

References totalEvents(), and totalEventsPassed().

Referenced by getTriggerReport().

{ return totalEvents() - 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 230 of file StreamSchedule.h.

References total_passed_.

Referenced by getTriggerReport(), and totalEventsFailed().

{ return total_passed_; }

Member Data Documentation

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

private

Definition at line 329 of file StreamSchedule.h.

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

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

private

Definition at line 345 of file StreamSchedule.h.

Referenced by initializeEarlyDelete(), and resetEarlyDelete().

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

private

Definition at line 355 of file StreamSchedule.h.

Referenced by initializeEarlyDelete(), and resetEarlyDelete().

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

private

Definition at line 352 of file StreamSchedule.h.

Referenced by initializeEarlyDelete(), and resetEarlyDelete().

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

private

Definition at line 340 of file StreamSchedule.h.

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

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

private

Definition at line 339 of file StreamSchedule.h.

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

TrigPaths edm::StreamSchedule::end_paths_

private

Definition at line 338 of file StreamSchedule.h.

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

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

private

Definition at line 335 of file StreamSchedule.h.

Referenced by makePathStatusInserters(), and processOneEventAsync().

unsigned int edm::StreamSchedule::number_of_unscheduled_modules_

private

Definition at line 359 of file StreamSchedule.h.

Referenced by numberOfUnscheduledModules(), and StreamSchedule().

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

private

Definition at line 334 of file StreamSchedule.h.

Referenced by makePathStatusInserters(), and processOneEventAsync().

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

private

Definition at line 331 of file StreamSchedule.h.

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

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

private

Definition at line 333 of file StreamSchedule.h.

Referenced by finishedPaths(), and StreamSchedule().

std::atomic<bool> edm::StreamSchedule::skippingEvent_

private

Definition at line 363 of file StreamSchedule.h.

Referenced by fillTrigPath(), and resetAll().

StreamContext edm::StreamSchedule::streamContext_

private

Definition at line 362 of file StreamSchedule.h.

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

StreamID edm::StreamSchedule::streamID_

private

Definition at line 361 of file StreamSchedule.h.

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

int edm::StreamSchedule::total_events_

private

Definition at line 357 of file StreamSchedule.h.

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

int edm::StreamSchedule::total_passed_

private

Definition at line 358 of file StreamSchedule.h.

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

TrigPaths edm::StreamSchedule::trig_paths_

private

Definition at line 337 of file StreamSchedule.h.

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

WorkerManager edm::StreamSchedule::workerManager_

private

Definition at line 328 of file StreamSchedule.h.

Referenced by actionTable(), addToAllWorkers(), allWorkers(), beginStream(), deleteModule(), endStream(), fillWorkers(), processOneEventAsync(), processOneStreamAsync(), and StreamSchedule().