edm::Worker Class Reference

#include <Worker.h>

Inheritance diagram for edm::Worker:

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

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

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

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

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

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

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

Friends
template<typename O >
class	workerhelper::CallImpl

Detailed Description

Definition at line 83 of file Worker.h.

Member Enumeration Documentation

enum edm::Worker::State

Enumerator
Ready
Pass
Fail
Exception

Definition at line 85 of file Worker.h.

{ Ready, Pass, Fail, Exception };

enum edm::Worker::Types

Enumerator
kAnalyzer
kFilter
kProducer
kOutputModule

Definition at line 86 of file Worker.h.

{ kAnalyzer, kFilter, kProducer, kOutputModule };

Constructor & Destructor Documentation

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

Definition at line 83 of file Worker.cc.

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

edm::Worker::~Worker ( )

virtual

Definition at line 100 of file Worker.cc.

{}

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

delete

Member Function Documentation

ActivityRegistry* edm::Worker::activityRegistry ( )

inlineprotected

Definition at line 284 of file Worker.h.

References edm::exceptionContext(), and benchmark_cfg::fb.

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

{ return actReg_.get(); }

void edm::Worker::addedToPath ( )

inline

Definition at line 225 of file Worker.h.

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

{ ++numberOfPathsOn_; }

void edm::Worker::beginJob

(

void

)

Definition at line 284 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 314 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 ( WaitingTask *  task )

inline

Definition at line 167 of file Worker.h.

References bk::beginJob().

Referenced by edm::SwitchProducerProductResolver::prefetchAsync_().

{ waitingTasks_.add(task); }

void edm::Worker::clearCounters ( )

inline

Definition at line 217 of file Worker.h.

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

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::descPtr ( ) const

inline

Definition at line 191 of file Worker.h.

References modifiedElectrons_cfi::processName, and AlCaHLTBitMon_QueryRunRegistry::string.

{ return moduleCallingContext_.moduleDescription(); }

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

inline

Definition at line 190 of file Worker.h.

Referenced by beginJob(), beginStream(), endJob(), endStream(), edm::fillModuleInPathSummary(), edm::StreamSchedule::fillWorkers(), edm::fillWorkerSummaryAux(), edm::WorkerT< T >::resolvePutIndicies(), and shouldRethrowException().

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

template<typename T >

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

Definition at line 961 of file Worker.h.

References Fail, edm::ServiceRegistry::instance(), edm::ModuleCallingContext::kInvalid, edm::ModuleCallingContext::kPrefetching, edm::make_empty_waiting_task(), Pass, edm::ServiceRegistry::presentToken(), createBeamHaloJobs::queue, Ready, and edm::ModuleCallingContext::setContext().

                                                        {
    if (T::isEvent_) {
      timesVisited_.fetch_add(1, std::memory_order_relaxed);
    }
    bool rc = false;

    switch (state_) {
      case Ready:
        break;
      case Pass:
        return true;
      case Fail:
        return false;
      case Exception: {
        std::rethrow_exception(cached_exception_);
      }
    }

    bool expected = false;
    if (not workStarted_.compare_exchange_strong(expected, true)) {
      //another thread beat us here
      auto waitTask = edm::make_empty_waiting_task();
      waitTask->increment_ref_count();

      waitingTasks_.add(waitTask.get());

      waitTask->wait_for_all();

      switch (state_) {
        case Ready:
          assert(false);
        case Pass:
          return true;
        case Fail:
          return false;
        case Exception: {
          std::rethrow_exception(cached_exception_);
        }
      }
    }

    //Need the context to be set until after any exception is resolved
    moduleCallingContext_.setContext(ModuleCallingContext::State::kPrefetching, parentContext, nullptr);

    auto resetContext = [](ModuleCallingContext* iContext) {
      iContext->setContext(ModuleCallingContext::State::kInvalid, ParentContext(), nullptr);
    };
    std::unique_ptr<ModuleCallingContext, decltype(resetContext)> prefetchSentry(&moduleCallingContext_, resetContext);

    if (T::isEvent_) {
      //if have TriggerResults based selection we want to reject the event before doing prefetching
      if (workerhelper::CallImpl<T>::needToRunSelection(this)) {
        auto waitTask = edm::make_empty_waiting_task();
        waitTask->set_ref_count(2);
        prePrefetchSelectionAsync(waitTask.get(), ServiceRegistry::instance().presentToken(), streamID, &ep);
        waitTask->decrement_ref_count();
        waitTask->wait_for_all();

        if (state() != Ready) {
          //The selection must have rejected this event
          return true;
        }
      }
      auto waitTask = edm::make_empty_waiting_task();
      {
        //Make sure signal is sent once the prefetching is done
        // [the 'pre' signal was sent in prefetchAsync]
        //The purpose of this block is to send the signal after wait_for_all
        auto sentryFunc = [this](void*) { emitPostModuleEventPrefetchingSignal(); };
        std::unique_ptr<ActivityRegistry, decltype(sentryFunc)> signalSentry(actReg_.get(), sentryFunc);

        //set count to 2 since wait_for_all requires value to not go to 0
        waitTask->set_ref_count(2);

        prefetchAsync(waitTask.get(), ServiceRegistry::instance().presentToken(), parentContext, ep);
        waitTask->decrement_ref_count();
        waitTask->wait_for_all();
      }
      if (waitTask->exceptionPtr() != nullptr) {
        TransitionIDValue<typename T::MyPrincipal> idValue(ep);
        if (shouldRethrowException(*waitTask->exceptionPtr(), parentContext, T::isEvent_, idValue)) {
          setException<T::isEvent_>(*waitTask->exceptionPtr());
          waitingTasks_.doneWaiting(cached_exception_);
          std::rethrow_exception(cached_exception_);
        } else {
          setPassed<T::isEvent_>();
          waitingTasks_.doneWaiting(nullptr);
          return true;
        }
      }
    }

    //successful prefetch so no reset necessary
    prefetchSentry.release();
    if (auto queue = serializeRunModule()) {
      auto serviceToken = ServiceRegistry::instance().presentToken();
      queue.pushAndWait([&]() {
        //Need to make the services available
        ServiceRegistry::Operate guard(serviceToken);
        try {
          rc = runModule<T>(ep, es, streamID, parentContext, context);
        } catch (...) {
        }
      });
    } else {
      try {
        rc = runModule<T>(ep, es, streamID, parentContext, context);
      } catch (...) {
      }
    }
    if (state_ == Exception) {
      waitingTasks_.doneWaiting(cached_exception_);
      std::rethrow_exception(cached_exception_);
    }

    waitingTasks_.doneWaiting(nullptr);
    return rc;
  }

template<typename T >

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

Definition at line 827 of file Worker.h.

References pyrootRender::destroy(), edm::ModuleCallingContext::kPrefetching, edm::make_waiting_task(), eostools::move(), AlCaHLTBitMon_ParallelJobs::p, fetchall_from_DQM_v2::release, edm::ModuleCallingContext::setContext(), and protons_cff::t.

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

                                                             {
    if (not workerhelper::CallImpl<T>::wantsTransition(this)) {
      return;
    }

    waitingTasks_.add(task);
    if (T::isEvent_) {
      timesVisited_.fetch_add(1, std::memory_order_relaxed);
    }

    bool expected = false;
    if (workStarted_.compare_exchange_strong(expected, true)) {
      moduleCallingContext_.setContext(ModuleCallingContext::State::kPrefetching, parentContext, nullptr);

      //if have TriggerResults based selection we want to reject the event before doing prefetching
      if (workerhelper::CallImpl<T>::needToRunSelection(this)) {
        //We need to run the selection in a different task so that
        // we can prefetch the data needed for the selection
        auto runTask =
            new (tbb::task::allocate_root()) RunModuleTask<T>(this, ep, es, token, streamID, parentContext, context);

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

          ~DestroyTask() {
            auto p = m_task.load();
            if (p) {
              tbb::task::destroy(*p);
            }
          }

          edm::WaitingTask* release() {
            auto t = m_task.load();
            m_task.store(nullptr);
            return t;
          }

          std::atomic<edm::WaitingTask*> m_task;
        };

        auto ownRunTask = std::make_shared<DestroyTask>(runTask);
        auto selectionTask =
            make_waiting_task(tbb::task::allocate_root(),
                              [ownRunTask, parentContext, &ep, token, this](std::exception_ptr const*) mutable {
                                ServiceRegistry::Operate guard(token);
                                prefetchAsync(ownRunTask->release(), token, parentContext, ep);
                              });
        prePrefetchSelectionAsync(selectionTask, token, streamID, &ep);
      } else {
        WaitingTask* moduleTask =
            new (tbb::task::allocate_root()) RunModuleTask<T>(this, ep, es, token, streamID, parentContext, context);
        if (T::isEvent_ && hasAcquire()) {
          WaitingTaskWithArenaHolder runTaskHolder(
              new (tbb::task::allocate_root()) HandleExternalWorkExceptionTask(this, moduleTask, parentContext));
          moduleTask = new (tbb::task::allocate_root())
              AcquireTask<T>(this, ep, es, token, parentContext, std::move(runTaskHolder));
        }
        prefetchAsync(moduleTask, token, parentContext, ep);
      }
    }
  }

template<typename T >

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

Definition at line 926 of file Worker.h.

References edm::make_functor_task(), cmsRelvalreport::principal(), and createBeamHaloJobs::queue.

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

                                                                          {
    if (not workerhelper::CallImpl<T>::wantsTransition(this)) {
      return;
    }
    waitingTasks_.add(task);
    bool expected = false;
    if (workStarted_.compare_exchange_strong(expected, true)) {
      auto toDo = [this, &principal, &es, streamID, parentContext, context, serviceToken]() {
        std::exception_ptr exceptionPtr;
        try {
          //Need to make the services available
          ServiceRegistry::Operate guard(serviceToken);

          this->runModule<T>(principal, es, streamID, parentContext, context);
        } catch (...) {
          exceptionPtr = std::current_exception();
        }
        this->waitingTasks_.doneWaiting(exceptionPtr);
      };
      if (auto queue = this->serializeRunModule()) {
        queue.push(toDo);
      } else {
        auto task = make_functor_task(tbb::task::allocate_root(), toDo);
        tbb::task::spawn(*task);
      }
    }
  }

void edm::Worker::emitPostModuleEventPrefetchingSignal ( )

inlineprivate

Definition at line 374 of file Worker.h.

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

void edm::Worker::endJob

(

void

)

Definition at line 299 of file Worker.cc.

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

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

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

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

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

staticprivate

Definition at line 104 of file Worker.cc.

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

Referenced by handleExternalWorkException(), and runAcquire().

                                                                                 {
    ModuleCallingContext const* imcc = mcc;
    while ((imcc->type() == ParentContext::Type::kModule) or (imcc->type() == ParentContext::Type::kInternal)) {
      std::ostringstream iost;
      if (imcc->state() == ModuleCallingContext::State::kPrefetching) {
        iost << "Prefetching for module ";
      } else {
        iost << "Calling method for module ";
      }
      iost << imcc->moduleDescription()->moduleName() << "/'" << imcc->moduleDescription()->moduleLabel() << "'";

      if (imcc->type() == ParentContext::Type::kInternal) {
        iost << " (probably inside some kind of mixing module)";
        imcc = imcc->internalContext()->moduleCallingContext();
      } else {
        imcc = imcc->moduleCallingContext();
      }
      ex.addContext(iost.str());
    }
    std::ostringstream ost;
    if (imcc->state() == ModuleCallingContext::State::kPrefetching) {
      ost << "Prefetching for module ";
    } else {
      ost << "Calling method for module ";
    }
    ost << imcc->moduleDescription()->moduleName() << "/'" << imcc->moduleDescription()->moduleLabel() << "'";
    ex.addContext(ost.str());

    if (imcc->type() == ParentContext::Type::kPlaceInPath) {
      ost.str("");
      ost << "Running path '";
      ost << imcc->placeInPathContext()->pathContext()->pathName() << "'";
      ex.addContext(ost.str());
      auto streamContext = imcc->placeInPathContext()->pathContext()->streamContext();
      if (streamContext) {
        ost.str("");
        edm::exceptionContext(ost, *streamContext);
        ex.addContext(ost.str());
      }
    } else {
      if (imcc->type() == ParentContext::Type::kStream) {
        ost.str("");
        edm::exceptionContext(ost, *(imcc->streamContext()));
        ex.addContext(ost.str());
      } else if (imcc->type() == ParentContext::Type::kGlobal) {
        ost.str("");
        edm::exceptionContext(ost, *(imcc->globalContext()));
        ex.addContext(ost.str());
      }
    }
  }

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

References exceptionContext(), moduleCallingContext_, ranAcquireWithoutException_, 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);
        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 >.

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 ( EventPrincipal const &  , EventSetupImpl const &  c, ModuleCallingContext const *  mcc  )

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

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

protectedpure virtual

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

Referenced by runAcquire().

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

protectedpure virtual

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

Referenced by prePrefetchSelectionAsync().

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

protectedpure virtual

Implemented in edm::WorkerT< T >.

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

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

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

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

privatepure virtual

Implemented in edm::WorkerT< T >.

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

privatepure virtual

Implemented in edm::WorkerT< T >.

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

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

pure virtual

Implemented in edm::WorkerT< T >.

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

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

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

delete

Referenced by edm::stream::EDFilterAdaptorBase::workerType(), and edm::stream::EDProducerAdaptorBase::workerType().

void edm::Worker::postDoEvent

(

EventPrincipal const &

iEvent

)

Definition at line 368 of file Worker.cc.

References earlyDeleteHelper_.

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

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

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

privatepure virtual

Implemented in edm::WorkerT< T >.

Referenced by prefetchAsync().

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

private

Definition at line 199 of file Worker.cc.

References actReg_, edm::Principal::branchType(), edm::ModuleCallingContext::getStreamContext(), edm::InEvent, mps_monitormerge::items, itemsToGetFrom(), edm::ModuleCallingContext::kPrefetching, moduleCallingContext_, preActionBeforeRunEventAsync(), edm::Principal::prefetchAsync(), edm::ProductResolverIndexAmbiguous, and edm::ModuleCallingContext::setContext().

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

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

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

    //Need to be sure the ref count isn't set to 0 immediately
    iTask->increment_ref_count();
    for (auto const& item : items) {
      ProductResolverIndex productResolverIndex = item.productResolverIndex();
      bool skipCurrentProcess = item.skipCurrentProcess();
      if (productResolverIndex != ProductResolverIndexAmbiguous) {
        iPrincipal.prefetchAsync(iTask, productResolverIndex, skipCurrentProcess, token, &moduleCallingContext_);
      }
    }

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

    if (0 == iTask->decrement_ref_count()) {
      //if everything finishes before we leave this routine, we need to launch the task
      tbb::task::spawn(*iTask);
    }
  }

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

Definition at line 232 of file Worker.cc.

References pyrootRender::destroy(), edm::WaitingTaskList::doneWaiting(), implDoPrePrefetchSelection(), mps_monitormerge::items, itemsToGetForSelection(), edm::make_waiting_task(), moduleCallingContext_, edm::Principal::prefetchAsync(), edm::ProductResolverIndexAmbiguous, timesRun_, and waitingTasks_.

                                                                           {
    successTask->increment_ref_count();

    auto choiceTask = edm::make_waiting_task(
        tbb::task::allocate_root(), [id, successTask, iPrincipal, this, token](std::exception_ptr const*) {
          ServiceRegistry::Operate guard(token);
          try {
            if (not implDoPrePrefetchSelection(id, *iPrincipal, &moduleCallingContext_)) {
              timesRun_.fetch_add(1, std::memory_order_relaxed);
              setPassed<true>();
              waitingTasks_.doneWaiting(nullptr);
              //TBB requires that destroyed tasks have count 0
              if (0 == successTask->decrement_ref_count()) {
                tbb::task::destroy(*successTask);
              }
              return;
            }
          } catch (...) {
          }
          if (0 == successTask->decrement_ref_count()) {
            tbb::task::spawn(*successTask);
          }
        });

    WaitingTaskHolder choiceHolder{choiceTask};

    std::vector<ProductResolverIndexAndSkipBit> items;
    itemsToGetForSelection(items);

    for (auto const& item : items) {
      ProductResolverIndex productResolverIndex = item.productResolverIndex();
      bool skipCurrentProcess = item.skipCurrentProcess();
      if (productResolverIndex != ProductResolverIndexAmbiguous) {
        iPrincipal->prefetchAsync(choiceTask, productResolverIndex, skipCurrentProcess, token, &moduleCallingContext_);
      }
    }
    choiceHolder.doneWaiting(std::exception_ptr{});
  }

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

inline

Definition at line 138 of file Worker.h.

                                                                                                         {
      assert(false);
    }

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

inline

Definition at line 176 of file Worker.h.

                                                                                                         {
      implRegisterThinnedAssociations(registry, helper);
    }

void edm::Worker::reset ( void  )

inline

Definition at line 180 of file Worker.h.

References Ready.

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

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

{ implRespondToCloseInputFile(fb); }

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

inline

Definition at line 173 of file Worker.h.

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

{ implRespondToOpenInputFile(fb); }

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

private

Definition at line 374 of file Worker.cc.

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

Referenced by runAcquireAfterAsyncPrefetch().

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

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

private

Definition at line 391 of file Worker.cc.

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

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

template<typename T >

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

private

Definition at line 1085 of file Worker.h.

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

                                                           {
    //unscheduled producers should advance this
    //if (T::isEvent_) {
    //  ++timesVisited_;
    //}
    ModuleContextSentry moduleContextSentry(&moduleCallingContext_, parentContext);
    if (T::isEvent_) {
      timesRun_.fetch_add(1, std::memory_order_relaxed);
    }

    bool rc = true;
    try {
      convertException::wrap([&]() {
        rc = workerhelper::CallImpl<T>::call(this, streamID, ep, es, actReg_.get(), &moduleCallingContext_, context);

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

    return rc;
  }

template<typename T >

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

private

Definition at line 897 of file Worker.h.

References edm::ModuleCallingContext::kInvalid, and edm::ModuleCallingContext::setContext().

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

template<typename T >

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

Definition at line 1126 of file Worker.h.

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, ep, es, streamID, parentContext, context);
  }

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

void edm::Worker::setActivityRegistry

(

std::shared_ptr< ActivityRegistry >

areg

)

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

Definition at line 102 of file Worker.cc.

References actReg_.

{ actReg_ = areg; }

void edm::Worker::setEarlyDeleteHelper

(

EarlyDeleteHelper *

iHelper

)

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

References Exception.

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

References Fail.

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

References Pass.

                     {
      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, TransitionIDValueBase const &  iID  ) const

private

Definition at line 156 of file Worker.cc.

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

Referenced by runAcquire(), and runAcquireAfterAsyncPrefetch().

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

(

)

Definition at line 362 of file Worker.cc.

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

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

                          {
    if (0 == --numberOfPathsLeftToRun_) {
      waitingTasks_.doneWaiting(cached_exception_);
    }
  }

State edm::Worker::state ( ) const

inline

Definition at line 232 of file Worker.h.

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

{ return state_; }

int edm::Worker::timesExcept ( ) const

inline

Definition at line 231 of file Worker.h.

Referenced by edm::fillWorkerSummaryAux().

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

int edm::Worker::timesFailed ( ) const

inline

Definition at line 230 of file Worker.h.

Referenced by edm::fillWorkerSummaryAux().

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

int edm::Worker::timesPass ( ) const

inline

Definition at line 234 of file Worker.h.

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

int edm::Worker::timesPassed ( ) const

inline

Definition at line 229 of file Worker.h.

Referenced by edm::fillWorkerSummaryAux().

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

int edm::Worker::timesRun ( ) const

inline

Definition at line 227 of file Worker.h.

Referenced by edm::fillWorkerSummaryAux().

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

int edm::Worker::timesVisited ( ) const

inline

Definition at line 228 of file Worker.h.

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

Member Data Documentation

ExceptionToActionTable const* edm::Worker::actions_

private

Definition at line 611 of file Worker.h.

Referenced by shouldRethrowException().

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

private

Definition at line 614 of file Worker.h.

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

std::exception_ptr edm::Worker::cached_exception_

private

Definition at line 612 of file Worker.h.

Referenced by skipOnPath().

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

private

Definition at line 616 of file Worker.h.

Referenced by postDoEvent(), and setEarlyDeleteHelper().

ModuleCallingContext edm::Worker::moduleCallingContext_

private

Definition at line 609 of file Worker.h.

Referenced by beginStream(), endStream(), handleExternalWorkException(), prefetchAsync(), prePrefetchSelectionAsync(), resetModuleDescription(), runAcquire(), and runAcquireAfterAsyncPrefetch().

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

private

Definition at line 607 of file Worker.h.

Referenced by skipOnPath().

int edm::Worker::numberOfPathsOn_

private

Definition at line 606 of file Worker.h.

bool edm::Worker::ranAcquireWithoutException_

private

Definition at line 620 of file Worker.h.

Referenced by handleExternalWorkException(), and runAcquireAfterAsyncPrefetch().

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

private

Definition at line 605 of file Worker.h.

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

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

private

Definition at line 604 of file Worker.h.

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

private

Definition at line 603 of file Worker.h.

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

private

Definition at line 602 of file Worker.h.

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

private

Definition at line 600 of file Worker.h.

Referenced by prePrefetchSelectionAsync(), and runAcquire().

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

private

Definition at line 601 of file Worker.h.

edm::WaitingTaskList edm::Worker::waitingTasks_

private

Definition at line 618 of file Worker.h.

Referenced by prePrefetchSelectionAsync(), and skipOnPath().

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

private

Definition at line 619 of file Worker.h.