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

Public Types
enum	ConcurrencyTypes {   kGlobal, kLimited, kOne, kStream,   kLegacy }
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 (WaitingTaskHolder task)
void	clearCounters ()
void	clearModule ()
virtual std::vector< ConsumesInfo >	consumesInfo () const =0
virtual void	convertCurrentProcessAlias (std::string const &processName)=0
ModuleDescription const *	description () const
template<typename T >
void	doWorkAsync (WaitingTaskHolder, typename T::TransitionInfoType const &, ServiceToken const &, StreamID, ParentContext const &, typename T::Context const *)
template<typename T >
void	doWorkNoPrefetchingAsync (WaitingTaskHolder, typename T::TransitionInfoType const &, ServiceToken const &, StreamID, ParentContext const &, typename T::Context const *)
void	endJob ()
void	endStream (StreamID id, StreamContext &streamContext)
virtual SerialTaskQueue *	globalLuminosityBlocksQueue ()=0
virtual SerialTaskQueue *	globalRunsQueue ()=0
virtual bool	hasAccumulator () const =0
virtual ConcurrencyTypes	moduleConcurrencyType () const =0
virtual void	modulesWhoseProductsAreConsumed (std::array< std::vector< ModuleDescription const * > *, NumBranchTypes > &modules, std::vector< ModuleProcessName > &modulesInPreviousProcesses, 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 (tbb::task_group &, WaitingTask *task, ServiceToken const &, StreamID stream, EventPrincipal const *)
void	prePrefetchSelectionAsync (tbb::task_group &, 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	respondToCloseOutputFile ()
void	respondToOpenInputFile (FileBlock const &fb)
template<typename T >
std::exception_ptr	runModuleDirectly (typename T::TransitionInfoType const &, StreamID, ParentContext const &, typename T::Context const *)
virtual void	selectInputProcessBlocks (ProductRegistry const &, ProcessBlockHelperBase const &)=0
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	wantsInputProcessBlocks () const =0
virtual bool	wantsProcessBlocks () 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	doClearModule ()=0
virtual void	implBeginJob ()=0
virtual void	implBeginStream (StreamID)=0
virtual bool	implDo (EventTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoAccessInputProcessBlock (ProcessBlockPrincipal const &, ModuleCallingContext const *)=0
virtual void	implDoAcquire (EventTransitionInfo const &, ModuleCallingContext const *, WaitingTaskWithArenaHolder &)=0
virtual bool	implDoBegin (RunTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoBegin (LumiTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoBeginProcessBlock (ProcessBlockPrincipal const &, ModuleCallingContext const *)=0
virtual bool	implDoEnd (RunTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoEnd (LumiTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoEndProcessBlock (ProcessBlockPrincipal const &, ModuleCallingContext const *)=0
virtual bool	implDoPrePrefetchSelection (StreamID, EventPrincipal const &, ModuleCallingContext const *)=0
virtual bool	implDoStreamBegin (StreamID, RunTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoStreamBegin (StreamID, LumiTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoStreamEnd (StreamID, RunTransitionInfo const &, ModuleCallingContext const *)=0
virtual bool	implDoStreamEnd (StreamID, LumiTransitionInfo const &, ModuleCallingContext const *)=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	edPrefetchAsync (WaitingTaskHolder, ServiceToken const &, Principal const &) const
void	emitPostModuleEventPrefetchingSignal ()
virtual std::vector < ESProxyIndex > const &	esItemsToGetFrom (Transition) const =0
void	esPrefetchAsync (WaitingTaskHolder, EventSetupImpl const &, Transition, ServiceToken const &)
virtual std::vector < ESRecordIndex > const &	esRecordsToGetFrom (Transition) const =0
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	implRespondToCloseOutputFile ()=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
bool	needsESPrefetching (Transition iTrans) const noexcept
virtual void	preActionBeforeRunEventAsync (WaitingTaskHolder iTask, ModuleCallingContext const &moduleCallingContext, Principal const &iPrincipal) const =0
template<typename T >
void	prefetchAsync (WaitingTaskHolder, ServiceToken const &, ParentContext const &, typename T::TransitionInfoType const &, Transition)
void	runAcquire (EventTransitionInfo const &, ParentContext const &, WaitingTaskWithArenaHolder &)
void	runAcquireAfterAsyncPrefetch (std::exception_ptr const *, EventTransitionInfo const &, ParentContext const &, WaitingTaskWithArenaHolder)
template<typename T >
bool	runModule (typename T::TransitionInfoType const &, StreamID, ParentContext const &, typename T::Context const *)
template<typename T >
std::exception_ptr	runModuleAfterAsyncPrefetch (std::exception_ptr const *, typename T::TransitionInfoType const &, StreamID, ParentContext const &, typename T::Context const *)
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) const

Private Attributes
ExceptionToActionTable const *	actions_
std::shared_ptr< ActivityRegistry >	actReg_
std::exception_ptr	cached_exception_
edm::propagate_const < EarlyDeleteHelper * >	earlyDeleteHelper_
ModuleCallingContext	moduleCallingContext_
bool	moduleValid_ = true
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 91 of file Worker.h.

Member Enumeration Documentation

enum edm::Worker::ConcurrencyTypes

Enumerator
kGlobal
kLimited
kOne
kStream
kLegacy

Definition at line 95 of file Worker.h.

{ kGlobal, kLimited, kOne, kStream, kLegacy };

enum edm::Worker::State

Enumerator
Ready
Pass
Fail
Exception

Definition at line 93 of file Worker.h.

{ Ready, Pass, Fail, Exception };

enum edm::Worker::Types

Enumerator
kAnalyzer
kFilter
kProducer
kOutputModule

Definition at line 94 of file Worker.h.

{ kAnalyzer, kFilter, kProducer, kOutputModule };

Constructor & Destructor Documentation

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

Definition at line 91 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) {}

edm::Worker::~Worker ( )

virtual

Definition at line 108 of file Worker.cc.

{}

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

delete

Member Function Documentation

ActivityRegistry* edm::Worker::activityRegistry ( )

inlineprotected

Definition at line 278 of file Worker.h.

References actReg_.

{ return actReg_.get(); }

void edm::Worker::addedToPath ( )

inline

Definition at line 230 of file Worker.h.

References numberOfPathsOn_.

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

{ ++numberOfPathsOn_; }

void edm::Worker::beginJob

(

void

)

Definition at line 295 of file Worker.cc.

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

Referenced by edm::WorkerManager::beginJob(), and edm::GlobalSchedule::replaceModule().

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

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

Definition at line 328 of file Worker.cc.

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().

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

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

void edm::Worker::callWhenDoneAsync ( WaitingTaskHolder  task )

inline

Definition at line 167 of file Worker.h.

References edm::WaitingTaskList::add(), eostools::move(), and waitingTasks_.

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

{ waitingTasks_.add(std::move(task)); }

void edm::Worker::clearCounters ( )

inline

Definition at line 222 of file Worker.h.

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

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

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

void edm::Worker::clearModule ( )

inline

Definition at line 122 of file Worker.h.

References doClearModule(), and moduleValid_.

                       {
      moduleValid_ = false;
      doClearModule();
    }

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

pure virtual

Implemented in edm::WorkerT< T >.

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

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

pure virtual

Implemented in edm::WorkerT< T >.

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

inline

Definition at line 188 of file Worker.h.

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

Referenced by beginJob(), beginStream(), endJob(), endStream(), edm::fillModuleInPathSummary(), edm::StreamSchedule::fillWorkers(), edm::fillWorkerSummaryAux(), registerThinnedAssociations(), and shouldRethrowException().

                                                 {
      if (moduleValid_) {
        return moduleCallingContext_.moduleDescription();
      }
      return nullptr;
    }

virtual void edm::Worker::doClearModule ( )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by clearModule().

template<typename T >

void edm::Worker::doWorkAsync	(	WaitingTaskHolder	task,
		typename T::TransitionInfoType const &	transitionInfo,
		ServiceToken const &	token,
		StreamID	streamID,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 931 of file Worker.h.

References edm::WaitingTaskList::add(), visDQMUpload::context, edm::WaitingTaskHolder::group(), watchdog::group, hasAcquire(), info(), edm::ModuleCallingContext::kPrefetching, edm::make_waiting_task(), moduleCallingContext_, eostools::move(), AlCaHLTBitMon_ParallelJobs::p, prePrefetchSelectionAsync(), fetchall_from_DQM_v2::release, alignCSCRings::s, edm::ModuleCallingContext::setContext(), submitPVValidationJobs::t, timesVisited_, unpackBuffers-CaloStage2::token, waitingTasks_, and workStarted_.

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

                                                             {
    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
        WaitingTask* moduleTask =
            new RunModuleTask<T>(this, transitionInfo, token, streamID, parentContext, context, task.group());

        //make sure the task is either run or destroyed
        struct DestroyTask {
          DestroyTask(edm::WaitingTask* iTask) : m_task(iTask) {}

          ~DestroyTask() {
            auto p = m_task.exchange(nullptr);
            if (p) {
              TaskSentry s{p};
            }
          }

          edm::WaitingTask* release() { return m_task.exchange(nullptr); }

        private:
          std::atomic<edm::WaitingTask*> m_task;
        };
        if constexpr (T::isEvent_) {
          if (hasAcquire()) {
            auto ownRunTask = std::make_shared<DestroyTask>(moduleTask);
            ServiceWeakToken weakToken = token;
            auto* group = task.group();
            moduleTask = make_waiting_task(
                [this, weakToken, transitionInfo, parentContext, ownRunTask, group](std::exception_ptr const* iExcept) {
                  WaitingTaskWithArenaHolder runTaskHolder(
                      *group, new HandleExternalWorkExceptionTask(this, group, ownRunTask->release(), parentContext));
                  AcquireTask<T> t(this, transitionInfo, weakToken.lock(), parentContext, runTaskHolder);
                  t.execute();
                });
          }
        }
        auto* group = task.group();
        auto ownModuleTask = std::make_shared<DestroyTask>(moduleTask);
        ServiceWeakToken weakToken = token;
        auto selectionTask =
            make_waiting_task([ownModuleTask, parentContext, info = transitionInfo, weakToken, group, this](
                                  std::exception_ptr const*) mutable {
              ServiceRegistry::Operate guard(weakToken.lock());
              prefetchAsync<T>(WaitingTaskHolder(*group, ownModuleTask->release()),
                               weakToken.lock(),
                               parentContext,
                               info,
                               T::transition_);
            });
        prePrefetchSelectionAsync(*group, selectionTask, token, streamID, &transitionInfo.principal());
      } else {
        WaitingTask* moduleTask =
            new RunModuleTask<T>(this, transitionInfo, token, streamID, parentContext, context, task.group());
        auto group = task.group();
        if constexpr (T::isEvent_) {
          if (hasAcquire()) {
            WaitingTaskWithArenaHolder runTaskHolder(
                *group, new HandleExternalWorkExceptionTask(this, group, moduleTask, parentContext));
            moduleTask = new AcquireTask<T>(this, transitionInfo, token, parentContext, std::move(runTaskHolder));
          }
        }
        prefetchAsync<T>(WaitingTaskHolder(*group, moduleTask), token, parentContext, transitionInfo, T::transition_);
      }
    }
  }

template<typename T >

void edm::Worker::doWorkNoPrefetchingAsync	(	WaitingTaskHolder	task,
		typename T::TransitionInfoType const &	transitionInfo,
		ServiceToken const &	serviceToken,
		StreamID	streamID,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 1047 of file Worker.h.

References edm::WaitingTaskList::add(), CMS_SA_ALLOW, visDQMUpload::context, edm::WaitingTaskList::doneWaiting(), esPrefetchAsync(), edm::WaitingTaskHolder::group(), watchdog::group, info(), edm::ServiceWeakToken::lock(), edm::make_waiting_task(), eostools::move(), needsESPrefetching(), serializeRunModule(), waitingTasks_, and workStarted_.

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

                                                                          {
    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) {
      ServiceWeakToken weakToken = serviceToken;
      auto toDo = [this, info = transitionInfo, streamID, parentContext, context, weakToken]() {
        std::exception_ptr exceptionPtr;
        // Caught exception is propagated via WaitingTaskList
        CMS_SA_ALLOW try {
          //Need to make the services available
          ServiceRegistry::Operate guard(weakToken.lock());

          this->runModule<T>(info, streamID, parentContext, context);
        } catch (...) {
          exceptionPtr = std::current_exception();
        }
        this->waitingTasks_.doneWaiting(exceptionPtr);
      };

      if (needsESPrefetching(T::transition_)) {
        auto group = task.group();
        auto afterPrefetch =
            edm::make_waiting_task([toDo = std::move(toDo), group, this](std::exception_ptr const* iExcept) {
              if (iExcept) {
                this->waitingTasks_.doneWaiting(*iExcept);
              } else {
                if (auto queue = this->serializeRunModule()) {
                  queue.push(*group, toDo);
                } else {
                  group->run(toDo);
                }
              }
            });
        esPrefetchAsync(
            WaitingTaskHolder(*group, afterPrefetch), transitionInfo.eventSetupImpl(), T::transition_, serviceToken);
      } else {
        auto group = task.group();
        if (auto queue = this->serializeRunModule()) {
          queue.push(*group, toDo);
        } else {
          group->run(toDo);
        }
      }
    }
  }

void edm::Worker::edPrefetchAsync ( WaitingTaskHolder  iTask, ServiceToken const &  token, Principal const &  iPrincipal  ) const

private

Definition at line 272 of file Worker.cc.

References edm::Principal::branchType(), B2GTnPMonitor_cfi::item, mps_monitormerge::items, itemsToGetFrom(), moduleCallingContext_, edm::Principal::prefetchAsync(), and edm::ProductResolverIndexAmbiguous.

Referenced by prefetchAsync().

                                                                                                                    {
    // Prefetch products the module declares it consumes
    std::vector<ProductResolverIndexAndSkipBit> const& items = itemsToGetFrom(iPrincipal.branchType());

    for (auto const& item : items) {
      ProductResolverIndex productResolverIndex = item.productResolverIndex();
      bool skipCurrentProcess = item.skipCurrentProcess();
      if (productResolverIndex != ProductResolverIndexAmbiguous) {
        iPrincipal.prefetchAsync(iTask, productResolverIndex, skipCurrentProcess, token, &moduleCallingContext_);
      }
    }
  }

void edm::Worker::emitPostModuleEventPrefetchingSignal ( )

inlineprivate

Definition at line 349 of file Worker.h.

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

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

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

void edm::Worker::endJob

(

void

)

Definition at line 311 of file Worker.cc.

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

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

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

Definition at line 352 of file Worker.cc.

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().

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

virtual std::vector<ESProxyIndex> const& edm::Worker::esItemsToGetFrom ( Transition  ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by esPrefetchAsync(), and needsESPrefetching().

void edm::Worker::esPrefetchAsync ( WaitingTaskHolder  iTask, EventSetupImpl const &  iImpl, Transition  iTrans, ServiceToken const &  iToken  )

private

Definition at line 201 of file Worker.cc.

References cms::cuda::assert(), esItemsToGetFrom(), esRecordsToGetFrom(), edm::EventSetupImpl::findImpl(), edm::WaitingTaskHolder::group(), watchdog::group, mps_fire::i, mps_monitormerge::items, edm::ServiceWeakToken::lock(), edm::make_waiting_task(), moduleCallingContext_, eostools::move(), edm::NumberOfEventSetupTransitions, edm::eventsetup::EventSetupRecordImpl::prefetchAsync(), edm::syncWait(), edm::EventSetupImpl::taskArena(), unpackBuffers-CaloStage2::token, and UNLIKELY.

Referenced by doWorkNoPrefetchingAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< EventPrincipal, BranchActionStreamBegin > >::esPrefetchAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalBegin > >::esPrefetchAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamBegin > >::esPrefetchAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalEnd > >::esPrefetchAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamEnd > >::esPrefetchAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalBegin > >::esPrefetchAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamBegin > >::esPrefetchAsync(), edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalEnd > >::esPrefetchAsync(), and edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamEnd > >::esPrefetchAsync().

                                                           {
    if (iTrans >= edm::Transition::NumberOfEventSetupTransitions) {
      return;
    }
    auto const& recs = esRecordsToGetFrom(iTrans);
    auto const& items = esItemsToGetFrom(iTrans);

    assert(items.size() == recs.size());
    if (items.empty()) {
      return;
    }

    //Thread case of 1 thread special. The master thread is doing a wait_for_all on the
    // default tbb arena. It will not process any tasks on the es arena. We need to add a
    // task that will synchronously do a wait_for_all in the es arena to be sure prefetching
    // will work.

    if UNLIKELY (tbb::global_control::active_value(tbb::global_control::max_allowed_parallelism) == 1) {
      auto taskGroup = iTask.group();
      ServiceWeakToken weakToken = iToken;
      taskGroup->run([this, task = std::move(iTask), iTrans, &iImpl, weakToken]() {
        std::exception_ptr exceptPtr{};
        iImpl.taskArena()->execute([this, iTrans, &iImpl, weakToken, &exceptPtr]() {
          exceptPtr = syncWait([&](WaitingTaskHolder&& iHolder) {
            auto const& recs = esRecordsToGetFrom(iTrans);
            auto const& items = esItemsToGetFrom(iTrans);
            auto hWaitTask = std::move(iHolder);
            auto token = weakToken.lock();
            for (size_t i = 0; i != items.size(); ++i) {
              if (recs[i] != ESRecordIndex{}) {
                auto rec = iImpl.findImpl(recs[i]);
                if (rec) {
                  rec->prefetchAsync(hWaitTask, items[i], &iImpl, token, ESParentContext(&moduleCallingContext_));
                }
              }
            }
          });  //syncWait
        });    //esTaskArena().execute
        //note use of a copy gets around declaring the lambda as mutable
        auto tempTask = task;
        tempTask.doneWaiting(exceptPtr);
      });  //group.run
    } else {
      auto group = iTask.group();
      //We need iTask to run in the default arena since it is not an ES task
      auto task = make_waiting_task(
          [holder = WaitingTaskWithArenaHolder(std::move(iTask))](std::exception_ptr const* iExcept) mutable {
            if (iExcept) {
              holder.doneWaiting(*iExcept);
            } else {
              holder.doneWaiting(std::exception_ptr{});
            }
          });

      WaitingTaskHolder tempH(*group, task);
      iImpl.taskArena()->execute([&]() {
        for (size_t i = 0; i != items.size(); ++i) {
          if (recs[i] != ESRecordIndex{}) {
            auto rec = iImpl.findImpl(recs[i]);
            if (rec) {
              rec->prefetchAsync(tempH, items[i], &iImpl, iToken, ESParentContext(&moduleCallingContext_));
            }
          }
        }
      });
    }
  }

virtual std::vector<ESRecordIndex> const& edm::Worker::esRecordsToGetFrom ( Transition  ) const

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by esPrefetchAsync().

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().

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().

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

private

Definition at line 440 of file Worker.cc.

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

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

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

pure virtual

Implemented in edm::WorkerT< T >.

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

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

privatepure virtual

Implemented in edm::WorkerT< T >.

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

virtual void edm::Worker::implBeginJob ( )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by beginJob().

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by beginStream().

virtual bool edm::Worker::implDo ( EventTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

virtual bool edm::Worker::implDoAccessInputProcessBlock ( ProcessBlockPrincipal const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< ProcessBlockPrincipal, BranchActionProcessBlockInput > >::call().

virtual void edm::Worker::implDoAcquire ( EventTransitionInfo const &  , ModuleCallingContext const *  , WaitingTaskWithArenaHolder &    )

protectedpure virtual

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

Referenced by runAcquire().

virtual bool edm::Worker::implDoBegin ( RunTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

virtual bool edm::Worker::implDoBegin ( LumiTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

virtual bool edm::Worker::implDoBeginProcessBlock ( ProcessBlockPrincipal const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< ProcessBlockPrincipal, BranchActionGlobalBegin > >::call().

virtual bool edm::Worker::implDoEnd ( RunTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

virtual bool edm::Worker::implDoEnd ( LumiTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

virtual bool edm::Worker::implDoEndProcessBlock ( ProcessBlockPrincipal const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< ProcessBlockPrincipal, BranchActionGlobalEnd > >::call().

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

protectedpure virtual

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

Referenced by prePrefetchSelectionAsync().

virtual bool edm::Worker::implDoStreamBegin ( StreamID  , RunTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

virtual bool edm::Worker::implDoStreamBegin ( StreamID  , LumiTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

virtual bool edm::Worker::implDoStreamEnd ( StreamID  , RunTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

virtual bool edm::Worker::implDoStreamEnd ( StreamID  , LumiTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

virtual void edm::Worker::implEndJob ( )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by endJob().

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by endStream().

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().

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

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by registerThinnedAssociations().

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

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by respondToCloseInputFile().

virtual void edm::Worker::implRespondToCloseOutputFile ( )

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by respondToCloseOutputFile().

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

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by respondToOpenInputFile().

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

privatepure virtual

Implemented in edm::WorkerT< T >.

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

privatepure virtual

Implemented in edm::WorkerT< T >.

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

privatepure virtual

Implemented in edm::WorkerT< T >.

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().

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

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by edPrefetchAsync().

virtual ConcurrencyTypes edm::Worker::moduleConcurrencyType ( ) 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 >.

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

pure virtual

Implemented in edm::WorkerT< T >.

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().

bool edm::Worker::needsESPrefetching ( Transition  iTrans ) const

inlineprivatenoexcept

Definition at line 345 of file Worker.h.

References esItemsToGetFrom(), and edm::NumberOfEventSetupTransitions.

Referenced by doWorkNoPrefetchingAsync().

                                                              {
      return iTrans < edm::Transition::NumberOfEventSetupTransitions ? not esItemsToGetFrom(iTrans).empty() : false;
    }

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

delete

void edm::Worker::postDoEvent

(

EventPrincipal const &

iEvent

)

Definition at line 394 of file Worker.cc.

References earlyDeleteHelper_.

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

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

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

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by prefetchAsync().

template<typename T >

void edm::Worker::prefetchAsync ( WaitingTaskHolder  iTask, ServiceToken const &  token, ParentContext const &  parentContext, typename T::TransitionInfoType const &  transitionInfo, Transition  iTransition  )

private

Definition at line 909 of file Worker.h.

References actReg_, edm::Principal::branchType(), edPrefetchAsync(), edm::ModuleCallingContext::getStreamContext(), edm::InEvent, edm::ModuleCallingContext::kPrefetching, moduleCallingContext_, preActionBeforeRunEventAsync(), and edm::ModuleCallingContext::setContext().

                                                     {
    Principal const& principal = transitionInfo.principal();

    moduleCallingContext_.setContext(ModuleCallingContext::State::kPrefetching, parentContext, nullptr);

    if (principal.branchType() == InEvent) {
      actReg_->preModuleEventPrefetchingSignal_.emit(*moduleCallingContext_.getStreamContext(), moduleCallingContext_);
    }

    workerhelper::CallImpl<T>::esPrefetchAsync(this, iTask, token, transitionInfo, iTransition);
    edPrefetchAsync(iTask, token, principal);

    if (principal.branchType() == InEvent) {
      preActionBeforeRunEventAsync(iTask, moduleCallingContext_, principal);
    }
  }

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

Definition at line 152 of file Worker.cc.

References CMS_SA_ALLOW, edm::TaskBase::decrement_ref_count(), edm::WaitingTaskList::doneWaiting(), edm::TaskBase::execute(), watchdog::group, implDoPrePrefetchSelection(), edm::TaskBase::increment_ref_count(), B2GTnPMonitor_cfi::item, mps_monitormerge::items, itemsToGetForSelection(), edm::ServiceWeakToken::lock(), edm::make_waiting_task(), moduleCallingContext_, edm::Principal::prefetchAsync(), edm::ProductResolverIndexAmbiguous, alignCSCRings::s, timesRun_, unpackBuffers-CaloStage2::token, and waitingTasks_.

Referenced by doWorkAsync().

                                                                           {
    successTask->increment_ref_count();

    ServiceWeakToken weakToken = token;
    auto choiceTask =
        edm::make_waiting_task([id, successTask, iPrincipal, this, weakToken, &group](std::exception_ptr const*) {
          ServiceRegistry::Operate guard(weakToken.lock());
          // 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()) {
                TaskSentry s(successTask);
              }
              return;
            }
          } catch (...) {
          }
          if (0 == successTask->decrement_ref_count()) {
            group.run([successTask]() {
              TaskSentry s(successTask);
              successTask->execute();
            });
          }
        });

    WaitingTaskHolder choiceHolder{group, 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(
            choiceHolder, productResolverIndex, skipCurrentProcess, token, &moduleCallingContext_);
      }
    }
    choiceHolder.doneWaiting(std::exception_ptr{});
  }

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

inline

Definition at line 140 of file Worker.h.

References cms::cuda::assert().

                                                                                              {
      assert(false);
    }

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

Definition at line 376 of file Worker.cc.

References cms::Exception::addContext(), description(), and implRegisterThinnedAssociations().

                                                                                                             {
    try {
      implRegisterThinnedAssociations(registry, helper);
    } catch (cms::Exception& ex) {
      ex.addContext("Calling registerThinnedAssociations() for module " + description()->moduleLabel());
      throw ex;
    }
  }

void edm::Worker::reset ( void  )

inline

Definition at line 178 of file Worker.h.

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

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

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

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

protected

Definition at line 287 of file Worker.cc.

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

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

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

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 >.

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

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

inline

Definition at line 174 of file Worker.h.

References implRespondToCloseInputFile().

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

{ implRespondToCloseInputFile(fb); }

void edm::Worker::respondToCloseOutputFile ( )

inline

Definition at line 175 of file Worker.h.

References implRespondToCloseOutputFile().

{ implRespondToCloseOutputFile(); }

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

inline

Definition at line 173 of file Worker.h.

References implRespondToOpenInputFile().

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

{ implRespondToOpenInputFile(fb); }

void edm::Worker::runAcquire ( EventTransitionInfo const &  info, ParentContext const &  parentContext, WaitingTaskWithArenaHolder &  holder  )

private

Definition at line 400 of file Worker.cc.

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

Referenced by runAcquireAfterAsyncPrefetch().

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

void edm::Worker::runAcquireAfterAsyncPrefetch ( std::exception_ptr const *  iEPtr, EventTransitionInfo const &  eventTransitionInfo, ParentContext const &  parentContext, WaitingTaskWithArenaHolder  holder  )

private

Definition at line 415 of file Worker.cc.

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

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

template<typename T >

bool edm::Worker::runModule ( typename T::TransitionInfoType const &  transitionInfo, StreamID  streamID, ParentContext const &  parentContext, typename T::Context const *  context  )

private

Definition at line 1106 of file Worker.h.

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

                                                           {
    //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, transitionInfo, actReg_.get(), &moduleCallingContext_, context);

        if (rc) {
          setPassed<T::isEvent_>();
        } else {
          setFailed<T::isEvent_>();
        }
      });
    } catch (cms::Exception& ex) {
      edm::exceptionContext(ex, moduleCallingContext_);
      if (shouldRethrowException(std::current_exception(), parentContext, T::isEvent_)) {
        assert(not cached_exception_);
        setException<T::isEvent_>(std::current_exception());
        std::rethrow_exception(cached_exception_);
      } else {
        rc = setPassed<T::isEvent_>();
      }
    }

    return rc;
  }

template<typename T >

std::exception_ptr edm::Worker::runModuleAfterAsyncPrefetch ( std::exception_ptr const *  iEPtr, typename T::TransitionInfoType const &  transitionInfo, StreamID  streamID, ParentContext const &  parentContext, typename T::Context const *  context  )

private

Definition at line 1021 of file Worker.h.

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

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

                                                                                           {
    std::exception_ptr exceptionPtr;
    if (iEPtr) {
      assert(*iEPtr);
      if (shouldRethrowException(*iEPtr, parentContext, T::isEvent_)) {
        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>(transitionInfo, streamID, parentContext, context); } catch (...) {
        exceptionPtr = std::current_exception();
      }
    }
    waitingTasks_.doneWaiting(exceptionPtr);
    return exceptionPtr;
  }

template<typename T >

std::exception_ptr edm::Worker::runModuleDirectly	(	typename T::TransitionInfoType const &	transitionInfo,
		StreamID	streamID,
		ParentContext const &	parentContext,
		typename T::Context const *	context
	)

Definition at line 1146 of file Worker.h.

References visDQMUpload::context, and timesVisited_.

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

                                                                                 {
    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, transitionInfo, streamID, parentContext, context);
  }

virtual void edm::Worker::selectInputProcessBlocks ( ProductRegistry const &  , ProcessBlockHelperBase const &    )

pure virtual

Implemented in edm::WorkerT< T >.

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 doWorkNoPrefetchingAsync(), and edm::Worker::RunModuleTask< T >::execute().

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 110 of file Worker.cc.

References actReg_, and edm::areg.

{ actReg_ = areg; }

void edm::Worker::setEarlyDeleteHelper

(

EarlyDeleteHelper *

iHelper

)

Definition at line 285 of file Worker.cc.

References earlyDeleteHelper_.

{ earlyDeleteHelper_ = iHelper; }

template<bool IS_EVENT>

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

inlineprivate

Definition at line 326 of file Worker.h.

References cached_exception_, Exception, state_, and timesExcept_.

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

template<bool IS_EVENT>

bool edm::Worker::setFailed ( )

inlineprivate

Definition at line 317 of file Worker.h.

References Fail, state_, and timesFailed_.

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

template<bool IS_EVENT>

bool edm::Worker::setPassed ( )

inlineprivate

Definition at line 308 of file Worker.h.

References Pass, state_, and timesPassed_.

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

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

private

Definition at line 112 of file Worker.cc.

References mps_fire::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 runAcquire(), runAcquireAfterAsyncPrefetch(), runModule(), and runModuleAfterAsyncPrefetch().

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

void edm::Worker::skipOnPath

(

EventPrincipal const &

iEvent

)

Definition at line 385 of file Worker.cc.

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

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

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

State edm::Worker::state ( ) const

inline

Definition at line 237 of file Worker.h.

References state_.

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

{ return state_; }

int edm::Worker::timesExcept ( ) const

inline

Definition at line 236 of file Worker.h.

References timesExcept_.

Referenced by edm::fillWorkerSummaryAux().

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

int edm::Worker::timesFailed ( ) const

inline

Definition at line 235 of file Worker.h.

References timesFailed_.

Referenced by edm::fillWorkerSummaryAux().

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

int edm::Worker::timesPass ( ) const

inline

Definition at line 239 of file Worker.h.

References timesPassed().

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

int edm::Worker::timesPassed ( ) const

inline

Definition at line 234 of file Worker.h.

References timesPassed_.

Referenced by edm::fillWorkerSummaryAux(), and timesPass().

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

int edm::Worker::timesRun ( ) const

inline

Definition at line 232 of file Worker.h.

References timesRun_.

Referenced by edm::fillWorkerSummaryAux().

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

int edm::Worker::timesVisited ( ) const

inline

Definition at line 233 of file Worker.h.

References timesVisited_.

Referenced by edm::fillWorkerSummaryAux().

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

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

pure virtual

Implemented in edm::WorkerT< T >.

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

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

pure virtual

Implemented in edm::WorkerT< T >.

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().

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().

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

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::workerhelper::CallImpl< OccurrenceTraits< ProcessBlockPrincipal, BranchActionProcessBlockInput > >::wantsTransition().

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

pure virtual

Implemented in edm::WorkerT< T >.

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

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().

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().

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

Friends And Related Function Documentation

template<typename O >

friend class workerhelper::CallImpl

friend

Definition at line 245 of file Worker.h.

Member Data Documentation

ExceptionToActionTable const* edm::Worker::actions_

private

Definition at line 578 of file Worker.h.

Referenced by shouldRethrowException().

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

private

Definition at line 581 of file Worker.h.

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

std::exception_ptr edm::Worker::cached_exception_

private

Definition at line 579 of file Worker.h.

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

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

private

Definition at line 583 of file Worker.h.

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

ModuleCallingContext edm::Worker::moduleCallingContext_

private

Definition at line 576 of file Worker.h.

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

bool edm::Worker::moduleValid_ = true

private

Definition at line 588 of file Worker.h.

Referenced by clearModule(), and description().

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

private

Definition at line 574 of file Worker.h.

Referenced by reset(), and skipOnPath().

int edm::Worker::numberOfPathsOn_

private

Definition at line 573 of file Worker.h.

Referenced by addedToPath(), and reset().

bool edm::Worker::ranAcquireWithoutException_

private

Definition at line 587 of file Worker.h.

Referenced by handleExternalWorkException(), and runAcquireAfterAsyncPrefetch().

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

private

Definition at line 572 of file Worker.h.

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

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

private

Definition at line 571 of file Worker.h.

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

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

private

Definition at line 570 of file Worker.h.

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

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

private

Definition at line 569 of file Worker.h.

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

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

private

Definition at line 567 of file Worker.h.

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

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

private

Definition at line 568 of file Worker.h.

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

edm::WaitingTaskList edm::Worker::waitingTasks_

private

Definition at line 585 of file Worker.h.

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

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

private

Definition at line 586 of file Worker.h.

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