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 }
enum	State { Ready, Pass, Fail, Exception }
enum	Types { kAnalyzer, kFilter, kProducer, kOutputModule }

Public Member Functions
void	addedToPath () noexcept
void	beginJob (GlobalContext const &)
void	beginStream (StreamID, StreamContext const &)
void	callWhenDoneAsync (WaitingTaskHolder task)
void	clearCounters () noexcept
void	clearModule ()
virtual std::vector< ConsumesInfo >	consumesInfo () const =0
virtual void	convertCurrentProcessAlias (std::string const &processName)=0
ModuleDescription const *	description () const noexcept
void	doTransformAsync (WaitingTaskHolder, size_t iTransformIndex, EventPrincipal const &, ServiceToken const &, StreamID, ModuleCallingContext const &, StreamContext const *) noexcept
template<typename T >
void	doWorkAsync (WaitingTaskHolder, typename T::TransitionInfoType const &, ServiceToken const &, StreamID, ParentContext const &, typename T::Context const *) noexcept
template<typename T >
void	doWorkNoPrefetchingAsync (WaitingTaskHolder, typename T::TransitionInfoType const &, ServiceToken const &, StreamID, ParentContext const &, typename T::Context const *) noexcept
void	endJob (GlobalContext const &)
void	endStream (StreamID, StreamContext const &)
virtual SerialTaskQueue *	globalLuminosityBlocksQueue ()=0
virtual SerialTaskQueue *	globalRunsQueue ()=0
virtual bool	hasAccumulator () const noexcept=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 (oneapi::tbb::task_group &, WaitingTask *task, ServiceToken const &, StreamID stream, EventPrincipal const *) noexcept
void	prePrefetchSelectionAsync (oneapi::tbb::task_group &, WaitingTask *task, ServiceToken const &, StreamID stream, void const *) noexcept
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 *) noexcept
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 noexcept
int	timesExcept () const noexcept
int	timesFailed () const noexcept
int	timesPass () const noexcept
int	timesPassed () const noexcept
int	timesRun () const noexcept
int	timesVisited () const noexcept
virtual size_t	transformIndex (edm::BranchDescription const &) const noexcept=0
virtual void	updateLookup (BranchType iBranchType, ProductResolverIndexHelper const &)=0
virtual void	updateLookup (eventsetup::ESRecordsToProductResolverIndices const &)=0
edm::WaitingTaskList &	waitingTaskList () noexcept
virtual bool	wantsGlobalLuminosityBlocks () const noexcept=0
virtual bool	wantsGlobalRuns () const noexcept=0
virtual bool	wantsInputProcessBlocks () const noexcept=0
virtual bool	wantsProcessBlocks () const noexcept=0
virtual bool	wantsStreamLuminosityBlocks () const noexcept=0
virtual bool	wantsStreamRuns () const noexcept=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	implDoTransformAsync (WaitingTaskHolder, size_t iTransformIndex, EventPrincipal const &, ParentContext const &, ServiceWeakToken const &) noexcept=0
virtual void	implEndJob ()=0
virtual void	implEndStream (StreamID)=0
virtual bool	implNeedToRunSelection () const noexcept=0
virtual void	itemsToGetForSelection (std::vector< ProductResolverIndexAndSkipBit > &) const =0
virtual ProductResolverIndex	itemToGetForTransform (size_t iTransformIndex) const noexcept=0
void	resetModuleDescription (ModuleDescription const *)
virtual std::string	workerType () const =0

Private Member Functions
void	checkForShouldTryToContinue (ModuleDescription const &)
void	edPrefetchAsync (WaitingTaskHolder, ServiceToken const &, Principal const &) const noexcept
void	emitPostModuleEventPrefetchingSignal ()
void	emitPostModuleGlobalPrefetchingSignal ()
void	emitPostModuleStreamPrefetchingSignal ()
virtual std::vector< ESResolverIndex > const &	esItemsToGetFrom (Transition) const =0
void	esPrefetchAsync (WaitingTaskHolder, EventSetupImpl const &, Transition, ServiceToken const &) noexcept
virtual std::vector< ESRecordIndex > const &	esRecordsToGetFrom (Transition) const =0
std::exception_ptr	handleExternalWorkException (std::exception_ptr iEPtr, ParentContext const &parentContext) noexcept
virtual bool	hasAcquire () const noexcept=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 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 noexcept=0
template<typename T >
void	prefetchAsync (WaitingTaskHolder, ServiceToken const &, ParentContext const &, typename T::TransitionInfoType const &, Transition) noexcept
void	runAcquire (EventTransitionInfo const &, ParentContext const &, WaitingTaskWithArenaHolder &)
void	runAcquireAfterAsyncPrefetch (std::exception_ptr, EventTransitionInfo const &, ParentContext const &, WaitingTaskWithArenaHolder) noexcept
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, typename T::TransitionInfoType const &, StreamID, ParentContext const &, typename T::Context const *) noexcept
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, bool isTryToContinue) const noexcept

Private Attributes
ExceptionToActionTable const *	actions_
std::shared_ptr< ActivityRegistry >	actReg_
bool	beginSucceeded_ = false
std::exception_ptr	cached_exception_
edm::propagate_const< EarlyDeleteHelper * >	earlyDeleteHelper_
ModuleCallingContext	moduleCallingContext_
bool	moduleValid_ = true
std::atomic< int >	numberOfPathsLeftToRun_
int	numberOfPathsOn_
bool	ranAcquireWithoutException_
bool	shouldTryToContinue_ = false
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

ConcurrencyTypes

enum edm::Worker::ConcurrencyTypes

Enumerator
kGlobal
kLimited
kOne
kStream

Definition at line 95 of file Worker.h.

{ kGlobal, kLimited, kOne, kStream };

State

enum edm::Worker::State

Enumerator
Ready
Pass
Fail
Exception

Definition at line 93 of file Worker.h.

{ Ready, Pass, Fail, Exception };

Types

enum edm::Worker::Types

Enumerator
kAnalyzer
kFilter
kProducer
kOutputModule

Definition at line 94 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 82 of file Worker.cc.

References checkForShouldTryToContinue().

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

~Worker()

edm::Worker::~Worker ( )

virtual

Definition at line 101 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 300 of file Worker.h.

References actReg_.

Referenced by beginJob(), beginStream(), endJob(), and endStream().

{ return actReg_.get(); }

addedToPath()

void edm::Worker::addedToPath ( )

inlinenoexcept

Definition at line 242 of file Worker.h.

References numberOfPathsOn_.

{ ++numberOfPathsOn_; }

beginJob()

void edm::Worker::beginJob

(

GlobalContext const &

globalContext

)

Definition at line 292 of file Worker.cc.

References activityRegistry(), beginSucceeded_, edm::exceptionContext(), implBeginJob(), moduleCallingContext_, and edm::convertException::wrap().

                                                          {
    ParentContext parentContext(&globalContext);
    ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
    ModuleSignalSentry<ModuleBeginJobTraits> sentry(activityRegistry(), &globalContext, &moduleCallingContext_);

    try {
      convertException::wrap([this, &sentry]() {
        beginSucceeded_ = false;
        sentry.preModuleSignal();
        implBeginJob();
        sentry.postModuleSignal();
        beginSucceeded_ = true;
      });
    } catch (cms::Exception& ex) {
      exceptionContext(ex, moduleCallingContext_);
      throw;
    }
  }

beginStream()

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

Definition at line 332 of file Worker.cc.

References activityRegistry(), beginSucceeded_, edm::exceptionContext(), implBeginStream(), moduleCallingContext_, and edm::convertException::wrap().

                                                                                {
    ParentContext parentContext(&streamContext);
    ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
    ModuleSignalSentry<ModuleBeginStreamTraits> sentry(activityRegistry(), &streamContext, &moduleCallingContext_);

    try {
      convertException::wrap([this, &sentry, streamID]() {
        beginSucceeded_ = false;
        sentry.preModuleSignal();
        implBeginStream(streamID);
        sentry.postModuleSignal();
        beginSucceeded_ = true;
      });
    } catch (cms::Exception& ex) {
      exceptionContext(ex, moduleCallingContext_);
      throw;
    }
  }

callWhenDoneAsync()

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

inline

Definition at line 179 of file Worker.h.

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

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

checkForShouldTryToContinue()

void edm::Worker::checkForShouldTryToContinue ( ModuleDescription const &  iDesc )

private

Definition at line 105 of file Worker.cc.

References edm::pset::Registry::getMapped(), edm::pset::Registry::instance(), edm::ModuleDescription::parameterSetID(), muonDTDigis_cfi::pset, and shouldTryToContinue_.

Referenced by resetModuleDescription(), and Worker().

                                                                         {
    auto pset = edm::pset::Registry::instance()->getMapped(iDesc.parameterSetID());
    if (pset and pset->exists("@shouldTryToContinue")) {
      shouldTryToContinue_ = true;
    }
  }

clearCounters()

void edm::Worker::clearCounters ( )

inlinenoexcept

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

clearModule()

void edm::Worker::clearModule ( )

inline

Definition at line 122 of file Worker.h.

References doClearModule(), and moduleValid_.

                       {
      moduleValid_ = false;
      doClearModule();
    }

consumesInfo()

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

pure virtual

Implemented in edm::WorkerT< T >.

Referenced by edm::PathsAndConsumesOfModules::doConsumesInfo(), and edm::StreamSchedule::tryToPlaceConditionalModules().

convertCurrentProcessAlias()

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

pure virtual

Implemented in edm::WorkerT< T >.

description()

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

inlinenoexcept

Definition at line 200 of file Worker.h.

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

Referenced by edm::StreamSchedule::fillWorkers(), registerThinnedAssociations(), and edm::StreamSchedule::tryToPlaceConditionalModules().

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

doClearModule()

virtual void edm::Worker::doClearModule ( )

protectedpure virtual

Implemented in edm::WorkerT< T >.

Referenced by clearModule().

doTransformAsync()

void edm::Worker::doTransformAsync ( WaitingTaskHolder  iTask, size_t  iTransformIndex, EventPrincipal const &  iPrincipal, ServiceToken const &  iToken, StreamID  , ModuleCallingContext const &  mcc, StreamContext const *    )

noexcept

Definition at line 254 of file Worker.cc.

References edm::make_waiting_task(), and TrackValidation_cff::task.

                                                               {
    ServiceWeakToken weakToken = iToken;

    //Need to make the services available early so other services can see them
    auto task = make_waiting_task(
        [this, iTask, weakToken, &iPrincipal, iTransformIndex, mcc](std::exception_ptr const* iExcept) mutable {
          //post prefetch signal
          actReg_->postModuleTransformPrefetchingSignal_.emit(*mcc.getStreamContext(), mcc);
          if (iExcept) {
            iTask.doneWaiting(*iExcept);
            return;
          }
          implDoTransformAsync(iTask, iTransformIndex, iPrincipal, mcc.parent(), weakToken);
        });

    //pre prefetch signal
    actReg_->preModuleTransformPrefetchingSignal_.emit(*mcc.getStreamContext(), mcc);
    iPrincipal.prefetchAsync(
        WaitingTaskHolder(*iTask.group(), task), itemToGetForTransform(iTransformIndex), false, iToken, &mcc);
  }

doWorkAsync()

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  )

noexcept

Definition at line 1078 of file Worker.h.

References ALPAKA_ACCELERATOR_NAMESPACE::brokenline::constexpr(), visDQMUpload::context, watchdog::group, info(), edm::ModuleCallingContext::kPrefetching, edm::make_waiting_task(), eostools::move(), AlCaHLTBitMon_ParallelJobs::p, fetchall_from_DQM_v2::release, alignCSCRings::s, submitPVValidationJobs::t, TrackValidation_cff::task, and unpackBuffers-CaloStage2::token.

                                                                      {
    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) noexcept : m_task(iTask) {}

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

          edm::WaitingTask* release() noexcept { 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_);
      }
    }
  }

doWorkNoPrefetchingAsync()

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  )

noexcept

Definition at line 1200 of file Worker.h.

References CMS_SA_ALLOW, visDQMUpload::context, watchdog::group, info(), edm::ModuleCallingContext::kPrefetching, edm::ServiceWeakToken::lock(), edm::make_waiting_task(), eostools::move(), createBeamHaloJobs::queue, and TrackValidation_cff::task.

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

                                                                                   {
    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);
                }
              }
            });
        moduleCallingContext_.setContext(ModuleCallingContext::State::kPrefetching, parentContext, nullptr);
        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);
        }
      }
    }
  }

edPrefetchAsync()

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

privatenoexcept

Definition at line 236 of file Worker.cc.

References B2GTnPMonitor_cfi::item, mps_monitormerge::items, edm::ProductResolverIndexAmbiguous, and unpackBuffers-CaloStage2::token.

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

emitPostModuleEventPrefetchingSignal()

void edm::Worker::emitPostModuleEventPrefetchingSignal ( )

inlineprivate

Definition at line 374 of file Worker.h.

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

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

emitPostModuleGlobalPrefetchingSignal()

void edm::Worker::emitPostModuleGlobalPrefetchingSignal ( )

inlineprivate

Definition at line 383 of file Worker.h.

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

                                                 {
      actReg_->postModuleGlobalPrefetchingSignal_.emit(*moduleCallingContext_.getGlobalContext(),
                                                       moduleCallingContext_);
    }

emitPostModuleStreamPrefetchingSignal()

void edm::Worker::emitPostModuleStreamPrefetchingSignal ( )

inlineprivate

Definition at line 378 of file Worker.h.

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

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

endJob()

void edm::Worker::endJob

(

GlobalContext const &

globalContext

)

Definition at line 311 of file Worker.cc.

References activityRegistry(), beginSucceeded_, edm::exceptionContext(), implEndJob(), moduleCallingContext_, and edm::convertException::wrap().

                                                        {
    if (beginSucceeded_) {
      beginSucceeded_ = false;

      ParentContext parentContext(&globalContext);
      ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
      ModuleSignalSentry<ModuleEndJobTraits> sentry(activityRegistry(), &globalContext, &moduleCallingContext_);

      try {
        convertException::wrap([this, &sentry]() {
          sentry.preModuleSignal();
          implEndJob();
          sentry.postModuleSignal();
        });
      } catch (cms::Exception& ex) {
        exceptionContext(ex, moduleCallingContext_);
        throw;
      }
    }
  }

endStream()

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

Definition at line 351 of file Worker.cc.

References activityRegistry(), beginSucceeded_, edm::exceptionContext(), implEndStream(), moduleCallingContext_, and edm::convertException::wrap().

                                                                        {
    if (beginSucceeded_) {
      beginSucceeded_ = false;

      ParentContext parentContext(&streamContext);
      ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
      ModuleSignalSentry<ModuleEndStreamTraits> sentry(activityRegistry(), &streamContext, &moduleCallingContext_);

      try {
        convertException::wrap([this, &sentry, id]() {
          sentry.preModuleSignal();
          implEndStream(id);
          sentry.postModuleSignal();
        });
      } catch (cms::Exception& ex) {
        exceptionContext(ex, moduleCallingContext_);
        throw;
      }
    }
  }

esItemsToGetFrom()

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

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by needsESPrefetching().

esPrefetchAsync()

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

privatenoexcept

Definition at line 211 of file Worker.cc.

References cms::cuda::assert(), mps_fire::i, mps_monitormerge::items, and edm::NumberOfEventSetupTransitions.

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

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

esRecordsToGetFrom()

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

privatepure virtual

Implemented in edm::WorkerT< T >.

globalLuminosityBlocksQueue()

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

pure virtual

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

globalRunsQueue()

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

pure virtual

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

handleExternalWorkException()

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

privatenoexcept

Definition at line 435 of file Worker.cc.

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

Referenced by edm::Worker::HandleExternalWorkExceptionTask::execute().

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

hasAccumulator()

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

pure virtualnoexcept

Implemented in edm::WorkerT< T >.

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

hasAcquire()

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

privatepure virtualnoexcept

Implemented in edm::WorkerT< T >.

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 ( EventTransitionInfo const &  , ModuleCallingContext const *    )

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

implDoAccessInputProcessBlock()

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

implDoAcquire()

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

implDoBegin() [1/2]

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

implDoBegin() [2/2]

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoBeginProcessBlock()

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

implDoEnd() [1/2]

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

implDoEnd() [2/2]

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoEndProcessBlock()

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

implDoPrePrefetchSelection()

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

implDoStreamBegin() [1/2]

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

implDoStreamBegin() [2/2]

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoStreamEnd() [1/2]

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

implDoStreamEnd() [2/2]

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

implDoTransformAsync()

virtual void edm::Worker::implDoTransformAsync ( WaitingTaskHolder  , size_t  iTransformIndex, EventPrincipal const &  , ParentContext const &  , ServiceWeakToken const &    )

protectedpure virtualnoexcept

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

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 virtualnoexcept

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

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

implRespondToCloseOutputFile()

virtual void edm::Worker::implRespondToCloseOutputFile ( )

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by respondToCloseOutputFile().

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

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

itemsToGetFrom()

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

privatepure virtual

Implemented in edm::WorkerT< T >.

itemToGetForTransform()

virtual ProductResolverIndex edm::Worker::itemToGetForTransform ( size_t  iTransformIndex ) const

protectedpure virtualnoexcept

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

moduleConcurrencyType()

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 >, and edm::WorkerT< T >.

modulesWhoseProductsAreConsumed()

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

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 >, and edm::WorkerT< T >.

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

needsESPrefetching()

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

inlineprivatenoexcept

Definition at line 370 of file Worker.h.

References esItemsToGetFrom(), and edm::NumberOfEventSetupTransitions.

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

operator=()

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

delete

postDoEvent()

void edm::Worker::postDoEvent

(

EventPrincipal const &

iEvent

)

Definition at line 390 of file Worker.cc.

References earlyDeleteHelper_, and iEvent.

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

preActionBeforeRunEventAsync()

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

privatepure virtualnoexcept

Implemented in edm::WorkerT< T >.

prefetchAsync()

template<typename T >

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

privatenoexcept

Definition at line 1052 of file Worker.h.

References edm::Principal::branchType(), ALPAKA_ACCELERATOR_NAMESPACE::brokenline::constexpr(), edm::InEvent, edm::ModuleCallingContext::kPrefetching, and unpackBuffers-CaloStage2::token.

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

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

    if constexpr (T::isEvent_) {
      actReg_->preModuleEventPrefetchingSignal_.emit(*moduleCallingContext_.getStreamContext(), moduleCallingContext_);
    } else if constexpr (std::is_same_v<typename T::Context, StreamContext>) {
      actReg_->preModuleStreamPrefetchingSignal_.emit(*moduleCallingContext_.getStreamContext(), moduleCallingContext_);
    } else if constexpr (std::is_same_v<typename T::Context, GlobalContext>) {
      actReg_->preModuleGlobalPrefetchingSignal_.emit(*moduleCallingContext_.getGlobalContext(), moduleCallingContext_);
    }

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

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

prePrefetchSelectionAsync() [1/2]

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

noexcept

Definition at line 147 of file Worker.cc.

References MillePedeFileConverter_cfg::e, edm::exceptionContext(), watchdog::group, B2GTnPMonitor_cfi::item, mps_monitormerge::items, edm::ServiceWeakToken::lock(), edm::make_waiting_task(), edm::ProductResolverIndexAmbiguous, edm::ProductResolverIndexInvalid, alignCSCRings::s, unpackBuffers-CaloStage2::token, and edm::convertException::wrap().

                                                                                    {
    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());
          try {
            bool selected = convertException::wrap([&]() {
              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 false;
              }
              return true;
            });
            if (not selected) {
              return;
            }

          } catch (cms::Exception& e) {
            edm::exceptionContext(e, moduleCallingContext_);
            setException<true>(std::current_exception());
            waitingTasks_.doneWaiting(std::current_exception());
            //TBB requires that destroyed tasks have count 0
            if (0 == successTask->decrement_ref_count()) {
              TaskSentry s(successTask);
            }
            return;
          }
          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 and
          productResolverIndex != ProductResolverIndexInvalid) {
        iPrincipal->prefetchAsync(
            choiceHolder, productResolverIndex, skipCurrentProcess, token, &moduleCallingContext_);
      }
    }
    choiceHolder.doneWaiting(std::exception_ptr{});
  }

prePrefetchSelectionAsync() [2/2]

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

inlinenoexcept

Definition at line 143 of file Worker.h.

References cms::cuda::assert().

                                                                                                             {
      assert(false);
    }

registerThinnedAssociations()

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

Definition at line 372 of file Worker.cc.

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

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

reset()

void edm::Worker::reset ( void  )

inline

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

resetModuleDescription()

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

protected

Definition at line 281 of file Worker.cc.

References cms::cuda::assert(), checkForShouldTryToContinue(), moduleCallingContext_, edm::ModuleCallingContext::parent(), edm::ModuleCallingContext::previousModuleOnThread(), edm::ModuleCallingContext::state(), and groupFilesInBlocks::temp.

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

                                                                    {
    ModuleCallingContext temp(iDesc,
                              0,
                              moduleCallingContext_.state(),
                              moduleCallingContext_.parent(),
                              moduleCallingContext_.previousModuleOnThread());
    moduleCallingContext_ = temp;
    assert(iDesc);
    checkForShouldTryToContinue(*iDesc);
  }

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 186 of file Worker.h.

References implRespondToCloseInputFile().

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

{ implRespondToCloseInputFile(fb); }

respondToCloseOutputFile()

void edm::Worker::respondToCloseOutputFile ( )

inline

Definition at line 187 of file Worker.h.

References implRespondToCloseOutputFile().

{ implRespondToCloseOutputFile(); }

respondToOpenInputFile()

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

inline

Definition at line 185 of file Worker.h.

References implRespondToOpenInputFile().

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

{ implRespondToOpenInputFile(fb); }

runAcquire()

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

private

Definition at line 396 of file Worker.cc.

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

                                                              {
    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, shouldTryToContinue_)) {
        timesRun_.fetch_add(1, std::memory_order_relaxed);
        throw;
      }
    }
  }

runAcquireAfterAsyncPrefetch()

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

privatenoexcept

Definition at line 411 of file Worker.cc.

References CMS_SA_ALLOW, and edm::ModuleCallingContext::kInvalid.

                                                                                        {
    ranAcquireWithoutException_ = false;
    std::exception_ptr exceptionPtr;
    if (iEPtr) {
      if (shouldRethrowException(iEPtr, parentContext, true, shouldTryToContinue_)) {
        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);
  }

runModule()

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 1260 of file Worker.h.

References actReg_, cms::cuda::assert(), cached_exception_, ALPAKA_ACCELERATOR_NAMESPACE::brokenline::constexpr(), visDQMUpload::context, edm::exceptionContext(), moduleCallingContext_, shouldRethrowException(), shouldTryToContinue_, 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_, shouldTryToContinue_)) {
        assert(not cached_exception_);
        setException<T::isEvent_>(std::current_exception());
        std::rethrow_exception(cached_exception_);
      } else {
        rc = setPassed<T::isEvent_>();
      }
    }

    return rc;
  }

runModuleAfterAsyncPrefetch()

template<typename T >

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

privatenoexcept

Definition at line 1168 of file Worker.h.

References CMS_SA_ALLOW, visDQMUpload::context, and edm::ModuleCallingContext::kInvalid.

                                                                                                    {
    std::exception_ptr exceptionPtr;
    bool shouldRun = true;
    if (iEPtr) {
      if (shouldRethrowException(iEPtr, parentContext, T::isEvent_, shouldTryToContinue_)) {
        exceptionPtr = iEPtr;
        setException<T::isEvent_>(exceptionPtr);
        shouldRun = false;
      } else {
        if (not shouldTryToContinue_) {
          setPassed<T::isEvent_>();
          shouldRun = false;
        }
      }
    }
    if (shouldRun) {
      // Caught exception is propagated via WaitingTaskList
      CMS_SA_ALLOW try { runModule<T>(transitionInfo, streamID, parentContext, context); } catch (...) {
        exceptionPtr = std::current_exception();
      }
    } else {
      moduleCallingContext_.setContext(ModuleCallingContext::State::kInvalid, ParentContext(), nullptr);
    }
    waitingTasks_.doneWaiting(exceptionPtr);
    return exceptionPtr;
  }

runModuleDirectly()

template<typename T >

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

noexcept

Definition at line 1300 of file Worker.h.

References visDQMUpload::context.

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

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

selectInputProcessBlocks()

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

pure virtual

Implemented in edm::WorkerT< T >.

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 >, and edm::WorkerT< T >.

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

References actReg_.

{ actReg_ = areg; }

setEarlyDeleteHelper()

void edm::Worker::setEarlyDeleteHelper

(

EarlyDeleteHelper *

iHelper

)

Definition at line 251 of file Worker.cc.

References earlyDeleteHelper_.

{ earlyDeleteHelper_ = iHelper; }

setException()

template<bool IS_EVENT>

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

inlineprivate

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

setFailed()

template<bool IS_EVENT>

bool edm::Worker::setFailed ( )

inlineprivate

Definition at line 342 of file Worker.h.

References Fail, state_, and timesFailed_.

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

setPassed()

template<bool IS_EVENT>

bool edm::Worker::setPassed ( )

inlineprivate

Definition at line 333 of file Worker.h.

References Pass, state_, and timesPassed_.

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

shouldRethrowException()

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

privatenoexcept

Definition at line 112 of file Worker.cc.

References writedatasetfile::action, cms::Exception::category(), edm::exception_actions::IgnoreCompletely, edm::printCmsExceptionWarning(), edm::exception_actions::Rethrow, edm::exception_actions::TryToContinue, and edm::convertException::wrap().

Referenced by runAcquire(), and runModule().

                                                                               {
    // 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;
      }
      if (action == exception_actions::TryToContinue) {
        if (shouldTryToContinue) {
          edm::printCmsExceptionWarning("TryToContinue", ex);
        }
        return not shouldTryToContinue;
      }
      if (action == exception_actions::IgnoreCompletely) {
        edm::printCmsExceptionWarning("IgnoreCompletely", ex);
        return false;
      }
    }
    return true;
  }

skipOnPath()

void edm::Worker::skipOnPath

(

EventPrincipal const &

iEvent

)

Definition at line 381 of file Worker.cc.

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

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

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

state()

State edm::Worker::state ( ) const

inlinenoexcept

Definition at line 249 of file Worker.h.

References state_.

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

{ return state_; }

timesExcept()

int edm::Worker::timesExcept ( ) const

inlinenoexcept

Definition at line 248 of file Worker.h.

References timesExcept_.

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

timesFailed()

int edm::Worker::timesFailed ( ) const

inlinenoexcept

Definition at line 247 of file Worker.h.

References timesFailed_.

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

timesPass()

int edm::Worker::timesPass ( ) const

inlinenoexcept

Definition at line 251 of file Worker.h.

References timesPassed().

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

timesPassed()

int edm::Worker::timesPassed ( ) const

inlinenoexcept

Definition at line 246 of file Worker.h.

References timesPassed_.

Referenced by timesPass().

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

timesRun()

int edm::Worker::timesRun ( ) const

inlinenoexcept

Definition at line 244 of file Worker.h.

References timesRun_.

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

timesVisited()

int edm::Worker::timesVisited ( ) const

inlinenoexcept

Definition at line 245 of file Worker.h.

References timesVisited_.

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

transformIndex()

size_t edm::Worker::transformIndex ( edm::BranchDescription const &  ) const

pure virtualnoexcept

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

Definition at line 253 of file Worker.cc.

Referenced by edm::TransformingProductResolver::setupUnscheduled().

{ return -1; }

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::ESRecordsToProductResolverIndices const &  )

pure virtual

Implemented in edm::WorkerT< T >.

waitingTaskList()

edm::WaitingTaskList& edm::Worker::waitingTaskList ( )

inlinenoexcept

Definition at line 256 of file Worker.h.

References waitingTasks_.

{ return waitingTasks_; }

wantsGlobalLuminosityBlocks()

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

pure virtualnoexcept

Implemented in edm::WorkerT< T >.

wantsGlobalRuns()

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

pure virtualnoexcept

Implemented in edm::WorkerT< T >.

wantsInputProcessBlocks()

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

pure virtualnoexcept

Implemented in edm::WorkerT< T >.

wantsProcessBlocks()

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

pure virtualnoexcept

Implemented in edm::WorkerT< T >.

wantsStreamLuminosityBlocks()

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

pure virtualnoexcept

Implemented in edm::WorkerT< T >.

wantsStreamRuns()

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

pure virtualnoexcept

Implemented in edm::WorkerT< T >.

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 260 of file Worker.h.

Member Data Documentation

actions_

ExceptionToActionTable const* edm::Worker::actions_

private

Definition at line 618 of file Worker.h.

actReg_

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

private

Definition at line 621 of file Worker.h.

Referenced by activityRegistry(), emitPostModuleEventPrefetchingSignal(), emitPostModuleGlobalPrefetchingSignal(), emitPostModuleStreamPrefetchingSignal(), runModule(), and setActivityRegistry().

beginSucceeded_

bool edm::Worker::beginSucceeded_ = false

private

Definition at line 630 of file Worker.h.

Referenced by beginJob(), beginStream(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalBegin > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamBegin > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionGlobalEnd > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< RunPrincipal, BranchActionStreamEnd > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalBegin > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamBegin > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionGlobalEnd > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< LuminosityBlockPrincipal, BranchActionStreamEnd > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< ProcessBlockPrincipal, BranchActionGlobalBegin > >::call(), edm::workerhelper::CallImpl< OccurrenceTraits< ProcessBlockPrincipal, BranchActionGlobalEnd > >::call(), endJob(), and endStream().

cached_exception_

std::exception_ptr edm::Worker::cached_exception_

private

Definition at line 619 of file Worker.h.

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

earlyDeleteHelper_

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

private

Definition at line 623 of file Worker.h.

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

moduleCallingContext_

ModuleCallingContext edm::Worker::moduleCallingContext_

private

Definition at line 616 of file Worker.h.

Referenced by beginJob(), beginStream(), description(), emitPostModuleEventPrefetchingSignal(), emitPostModuleGlobalPrefetchingSignal(), emitPostModuleStreamPrefetchingSignal(), endJob(), endStream(), resetModuleDescription(), runAcquire(), and runModule().

moduleValid_

bool edm::Worker::moduleValid_ = true

private

Definition at line 628 of file Worker.h.

Referenced by clearModule(), and description().

numberOfPathsLeftToRun_

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

private

Definition at line 614 of file Worker.h.

Referenced by reset(), and skipOnPath().

numberOfPathsOn_

int edm::Worker::numberOfPathsOn_

private

Definition at line 613 of file Worker.h.

Referenced by addedToPath(), and reset().

ranAcquireWithoutException_

bool edm::Worker::ranAcquireWithoutException_

private

Definition at line 627 of file Worker.h.

shouldTryToContinue_

bool edm::Worker::shouldTryToContinue_ = false

private

Definition at line 629 of file Worker.h.

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

state_

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

private

Definition at line 612 of file Worker.h.

Referenced by reset(), setException(), setFailed(), setPassed(), and state().

timesExcept_

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

private

Definition at line 611 of file Worker.h.

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

timesFailed_

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

private

Definition at line 610 of file Worker.h.

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

timesPassed_

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

private

Definition at line 609 of file Worker.h.

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

timesRun_

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

private

Definition at line 607 of file Worker.h.

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

timesVisited_

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

private

Definition at line 608 of file Worker.h.

Referenced by clearCounters(), and timesVisited().

waitingTasks_

edm::WaitingTaskList edm::Worker::waitingTasks_

private

Definition at line 625 of file Worker.h.

Referenced by callWhenDoneAsync(), reset(), skipOnPath(), and waitingTaskList().

workStarted_

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

private

Definition at line 626 of file Worker.h.

Referenced by reset().