edm::Worker Class Reference

#include <Worker.h>

Inheritance diagram for edm::Worker:

Classes
class	AcquireTask
class	AcquireTask< OccurrenceTraits< EventPrincipal, BranchActionStreamBegin >, DUMMY >
class	HandleExternalWorkExceptionTask
class	RunModuleTask
struct	TaskQueueAdaptor
class	TransitionIDValue
class	TransitionIDValueBase

Public Types
enum	State { Ready, Pass, Fail, Exception }
enum	Types { kAnalyzer, kFilter, kProducer, kOutputModule }

Public Member Functions
void	addedToPath ()
void	beginJob ()
void	beginStream (StreamID id, StreamContext &streamContext)
void	callWhenDoneAsync (WaitingTask *task)
void	clearCounters ()
virtual std::vector< ConsumesInfo >	consumesInfo () const =0
virtual void	convertCurrentProcessAlias (std::string const &processName)=0
ModuleDescription const *	descPtr () const
ModuleDescription const &	description () const
template<typename T >
bool	doWork (typename T::MyPrincipal const &, EventSetupImpl const &c, StreamID stream, ParentContext const &parentContext, typename T::Context const *context)
template<typename T >
void	doWorkAsync (WaitingTask *task, typename T::MyPrincipal const &, EventSetupImpl const &c, ServiceToken const &token, StreamID stream, ParentContext const &parentContext, typename T::Context const *context)
template<typename T >
void	doWorkNoPrefetchingAsync (WaitingTask *task, typename T::MyPrincipal const &, EventSetupImpl const &c, ServiceToken const &token, StreamID stream, ParentContext const &parentContext, typename T::Context const *context)
void	endJob ()
void	endStream (StreamID id, StreamContext &streamContext)
virtual SerialTaskQueue *	globalLuminosityBlocksQueue ()=0
virtual SerialTaskQueue *	globalRunsQueue ()=0
virtual bool	hasAccumulator () const =0
virtual void	modulesWhoseProductsAreConsumed (std::vector< ModuleDescription const * > &modules, ProductRegistry const &preg, std::map< std::string, ModuleDescription const * > const &labelsToDesc) const =0
virtual Types	moduleType () const =0
Worker &	operator= (Worker const &)=delete
void	postDoEvent (EventPrincipal const &)
void	prePrefetchSelectionAsync (WaitingTask *task, ServiceToken const &, StreamID stream, EventPrincipal const *)
void	prePrefetchSelectionAsync (WaitingTask *task, ServiceToken const &, StreamID stream, void const *)
void	registerThinnedAssociations (ProductRegistry const &registry, ThinnedAssociationsHelper &helper)
void	reset ()
virtual void	resolvePutIndicies (BranchType iBranchType, std::unordered_multimap< std::string, std::tuple< TypeID const *, const char *, edm::ProductResolverIndex >> const &iIndicies)=0
void	respondToCloseInputFile (FileBlock const &fb)
void	respondToOpenInputFile (FileBlock const &fb)
template<typename T >
std::exception_ptr	runModuleDirectly (typename T::MyPrincipal const &ep, EventSetupImpl const &es, StreamID streamID, ParentContext const &parentContext, typename T::Context const *context)
void	setActivityRegistry (std::shared_ptr< ActivityRegistry > areg)
void	setEarlyDeleteHelper (EarlyDeleteHelper *iHelper)
void	skipOnPath (EventPrincipal const &iEvent)
State	state () const
int	timesExcept () const
int	timesFailed () const
int	timesPass () const
int	timesPassed () const
int	timesRun () const
int	timesVisited () const
virtual void	updateLookup (BranchType iBranchType, ProductResolverIndexHelper const &)=0
virtual void	updateLookup (eventsetup::ESRecordsToProxyIndices const &)=0
virtual bool	wantsGlobalLuminosityBlocks () const =0
virtual bool	wantsGlobalRuns () const =0
virtual bool	wantsStreamLuminosityBlocks () const =0
virtual bool	wantsStreamRuns () const =0
	Worker (ModuleDescription const &iMD, ExceptionToActionTable const *iActions)
	Worker (Worker const &)=delete
virtual	~Worker ()

Protected Member Functions
ActivityRegistry *	activityRegistry ()
virtual void	implBeginJob ()=0
virtual void	implBeginStream (StreamID)=0
virtual bool	implDo (EventPrincipal const &, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual void	implDoAcquire (EventPrincipal const &, EventSetupImpl const &c, ModuleCallingContext const *mcc, WaitingTaskWithArenaHolder &holder)=0
virtual bool	implDoBegin (LuminosityBlockPrincipal const &lbp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual bool	implDoBegin (RunPrincipal const &rp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual bool	implDoEnd (LuminosityBlockPrincipal const &lbp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual bool	implDoEnd (RunPrincipal const &rp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual bool	implDoPrePrefetchSelection (StreamID id, EventPrincipal const &ep, ModuleCallingContext const *mcc)=0
virtual bool	implDoStreamBegin (StreamID id, LuminosityBlockPrincipal const &lbp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual bool	implDoStreamBegin (StreamID id, RunPrincipal const &rp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual bool	implDoStreamEnd (StreamID id, LuminosityBlockPrincipal const &lbp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual bool	implDoStreamEnd (StreamID id, RunPrincipal const &rp, EventSetupImpl const &c, ModuleCallingContext const *mcc)=0
virtual void	implEndJob ()=0
virtual void	implEndStream (StreamID)=0
virtual bool	implNeedToRunSelection () const =0
virtual void	itemsToGetForSelection (std::vector< ProductResolverIndexAndSkipBit > &) const =0
void	resetModuleDescription (ModuleDescription const *)
virtual std::string	workerType () const =0

Private Member Functions
void	emitPostModuleEventPrefetchingSignal ()
std::exception_ptr	handleExternalWorkException (std::exception_ptr const *iEPtr, ParentContext const &parentContext)
virtual bool	hasAcquire () const =0
virtual void	implRegisterThinnedAssociations (ProductRegistry const &, ThinnedAssociationsHelper &)=0
virtual void	implRespondToCloseInputFile (FileBlock const &fb)=0
virtual void	implRespondToOpenInputFile (FileBlock const &fb)=0
virtual void	itemsMayGet (BranchType, std::vector< ProductResolverIndexAndSkipBit > &) const =0
virtual std::vector< ProductResolverIndex > const &	itemsShouldPutInEvent () const =0
virtual void	itemsToGet (BranchType, std::vector< ProductResolverIndexAndSkipBit > &) const =0
virtual std::vector< ProductResolverIndexAndSkipBit > const &	itemsToGetFrom (BranchType) const =0
virtual void	preActionBeforeRunEventAsync (WaitingTask *iTask, ModuleCallingContext const &moduleCallingContext, Principal const &iPrincipal) const =0
void	prefetchAsync (WaitingTask *, ServiceToken const &, ParentContext const &parentContext, Principal const &)
void	runAcquire (EventPrincipal const &ep, EventSetupImpl const &es, ParentContext const &parentContext, WaitingTaskWithArenaHolder &holder)
void	runAcquireAfterAsyncPrefetch (std::exception_ptr const *iEPtr, EventPrincipal const &ep, EventSetupImpl const &es, ParentContext const &parentContext, WaitingTaskWithArenaHolder holder)
template<typename T >
bool	runModule (typename T::MyPrincipal const &, EventSetupImpl const &c, StreamID stream, ParentContext const &parentContext, typename T::Context const *context)
template<typename T >
std::exception_ptr	runModuleAfterAsyncPrefetch (std::exception_ptr const *iEPtr, typename T::MyPrincipal const &ep, EventSetupImpl const &es, StreamID streamID, ParentContext const &parentContext, typename T::Context const *context)
virtual TaskQueueAdaptor	serializeRunModule ()=0
template<bool IS_EVENT>
std::exception_ptr	setException (std::exception_ptr iException)
template<bool IS_EVENT>
bool	setFailed ()
template<bool IS_EVENT>
bool	setPassed ()
bool	shouldRethrowException (std::exception_ptr iPtr, ParentContext const &parentContext, bool isEvent, TransitionIDValueBase const &iID) const

Static Private Member Functions
static void	exceptionContext (cms::Exception &ex, ModuleCallingContext const *mcc)

Private Attributes
ExceptionToActionTable const *	actions_
std::shared_ptr< ActivityRegistry >	actReg_
std::exception_ptr	cached_exception_
edm::propagate_const< EarlyDeleteHelper * >	earlyDeleteHelper_
ModuleCallingContext	moduleCallingContext_
std::atomic< int >	numberOfPathsLeftToRun_
int	numberOfPathsOn_
bool	ranAcquireWithoutException_
std::atomic< State >	state_
std::atomic< int >	timesExcept_
std::atomic< int >	timesFailed_
std::atomic< int >	timesPassed_
std::atomic< int >	timesRun_
std::atomic< int >	timesVisited_
edm::WaitingTaskList	waitingTasks_
std::atomic< bool >	workStarted_

Friends
template<typename O >
class	workerhelper::CallImpl

Detailed Description

Definition at line 83 of file Worker.h.

Member Enumeration Documentation

State

enum edm::Worker::State

Enumerator
Ready
Pass
Fail
Exception

Definition at line 85 of file Worker.h.

{ Ready, Pass, Fail, Exception };

Types

enum edm::Worker::Types

Enumerator
kAnalyzer
kFilter
kProducer
kOutputModule

Definition at line 86 of file Worker.h.

{ kAnalyzer, kFilter, kProducer, kOutputModule };

Constructor & Destructor Documentation

Worker() [1/2]

edm::Worker::Worker	(	ModuleDescription const &	iMD,
		ExceptionToActionTable const *	iActions
	)

Definition at line 83 of file Worker.cc.

      : timesRun_(0),
        timesVisited_(0),
        timesPassed_(0),
        timesFailed_(0),
        timesExcept_(0),
        state_(Ready),
        numberOfPathsOn_(0),
        numberOfPathsLeftToRun_(0),
        moduleCallingContext_(&iMD),
        actions_(iActions),
        cached_exception_(),
        actReg_(),
        earlyDeleteHelper_(nullptr),
        workStarted_(false),
        ranAcquireWithoutException_(false) {}

~Worker()

edm::Worker::~Worker ( )

virtual

Definition at line 100 of file Worker.cc.

{}

Worker() [2/2]

edm::Worker::Worker ( Worker const &  )

delete

Member Function Documentation

activityRegistry()

ActivityRegistry* edm::Worker::activityRegistry ( )

inlineprotected

Definition at line 284 of file Worker.h.

{ return actReg_.get(); }

References actReg_.

addedToPath()

void edm::Worker::addedToPath ( )

inline

Definition at line 225 of file Worker.h.

{ ++numberOfPathsOn_; }

References numberOfPathsOn_.

beginJob()

void edm::Worker::beginJob

(

void

)

Definition at line 285 of file Worker.cc.

                        {
    try {
      convertException::wrap([&]() {
        ModuleBeginJobSignalSentry cpp(actReg_.get(), description());
        implBeginJob();
      });
    } catch (cms::Exception& ex) {
      state_ = Exception;
      std::ostringstream ost;
      ost << "Calling beginJob for module " << description().moduleName() << "/'" << description().moduleLabel() << "'";
      ex.addContext(ost.str());
      throw;
    }
  }

References actReg_, cms::Exception::addContext(), description(), Exception, implBeginJob(), edm::ModuleDescription::moduleLabel(), edm::ModuleDescription::moduleName(), state_, and edm::convertException::wrap().

Referenced by edm::WorkerManager::beginJob().

beginStream()

void edm::Worker::beginStream	(	StreamID	id,
		StreamContext &	streamContext
	)

Definition at line 315 of file Worker.cc.

                                                                    {
    try {
      convertException::wrap([&]() {
        streamContext.setTransition(StreamContext::Transition::kBeginStream);
        streamContext.setEventID(EventID(0, 0, 0));
        streamContext.setRunIndex(RunIndex::invalidRunIndex());
        streamContext.setLuminosityBlockIndex(LuminosityBlockIndex::invalidLuminosityBlockIndex());
        streamContext.setTimestamp(Timestamp());
        ParentContext parentContext(&streamContext);
        ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
        moduleCallingContext_.setState(ModuleCallingContext::State::kRunning);
        ModuleBeginStreamSignalSentry beginSentry(actReg_.get(), streamContext, moduleCallingContext_);
        implBeginStream(id);
      });
    } catch (cms::Exception& ex) {
      state_ = Exception;
      std::ostringstream ost;
      ost << "Calling beginStream for module " << description().moduleName() << "/'" << description().moduleLabel()
          << "'";
      ex.addContext(ost.str());
      throw;
    }
  }

References actReg_, cms::Exception::addContext(), description(), Exception, implBeginStream(), edm::LuminosityBlockIndex::invalidLuminosityBlockIndex(), edm::RunIndex::invalidRunIndex(), edm::StreamContext::kBeginStream, edm::ModuleCallingContext::kRunning, moduleCallingContext_, edm::ModuleDescription::moduleLabel(), edm::ModuleDescription::moduleName(), edm::StreamContext::setEventID(), edm::StreamContext::setLuminosityBlockIndex(), edm::StreamContext::setRunIndex(), edm::ModuleCallingContext::setState(), edm::StreamContext::setTimestamp(), edm::StreamContext::setTransition(), state_, and edm::convertException::wrap().

callWhenDoneAsync()

void edm::Worker::callWhenDoneAsync ( WaitingTask *  task )

inline

Definition at line 167 of file Worker.h.

{ waitingTasks_.add(task); }

References edm::WaitingTaskList::add(), TrackValidation_cff::task, and waitingTasks_.

Referenced by edm::PuttableProductResolver::prefetchAsync_(), and edm::SwitchProducerProductResolver::prefetchAsync_().

clearCounters()

void edm::Worker::clearCounters ( )

inline

Definition at line 217 of file Worker.h.

                         {
      timesRun_.store(0, std::memory_order_release);
      timesVisited_.store(0, std::memory_order_release);
      timesPassed_.store(0, std::memory_order_release);
      timesFailed_.store(0, std::memory_order_release);
      timesExcept_.store(0, std::memory_order_release);
    }

References timesExcept_, timesFailed_, timesPassed_, timesRun_, and timesVisited_.

Referenced by edm::StreamSchedule::clearCounters().

consumesInfo()

virtual std::vector<ConsumesInfo> edm::Worker::consumesInfo ( ) const

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::PathsAndConsumesOfModules::doConsumesInfo().

convertCurrentProcessAlias()

virtual void edm::Worker::convertCurrentProcessAlias ( std::string const &  processName )

pure virtual

Implemented in edm::WorkerT< T >.

descPtr()

ModuleDescription const* edm::Worker::descPtr ( ) const

inline

Definition at line 191 of file Worker.h.

{ return moduleCallingContext_.moduleDescription(); }

References moduleCallingContext_, and edm::ModuleCallingContext::moduleDescription().

description()

ModuleDescription const& edm::Worker::description ( ) const

inline

Definition at line 190 of file Worker.h.

{ return *(moduleCallingContext_.moduleDescription()); }

References moduleCallingContext_, and edm::ModuleCallingContext::moduleDescription().

Referenced by beginJob(), beginStream(), endJob(), endStream(), edm::StreamSchedule::fillWorkers(), edm::UnscheduledProductResolver::resolveProduct_(), and shouldRethrowException().

doWork()

template<typename T >

bool edm::Worker::doWork	(	typename T::MyPrincipal const &	ep,
		EventSetupImpl const &	c,
		StreamID	stream,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 974 of file Worker.h.

                                                        {
    if constexpr (T::isEvent_) {
      timesVisited_.fetch_add(1, std::memory_order_relaxed);
    }
    bool rc = false;
 
    switch (state_) {
      case Ready:
        break;
      case Pass:
        return true;
      case Fail:
        return false;
      case Exception: {
        std::rethrow_exception(cached_exception_);
      }
    }
 
    bool expected = false;
    if (not workStarted_.compare_exchange_strong(expected, true)) {
      //another thread beat us here
      auto waitTask = edm::make_empty_waiting_task();
      waitTask->increment_ref_count();
 
      waitingTasks_.add(waitTask.get());
 
      waitTask->wait_for_all();
 
      switch (state_) {
        case Ready:
          assert(false);
        case Pass:
          return true;
        case Fail:
          return false;
        case Exception: {
          std::rethrow_exception(cached_exception_);
        }
      }
    }
 
    //Need the context to be set until after any exception is resolved
    moduleCallingContext_.setContext(ModuleCallingContext::State::kPrefetching, parentContext, nullptr);
 
    auto resetContext = [](ModuleCallingContext* iContext) {
      iContext->setContext(ModuleCallingContext::State::kInvalid, ParentContext(), nullptr);
    };
    std::unique_ptr<ModuleCallingContext, decltype(resetContext)> prefetchSentry(&moduleCallingContext_, resetContext);
 
    if constexpr (T::isEvent_) {
      //if have TriggerResults based selection we want to reject the event before doing prefetching
      if (workerhelper::CallImpl<T>::needToRunSelection(this)) {
        auto waitTask = edm::make_empty_waiting_task();
        waitTask->set_ref_count(2);
        prePrefetchSelectionAsync(waitTask.get(), ServiceRegistry::instance().presentToken(), streamID, &ep);
        waitTask->decrement_ref_count();
        waitTask->wait_for_all();
 
        if (state() != Ready) {
          //The selection must have rejected this event
          return true;
        }
      }
      auto waitTask = edm::make_empty_waiting_task();
      {
        //Make sure signal is sent once the prefetching is done
        // [the 'pre' signal was sent in prefetchAsync]
        //The purpose of this block is to send the signal after wait_for_all
        auto sentryFunc = [this](void*) { emitPostModuleEventPrefetchingSignal(); };
        std::unique_ptr<ActivityRegistry, decltype(sentryFunc)> signalSentry(actReg_.get(), sentryFunc);
 
        //set count to 2 since wait_for_all requires value to not go to 0
        waitTask->set_ref_count(2);
 
        prefetchAsync(waitTask.get(), ServiceRegistry::instance().presentToken(), parentContext, ep);
        waitTask->decrement_ref_count();
        waitTask->wait_for_all();
      }
      if (waitTask->exceptionPtr() != nullptr) {
        TransitionIDValue<typename T::MyPrincipal> idValue(ep);
        if (shouldRethrowException(*waitTask->exceptionPtr(), parentContext, T::isEvent_, idValue)) {
          setException<T::isEvent_>(*waitTask->exceptionPtr());
          waitingTasks_.doneWaiting(cached_exception_);
          std::rethrow_exception(cached_exception_);
        } else {
          setPassed<T::isEvent_>();
          waitingTasks_.doneWaiting(nullptr);
          return true;
        }
      }
    }
 
    //successful prefetch so no reset necessary
    prefetchSentry.release();
    if (auto queue = serializeRunModule()) {
      auto serviceToken = ServiceRegistry::instance().presentToken();
      queue.pushAndWait([&]() {
        //Need to make the services available
        ServiceRegistry::Operate guard(serviceToken);
        // This try-catch is primarily for paranoia: runModule() deals internally with exceptions, except for those coming from Service signal actions, which are not supposed to throw exceptions
        CMS_SA_ALLOW try { rc = runModule<T>(ep, es, streamID, parentContext, context); } catch (...) {
        }
      });
    } else {
      // This try-catch is primarily for paranoia: runModule() deals internally with exceptions, except for those coming from Service signal actions, which are not supposed to throw exceptions
      CMS_SA_ALLOW try { rc = runModule<T>(ep, es, streamID, parentContext, context); } catch (...) {
      }
    }
    if (state_ == Exception) {
      waitingTasks_.doneWaiting(cached_exception_);
      std::rethrow_exception(cached_exception_);
    }
 
    waitingTasks_.doneWaiting(nullptr);
    return rc;
  }

References actReg_, edm::WaitingTaskList::add(), cms::cuda::assert(), cached_exception_, CMS_SA_ALLOW, edm::WaitingTaskList::doneWaiting(), emitPostModuleEventPrefetchingSignal(), SiStripBadComponentsDQMServiceTemplate_cfg::ep, Exception, Fail, edm::ServiceRegistry::instance(), edm::ModuleCallingContext::kInvalid, edm::ModuleCallingContext::kPrefetching, edm::make_empty_waiting_task(), moduleCallingContext_, Pass, prefetchAsync(), prePrefetchSelectionAsync(), edm::ServiceRegistry::presentToken(), createBeamHaloJobs::queue, Ready, serializeRunModule(), edm::ModuleCallingContext::setContext(), shouldRethrowException(), state(), state_, timesVisited_, waitingTasks_, and workStarted_.

Referenced by edm::UnscheduledProductResolver::resolveProduct_().

doWorkAsync()

template<typename T >

void edm::Worker::doWorkAsync	(	WaitingTask *	task,
		typename T::MyPrincipal const &	ep,
		EventSetupImpl const &	c,
		ServiceToken const &	token,
		StreamID	stream,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 831 of file Worker.h.

                                                             {
    if (not workerhelper::CallImpl<T>::wantsTransition(this)) {
      return;
    }
 
    //Need to check workStarted_ before adding to waitingTasks_
    bool expected = false;
    bool workStarted = workStarted_.compare_exchange_strong(expected, true);
 
    waitingTasks_.add(task);
    if constexpr (T::isEvent_) {
      timesVisited_.fetch_add(1, std::memory_order_relaxed);
    }
 
    if (workStarted) {
      moduleCallingContext_.setContext(ModuleCallingContext::State::kPrefetching, parentContext, nullptr);
 
      //if have TriggerResults based selection we want to reject the event before doing prefetching
      if (workerhelper::CallImpl<T>::needToRunSelection(this)) {
        //We need to run the selection in a different task so that
        // we can prefetch the data needed for the selection
        auto runTask =
            new (tbb::task::allocate_root()) RunModuleTask<T>(this, ep, es, token, streamID, parentContext, context);
 
        //make sure the task is either run or destroyed
        struct DestroyTask {
          DestroyTask(edm::WaitingTask* iTask) : m_task(iTask) {}
 
          ~DestroyTask() {
            auto p = m_task.load();
            if (p) {
              tbb::task::destroy(*p);
            }
          }
 
          edm::WaitingTask* release() {
            auto t = m_task.load();
            m_task.store(nullptr);
            return t;
          }
 
          std::atomic<edm::WaitingTask*> m_task;
        };
 
        auto ownRunTask = std::make_shared<DestroyTask>(runTask);
        auto selectionTask =
            make_waiting_task(tbb::task::allocate_root(),
                              [ownRunTask, parentContext, &ep, token, this](std::exception_ptr const*) mutable {
                                ServiceRegistry::Operate guard(token);
                                prefetchAsync(ownRunTask->release(), token, parentContext, ep);
                              });
        prePrefetchSelectionAsync(selectionTask, token, streamID, &ep);
      } else {
        WaitingTask* moduleTask =
            new (tbb::task::allocate_root()) RunModuleTask<T>(this, ep, es, token, streamID, parentContext, context);
        if constexpr (T::isEvent_) {
          if (hasAcquire()) {
            WaitingTaskWithArenaHolder runTaskHolder(
                new (tbb::task::allocate_root()) HandleExternalWorkExceptionTask(this, moduleTask, parentContext));
            moduleTask = new (tbb::task::allocate_root())
                AcquireTask<T>(this, ep, es, token, parentContext, std::move(runTaskHolder));
          }
        }
        prefetchAsync(moduleTask, token, parentContext, ep);
      }
    }
  }

References edm::WaitingTaskList::add(), SiStripBadComponentsDQMServiceTemplate_cfg::ep, hasAcquire(), edm::ModuleCallingContext::kPrefetching, edm::make_waiting_task(), moduleCallingContext_, eostools::move(), AlCaHLTBitMon_ParallelJobs::p, prefetchAsync(), prePrefetchSelectionAsync(), fetchall_from_DQM_v2::release, edm::ModuleCallingContext::setContext(), OrderedSet::t, TrackValidation_cff::task, timesVisited_, unpackBuffers-CaloStage2::token, waitingTasks_, and workStarted_.

Referenced by edm::UnscheduledProductResolver::prefetchAsync_(), and edm::WorkerInPath::runWorkerAsync().

doWorkNoPrefetchingAsync()

template<typename T >

void edm::Worker::doWorkNoPrefetchingAsync	(	WaitingTask *	task,
		typename T::MyPrincipal const &	principal,
		EventSetupImpl const &	c,
		ServiceToken const &	token,
		StreamID	stream,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 934 of file Worker.h.

                                                                          {
    if (not workerhelper::CallImpl<T>::wantsTransition(this)) {
      return;
    }
 
    //Need to check workStarted_ before adding to waitingTasks_
    bool expected = false;
    auto workStarted = workStarted_.compare_exchange_strong(expected, true);
 
    waitingTasks_.add(task);
    if (workStarted) {
      auto toDo = [this, &principal, &es, streamID, parentContext, context, serviceToken]() {
        std::exception_ptr exceptionPtr;
        // Caught exception is propagated via WaitingTaskList
        CMS_SA_ALLOW try {
          //Need to make the services available
          ServiceRegistry::Operate guard(serviceToken);
 
          this->runModule<T>(principal, es, streamID, parentContext, context);
        } catch (...) {
          exceptionPtr = std::current_exception();
        }
        this->waitingTasks_.doneWaiting(exceptionPtr);
      };
      if (auto queue = this->serializeRunModule()) {
        queue.push(toDo);
      } else {
        auto taskToDo = make_functor_task(tbb::task::allocate_root(), toDo);
        tbb::task::spawn(*taskToDo);
      }
    }
  }

References edm::WaitingTaskList::add(), CMS_SA_ALLOW, edm::WaitingTaskList::doneWaiting(), edm::make_functor_task(), createBeamHaloJobs::queue, serializeRunModule(), TrackValidation_cff::task, waitingTasks_, and workStarted_.

Referenced by edm::WorkerInPath::runWorkerAsync().

emitPostModuleEventPrefetchingSignal()

void edm::Worker::emitPostModuleEventPrefetchingSignal ( )

inlineprivate

Definition at line 374 of file Worker.h.

                                                {
      actReg_->postModuleEventPrefetchingSignal_.emit(*moduleCallingContext_.getStreamContext(), moduleCallingContext_);
    }

References actReg_, edm::ModuleCallingContext::getStreamContext(), and moduleCallingContext_.

Referenced by doWork(), and edm::Worker::RunModuleTask< T >::execute().

endJob()

void edm::Worker::endJob

(

void

)

Definition at line 300 of file Worker.cc.

                      {
    try {
      convertException::wrap([&]() {
        ModuleEndJobSignalSentry cpp(actReg_.get(), description());
        implEndJob();
      });
    } catch (cms::Exception& ex) {
      state_ = Exception;
      std::ostringstream ost;
      ost << "Calling endJob for module " << description().moduleName() << "/'" << description().moduleLabel() << "'";
      ex.addContext(ost.str());
      throw;
    }
  }

References actReg_, cms::Exception::addContext(), description(), Exception, implEndJob(), edm::ModuleDescription::moduleLabel(), edm::ModuleDescription::moduleName(), state_, and edm::convertException::wrap().

endStream()

void edm::Worker::endStream	(	StreamID	id,
		StreamContext &	streamContext
	)

Definition at line 339 of file Worker.cc.

                                                                  {
    try {
      convertException::wrap([&]() {
        streamContext.setTransition(StreamContext::Transition::kEndStream);
        streamContext.setEventID(EventID(0, 0, 0));
        streamContext.setRunIndex(RunIndex::invalidRunIndex());
        streamContext.setLuminosityBlockIndex(LuminosityBlockIndex::invalidLuminosityBlockIndex());
        streamContext.setTimestamp(Timestamp());
        ParentContext parentContext(&streamContext);
        ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
        moduleCallingContext_.setState(ModuleCallingContext::State::kRunning);
        ModuleEndStreamSignalSentry endSentry(actReg_.get(), streamContext, moduleCallingContext_);
        implEndStream(id);
      });
    } catch (cms::Exception& ex) {
      state_ = Exception;
      std::ostringstream ost;
      ost << "Calling endStream for module " << description().moduleName() << "/'" << description().moduleLabel()
          << "'";
      ex.addContext(ost.str());
      throw;
    }
  }

References actReg_, cms::Exception::addContext(), description(), Exception, implEndStream(), edm::LuminosityBlockIndex::invalidLuminosityBlockIndex(), edm::RunIndex::invalidRunIndex(), edm::StreamContext::kEndStream, edm::ModuleCallingContext::kRunning, moduleCallingContext_, edm::ModuleDescription::moduleLabel(), edm::ModuleDescription::moduleName(), edm::StreamContext::setEventID(), edm::StreamContext::setLuminosityBlockIndex(), edm::StreamContext::setRunIndex(), edm::ModuleCallingContext::setState(), edm::StreamContext::setTimestamp(), edm::StreamContext::setTransition(), state_, and edm::convertException::wrap().

exceptionContext()

void edm::Worker::exceptionContext ( cms::Exception &  ex, ModuleCallingContext const *  mcc  )

staticprivate

Definition at line 104 of file Worker.cc.

                                                                                 {
    ModuleCallingContext const* imcc = mcc;
    while ((imcc->type() == ParentContext::Type::kModule) or (imcc->type() == ParentContext::Type::kInternal)) {
      std::ostringstream iost;
      if (imcc->state() == ModuleCallingContext::State::kPrefetching) {
        iost << "Prefetching for module ";
      } else {
        iost << "Calling method for module ";
      }
      iost << imcc->moduleDescription()->moduleName() << "/'" << imcc->moduleDescription()->moduleLabel() << "'";
 
      if (imcc->type() == ParentContext::Type::kInternal) {
        iost << " (probably inside some kind of mixing module)";
        imcc = imcc->internalContext()->moduleCallingContext();
      } else {
        imcc = imcc->moduleCallingContext();
      }
      ex.addContext(iost.str());
    }
    std::ostringstream ost;
    if (imcc->state() == ModuleCallingContext::State::kPrefetching) {
      ost << "Prefetching for module ";
    } else {
      ost << "Calling method for module ";
    }
    ost << imcc->moduleDescription()->moduleName() << "/'" << imcc->moduleDescription()->moduleLabel() << "'";
    ex.addContext(ost.str());
 
    if (imcc->type() == ParentContext::Type::kPlaceInPath) {
      ost.str("");
      ost << "Running path '";
      ost << imcc->placeInPathContext()->pathContext()->pathName() << "'";
      ex.addContext(ost.str());
      auto streamContext = imcc->placeInPathContext()->pathContext()->streamContext();
      if (streamContext) {
        ost.str("");
        edm::exceptionContext(ost, *streamContext);
        ex.addContext(ost.str());
      }
    } else {
      if (imcc->type() == ParentContext::Type::kStream) {
        ost.str("");
        edm::exceptionContext(ost, *(imcc->streamContext()));
        ex.addContext(ost.str());
      } else if (imcc->type() == ParentContext::Type::kGlobal) {
        ost.str("");
        edm::exceptionContext(ost, *(imcc->globalContext()));
        ex.addContext(ost.str());
      }
    }
  }

References cms::Exception::addContext(), edm::exceptionContext(), edm::ModuleCallingContext::globalContext(), edm::ModuleCallingContext::internalContext(), edm::ParentContext::kGlobal, edm::ParentContext::kInternal, edm::ParentContext::kModule, edm::ParentContext::kPlaceInPath, edm::ModuleCallingContext::kPrefetching, edm::ParentContext::kStream, edm::InternalContext::moduleCallingContext(), edm::ModuleCallingContext::moduleCallingContext(), edm::ModuleCallingContext::moduleDescription(), edm::ModuleDescription::moduleLabel(), edm::ModuleDescription::moduleName(), or, edm::PlaceInPathContext::pathContext(), edm::PathContext::pathName(), edm::ModuleCallingContext::placeInPathContext(), edm::ModuleCallingContext::state(), edm::PathContext::streamContext(), edm::ModuleCallingContext::streamContext(), and edm::ModuleCallingContext::type().

Referenced by handleExternalWorkException(), runAcquire(), and runModule().

globalLuminosityBlocksQueue()

virtual SerialTaskQueue* edm::Worker::globalLuminosityBlocksQueue ( )

pure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalEnd > >::enableGlobalQueue(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalBegin > >::pauseGlobalQueue().

globalRunsQueue()

virtual SerialTaskQueue* edm::Worker::globalRunsQueue ( )

pure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalEnd > >::enableGlobalQueue(), and edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalBegin > >::pauseGlobalQueue().

handleExternalWorkException()

std::exception_ptr edm::Worker::handleExternalWorkException ( std::exception_ptr const *  iEPtr, ParentContext const &  parentContext  )

private

Definition at line 422 of file Worker.cc.

                                                                                             {
    if (ranAcquireWithoutException_) {
      try {
        convertException::wrap([iEPtr]() { std::rethrow_exception(*iEPtr); });
      } catch (cms::Exception& ex) {
        ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
        exceptionContext(ex, &moduleCallingContext_);
        return std::current_exception();
      }
    }
    return *iEPtr;
  }

References exceptionContext(), moduleCallingContext_, ranAcquireWithoutException_, and edm::convertException::wrap().

hasAccumulator()

virtual bool edm::Worker::hasAccumulator ( ) const

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::UnscheduledCallProducer::addWorker().

hasAcquire()

virtual bool edm::Worker::hasAcquire ( ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by doWorkAsync(), and edm::Worker::RunModuleTask< T >::execute().

implBeginJob()

virtual void edm::Worker::implBeginJob ( )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by beginJob().

implBeginStream()

virtual void edm::Worker::implBeginStream ( StreamID  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by beginStream().

implDo()

virtual bool edm::Worker::implDo ( EventPrincipal const &  , EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< EventPrincipal, BranchActionStreamBegin > >::call().

implDoAcquire()

virtual void edm::Worker::implDoAcquire ( EventPrincipal const &  , EventSetupImpl const &  c, ModuleCallingContext const *  mcc, WaitingTaskWithArenaHolder &  holder  )

protectedpure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by runAcquire().

implDoBegin() [1/2]

virtual bool edm::Worker::implDoBegin ( LuminosityBlockPrincipal const &  lbp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoBegin() [2/2]

virtual bool edm::Worker::implDoBegin ( RunPrincipal const &  rp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalBegin > >::call(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalBegin > >::call().

implDoEnd() [1/2]

virtual bool edm::Worker::implDoEnd ( LuminosityBlockPrincipal const &  lbp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoEnd() [2/2]

virtual bool edm::Worker::implDoEnd ( RunPrincipal const &  rp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalEnd > >::call(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalEnd > >::call().

implDoPrePrefetchSelection()

virtual bool edm::Worker::implDoPrePrefetchSelection ( StreamID  id, EventPrincipal const &  ep, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by prePrefetchSelectionAsync().

implDoStreamBegin() [1/2]

virtual bool edm::Worker::implDoStreamBegin ( StreamID  id, LuminosityBlockPrincipal const &  lbp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoStreamBegin() [2/2]

virtual bool edm::Worker::implDoStreamBegin ( StreamID  id, RunPrincipal const &  rp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamBegin > >::call(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamBegin > >::call().

implDoStreamEnd() [1/2]

virtual bool edm::Worker::implDoStreamEnd ( StreamID  id, LuminosityBlockPrincipal const &  lbp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoStreamEnd() [2/2]

virtual bool edm::Worker::implDoStreamEnd ( StreamID  id, RunPrincipal const &  rp, EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamEnd > >::call(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamEnd > >::call().

implEndJob()

virtual void edm::Worker::implEndJob ( )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by endJob().

implEndStream()

virtual void edm::Worker::implEndStream ( StreamID  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by endStream().

implNeedToRunSelection()

virtual bool edm::Worker::implNeedToRunSelection ( ) const

protectedpure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< EventPrincipal, BranchActionStreamBegin > >::needToRunSelection().

implRegisterThinnedAssociations()

virtual void edm::Worker::implRegisterThinnedAssociations ( ProductRegistry const &  , ThinnedAssociationsHelper &    )

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by registerThinnedAssociations().

implRespondToCloseInputFile()

virtual void edm::Worker::implRespondToCloseInputFile ( FileBlock const &  fb )

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by respondToCloseInputFile().

implRespondToOpenInputFile()

virtual void edm::Worker::implRespondToOpenInputFile ( FileBlock const &  fb )

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by respondToOpenInputFile().

itemsMayGet()

virtual void edm::Worker::itemsMayGet ( BranchType  , std::vector< ProductResolverIndexAndSkipBit > &    ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

itemsShouldPutInEvent()

virtual std::vector<ProductResolverIndex> const& edm::Worker::itemsShouldPutInEvent ( ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

itemsToGet()

virtual void edm::Worker::itemsToGet ( BranchType  , std::vector< ProductResolverIndexAndSkipBit > &    ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

itemsToGetForSelection()

virtual void edm::Worker::itemsToGetForSelection ( std::vector< ProductResolverIndexAndSkipBit > &  ) const

protectedpure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by prePrefetchSelectionAsync().

itemsToGetFrom()

virtual std::vector<ProductResolverIndexAndSkipBit> const& edm::Worker::itemsToGetFrom ( BranchType  ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by prefetchAsync().

modulesWhoseProductsAreConsumed()

virtual void edm::Worker::modulesWhoseProductsAreConsumed ( std::vector< ModuleDescription const * > &  modules, ProductRegistry const &  preg, std::map< std::string, ModuleDescription const * > const &  labelsToDesc  ) const

pure virtual

Implemented in edm::WorkerT< T >.

moduleType()

virtual Types edm::Worker::moduleType ( ) const

pure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by edm::WorkerManager::addToUnscheduledWorkers(), and edm::StreamSchedule::fillWorkers().

operator=()

Worker& edm::Worker::operator= ( Worker const &  )

delete

postDoEvent()

void edm::Worker::postDoEvent

(

EventPrincipal const &

iEvent

)

Definition at line 372 of file Worker.cc.

                                                       {
    if (earlyDeleteHelper_) {
      earlyDeleteHelper_->moduleRan(iEvent);
    }
  }

References earlyDeleteHelper_, and iEvent.

preActionBeforeRunEventAsync()

virtual void edm::Worker::preActionBeforeRunEventAsync ( WaitingTask *  iTask, ModuleCallingContext const &  moduleCallingContext, Principal const &  iPrincipal  ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by prefetchAsync().

prefetchAsync()

void edm::Worker::prefetchAsync ( WaitingTask *  iTask, ServiceToken const &  token, ParentContext const &  parentContext, Principal const &  iPrincipal  )

private

Definition at line 199 of file Worker.cc.

                                                          {
    // Prefetch products the module declares it consumes (not including the products it maybe consumes)
    std::vector<ProductResolverIndexAndSkipBit> const& items = itemsToGetFrom(iPrincipal.branchType());
 
    moduleCallingContext_.setContext(ModuleCallingContext::State::kPrefetching, parentContext, nullptr);
 
    if (iPrincipal.branchType() == InEvent) {
      actReg_->preModuleEventPrefetchingSignal_.emit(*moduleCallingContext_.getStreamContext(), moduleCallingContext_);
    }
 
    //Need to be sure the ref count isn't set to 0 immediately
    iTask->increment_ref_count();
    for (auto const& item : items) {
      ProductResolverIndex productResolverIndex = item.productResolverIndex();
      bool skipCurrentProcess = item.skipCurrentProcess();
      if (productResolverIndex != ProductResolverIndexAmbiguous) {
        iPrincipal.prefetchAsync(iTask, productResolverIndex, skipCurrentProcess, token, &moduleCallingContext_);
      }
    }
 
    if (iPrincipal.branchType() == InEvent) {
      preActionBeforeRunEventAsync(iTask, moduleCallingContext_, iPrincipal);
    }
 
    if (0 == iTask->decrement_ref_count()) {
      //if everything finishes before we leave this routine, we need to launch the task
      tbb::task::spawn(*iTask);
    }
  }

References actReg_, edm::Principal::branchType(), edm::ModuleCallingContext::getStreamContext(), edm::InEvent, B2GTnPMonitor_cfi::item, mps_monitormerge::items, itemsToGetFrom(), edm::ModuleCallingContext::kPrefetching, moduleCallingContext_, preActionBeforeRunEventAsync(), edm::Principal::prefetchAsync(), edm::ProductResolverIndexAmbiguous, edm::ModuleCallingContext::setContext(), and unpackBuffers-CaloStage2::token.

Referenced by doWork(), and doWorkAsync().

prePrefetchSelectionAsync() [1/2]

void edm::Worker::prePrefetchSelectionAsync	(	WaitingTask *	task,
		ServiceToken const &	token,
		StreamID	stream,
		EventPrincipal const *	iPrincipal
	)

Definition at line 232 of file Worker.cc.

                                                                           {
    successTask->increment_ref_count();
 
    auto choiceTask = edm::make_waiting_task(
        tbb::task::allocate_root(), [id, successTask, iPrincipal, this, token](std::exception_ptr const*) {
          ServiceRegistry::Operate guard(token);
          // There is no reasonable place to rethrow, and implDoPrePrefetchSelection() should not throw in the first place.
          CMS_SA_ALLOW try {
            if (not implDoPrePrefetchSelection(id, *iPrincipal, &moduleCallingContext_)) {
              timesRun_.fetch_add(1, std::memory_order_relaxed);
              setPassed<true>();
              waitingTasks_.doneWaiting(nullptr);
              //TBB requires that destroyed tasks have count 0
              if (0 == successTask->decrement_ref_count()) {
                tbb::task::destroy(*successTask);
              }
              return;
            }
          } catch (...) {
          }
          if (0 == successTask->decrement_ref_count()) {
            tbb::task::spawn(*successTask);
          }
        });
 
    WaitingTaskHolder choiceHolder{choiceTask};
 
    std::vector<ProductResolverIndexAndSkipBit> items;
    itemsToGetForSelection(items);
 
    for (auto const& item : items) {
      ProductResolverIndex productResolverIndex = item.productResolverIndex();
      bool skipCurrentProcess = item.skipCurrentProcess();
      if (productResolverIndex != ProductResolverIndexAmbiguous) {
        iPrincipal->prefetchAsync(choiceTask, productResolverIndex, skipCurrentProcess, token, &moduleCallingContext_);
      }
    }
    choiceHolder.doneWaiting(std::exception_ptr{});
  }

References CMS_SA_ALLOW, edm::WaitingTaskList::doneWaiting(), implDoPrePrefetchSelection(), B2GTnPMonitor_cfi::item, mps_monitormerge::items, itemsToGetForSelection(), edm::make_waiting_task(), moduleCallingContext_, edm::Principal::prefetchAsync(), edm::ProductResolverIndexAmbiguous, timesRun_, unpackBuffers-CaloStage2::token, and waitingTasks_.

Referenced by doWork(), and doWorkAsync().

prePrefetchSelectionAsync() [2/2]

void edm::Worker::prePrefetchSelectionAsync ( WaitingTask *  task, ServiceToken const &  , StreamID  stream, void const *    )

inline

Definition at line 138 of file Worker.h.

                                                                                                         {
      assert(false);
    }

References cms::cuda::assert().

registerThinnedAssociations()

void edm::Worker::registerThinnedAssociations ( ProductRegistry const &  registry, ThinnedAssociationsHelper &  helper  )

inline

Definition at line 176 of file Worker.h.

                                                                                                         {
      implRegisterThinnedAssociations(registry, helper);
    }

References implRegisterThinnedAssociations().

reset()

void edm::Worker::reset ( void  )

inline

Definition at line 180 of file Worker.h.

                 {
      cached_exception_ = std::exception_ptr();
      state_ = Ready;
      waitingTasks_.reset();
      workStarted_ = false;
      numberOfPathsLeftToRun_ = numberOfPathsOn_;
    }

References cached_exception_, numberOfPathsLeftToRun_, numberOfPathsOn_, Ready, edm::WaitingTaskList::reset(), state_, waitingTasks_, and workStarted_.

Referenced by edm::WorkerManager::resetAll().

resetModuleDescription()

void edm::Worker::resetModuleDescription ( ModuleDescription const *  iDesc )

protected

Definition at line 277 of file Worker.cc.

                                                                    {
    ModuleCallingContext temp(iDesc,
                              moduleCallingContext_.state(),
                              moduleCallingContext_.parent(),
                              moduleCallingContext_.previousModuleOnThread());
    moduleCallingContext_ = temp;
  }

References moduleCallingContext_, edm::ModuleCallingContext::parent(), edm::ModuleCallingContext::previousModuleOnThread(), edm::ModuleCallingContext::state(), and groupFilesInBlocks::temp.

Referenced by edm::WorkerT< T >::setModule().

resolvePutIndicies()

virtual void edm::Worker::resolvePutIndicies ( BranchType  iBranchType, std::unordered_multimap< std::string, std::tuple< TypeID const *, const char *, edm::ProductResolverIndex >> const &  iIndicies  )

pure virtual

Implemented in edm::WorkerT< T >.

respondToCloseInputFile()

void edm::Worker::respondToCloseInputFile ( FileBlock const &  fb )

inline

Definition at line 174 of file Worker.h.

{ implRespondToCloseInputFile(fb); }

References benchmark_cfg::fb, and implRespondToCloseInputFile().

Referenced by edm::Schedule::respondToCloseInputFile().

respondToOpenInputFile()

void edm::Worker::respondToOpenInputFile ( FileBlock const &  fb )

inline

Definition at line 173 of file Worker.h.

{ implRespondToOpenInputFile(fb); }

References benchmark_cfg::fb, and implRespondToOpenInputFile().

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

runAcquire()

void edm::Worker::runAcquire ( EventPrincipal const &  ep, EventSetupImpl const &  es, ParentContext const &  parentContext, WaitingTaskWithArenaHolder &  holder  )

private

Definition at line 378 of file Worker.cc.

                                                              {
    ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
    try {
      convertException::wrap([&]() { this->implDoAcquire(ep, es, &moduleCallingContext_, holder); });
    } catch (cms::Exception& ex) {
      exceptionContext(ex, &moduleCallingContext_);
      TransitionIDValue<EventPrincipal> idValue(ep);
      if (shouldRethrowException(std::current_exception(), parentContext, true, idValue)) {
        timesRun_.fetch_add(1, std::memory_order_relaxed);
        throw;
      }
    }
  }

References SiStripBadComponentsDQMServiceTemplate_cfg::ep, exceptionContext(), implDoAcquire(), moduleCallingContext_, shouldRethrowException(), timesRun_, and edm::convertException::wrap().

Referenced by runAcquireAfterAsyncPrefetch().

runAcquireAfterAsyncPrefetch()

void edm::Worker::runAcquireAfterAsyncPrefetch ( std::exception_ptr const *  iEPtr, EventPrincipal const &  ep, EventSetupImpl const &  es, ParentContext const &  parentContext, WaitingTaskWithArenaHolder  holder  )

private

Definition at line 395 of file Worker.cc.

                                                                               {
    ranAcquireWithoutException_ = false;
    std::exception_ptr exceptionPtr;
    if (iEPtr) {
      assert(*iEPtr);
      TransitionIDValue<EventPrincipal> idValue(ep);
      if (shouldRethrowException(*iEPtr, parentContext, true, idValue)) {
        exceptionPtr = *iEPtr;
      }
      moduleCallingContext_.setContext(ModuleCallingContext::State::kInvalid, ParentContext(), nullptr);
    } else {
      // Caught exception is propagated via WaitingTaskWithArenaHolder
      CMS_SA_ALLOW try {
        runAcquire(ep, es, parentContext, holder);
        ranAcquireWithoutException_ = true;
      } catch (...) {
        exceptionPtr = std::current_exception();
      }
    }
    // It is important this is after runAcquire completely finishes
    holder.doneWaiting(exceptionPtr);
  }

References cms::cuda::assert(), CMS_SA_ALLOW, edm::WaitingTaskWithArenaHolder::doneWaiting(), SiStripBadComponentsDQMServiceTemplate_cfg::ep, edm::ModuleCallingContext::kInvalid, moduleCallingContext_, ranAcquireWithoutException_, runAcquire(), edm::ModuleCallingContext::setContext(), and shouldRethrowException().

runModule()

template<typename T >

bool edm::Worker::runModule ( typename T::MyPrincipal const &  ep, EventSetupImpl const &  c, StreamID  stream, ParentContext const &  parentContext, typename T::Context const *  context  )

private

Definition at line 1096 of file Worker.h.

                                                           {
    //unscheduled producers should advance this
    //if (T::isEvent_) {
    //  ++timesVisited_;
    //}
    ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
    if constexpr (T::isEvent_) {
      timesRun_.fetch_add(1, std::memory_order_relaxed);
    }
 
    bool rc = true;
    try {
      convertException::wrap([&]() {
        rc = workerhelper::CallImpl<T>::call(this, streamID, ep, es, actReg_.get(), &moduleCallingContext_, context);
 
        if (rc) {
          setPassed<T::isEvent_>();
        } else {
          setFailed<T::isEvent_>();
        }
      });
    } catch (cms::Exception& ex) {
      exceptionContext(ex, &moduleCallingContext_);
      TransitionIDValue<typename T::MyPrincipal> idValue(ep);
      if (shouldRethrowException(std::current_exception(), parentContext, T::isEvent_, idValue)) {
        assert(not cached_exception_);
        setException<T::isEvent_>(std::current_exception());
        std::rethrow_exception(cached_exception_);
      } else {
        rc = setPassed<T::isEvent_>();
      }
    }
 
    return rc;
  }

References actReg_, cms::cuda::assert(), cached_exception_, SiStripBadComponentsDQMServiceTemplate_cfg::ep, exceptionContext(), moduleCallingContext_, shouldRethrowException(), timesRun_, and edm::convertException::wrap().

runModuleAfterAsyncPrefetch()

template<typename T >

std::exception_ptr edm::Worker::runModuleAfterAsyncPrefetch ( std::exception_ptr const *  iEPtr, typename T::MyPrincipal const &  ep, EventSetupImpl const &  es, StreamID  streamID, ParentContext const &  parentContext, typename T::Context const *  context  )

private

Definition at line 906 of file Worker.h.

                                                                                           {
    std::exception_ptr exceptionPtr;
    if (iEPtr) {
      assert(*iEPtr);
      TransitionIDValue<typename T::MyPrincipal> idValue(ep);
      if (shouldRethrowException(*iEPtr, parentContext, T::isEvent_, idValue)) {
        exceptionPtr = *iEPtr;
        setException<T::isEvent_>(exceptionPtr);
      } else {
        setPassed<T::isEvent_>();
      }
      moduleCallingContext_.setContext(ModuleCallingContext::State::kInvalid, ParentContext(), nullptr);
    } else {
      // Caught exception is propagated via WaitingTaskList
      CMS_SA_ALLOW try { runModule<T>(ep, es, streamID, parentContext, context); } catch (...) {
        exceptionPtr = std::current_exception();
      }
    }
    waitingTasks_.doneWaiting(exceptionPtr);
    return exceptionPtr;
  }

References cms::cuda::assert(), CMS_SA_ALLOW, edm::WaitingTaskList::doneWaiting(), SiStripBadComponentsDQMServiceTemplate_cfg::ep, edm::ModuleCallingContext::kInvalid, moduleCallingContext_, edm::ModuleCallingContext::setContext(), shouldRethrowException(), and waitingTasks_.

Referenced by edm::Worker::RunModuleTask< T >::execute().

runModuleDirectly()

template<typename T >

std::exception_ptr edm::Worker::runModuleDirectly	(	typename T::MyPrincipal const &	ep,
		EventSetupImpl const &	es,
		StreamID	streamID,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 1137 of file Worker.h.

                                                                                 {
    timesVisited_.fetch_add(1, std::memory_order_relaxed);
    std::exception_ptr const* prefetchingException = nullptr;  // null because there was no prefetching to do
    return runModuleAfterAsyncPrefetch<T>(prefetchingException, ep, es, streamID, parentContext, context);
  }

References SiStripBadComponentsDQMServiceTemplate_cfg::ep, and timesVisited_.

Referenced by edm::Path::finished().

serializeRunModule()

virtual TaskQueueAdaptor edm::Worker::serializeRunModule ( )

privatepure virtual

Implemented in edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, edm::WorkerT< T >, and edm::WorkerT< T >.

Referenced by doWork(), doWorkNoPrefetchingAsync(), and edm::Worker::RunModuleTask< T >::execute().

setActivityRegistry()

void edm::Worker::setActivityRegistry

(

std::shared_ptr< ActivityRegistry >

areg

)

The signals are required to live longer than the last call to 'doWork' this was done to improve performance based on profiling

Definition at line 102 of file Worker.cc.

{ actReg_ = areg; }

References actReg_.

setEarlyDeleteHelper()

void edm::Worker::setEarlyDeleteHelper

(

EarlyDeleteHelper *

iHelper

)

Definition at line 275 of file Worker.cc.

{ earlyDeleteHelper_ = iHelper; }

References earlyDeleteHelper_.

setException()

template<bool IS_EVENT>

std::exception_ptr edm::Worker::setException ( std::exception_ptr  iException )

inlineprivate

Definition at line 363 of file Worker.h.

                                                               {
      if (IS_EVENT) {
        timesExcept_.fetch_add(1, std::memory_order_relaxed);
      }
      cached_exception_ = iException;  // propagate_const<T> has no reset() function
      state_ = Exception;
      return cached_exception_;
    }

References cached_exception_, Exception, state_, and timesExcept_.

setFailed()

template<bool IS_EVENT>

bool edm::Worker::setFailed ( )

inlineprivate

Definition at line 354 of file Worker.h.

                     {
      if (IS_EVENT) {
        timesFailed_.fetch_add(1, std::memory_order_relaxed);
      }
      state_ = Fail;
      return false;
    }

References Fail, state_, and timesFailed_.

setPassed()

template<bool IS_EVENT>

bool edm::Worker::setPassed ( )

inlineprivate

Definition at line 345 of file Worker.h.

                     {
      if (IS_EVENT) {
        timesPassed_.fetch_add(1, std::memory_order_relaxed);
      }
      state_ = Pass;
      return true;
    }

References Pass, state_, and timesPassed_.

shouldRethrowException()

bool edm::Worker::shouldRethrowException ( std::exception_ptr  iPtr, ParentContext const &  parentContext, bool  isEvent, TransitionIDValueBase const &  iID  ) const

private

Definition at line 156 of file Worker.cc.

                                                                              {
    // NOTE: the warning printed as a result of ignoring or failing
    // a module will only be printed during the full true processing
    // pass of this module
 
    // Get the action corresponding to this exception.  However, if processing
    // something other than an event (e.g. run, lumi) always rethrow.
    if (not isEvent) {
      return true;
    }
    try {
      convertException::wrap([&]() { std::rethrow_exception(iPtr); });
    } catch (cms::Exception& ex) {
      exception_actions::ActionCodes action = actions_->find(ex.category());
 
      if (action == exception_actions::Rethrow) {
        return true;
      }
 
      ModuleCallingContext tempContext(&description(), ModuleCallingContext::State::kInvalid, parentContext, nullptr);
 
      // If we are processing an endpath and the module was scheduled, treat SkipEvent or FailPath
      // as IgnoreCompletely, so any subsequent OutputModules are still run.
      // For unscheduled modules only treat FailPath as IgnoreCompletely but still allow SkipEvent to throw
      ModuleCallingContext const* top_mcc = tempContext.getTopModuleCallingContext();
      if (top_mcc->type() == ParentContext::Type::kPlaceInPath &&
          top_mcc->placeInPathContext()->pathContext()->isEndPath()) {
        if ((action == exception_actions::SkipEvent && tempContext.type() == ParentContext::Type::kPlaceInPath) ||
            action == exception_actions::FailPath) {
          action = exception_actions::IgnoreCompletely;
        }
      }
      if (action == exception_actions::IgnoreCompletely) {
        edm::printCmsExceptionWarning("IgnoreCompletely", ex);
        return false;
      }
    }
    return true;
  }

References writedatasetfile::action, actions_, cms::Exception::category(), description(), edm::exception_actions::FailPath, edm::ExceptionToActionTable::find(), edm::ModuleCallingContext::getTopModuleCallingContext(), edm::exception_actions::IgnoreCompletely, edm::PathContext::isEndPath(), edm::ModuleCallingContext::kInvalid, edm::ParentContext::kPlaceInPath, edm::PlaceInPathContext::pathContext(), edm::ModuleCallingContext::placeInPathContext(), edm::printCmsExceptionWarning(), edm::exception_actions::Rethrow, edm::exception_actions::SkipEvent, edm::ModuleCallingContext::type(), and edm::convertException::wrap().

Referenced by doWork(), runAcquire(), runAcquireAfterAsyncPrefetch(), runModule(), and runModuleAfterAsyncPrefetch().

skipOnPath()

void edm::Worker::skipOnPath

(

EventPrincipal const &

iEvent

)

Definition at line 363 of file Worker.cc.

                                                      {
    if (earlyDeleteHelper_) {
      earlyDeleteHelper_->pathFinished(iEvent);
    }
    if (0 == --numberOfPathsLeftToRun_) {
      waitingTasks_.doneWaiting(cached_exception_);
    }
  }

References cached_exception_, edm::WaitingTaskList::doneWaiting(), earlyDeleteHelper_, iEvent, numberOfPathsLeftToRun_, and waitingTasks_.

Referenced by edm::WorkerInPath::skipWorker().

state()

State edm::Worker::state ( ) const

inline

Definition at line 232 of file Worker.h.

{ return state_; }

References state_.

Referenced by edm::WorkerInPath::checkResultsOfRunWorker(), and doWork().

timesExcept()

int edm::Worker::timesExcept ( ) const

inline

Definition at line 231 of file Worker.h.

{ return timesExcept_.load(std::memory_order_acquire); }

References timesExcept_.

timesFailed()

int edm::Worker::timesFailed ( ) const

inline

Definition at line 230 of file Worker.h.

{ return timesFailed_.load(std::memory_order_acquire); }

References timesFailed_.

timesPass()

int edm::Worker::timesPass ( ) const

inline

Definition at line 234 of file Worker.h.

{ return timesPassed(); }  // for backward compatibility only - to be removed soon

References timesPassed().

timesPassed()

int edm::Worker::timesPassed ( ) const

inline

Definition at line 229 of file Worker.h.

{ return timesPassed_.load(std::memory_order_acquire); }

References timesPassed_.

Referenced by timesPass().

timesRun()

int edm::Worker::timesRun ( ) const

inline

Definition at line 227 of file Worker.h.

{ return timesRun_.load(std::memory_order_acquire); }

References timesRun_.

timesVisited()

int edm::Worker::timesVisited ( ) const

inline

Definition at line 228 of file Worker.h.

{ return timesVisited_.load(std::memory_order_acquire); }

References timesVisited_.

updateLookup() [1/2]

virtual void edm::Worker::updateLookup ( BranchType  iBranchType, ProductResolverIndexHelper const &    )

pure virtual

Implemented in edm::WorkerT< T >.

updateLookup() [2/2]

virtual void edm::Worker::updateLookup ( eventsetup::ESRecordsToProxyIndices const &  )

pure virtual

Implemented in edm::WorkerT< T >.

wantsGlobalLuminosityBlocks()

virtual bool edm::Worker::wantsGlobalLuminosityBlocks ( ) const

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalBegin > >::wantsTransition(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalEnd > >::wantsTransition().

wantsGlobalRuns()

virtual bool edm::Worker::wantsGlobalRuns ( ) const

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalBegin > >::wantsTransition(), and edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalEnd > >::wantsTransition().

wantsStreamLuminosityBlocks()

virtual bool edm::Worker::wantsStreamLuminosityBlocks ( ) const

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamBegin > >::wantsTransition(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamEnd > >::wantsTransition().

wantsStreamRuns()

virtual bool edm::Worker::wantsStreamRuns ( ) const

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamBegin > >::wantsTransition(), and edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamEnd > >::wantsTransition().

workerType()

virtual std::string edm::Worker::workerType ( ) const

protectedpure virtual

Implemented in edm::WorkerT< T >.

Friends And Related Function Documentation

workerhelper::CallImpl

template<typename O >

friend class workerhelper::CallImpl

friend

Definition at line 240 of file Worker.h.

Member Data Documentation

actions_

ExceptionToActionTable const* edm::Worker::actions_

private

Definition at line 615 of file Worker.h.

Referenced by shouldRethrowException().

actReg_

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

private

Definition at line 618 of file Worker.h.

Referenced by activityRegistry(), beginJob(), beginStream(), doWork(), emitPostModuleEventPrefetchingSignal(), endJob(), endStream(), prefetchAsync(), runModule(), and setActivityRegistry().

cached_exception_

std::exception_ptr edm::Worker::cached_exception_

private

Definition at line 616 of file Worker.h.

Referenced by doWork(), reset(), runModule(), setException(), and skipOnPath().

earlyDeleteHelper_

edm::propagate_const<EarlyDeleteHelper*> edm::Worker::earlyDeleteHelper_

private

Definition at line 620 of file Worker.h.

Referenced by postDoEvent(), setEarlyDeleteHelper(), and skipOnPath().

moduleCallingContext_

ModuleCallingContext edm::Worker::moduleCallingContext_

private

Definition at line 613 of file Worker.h.

Referenced by beginStream(), descPtr(), description(), doWork(), doWorkAsync(), emitPostModuleEventPrefetchingSignal(), endStream(), handleExternalWorkException(), prefetchAsync(), prePrefetchSelectionAsync(), resetModuleDescription(), runAcquire(), runAcquireAfterAsyncPrefetch(), runModule(), and runModuleAfterAsyncPrefetch().

numberOfPathsLeftToRun_

std::atomic<int> edm::Worker::numberOfPathsLeftToRun_

private

Definition at line 611 of file Worker.h.

Referenced by reset(), and skipOnPath().

numberOfPathsOn_

int edm::Worker::numberOfPathsOn_

private

Definition at line 610 of file Worker.h.

Referenced by addedToPath(), and reset().

ranAcquireWithoutException_

bool edm::Worker::ranAcquireWithoutException_

private

Definition at line 624 of file Worker.h.

Referenced by handleExternalWorkException(), and runAcquireAfterAsyncPrefetch().

state_

std::atomic<State> edm::Worker::state_

private

Definition at line 609 of file Worker.h.

Referenced by beginJob(), beginStream(), doWork(), endJob(), endStream(), reset(), setException(), setFailed(), setPassed(), and state().

timesExcept_

std::atomic<int> edm::Worker::timesExcept_

private

Definition at line 608 of file Worker.h.

Referenced by clearCounters(), setException(), and timesExcept().

timesFailed_

std::atomic<int> edm::Worker::timesFailed_

private

Definition at line 607 of file Worker.h.

Referenced by clearCounters(), setFailed(), and timesFailed().

timesPassed_

std::atomic<int> edm::Worker::timesPassed_

private

Definition at line 606 of file Worker.h.

Referenced by clearCounters(), setPassed(), and timesPassed().

timesRun_

std::atomic<int> edm::Worker::timesRun_

private

Definition at line 604 of file Worker.h.

Referenced by clearCounters(), prePrefetchSelectionAsync(), runAcquire(), runModule(), and timesRun().

timesVisited_

std::atomic<int> edm::Worker::timesVisited_

private

Definition at line 605 of file Worker.h.

Referenced by clearCounters(), doWork(), doWorkAsync(), runModuleDirectly(), and timesVisited().

waitingTasks_

edm::WaitingTaskList edm::Worker::waitingTasks_

private

Definition at line 622 of file Worker.h.

Referenced by callWhenDoneAsync(), doWork(), doWorkAsync(), doWorkNoPrefetchingAsync(), prePrefetchSelectionAsync(), reset(), runModuleAfterAsyncPrefetch(), and skipOnPath().

workStarted_

std::atomic<bool> edm::Worker::workStarted_

private

Definition at line 623 of file Worker.h.

Referenced by doWork(), doWorkAsync(), doWorkNoPrefetchingAsync(), and reset().