edm::EventProcessor Class Reference

#include <EventProcessor.h>

Public Types
using	ProcessBlockType = PrincipalCache::ProcessBlockType
enum	StatusCode {   epSuccess = 0, epException = 1, epOther = 2, epSignal = 3,   epInputComplete = 4, epTimedOut = 5, epCountComplete = 6 }

Public Member Functions
void	beginJob ()
void	beginLumiAsync (edm::IOVSyncValue const &iSyncValue, std::shared_ptr< void > const &iRunResource, edm::WaitingTaskHolder iHolder)
void	beginProcessBlock (bool &beginProcessBlockSucceeded)
void	beginRun (ProcessHistoryID const &phid, RunNumber_t run, bool &globalBeginSucceeded, bool &eventSetupForInstanceSucceeded)
void	clearCounters ()
	Clears counters used by trigger report. More...
void	closeInputFile (bool cleaningUpAfterException)
void	closeOutputFiles ()
void	continueLumiAsync (edm::WaitingTaskHolder iHolder)
void	deleteLumiFromCache (LuminosityBlockProcessingStatus &)
void	deleteRunFromCache (ProcessHistoryID const &phid, RunNumber_t run)
void	doErrorStuff ()
void	enableEndPaths (bool active)
void	endJob ()
bool	endOfLoop ()
bool	endPathsEnabled () const
void	endProcessBlock (bool cleaningUpAfterException, bool beginProcessBlockSucceeded)
void	endRun (ProcessHistoryID const &phid, RunNumber_t run, bool globalBeginSucceeded, bool cleaningUpAfterException)
void	endUnfinishedLumi ()
void	endUnfinishedRun (ProcessHistoryID const &phid, RunNumber_t run, bool globalBeginSucceeded, bool cleaningUpAfterException, bool eventSetupForInstanceSucceeded)
	EventProcessor (EventProcessor const &)=delete
	EventProcessor (std::shared_ptr< ProcessDesc > processDesc, ServiceToken const &token, serviceregistry::ServiceLegacy legacy)
	EventProcessor (std::unique_ptr< ParameterSet > parameterSet, ServiceToken const &token=ServiceToken(), serviceregistry::ServiceLegacy=serviceregistry::kOverlapIsError, std::vector< std::string > const &defaultServices=std::vector< std::string >(), std::vector< std::string > const &forcedServices=std::vector< std::string >())
	EventProcessor (std::unique_ptr< ParameterSet > parameterSet, std::vector< std::string > const &defaultServices, std::vector< std::string > const &forcedServices=std::vector< std::string >())
std::vector< ModuleDescription const * >	getAllModuleDescriptions () const
ServiceToken	getToken ()
void	globalEndLumiAsync (edm::WaitingTaskHolder iTask, std::shared_ptr< LuminosityBlockProcessingStatus > iLumiStatus)
void	handleEndLumiExceptions (std::exception_ptr const *iPtr, WaitingTaskHolder &holder)
void	inputProcessBlocks ()
InputSource::ItemType	lastTransitionType () const
edm::LuminosityBlockNumber_t	nextLuminosityBlockID ()
std::pair< edm::ProcessHistoryID, edm::RunNumber_t >	nextRunID ()
InputSource::ItemType	nextTransitionType ()
void	openOutputFiles ()
EventProcessor &	operator= (EventProcessor const &)=delete
void	prepareForNextLoop ()
ProcessConfiguration const &	processConfiguration () const
InputSource::ItemType	processLumis (std::shared_ptr< void > const &iRunResource)
int	readAndMergeLumi (LuminosityBlockProcessingStatus &)
std::pair< ProcessHistoryID, RunNumber_t >	readAndMergeRun ()
void	readFile ()
void	readLuminosityBlock (LuminosityBlockProcessingStatus &)
std::pair< ProcessHistoryID, RunNumber_t >	readRun ()
void	respondToCloseInputFile ()
void	respondToOpenInputFile ()
void	rewindInput ()
StatusCode	run ()
StatusCode	runToCompletion ()
bool	setDeferredException (std::exception_ptr)
void	setExceptionMessageFiles (std::string &message)
void	setExceptionMessageLumis ()
void	setExceptionMessageRuns (std::string &message)
bool	shouldWeCloseOutput () const
bool	shouldWeStop () const
void	startingNewLoop ()
void	streamEndLumiAsync (edm::WaitingTaskHolder iTask, unsigned int iStreamIndex)
void	taskCleanup ()
int	totalEvents () const
int	totalEventsFailed () const
int	totalEventsPassed () const
void	writeLumiAsync (WaitingTaskHolder, LuminosityBlockPrincipal &lumiPrincipal)
void	writeProcessBlockAsync (WaitingTaskHolder, ProcessBlockType)
void	writeRunAsync (WaitingTaskHolder, ProcessHistoryID const &phid, RunNumber_t run, MergeableRunProductMetadata const *)
	~EventProcessor ()

Private Types
typedef std::set< std::pair< std::string, std::string > >	ExcludedData
typedef std::map< std::string, ExcludedData >	ExcludedDataMap

Private Member Functions
std::shared_ptr< BranchIDListHelper > &	branchIDListHelper ()
std::shared_ptr< BranchIDListHelper const >	branchIDListHelper () const
bool	checkForAsyncStopRequest (StatusCode &)
void	handleNextEventForStreamAsync (WaitingTaskHolder iTask, unsigned int iStreamIndex)
void	init (std::shared_ptr< ProcessDesc > &processDesc, ServiceToken const &token, serviceregistry::ServiceLegacy)
std::shared_ptr< EDLooperBase > &	looper ()
std::shared_ptr< EDLooperBase const >	looper () const
std::shared_ptr< ProductRegistry > &	preg ()
std::shared_ptr< ProductRegistry const >	preg () const
void	processEventAsync (WaitingTaskHolder iHolder, unsigned int iStreamIndex)
void	processEventAsyncImpl (WaitingTaskHolder iHolder, unsigned int iStreamIndex)
void	processEventWithLooper (EventPrincipal &, unsigned int iStreamIndex)
void	readEvent (unsigned int iStreamIndex)
bool	readNextEventForStream (unsigned int iStreamIndex, LuminosityBlockProcessingStatus &iLumiStatus)
std::shared_ptr< ThinnedAssociationsHelper > &	thinnedAssociationsHelper ()
std::shared_ptr< ThinnedAssociationsHelper const >	thinnedAssociationsHelper () const
void	warnAboutModulesRequiringLuminosityBLockSynchronization () const

Private Attributes
std::unique_ptr< ExceptionToActionTable const >	act_table_
std::shared_ptr< ActivityRegistry >	actReg_
bool	asyncStopRequestedWhileProcessingEvents_
StatusCode	asyncStopStatusCodeFromProcessingEvents_
bool	beginJobCalled_
edm::propagate_const< std::shared_ptr< BranchIDListHelper > >	branchIDListHelper_
std::exception_ptr	deferredExceptionPtr_
std::atomic< bool >	deferredExceptionPtrIsSet_
edm::propagate_const< std::shared_ptr< eventsetup::EventSetupProvider > >	esp_
edm::propagate_const< std::unique_ptr< eventsetup::EventSetupsController > >	espController_
ExcludedDataMap	eventSetupDataToExcludeFromPrefetching_
std::string	exceptionMessageFiles_
std::atomic< bool >	exceptionMessageLumis_
std::string	exceptionMessageRuns_
edm::propagate_const< std::unique_ptr< FileBlock > >	fb_
bool	fileModeNoMerge_
bool	firstEventInBlock_ = true
bool	forceESCacheClearOnNewRun_
bool	forceLooperToEnd_
edm::propagate_const< std::unique_ptr< HistoryAppender > >	historyAppender_
edm::propagate_const< std::unique_ptr< InputSource > >	input_
InputSource::ItemType	lastSourceTransition_
edm::propagate_const< std::shared_ptr< EDLooperBase > >	looper_
bool	looperBeginJobRun_
std::unique_ptr< edm::LimitedTaskQueue >	lumiQueue_
MergeableRunProductProcesses	mergeableRunProductProcesses_
PathsAndConsumesOfModules	pathsAndConsumesOfModules_
PreallocationConfiguration	preallocations_
edm::propagate_const< std::shared_ptr< ProductRegistry > >	preg_
PrincipalCache	principalCache_
bool	printDependencies_ = false
std::shared_ptr< ProcessConfiguration const >	processConfiguration_
ProcessContext	processContext_
edm::SerialTaskQueue	queueWhichWaitsForIOVsToFinish_
edm::propagate_const< std::unique_ptr< Schedule > >	schedule_
ServiceToken	serviceToken_
bool	shouldWeStop_
std::shared_ptr< std::recursive_mutex >	sourceMutex_
SharedResourcesAcquirer	sourceResourcesAcquirer_
std::atomic< unsigned int >	streamLumiActive_ {0}
std::vector< std::shared_ptr< LuminosityBlockProcessingStatus > >	streamLumiStatus_
std::vector< edm::SerialTaskQueue >	streamQueues_
std::vector< SubProcess >	subProcesses_
edm::propagate_const< std::shared_ptr< ThinnedAssociationsHelper > >	thinnedAssociationsHelper_

Detailed Description

Definition at line 64 of file EventProcessor.h.

Member Typedef Documentation

ExcludedData

typedef std::set<std::pair<std::string, std::string> > edm::EventProcessor::ExcludedData

private

Definition at line 362 of file EventProcessor.h.

ExcludedDataMap

typedef std::map<std::string, ExcludedData> edm::EventProcessor::ExcludedDataMap

private

Definition at line 363 of file EventProcessor.h.

ProcessBlockType

using edm::EventProcessor::ProcessBlockType = PrincipalCache::ProcessBlockType

Definition at line 247 of file EventProcessor.h.

Member Enumeration Documentation

StatusCode

enum edm::EventProcessor::StatusCode

Enumerator
epSuccess
epException
epOther
epSignal
epInputComplete
epTimedOut
epCountComplete

Definition at line 74 of file EventProcessor.h.

                    {
      epSuccess = 0,
      epException = 1,
      epOther = 2,
      epSignal = 3,
      epInputComplete = 4,
      epTimedOut = 5,
      epCountComplete = 6
    };

Constructor & Destructor Documentation

EventProcessor() [1/4]

edm::EventProcessor::EventProcessor ( std::unique_ptr< ParameterSet >  parameterSet, ServiceToken const &  token = ServiceToken(), serviceregistry::ServiceLegacy  iLegacy = serviceregistry::kOverlapIsError, std::vector< std::string > const &  defaultServices = std::vector<std::string>(), std::vector< std::string > const &  forcedServices = std::vector<std::string>()  )

explicit

Definition at line 218 of file EventProcessor.cc.

      : actReg_(),
        preg_(),
        branchIDListHelper_(),
        serviceToken_(),
        input_(),
        espController_(new eventsetup::EventSetupsController),
        esp_(),
        act_table_(),
        processConfiguration_(),
        schedule_(),
        subProcesses_(),
        historyAppender_(new HistoryAppender),
        fb_(),
        looper_(),
        deferredExceptionPtrIsSet_(false),
        sourceResourcesAcquirer_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().first),
        sourceMutex_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().second),
        principalCache_(),
        beginJobCalled_(false),
        shouldWeStop_(false),
        fileModeNoMerge_(false),
        exceptionMessageFiles_(),
        exceptionMessageRuns_(),
        exceptionMessageLumis_(false),
        forceLooperToEnd_(false),
        looperBeginJobRun_(false),
        forceESCacheClearOnNewRun_(false),
        eventSetupDataToExcludeFromPrefetching_() {
    auto processDesc = std::make_shared<ProcessDesc>(std::move(parameterSet));
    processDesc->addServices(defaultServices, forcedServices);
    init(processDesc, iToken, iLegacy);
  }

References init(), eostools::move(), and edm::parameterSet().

EventProcessor() [2/4]

edm::EventProcessor::EventProcessor	(	std::unique_ptr< ParameterSet >	parameterSet,
		std::vector< std::string > const &	defaultServices,
		std::vector< std::string > const &	forcedServices = std::vector<std::string>()
	)

Definition at line 256 of file EventProcessor.cc.

      : actReg_(),
        preg_(),
        branchIDListHelper_(),
        serviceToken_(),
        input_(),
        espController_(new eventsetup::EventSetupsController),
        esp_(),
        act_table_(),
        processConfiguration_(),
        schedule_(),
        subProcesses_(),
        historyAppender_(new HistoryAppender),
        fb_(),
        looper_(),
        deferredExceptionPtrIsSet_(false),
        sourceResourcesAcquirer_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().first),
        sourceMutex_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().second),
        principalCache_(),
        beginJobCalled_(false),
        shouldWeStop_(false),
        fileModeNoMerge_(false),
        exceptionMessageFiles_(),
        exceptionMessageRuns_(),
        exceptionMessageLumis_(false),
        forceLooperToEnd_(false),
        looperBeginJobRun_(false),
        forceESCacheClearOnNewRun_(false),
        asyncStopRequestedWhileProcessingEvents_(false),
        eventSetupDataToExcludeFromPrefetching_() {
    auto processDesc = std::make_shared<ProcessDesc>(std::move(parameterSet));
    processDesc->addServices(defaultServices, forcedServices);
    init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
  }

References init(), edm::serviceregistry::kOverlapIsError, eostools::move(), and edm::parameterSet().

EventProcessor() [3/4]

edm::EventProcessor::EventProcessor	(	std::shared_ptr< ProcessDesc >	processDesc,
		ServiceToken const &	token,
		serviceregistry::ServiceLegacy	legacy
	)

Definition at line 293 of file EventProcessor.cc.

      : actReg_(),
        preg_(),
        branchIDListHelper_(),
        serviceToken_(),
        input_(),
        espController_(new eventsetup::EventSetupsController),
        esp_(),
        act_table_(),
        processConfiguration_(),
        schedule_(),
        subProcesses_(),
        historyAppender_(new HistoryAppender),
        fb_(),
        looper_(),
        deferredExceptionPtrIsSet_(false),
        sourceResourcesAcquirer_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().first),
        sourceMutex_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().second),
        principalCache_(),
        beginJobCalled_(false),
        shouldWeStop_(false),
        fileModeNoMerge_(false),
        exceptionMessageFiles_(),
        exceptionMessageRuns_(),
        exceptionMessageLumis_(false),
        forceLooperToEnd_(false),
        looperBeginJobRun_(false),
        forceESCacheClearOnNewRun_(false),
        asyncStopRequestedWhileProcessingEvents_(false),
        eventSetupDataToExcludeFromPrefetching_() {
    init(processDesc, token, legacy);
  }

References init(), and unpackBuffers-CaloStage2::token.

~EventProcessor()

edm::EventProcessor::~EventProcessor

(

)

Definition at line 515 of file EventProcessor.cc.

                                  {
    // Make the services available while everything is being deleted.
    ServiceToken token = getToken();
    ServiceRegistry::Operate op(token);
 
    // manually destroy all these thing that may need the services around
    // propagate_const<T> has no reset() function
    espController_ = nullptr;
    esp_ = nullptr;
    schedule_ = nullptr;
    input_ = nullptr;
    looper_ = nullptr;
    actReg_ = nullptr;
 
    pset::Registry::instance()->clear();
    ParentageRegistry::instance()->clear();
  }

References actReg_, edm::ParentageRegistry::clear(), edm::pset::Registry::clear(), esp_, espController_, getToken(), input_, edm::ParentageRegistry::instance(), edm::pset::Registry::instance(), looper_, schedule_, and unpackBuffers-CaloStage2::token.

EventProcessor() [4/4]

edm::EventProcessor::EventProcessor ( EventProcessor const &  )

delete

Member Function Documentation

beginJob()

void edm::EventProcessor::beginJob

(

void

)

This should be called before the first call to 'run' If this is not called in time, it will automatically be called the first time 'run' is called

Definition at line 535 of file EventProcessor.cc.

                                {
    if (beginJobCalled_)
      return;
    beginJobCalled_ = true;
    bk::beginJob();
 
    // StateSentry toerror(this); // should we add this ?
    //make the services available
    ServiceRegistry::Operate operate(serviceToken_);
 
    service::SystemBounds bounds(preallocations_.numberOfStreams(),
                                 preallocations_.numberOfLuminosityBlocks(),
                                 preallocations_.numberOfRuns(),
                                 preallocations_.numberOfThreads());
    actReg_->preallocateSignal_(bounds);
    schedule_->convertCurrentProcessAlias(processConfiguration_->processName());
    pathsAndConsumesOfModules_.initialize(schedule_.get(), preg());
 
    //NOTE: this may throw
    checkForModuleDependencyCorrectness(pathsAndConsumesOfModules_, printDependencies_);
    actReg_->preBeginJobSignal_(pathsAndConsumesOfModules_, processContext_);
 
    if (preallocations_.numberOfLuminosityBlocks() > 1) {
      warnAboutModulesRequiringLuminosityBLockSynchronization();
    }
    //NOTE:  This implementation assumes 'Job' means one call
    // the EventProcessor::run
    // If it really means once per 'application' then this code will
    // have to be changed.
    // Also have to deal with case where have 'run' then new Module
    // added and do 'run'
    // again.  In that case the newly added Module needs its 'beginJob'
    // to be called.
 
    //NOTE: in future we should have a beginOfJob for looper that takes no arguments
    //  For now we delay calling beginOfJob until first beginOfRun
    //if(looper_) {
    //   looper_->beginOfJob(es);
    //}
    try {
      convertException::wrap([&]() { input_->doBeginJob(); });
    } catch (cms::Exception& ex) {
      ex.addContext("Calling beginJob for the source");
      throw;
    }
    espController_->finishConfiguration();
    schedule_->beginJob(*preg_, esp_->recordsToProxyIndices());
    // toerror.succeeded(); // should we add this?
    for_all(subProcesses_, [](auto& subProcess) { subProcess.doBeginJob(); });
    actReg_->postBeginJobSignal_();
 
    for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
      schedule_->beginStream(i);
      for_all(subProcesses_, [i](auto& subProcess) { subProcess.doBeginStream(i); });
    }
  }

References actReg_, cms::Exception::addContext(), bk::beginJob(), beginJobCalled_, edm::checkForModuleDependencyCorrectness(), esp_, espController_, edm::for_all(), mps_fire::i, edm::PathsAndConsumesOfModules::initialize(), input_, edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfRuns(), edm::PreallocationConfiguration::numberOfStreams(), edm::PreallocationConfiguration::numberOfThreads(), pathsAndConsumesOfModules_, preallocations_, preg(), preg_, printDependencies_, processConfiguration_, processContext_, schedule_, serviceToken_, subProcesses_, warnAboutModulesRequiringLuminosityBLockSynchronization(), and edm::convertException::wrap().

Referenced by runToCompletion().

beginLumiAsync()

void edm::EventProcessor::beginLumiAsync	(	edm::IOVSyncValue const &	iSyncValue,
		std::shared_ptr< void > const &	iRunResource,
		edm::WaitingTaskHolder	iHolder
	)

Definition at line 1155 of file EventProcessor.cc.

                                                                    {
    if (iHolder.taskHasFailed()) {
      return;
    }
 
    // We must be careful with the status object here and in code this function calls. IF we want
    // endRun to be called, then we must call resetResources before the things waiting on
    // iHolder are allowed to proceed. Otherwise, there will be race condition (possibly causing
    // endRun to be called much later than it should be, because status is holding iRunResource).
    // Note that this must be done explicitly. Relying on the destructor does not work well
    // because the LimitedTaskQueue for the lumiWork holds the shared_ptr in each of its internal
    // queues, plus it is difficult to guarantee the destructor is called  before iHolder gets
    // destroyed inside this function and lumiWork.
    auto status =
        std::make_shared<LuminosityBlockProcessingStatus>(this, preallocations_.numberOfStreams(), iRunResource);
 
    auto lumiWork = [this, iHolder, status](edm::LimitedTaskQueue::Resumer iResumer) mutable {
      if (iHolder.taskHasFailed()) {
        status->resetResources();
        return;
      }
 
      status->setResumer(std::move(iResumer));
 
      sourceResourcesAcquirer_.serialQueueChain().push([this, iHolder, status = std::move(status)]() mutable {
        //make the services available
        ServiceRegistry::Operate operate(serviceToken_);
        // Caught exception is propagated via WaitingTaskHolder
        CMS_SA_ALLOW try {
          readLuminosityBlock(*status);
 
          LuminosityBlockPrincipal& lumiPrincipal = *status->lumiPrincipal();
          {
            SendSourceTerminationSignalIfException sentry(actReg_.get());
 
            input_->doBeginLumi(lumiPrincipal, &processContext_);
            sentry.completedSuccessfully();
          }
 
          Service<RandomNumberGenerator> rng;
          if (rng.isAvailable()) {
            LuminosityBlock lb(lumiPrincipal, ModuleDescription(), nullptr, false);
            rng->preBeginLumi(lb);
          }
 
          //Task to start the stream beginLumis
          auto beginStreamsTask = make_waiting_task(
              tbb::task::allocate_root(), [this, holder = iHolder, status](std::exception_ptr const* iPtr) mutable {
                if (iPtr) {
                  status->resetResources();
                  holder.doneWaiting(*iPtr);
                } else {
                  status->globalBeginDidSucceed();
                  EventSetupImpl const& es = status->eventSetupImpl(esp_->subProcessIndex());
 
                  if (looper_) {
                    // Caught exception is propagated via WaitingTaskHolder
                    CMS_SA_ALLOW try {
                      //make the services available
                      ServiceRegistry::Operate operateLooper(serviceToken_);
                      looper_->doBeginLuminosityBlock(*(status->lumiPrincipal()), es, &processContext_);
                    } catch (...) {
                      status->resetResources();
                      holder.doneWaiting(std::current_exception());
                      return;
                    }
                  }
                  using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin>;
 
                  for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
                    streamQueues_[i].push([this, i, status, holder, &es]() mutable {
                      streamQueues_[i].pause();
 
                      auto eventTask =
                          edm::make_waiting_task(tbb::task::allocate_root(),
                                                 [this, i, h = std::move(holder)](
                                                     std::exception_ptr const* exceptionFromBeginStreamLumi) mutable {
                                                   if (exceptionFromBeginStreamLumi) {
                                                     WaitingTaskHolder tmp(h);
                                                     tmp.doneWaiting(*exceptionFromBeginStreamLumi);
                                                     streamEndLumiAsync(h, i);
                                                   } else {
                                                     handleNextEventForStreamAsync(std::move(h), i);
                                                   }
                                                 });
                      auto& event = principalCache_.eventPrincipal(i);
                      //We need to be sure that 'status' and its internal shared_ptr<LuminosityBlockPrincipal> are only
                      // held by the container as this lambda may not finish executing before all the tasks it
                      // spawns have already started to run.
                      auto eventSetupImpls = &status->eventSetupImpls();
                      auto lp = status->lumiPrincipal().get();
                      streamLumiStatus_[i] = std::move(status);
                      ++streamLumiActive_;
                      event.setLuminosityBlockPrincipal(lp);
                      LumiTransitionInfo transitionInfo(*lp, es, eventSetupImpls);
                      beginStreamTransitionAsync<Traits>(
                          WaitingTaskHolder{eventTask}, *schedule_, i, transitionInfo, serviceToken_, subProcesses_);
                    });
                  }
                }
              });  // beginStreamTask
 
          //task to start the global begin lumi
          WaitingTaskHolder beginStreamsHolder{beginStreamsTask};
 
          EventSetupImpl const& es = status->eventSetupImpl(esp_->subProcessIndex());
          {
            LumiTransitionInfo transitionInfo(lumiPrincipal, es, &status->eventSetupImpls());
            using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin>;
            beginGlobalTransitionAsync<Traits>(
                beginStreamsHolder, *schedule_, transitionInfo, serviceToken_, subProcesses_);
          }
        } catch (...) {
          status->resetResources();
          iHolder.doneWaiting(std::current_exception());
        }
      });  // task in sourceResourcesAcquirer
    };     // end lumiWork
 
    auto queueLumiWorkTask = make_waiting_task(
        tbb::task::allocate_root(),
        [this, lumiWorkLambda = std::move(lumiWork), iHolder](std::exception_ptr const* iPtr) mutable {
          if (iPtr) {
            iHolder.doneWaiting(*iPtr);
          }
          lumiQueue_->pushAndPause(std::move(lumiWorkLambda));
        });
 
    if (espController_->doWeNeedToWaitForIOVsToFinish(iSync)) {
      // We only get here inside this block if there is an EventSetup
      // module not able to handle concurrent IOVs (usually an ESSource)
      // and the new sync value is outside the current IOV of that module.
 
      WaitingTaskHolder queueLumiWorkTaskHolder{queueLumiWorkTask};
 
      queueWhichWaitsForIOVsToFinish_.push([this, queueLumiWorkTaskHolder, iSync, status]() mutable {
        // Caught exception is propagated via WaitingTaskHolder
        CMS_SA_ALLOW try {
          SendSourceTerminationSignalIfException sentry(actReg_.get());
          // Pass in iSync to let the EventSetup system know which run and lumi
          // need to be processed and prepare IOVs for it.
          // Pass in the endIOVWaitingTasks so the lumi can notify them when the
          // lumi is done and no longer needs its EventSetup IOVs.
          espController_->eventSetupForInstance(
              iSync, queueLumiWorkTaskHolder, status->endIOVWaitingTasks(), status->eventSetupImpls());
          sentry.completedSuccessfully();
        } catch (...) {
          queueLumiWorkTaskHolder.doneWaiting(std::current_exception());
        }
        queueWhichWaitsForIOVsToFinish_.pause();
      });
 
    } else {
      queueWhichWaitsForIOVsToFinish_.pause();
 
      // This holder will be used to wait until the EventSetup IOVs are ready
      WaitingTaskHolder queueLumiWorkTaskHolder{queueLumiWorkTask};
      // Caught exception is propagated via WaitingTaskHolder
      CMS_SA_ALLOW try {
        SendSourceTerminationSignalIfException sentry(actReg_.get());
 
        // Pass in iSync to let the EventSetup system know which run and lumi
        // need to be processed and prepare IOVs for it.
        // Pass in the endIOVWaitingTasks so the lumi can notify them when the
        // lumi is done and no longer needs its EventSetup IOVs.
        espController_->eventSetupForInstance(
            iSync, queueLumiWorkTaskHolder, status->endIOVWaitingTasks(), status->eventSetupImpls());
        sentry.completedSuccessfully();
 
      } catch (...) {
        queueLumiWorkTaskHolder.doneWaiting(std::current_exception());
      }
    }
  }

References actReg_, CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), esp_, espController_, edm::PrincipalCache::eventPrincipal(), handleNextEventForStreamAsync(), mps_fire::i, input_, edm::Service< T >::isAvailable(), looper_, lumiQueue_, edm::make_waiting_task(), eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), edm::SerialTaskQueue::pause(), preallocations_, principalCache_, processContext_, edm::SerialTaskQueueChain::push(), edm::SerialTaskQueue::push(), queueWhichWaitsForIOVsToFinish_, readLuminosityBlock(), schedule_, edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, sourceResourcesAcquirer_, mps_update::status, streamEndLumiAsync(), streamLumiActive_, streamLumiStatus_, streamQueues_, subProcesses_, edm::WaitingTaskHolder::taskHasFailed(), and createJobs::tmp.

Referenced by handleNextEventForStreamAsync(), and processLumis().

beginProcessBlock()

void edm::EventProcessor::beginProcessBlock

(

bool &

beginProcessBlockSucceeded

)

Definition at line 876 of file EventProcessor.cc.

                                                                         {
    ProcessBlockPrincipal& processBlockPrincipal = principalCache_.processBlockPrincipal();
    processBlockPrincipal.fillProcessBlockPrincipal(processConfiguration_->processName());
 
    using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionGlobalBegin>;
    auto globalWaitTask = make_empty_waiting_task();
    globalWaitTask->increment_ref_count();
 
    ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
    beginGlobalTransitionAsync<Traits>(
        WaitingTaskHolder(globalWaitTask.get()), *schedule_, transitionInfo, serviceToken_, subProcesses_);
 
    globalWaitTask->wait_for_all();
    if (globalWaitTask->exceptionPtr() != nullptr) {
      std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
    }
    beginProcessBlockSucceeded = true;
  }

References edm::ProcessBlockPrincipal::fillProcessBlockPrincipal(), edm::make_empty_waiting_task(), principalCache_, edm::PrincipalCache::processBlockPrincipal(), processConfiguration_, schedule_, serviceToken_, and subProcesses_.

beginRun()

void edm::EventProcessor::beginRun	(	ProcessHistoryID const &	phid,
		RunNumber_t	run,
		bool &	globalBeginSucceeded,
		bool &	eventSetupForInstanceSucceeded
	)

Definition at line 970 of file EventProcessor.cc.

                                                                      {
    globalBeginSucceeded = false;
    RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, run);
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
 
      input_->doBeginRun(runPrincipal, &processContext_);
      sentry.completedSuccessfully();
    }
 
    IOVSyncValue ts(EventID(runPrincipal.run(), 0, 0), runPrincipal.beginTime());
    if (forceESCacheClearOnNewRun_) {
      espController_->forceCacheClear();
    }
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      espController_->eventSetupForInstance(ts);
      eventSetupForInstanceSucceeded = true;
      sentry.completedSuccessfully();
    }
    auto const& es = esp_->eventSetupImpl();
    if (looper_ && looperBeginJobRun_ == false) {
      looper_->copyInfo(ScheduleInfo(schedule_.get()));
      looper_->beginOfJob(es);
      looperBeginJobRun_ = true;
      looper_->doStartingNewLoop();
    }
    {
      using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin>;
      auto globalWaitTask = make_empty_waiting_task();
      globalWaitTask->increment_ref_count();
      RunTransitionInfo transitionInfo(runPrincipal, es);
      beginGlobalTransitionAsync<Traits>(
          WaitingTaskHolder(globalWaitTask.get()), *schedule_, transitionInfo, serviceToken_, subProcesses_);
      globalWaitTask->wait_for_all();
      if (globalWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
      }
    }
    globalBeginSucceeded = true;
    FDEBUG(1) << "\tbeginRun " << run << "\n";
    if (looper_) {
      looper_->doBeginRun(runPrincipal, es, &processContext_);
    }
    {
      //To wait, the ref count has to be 1+#streams
      auto streamLoopWaitTask = make_empty_waiting_task();
      streamLoopWaitTask->increment_ref_count();
 
      using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamBegin>;
 
      RunTransitionInfo transitionInfo(runPrincipal, es);
      beginStreamsTransitionAsync<Traits>(streamLoopWaitTask.get(),
                                          *schedule_,
                                          preallocations_.numberOfStreams(),
                                          transitionInfo,
                                          serviceToken_,
                                          subProcesses_);
 
      streamLoopWaitTask->wait_for_all();
      if (streamLoopWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(*(streamLoopWaitTask->exceptionPtr()));
      }
    }
    FDEBUG(1) << "\tstreamBeginRun " << run << "\n";
    if (looper_) {
      //looper_->doStreamBeginRun(schedule_->streamID(),runPrincipal, es);
    }
  }

References actReg_, edm::RunPrincipal::beginTime(), esp_, espController_, FDEBUG, forceESCacheClearOnNewRun_, input_, looper_, looperBeginJobRun_, edm::make_empty_waiting_task(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, edm::RunPrincipal::run(), run(), edm::PrincipalCache::runPrincipal(), schedule_, serviceToken_, and subProcesses_.

branchIDListHelper() [1/2]

std::shared_ptr<BranchIDListHelper>& edm::EventProcessor::branchIDListHelper ( )

inlineprivate

Definition at line 294 of file EventProcessor.h.

{ return get_underlying_safe(branchIDListHelper_); }

References branchIDListHelper_, and edm::get_underlying_safe().

branchIDListHelper() [2/2]

std::shared_ptr<BranchIDListHelper const> edm::EventProcessor::branchIDListHelper ( ) const

inlineprivate

Definition at line 291 of file EventProcessor.h.

                                                                       {
      return get_underlying_safe(branchIDListHelper_);
    }

References branchIDListHelper_, and edm::get_underlying_safe().

Referenced by init().

checkForAsyncStopRequest()

bool edm::EventProcessor::checkForAsyncStopRequest ( StatusCode &  returnCode )

private

Definition at line 645 of file EventProcessor.cc.

                                                                      {
    bool returnValue = false;
 
    // Look for a shutdown signal
    if (shutdown_flag.load(std::memory_order_acquire)) {
      returnValue = true;
      returnCode = epSignal;
    }
    return returnValue;
  }

References epSignal, runEdmFileComparison::returnCode, and edm::shutdown_flag.

Referenced by nextTransitionType().

clearCounters()

void edm::EventProcessor::clearCounters

(

)

Clears counters used by trigger report.

Definition at line 639 of file EventProcessor.cc.

{ schedule_->clearCounters(); }

References schedule_.

closeInputFile()

void edm::EventProcessor::closeInputFile

(

bool

cleaningUpAfterException

)

Definition at line 775 of file EventProcessor.cc.

                                                                   {
    if (fb_.get() != nullptr) {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->closeFile(fb_.get(), cleaningUpAfterException);
      sentry.completedSuccessfully();
    }
    FDEBUG(1) << "\tcloseInputFile\n";
  }

References actReg_, fb_, FDEBUG, and input_.

closeOutputFiles()

void edm::EventProcessor::closeOutputFiles

(

)

Definition at line 792 of file EventProcessor.cc.

                                        {
    if (fb_.get() != nullptr) {
      schedule_->closeOutputFiles();
      for_all(subProcesses_, [](auto& subProcess) { subProcess.closeOutputFiles(); });
    }
    FDEBUG(1) << "\tcloseOutputFiles\n";
  }

References fb_, FDEBUG, edm::for_all(), schedule_, and subProcesses_.

continueLumiAsync()

void edm::EventProcessor::continueLumiAsync

(

edm::WaitingTaskHolder

iHolder

)

Definition at line 1332 of file EventProcessor.cc.

                                                                     {
    {
      //all streams are sharing the same status at the moment
      auto status = streamLumiStatus_[0];  //read from streamLumiActive_ happened in calling routine
      status->needToContinueLumi();
      status->startProcessingEvents();
    }
 
    unsigned int streamIndex = 0;
    for (; streamIndex < preallocations_.numberOfStreams() - 1; ++streamIndex) {
      tbb::task::enqueue(*edm::make_functor_task(tbb::task::allocate_root(), [this, streamIndex, h = iHolder]() {
        handleNextEventForStreamAsync(h, streamIndex);
      }));
    }
    tbb::task::spawn(*edm::make_functor_task(tbb::task::allocate_root(), [this, streamIndex, h = std::move(iHolder)]() {
      handleNextEventForStreamAsync(h, streamIndex);
    }));
  }

References handleNextEventForStreamAsync(), edm::make_functor_task(), eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, mps_update::status, and streamLumiStatus_.

Referenced by processLumis().

deleteLumiFromCache()

void edm::EventProcessor::deleteLumiFromCache

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 1653 of file EventProcessor.cc.

                                                                                   {
    for (auto& s : subProcesses_) {
      s.deleteLumiFromCache(*iStatus.lumiPrincipal());
    }
    iStatus.lumiPrincipal()->clearPrincipal();
    //FDEBUG(1) << "\tdeleteLumiFromCache " << run << "/" << lumi << "\n";
  }

References edm::LuminosityBlockProcessingStatus::lumiPrincipal(), alignCSCRings::s, and subProcesses_.

Referenced by globalEndLumiAsync().

deleteRunFromCache()

void edm::EventProcessor::deleteRunFromCache	(	ProcessHistoryID const &	phid,
		RunNumber_t	run
	)

Definition at line 1628 of file EventProcessor.cc.

                                                                                       {
    principalCache_.deleteRun(phid, run);
    for_all(subProcesses_, [run, phid](auto& subProcess) { subProcess.deleteRunFromCache(phid, run); });
    FDEBUG(1) << "\tdeleteRunFromCache " << run << "\n";
  }

References edm::PrincipalCache::deleteRun(), FDEBUG, edm::for_all(), principalCache_, run(), and subProcesses_.

Referenced by endUnfinishedRun().

doErrorStuff()

void edm::EventProcessor::doErrorStuff

(

)

Definition at line 867 of file EventProcessor.cc.

                                    {
    FDEBUG(1) << "\tdoErrorStuff\n";
    LogError("StateMachine") << "The EventProcessor state machine encountered an unexpected event\n"
                             << "and went to the error state\n"
                             << "Will attempt to terminate processing normally\n"
                             << "(IF using the looper the next loop will be attempted)\n"
                             << "This likely indicates a bug in an input module or corrupted input or both\n";
  }

References FDEBUG.

Referenced by runToCompletion().

enableEndPaths()

void edm::EventProcessor::enableEndPaths

(

bool

active

)

Turn end_paths "off" if "active" is false; turn end_paths "on" if "active" is true.

Definition at line 635 of file EventProcessor.cc.

{ schedule_->enableEndPaths(active); }

References schedule_.

endJob()

void edm::EventProcessor::endJob

(

void

)

This should be called before the EventProcessor is destroyed throws if any module's endJob throws an exception.

Definition at line 592 of file EventProcessor.cc.

                              {
    // Collects exceptions, so we don't throw before all operations are performed.
    ExceptionCollector c(
        "Multiple exceptions were thrown while executing endJob. An exception message follows for each.\n");
 
    //make the services available
    ServiceRegistry::Operate operate(serviceToken_);
 
    //NOTE: this really should go elsewhere in the future
    for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
      c.call([this, i]() { this->schedule_->endStream(i); });
      for (auto& subProcess : subProcesses_) {
        c.call([&subProcess, i]() { subProcess.doEndStream(i); });
      }
    }
    auto actReg = actReg_.get();
    c.call([actReg]() { actReg->preEndJobSignal_(); });
    schedule_->endJob(c);
    for (auto& subProcess : subProcesses_) {
      c.call(std::bind(&SubProcess::doEndJob, &subProcess));
    }
    c.call(std::bind(&InputSource::doEndJob, input_.get()));
    if (looper_) {
      c.call(std::bind(&EDLooperBase::endOfJob, looper()));
    }
    c.call([actReg]() { actReg->postEndJobSignal_(); });
    if (c.hasThrown()) {
      c.rethrow();
    }
  }

References actReg_, HltBtagPostValidation_cff::c, edm::SubProcess::doEndJob(), edm::InputSource::doEndJob(), edm::EDLooperBase::endOfJob(), mps_fire::i, input_, looper(), looper_, edm::PreallocationConfiguration::numberOfStreams(), preallocations_, schedule_, serviceToken_, and subProcesses_.

Referenced by PythonEventProcessor::~PythonEventProcessor().

endOfLoop()

bool edm::EventProcessor::endOfLoop

(

)

Definition at line 828 of file EventProcessor.cc.

                                 {
    if (looper_) {
      ModuleChanger changer(schedule_.get(), preg_.get(), esp_->recordsToProxyIndices());
      looper_->setModuleChanger(&changer);
      EDLooperBase::Status status = looper_->doEndOfLoop(esp_->eventSetupImpl());
      looper_->setModuleChanger(nullptr);
      if (status != EDLooperBase::kContinue || forceLooperToEnd_)
        return true;
      else
        return false;
    }
    FDEBUG(1) << "\tendOfLoop\n";
    return true;
  }

References esp_, FDEBUG, forceLooperToEnd_, edm::EDLooperBase::kContinue, looper_, preg_, schedule_, and mps_update::status.

Referenced by runToCompletion().

endPathsEnabled()

bool edm::EventProcessor::endPathsEnabled

(

)

const

Return true if end_paths are active, and false if they are inactive.

Definition at line 637 of file EventProcessor.cc.

{ return schedule_->endPathsEnabled(); }

References schedule_.

endProcessBlock()

void edm::EventProcessor::endProcessBlock	(	bool	cleaningUpAfterException,
		bool	beginProcessBlockSucceeded
	)

Definition at line 934 of file EventProcessor.cc.

                                                                                                     {
    ProcessBlockPrincipal& processBlockPrincipal = principalCache_.processBlockPrincipal();
 
    using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionGlobalEnd>;
    auto globalWaitTask = make_empty_waiting_task();
    globalWaitTask->increment_ref_count();
 
    ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
    endGlobalTransitionAsync<Traits>(WaitingTaskHolder(globalWaitTask.get()),
                                     *schedule_,
                                     transitionInfo,
                                     serviceToken_,
                                     subProcesses_,
                                     cleaningUpAfterException);
 
    globalWaitTask->wait_for_all();
    if (globalWaitTask->exceptionPtr() != nullptr) {
      std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
    }
 
    if (beginProcessBlockSucceeded) {
      auto writeWaitTask = edm::make_empty_waiting_task();
      writeWaitTask->increment_ref_count();
      writeProcessBlockAsync(edm::WaitingTaskHolder{writeWaitTask.get()}, ProcessBlockType::New);
      writeWaitTask->wait_for_all();
      if (writeWaitTask->exceptionPtr()) {
        std::rethrow_exception(*writeWaitTask->exceptionPtr());
      }
    }
 
    processBlockPrincipal.clearPrincipal();
    for (auto& s : subProcesses_) {
      s.clearProcessBlockPrincipal(ProcessBlockType::New);
    }
  }

References edm::Principal::clearPrincipal(), edm::make_empty_waiting_task(), edm::PrincipalCache::New, principalCache_, edm::PrincipalCache::processBlockPrincipal(), alignCSCRings::s, schedule_, serviceToken_, subProcesses_, and writeProcessBlockAsync().

endRun()

void edm::EventProcessor::endRun	(	ProcessHistoryID const &	phid,
		RunNumber_t	run,
		bool	globalBeginSucceeded,
		bool	cleaningUpAfterException
	)

Definition at line 1069 of file EventProcessor.cc.

                                                             {
    RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, run);
    runPrincipal.setEndTime(input_->timestamp());
 
    IOVSyncValue ts(
        EventID(runPrincipal.run(), LuminosityBlockID::maxLuminosityBlockNumber(), EventID::maxEventNumber()),
        runPrincipal.endTime());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      espController_->eventSetupForInstance(ts);
      sentry.completedSuccessfully();
    }
    auto const& es = esp_->eventSetupImpl();
    if (globalBeginSucceeded) {
      //To wait, the ref count has to be 1+#streams
      auto streamLoopWaitTask = make_empty_waiting_task();
      streamLoopWaitTask->increment_ref_count();
 
      using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamEnd>;
 
      RunTransitionInfo transitionInfo(runPrincipal, es);
      endStreamsTransitionAsync<Traits>(WaitingTaskHolder(streamLoopWaitTask.get()),
                                        *schedule_,
                                        preallocations_.numberOfStreams(),
                                        transitionInfo,
                                        serviceToken_,
                                        subProcesses_,
                                        cleaningUpAfterException);
 
      streamLoopWaitTask->wait_for_all();
      if (streamLoopWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(*(streamLoopWaitTask->exceptionPtr()));
      }
    }
    FDEBUG(1) << "\tstreamEndRun " << run << "\n";
    if (looper_) {
      //looper_->doStreamEndRun(schedule_->streamID(),runPrincipal, es);
    }
    {
      auto globalWaitTask = make_empty_waiting_task();
      globalWaitTask->increment_ref_count();
 
      RunTransitionInfo transitionInfo(runPrincipal, es);
      using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd>;
      endGlobalTransitionAsync<Traits>(WaitingTaskHolder(globalWaitTask.get()),
                                       *schedule_,
                                       transitionInfo,
                                       serviceToken_,
                                       subProcesses_,
                                       cleaningUpAfterException);
      globalWaitTask->wait_for_all();
      if (globalWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
      }
    }
    FDEBUG(1) << "\tendRun " << run << "\n";
    if (looper_) {
      looper_->doEndRun(runPrincipal, es, &processContext_);
    }
  }

References actReg_, edm::RunPrincipal::endTime(), esp_, espController_, FDEBUG, input_, looper_, edm::make_empty_waiting_task(), edm::EventID::maxEventNumber(), edm::LuminosityBlockID::maxLuminosityBlockNumber(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, edm::RunPrincipal::run(), run(), edm::PrincipalCache::runPrincipal(), schedule_, serviceToken_, edm::RunPrincipal::setEndTime(), and subProcesses_.

Referenced by endUnfinishedRun().

endUnfinishedLumi()

void edm::EventProcessor::endUnfinishedLumi

(

)

Definition at line 1495 of file EventProcessor.cc.

                                         {
    if (streamLumiActive_.load() > 0) {
      auto globalWaitTask = make_empty_waiting_task();
      globalWaitTask->increment_ref_count();
      {
        WaitingTaskHolder globalTaskHolder{globalWaitTask.get()};
        for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
          if (streamLumiStatus_[i]) {
            streamEndLumiAsync(globalTaskHolder, i);
          }
        }
      }
      globalWaitTask->wait_for_all();
      if (globalWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
      }
    }
  }

References mps_fire::i, edm::make_empty_waiting_task(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, streamEndLumiAsync(), streamLumiActive_, and streamLumiStatus_.

endUnfinishedRun()

void edm::EventProcessor::endUnfinishedRun	(	ProcessHistoryID const &	phid,
		RunNumber_t	run,
		bool	globalBeginSucceeded,
		bool	cleaningUpAfterException,
		bool	eventSetupForInstanceSucceeded
	)

Definition at line 1043 of file EventProcessor.cc.

                                                                             {
    if (eventSetupForInstanceSucceeded) {
      //If we skip empty runs, this would be called conditionally
      endRun(phid, run, globalBeginSucceeded, cleaningUpAfterException);
 
      if (globalBeginSucceeded) {
        auto t = edm::make_empty_waiting_task();
        t->increment_ref_count();
        RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, run);
        MergeableRunProductMetadata* mergeableRunProductMetadata = runPrincipal.mergeableRunProductMetadata();
        mergeableRunProductMetadata->preWriteRun();
        writeRunAsync(edm::WaitingTaskHolder{t.get()}, phid, run, mergeableRunProductMetadata);
        t->wait_for_all();
        mergeableRunProductMetadata->postWriteRun();
        if (t->exceptionPtr()) {
          std::rethrow_exception(*t->exceptionPtr());
        }
      }
    }
    deleteRunFromCache(phid, run);
  }

References deleteRunFromCache(), endRun(), edm::make_empty_waiting_task(), edm::RunPrincipal::mergeableRunProductMetadata(), edm::MergeableRunProductMetadata::postWriteRun(), edm::MergeableRunProductMetadata::preWriteRun(), principalCache_, run(), edm::PrincipalCache::runPrincipal(), submitPVValidationJobs::t, and writeRunAsync().

getAllModuleDescriptions()

std::vector< ModuleDescription const * > edm::EventProcessor::getAllModuleDescriptions

(

)

const

Return a vector allowing const access to all the ModuleDescriptions for this EventProccessor. *** N.B. *** Ownership of the ModuleDescriptions is not *** passed to the caller. Do not call delete on these *** pointers!

Definition at line 625 of file EventProcessor.cc.

                                                                                     {
    return schedule_->getAllModuleDescriptions();
  }

References schedule_.

getToken()

ServiceToken edm::EventProcessor::getToken

(

)

Definition at line 623 of file EventProcessor.cc.

{ return serviceToken_; }

References serviceToken_.

Referenced by ~EventProcessor().

globalEndLumiAsync()

void edm::EventProcessor::globalEndLumiAsync	(	edm::WaitingTaskHolder	iTask,
		std::shared_ptr< LuminosityBlockProcessingStatus >	iLumiStatus
	)

Definition at line 1360 of file EventProcessor.cc.

                                                                                                      {
    // Get some needed info out of the status object before moving
    // it into finalTaskForThisLumi.
    auto& lp = *(iLumiStatus->lumiPrincipal());
    bool didGlobalBeginSucceed = iLumiStatus->didGlobalBeginSucceed();
    bool cleaningUpAfterException = iLumiStatus->cleaningUpAfterException();
    EventSetupImpl const& es = iLumiStatus->eventSetupImpl(esp_->subProcessIndex());
    std::vector<std::shared_ptr<const EventSetupImpl>> const* eventSetupImpls = &iLumiStatus->eventSetupImpls();
 
    auto finalTaskForThisLumi = edm::make_waiting_task(
        tbb::task::allocate_root(),
        [status = std::move(iLumiStatus), iTask = std::move(iTask), this](std::exception_ptr const* iPtr) mutable {
          std::exception_ptr ptr;
          if (iPtr) {
            handleEndLumiExceptions(iPtr, iTask);
          } else {
            // Caught exception is passed to handleEndLumiExceptions()
            CMS_SA_ALLOW try {
              ServiceRegistry::Operate operate(serviceToken_);
              if (looper_) {
                auto& lumiPrincipal = *(status->lumiPrincipal());
                EventSetupImpl const& eventSetupImpl = status->eventSetupImpl(esp_->subProcessIndex());
                looper_->doEndLuminosityBlock(lumiPrincipal, eventSetupImpl, &processContext_);
              }
            } catch (...) {
              ptr = std::current_exception();
            }
          }
          ServiceRegistry::Operate operate(serviceToken_);
 
          // Try hard to clean up resources so the
          // process can terminate in a controlled
          // fashion even after exceptions have occurred.
          // Caught exception is passed to handleEndLumiExceptions()
          CMS_SA_ALLOW try { deleteLumiFromCache(*status); } catch (...) {
            if (not ptr) {
              ptr = std::current_exception();
            }
          }
          // Caught exception is passed to handleEndLumiExceptions()
          CMS_SA_ALLOW try {
            status->resumeGlobalLumiQueue();
            queueWhichWaitsForIOVsToFinish_.resume();
          } catch (...) {
            if (not ptr) {
              ptr = std::current_exception();
            }
          }
          // Caught exception is passed to handleEndLumiExceptions()
          CMS_SA_ALLOW try {
            // This call to status.resetResources() must occur before iTask is destroyed.
            // Otherwise there will be a data race which could result in endRun
            // being delayed until it is too late to successfully call it.
            status->resetResources();
            status.reset();
          } catch (...) {
            if (not ptr) {
              ptr = std::current_exception();
            }
          }
 
          if (ptr) {
            handleEndLumiExceptions(&ptr, iTask);
          }
        });
 
    auto writeT = edm::make_waiting_task(
        tbb::task::allocate_root(),
        [this, didGlobalBeginSucceed, &lumiPrincipal = lp, task = WaitingTaskHolder(finalTaskForThisLumi)](
            std::exception_ptr const* iExcept) mutable {
          if (iExcept) {
            task.doneWaiting(*iExcept);
          } else {
            //Only call writeLumi if beginLumi succeeded
            if (didGlobalBeginSucceed) {
              writeLumiAsync(std::move(task), lumiPrincipal);
            }
          }
        });
 
    IOVSyncValue ts(EventID(lp.run(), lp.luminosityBlock(), EventID::maxEventNumber()), lp.beginTime());
 
    LumiTransitionInfo transitionInfo(lp, es, eventSetupImpls);
    using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd>;
    endGlobalTransitionAsync<Traits>(
        WaitingTaskHolder(writeT), *schedule_, transitionInfo, serviceToken_, subProcesses_, cleaningUpAfterException);
  }

References CMS_SA_ALLOW, deleteLumiFromCache(), esp_, handleEndLumiExceptions(), looper_, edm::make_waiting_task(), edm::EventID::maxEventNumber(), eostools::move(), processContext_, queueWhichWaitsForIOVsToFinish_, edm::SerialTaskQueue::resume(), schedule_, serviceToken_, mps_update::status, subProcesses_, TrackValidation_cff::task, and writeLumiAsync().

Referenced by streamEndLumiAsync().

handleEndLumiExceptions()

void edm::EventProcessor::handleEndLumiExceptions	(	std::exception_ptr const *	iPtr,
		WaitingTaskHolder &	holder
	)

Definition at line 1351 of file EventProcessor.cc.

                                                                                                      {
    if (setDeferredException(*iPtr)) {
      WaitingTaskHolder tmp(holder);
      tmp.doneWaiting(*iPtr);
    } else {
      setExceptionMessageLumis();
    }
  }

References setDeferredException(), setExceptionMessageLumis(), and createJobs::tmp.

Referenced by globalEndLumiAsync(), and streamEndLumiAsync().

handleNextEventForStreamAsync()

void edm::EventProcessor::handleNextEventForStreamAsync ( WaitingTaskHolder  iTask, unsigned int  iStreamIndex  )

private

Definition at line 1723 of file EventProcessor.cc.

                                                                                                       {
    sourceResourcesAcquirer_.serialQueueChain().push([this, iTask, iStreamIndex]() mutable {
      ServiceRegistry::Operate operate(serviceToken_);
      //we do not want to extend the lifetime of the shared_ptr to the end of this function
      // as steramEndLumiAsync may clear the value from streamLumiStatus_[iStreamIndex]
      auto status = streamLumiStatus_[iStreamIndex].get();
      // Caught exception is propagated to EventProcessor::runToCompletion() via deferredExceptionPtr_
      CMS_SA_ALLOW try {
        if (readNextEventForStream(iStreamIndex, *status)) {
          auto recursionTask = make_waiting_task(
              tbb::task::allocate_root(), [this, iTask, iStreamIndex](std::exception_ptr const* iPtr) mutable {
                if (iPtr) {
                  // Try to end the stream properly even if an exception was
                  // thrown on an event.
                  bool expected = false;
                  if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
                    // This is the case where the exception in iPtr is the primary
                    // exception and we want to see its message.
                    deferredExceptionPtr_ = *iPtr;
                    WaitingTaskHolder tempHolder(iTask);
                    tempHolder.doneWaiting(*iPtr);
                  }
                  streamEndLumiAsync(std::move(iTask), iStreamIndex);
                  //the stream will stop now
                  return;
                }
                handleNextEventForStreamAsync(std::move(iTask), iStreamIndex);
              });
 
          processEventAsync(WaitingTaskHolder(recursionTask), iStreamIndex);
        } else {
          //the stream will stop now
          if (status->isLumiEnding()) {
            if (lastTransitionType() == InputSource::IsLumi and not status->haveStartedNextLumi()) {
              status->startNextLumi();
              beginLumiAsync(status->nextSyncValue(), status->runResource(), iTask);
            }
            streamEndLumiAsync(std::move(iTask), iStreamIndex);
          } else {
            iTask.doneWaiting(std::exception_ptr{});
          }
        }
      } catch (...) {
        // It is unlikely we will ever get in here ...
        // But if we do try to clean up and propagate the exception
        if (streamLumiStatus_[iStreamIndex]) {
          streamEndLumiAsync(iTask, iStreamIndex);
        }
        bool expected = false;
        if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
          auto e = std::current_exception();
          deferredExceptionPtr_ = e;
          iTask.doneWaiting(e);
        }
      }
    });
  }

References beginLumiAsync(), CMS_SA_ALLOW, deferredExceptionPtr_, deferredExceptionPtrIsSet_, edm::WaitingTaskHolder::doneWaiting(), MillePedeFileConverter_cfg::e, edm::InputSource::IsLumi, lastTransitionType(), edm::make_waiting_task(), eostools::move(), processEventAsync(), edm::SerialTaskQueueChain::push(), readNextEventForStream(), edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, sourceResourcesAcquirer_, mps_update::status, streamEndLumiAsync(), and streamLumiStatus_.

Referenced by beginLumiAsync(), and continueLumiAsync().

init()

void edm::EventProcessor::init ( std::shared_ptr< ProcessDesc > &  processDesc, ServiceToken const &  token, serviceregistry::ServiceLegacy  iLegacy  )

private

Definition at line 328 of file EventProcessor.cc.

                                                                  {
    //std::cerr << processDesc->dump() << std::endl;
 
    // register the empty parentage vector , once and for all
    ParentageRegistry::instance()->insertMapped(Parentage());
 
    // register the empty parameter set, once and for all.
    ParameterSet().registerIt();
 
    std::shared_ptr<ParameterSet> parameterSet = processDesc->getProcessPSet();
 
    // If there are subprocesses, pop the subprocess parameter sets out of the process parameter set
    auto subProcessVParameterSet = popSubProcessVParameterSet(*parameterSet);
    bool const hasSubProcesses = !subProcessVParameterSet.empty();
 
    // Validates the parameters in the 'options', 'maxEvents', 'maxLuminosityBlocks',
    // and 'maxSecondsUntilRampdown' top level parameter sets. Default values are also
    // set in here if the parameters were not explicitly set.
    validateTopLevelParameterSets(parameterSet.get());
 
    // Now set some parameters specific to the main process.
    ParameterSet const& optionsPset(parameterSet->getUntrackedParameterSet("options"));
    auto const& fileMode = optionsPset.getUntrackedParameter<std::string>("fileMode");
    if (fileMode != "NOMERGE" and fileMode != "FULLMERGE") {
      throw Exception(errors::Configuration, "Illegal fileMode parameter value: ")
          << fileMode << ".\n"
          << "Legal values are 'NOMERGE' and 'FULLMERGE'.\n";
    } else {
      fileModeNoMerge_ = (fileMode == "NOMERGE");
    }
    forceESCacheClearOnNewRun_ = optionsPset.getUntrackedParameter<bool>("forceEventSetupCacheClearOnNewRun");
 
    //threading
    unsigned int nThreads = optionsPset.getUntrackedParameter<unsigned int>("numberOfThreads");
 
    // Even if numberOfThreads was set to zero in the Python configuration, the code
    // in cmsRun.cpp should have reset it to something else.
    assert(nThreads != 0);
 
    unsigned int nStreams = optionsPset.getUntrackedParameter<unsigned int>("numberOfStreams");
    if (nStreams == 0) {
      nStreams = nThreads;
    }
    if (nThreads > 1 or nStreams > 1) {
      edm::LogInfo("ThreadStreamSetup") << "setting # threads " << nThreads << "\nsetting # streams " << nStreams;
    }
    unsigned int nConcurrentRuns = optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentRuns");
    if (nConcurrentRuns != 1) {
      throw Exception(errors::Configuration, "Illegal value nConcurrentRuns : ")
          << "Although the plan is to change this in the future, currently nConcurrentRuns must always be 1.\n";
    }
    unsigned int nConcurrentLumis =
        optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentLuminosityBlocks");
    if (nConcurrentLumis == 0) {
      nConcurrentLumis = nConcurrentRuns;
    }
 
    //Check that relationships between threading parameters makes sense
    /*
      if(nThreads<nStreams) {
      //bad
      }
      if(nConcurrentRuns>nStreams) {
      //bad
      }
      if(nConcurrentRuns>nConcurrentLumis) {
      //bad
      }
    */
    IllegalParameters::setThrowAnException(optionsPset.getUntrackedParameter<bool>("throwIfIllegalParameter"));
 
    printDependencies_ = optionsPset.getUntrackedParameter<bool>("printDependencies");
 
    // Now do general initialization
    ScheduleItems items;
 
    //initialize the services
    auto& serviceSets = processDesc->getServicesPSets();
    ServiceToken token = items.initServices(serviceSets, *parameterSet, iToken, iLegacy, true);
    serviceToken_ = items.addCPRandTNS(*parameterSet, token);
 
    //make the services available
    ServiceRegistry::Operate operate(serviceToken_);
 
    if (nStreams > 1) {
      edm::Service<RootHandlers> handler;
      handler->willBeUsingThreads();
    }
 
    // intialize miscellaneous items
    std::shared_ptr<CommonParams> common(items.initMisc(*parameterSet));
 
    // intialize the event setup provider
    ParameterSet const& eventSetupPset(optionsPset.getUntrackedParameterSet("eventSetup"));
    esp_ = espController_->makeProvider(*parameterSet, items.actReg_.get(), &eventSetupPset);
 
    // initialize the looper, if any
    looper_ = fillLooper(*espController_, *esp_, *parameterSet);
    if (looper_) {
      looper_->setActionTable(items.act_table_.get());
      looper_->attachTo(*items.actReg_);
 
      //For now loopers make us run only 1 transition at a time
      nStreams = 1;
      nConcurrentLumis = 1;
      nConcurrentRuns = 1;
    }
    espController_->setMaxConcurrentIOVs(nStreams, nConcurrentLumis);
 
    preallocations_ = PreallocationConfiguration{nThreads, nStreams, nConcurrentLumis, nConcurrentRuns};
 
    lumiQueue_ = std::make_unique<LimitedTaskQueue>(nConcurrentLumis);
    streamQueues_.resize(nStreams);
    streamLumiStatus_.resize(nStreams);
 
    // initialize the input source
    input_ = makeInput(*parameterSet,
                       *common,
                       items.preg(),
                       items.branchIDListHelper(),
                       items.thinnedAssociationsHelper(),
                       items.actReg_,
                       items.processConfiguration(),
                       preallocations_);
 
    // intialize the Schedule
    schedule_ = items.initSchedule(*parameterSet, hasSubProcesses, preallocations_, &processContext_);
 
    // set the data members
    act_table_ = std::move(items.act_table_);
    actReg_ = items.actReg_;
    preg_ = items.preg();
    mergeableRunProductProcesses_.setProcessesWithMergeableRunProducts(*preg_);
    branchIDListHelper_ = items.branchIDListHelper();
    thinnedAssociationsHelper_ = items.thinnedAssociationsHelper();
    processConfiguration_ = items.processConfiguration();
    processContext_.setProcessConfiguration(processConfiguration_.get());
    principalCache_.setProcessHistoryRegistry(input_->processHistoryRegistry());
 
    FDEBUG(2) << parameterSet << std::endl;
 
    principalCache_.setNumberOfConcurrentPrincipals(preallocations_);
    for (unsigned int index = 0; index < preallocations_.numberOfStreams(); ++index) {
      // Reusable event principal
      auto ep = std::make_shared<EventPrincipal>(preg(),
                                                 branchIDListHelper(),
                                                 thinnedAssociationsHelper(),
                                                 *processConfiguration_,
                                                 historyAppender_.get(),
                                                 index);
      principalCache_.insert(std::move(ep));
    }
 
    for (unsigned int index = 0; index < preallocations_.numberOfLuminosityBlocks(); ++index) {
      auto lp =
          std::make_unique<LuminosityBlockPrincipal>(preg(), *processConfiguration_, historyAppender_.get(), index);
      principalCache_.insert(std::move(lp));
    }
 
    {
      auto pb = std::make_unique<ProcessBlockPrincipal>(preg(), *processConfiguration_);
      principalCache_.insert(std::move(pb));
 
      auto pbForInput = std::make_unique<ProcessBlockPrincipal>(preg(), *processConfiguration_);
      principalCache_.insertForInput(std::move(pbForInput));
    }
 
    // fill the subprocesses, if there are any
    subProcesses_.reserve(subProcessVParameterSet.size());
    for (auto& subProcessPSet : subProcessVParameterSet) {
      subProcesses_.emplace_back(subProcessPSet,
                                 *parameterSet,
                                 preg(),
                                 branchIDListHelper(),
                                 *thinnedAssociationsHelper_,
                                 SubProcessParentageHelper(),
                                 *espController_,
                                 *actReg_,
                                 token,
                                 serviceregistry::kConfigurationOverrides,
                                 preallocations_,
                                 &processContext_);
    }
  }

References act_table_, actReg_, cms::cuda::assert(), branchIDListHelper(), branchIDListHelper_, trackingPlots::common, edm::errors::Configuration, SiStripBadComponentsDQMServiceTemplate_cfg::ep, esp_, espController_, Exception, FDEBUG, processOptions_cff::fileMode, fileModeNoMerge_, edm::fillLooper(), forceESCacheClearOnNewRun_, edm::ParameterSet::getUntrackedParameterSet(), historyAppender_, input_, edm::PrincipalCache::insert(), edm::PrincipalCache::insertForInput(), edm::ParentageRegistry::insertMapped(), edm::ParentageRegistry::instance(), mps_monitormerge::items, edm::serviceregistry::kConfigurationOverrides, looper_, lumiQueue_, edm::makeInput(), mergeableRunProductProcesses_, eostools::move(), runTheMatrix::nStreams, runTheMatrix::nThreads, edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfStreams(), or, edm::parameterSet(), edm::popSubProcessVParameterSet(), preallocations_, preg(), preg_, principalCache_, printDependencies_, processConfiguration_, processContext_, edm::ParameterSet::registerIt(), schedule_, serviceToken_, edm::PrincipalCache::setNumberOfConcurrentPrincipals(), edm::ProcessContext::setProcessConfiguration(), edm::MergeableRunProductProcesses::setProcessesWithMergeableRunProducts(), edm::PrincipalCache::setProcessHistoryRegistry(), edm::IllegalParameters::setThrowAnException(), streamLumiStatus_, streamQueues_, AlCaHLTBitMon_QueryRunRegistry::string, subProcesses_, thinnedAssociationsHelper(), thinnedAssociationsHelper_, unpackBuffers-CaloStage2::token, and edm::validateTopLevelParameterSets().

Referenced by EventProcessor().

inputProcessBlocks()

void edm::EventProcessor::inputProcessBlocks

(

)

Definition at line 895 of file EventProcessor.cc.

                                          {
    ProcessBlockPrincipal& processBlockPrincipal = principalCache_.inputProcessBlockPrincipal();
    // For now the input source always returns false from readProcessBlock,
    // so this does nothing at all.
    // Eventually the ProcessBlockPrincipal needs to be properly filled
    // and cleared. The delayed reader needs to be set. The correct process name
    // needs to be supplied.
    while (input_->readProcessBlock()) {
      DelayedReader* reader = nullptr;
      processBlockPrincipal.fillProcessBlockPrincipal(processConfiguration_->processName(), reader);
 
      using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionProcessBlockInput>;
      auto globalWaitTask = make_empty_waiting_task();
      globalWaitTask->increment_ref_count();
 
      ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
      beginGlobalTransitionAsync<Traits>(
          WaitingTaskHolder(globalWaitTask.get()), *schedule_, transitionInfo, serviceToken_, subProcesses_);
 
      globalWaitTask->wait_for_all();
      if (globalWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
      }
 
      auto writeWaitTask = edm::make_empty_waiting_task();
      writeWaitTask->increment_ref_count();
      writeProcessBlockAsync(edm::WaitingTaskHolder{writeWaitTask.get()}, ProcessBlockType::Input);
      writeWaitTask->wait_for_all();
      if (writeWaitTask->exceptionPtr()) {
        std::rethrow_exception(*writeWaitTask->exceptionPtr());
      }
 
      processBlockPrincipal.clearPrincipal();
      for (auto& s : subProcesses_) {
        s.clearProcessBlockPrincipal(ProcessBlockType::Input);
      }
    }
  }

References edm::Principal::clearPrincipal(), edm::ProcessBlockPrincipal::fillProcessBlockPrincipal(), edm::PrincipalCache::Input, input_, edm::PrincipalCache::inputProcessBlockPrincipal(), edm::make_empty_waiting_task(), principalCache_, processConfiguration_, DQM::reader, alignCSCRings::s, schedule_, serviceToken_, subProcesses_, and writeProcessBlockAsync().

lastTransitionType()

InputSource::ItemType edm::EventProcessor::lastTransitionType ( ) const

inline

Definition at line 192 of file EventProcessor.h.

                                                   {
      if (deferredExceptionPtrIsSet_) {
        return InputSource::IsStop;
      }
      return lastSourceTransition_;
    }

References deferredExceptionPtrIsSet_, edm::InputSource::IsStop, and lastSourceTransition_.

Referenced by handleNextEventForStreamAsync(), and processLumis().

looper() [1/2]

std::shared_ptr<EDLooperBase>& edm::EventProcessor::looper ( )

inlineprivate

Definition at line 302 of file EventProcessor.h.

{ return get_underlying_safe(looper_); }

References edm::get_underlying_safe(), and looper_.

looper() [2/2]

std::shared_ptr<EDLooperBase const> edm::EventProcessor::looper ( ) const

inlineprivate

Definition at line 301 of file EventProcessor.h.

{ return get_underlying_safe(looper_); }

References edm::get_underlying_safe(), and looper_.

Referenced by endJob().

nextLuminosityBlockID()

edm::LuminosityBlockNumber_t edm::EventProcessor::nextLuminosityBlockID

(

)

Definition at line 686 of file EventProcessor.cc.

{ return input_->luminosityBlock(); }

References input_.

Referenced by readNextEventForStream().

nextRunID()

std::pair< edm::ProcessHistoryID, edm::RunNumber_t > edm::EventProcessor::nextRunID

(

)

Definition at line 682 of file EventProcessor.cc.

                                                                         {
    return std::make_pair(input_->reducedProcessHistoryID(), input_->run());
  }

References input_.

nextTransitionType()

InputSource::ItemType edm::EventProcessor::nextTransitionType

(

)

Definition at line 656 of file EventProcessor.cc.

                                                         {
    if (deferredExceptionPtrIsSet_.load()) {
      lastSourceTransition_ = InputSource::IsStop;
      return InputSource::IsStop;
    }
 
    SendSourceTerminationSignalIfException sentry(actReg_.get());
    InputSource::ItemType itemType;
    //For now, do nothing with InputSource::IsSynchronize
    do {
      itemType = input_->nextItemType();
    } while (itemType == InputSource::IsSynchronize);
 
    lastSourceTransition_ = itemType;
    sentry.completedSuccessfully();
 
    StatusCode returnCode = epSuccess;
 
    if (checkForAsyncStopRequest(returnCode)) {
      actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
      lastSourceTransition_ = InputSource::IsStop;
    }
 
    return lastSourceTransition_;
  }

References actReg_, checkForAsyncStopRequest(), deferredExceptionPtrIsSet_, epSuccess, edm::ExternalSignal, input_, edm::InputSource::IsStop, edm::InputSource::IsSynchronize, lastSourceTransition_, and runEdmFileComparison::returnCode.

Referenced by readNextEventForStream().

openOutputFiles()

void edm::EventProcessor::openOutputFiles

(

)

Definition at line 784 of file EventProcessor.cc.

                                       {
    if (fb_.get() != nullptr) {
      schedule_->openOutputFiles(*fb_);
      for_all(subProcesses_, [this](auto& subProcess) { subProcess.openOutputFiles(*fb_); });
    }
    FDEBUG(1) << "\topenOutputFiles\n";
  }

References fb_, FDEBUG, edm::for_all(), schedule_, and subProcesses_.

operator=()

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

delete

preg() [1/2]

std::shared_ptr<ProductRegistry>& edm::EventProcessor::preg ( )

inlineprivate

Definition at line 290 of file EventProcessor.h.

{ return get_underlying_safe(preg_); }

References edm::get_underlying_safe(), and preg_.

preg() [2/2]

std::shared_ptr<ProductRegistry const> edm::EventProcessor::preg ( ) const

inlineprivate

Definition at line 289 of file EventProcessor.h.

{ return get_underlying_safe(preg_); }

References edm::get_underlying_safe(), and preg_.

Referenced by beginJob(), init(), readAndMergeLumi(), readAndMergeRun(), readFile(), and readRun().

prepareForNextLoop()

void edm::EventProcessor::prepareForNextLoop

(

)

Definition at line 849 of file EventProcessor.cc.

                                          {
    looper_->prepareForNextLoop(esp_.get());
    FDEBUG(1) << "\tprepareForNextLoop\n";
  }

References esp_, FDEBUG, and looper_.

Referenced by runToCompletion().

processConfiguration()

ProcessConfiguration const& edm::EventProcessor::processConfiguration ( ) const

inline

Definition at line 138 of file EventProcessor.h.

{ return *processConfiguration_; }

References processConfiguration_.

processEventAsync()

void edm::EventProcessor::processEventAsync ( WaitingTaskHolder  iHolder, unsigned int  iStreamIndex  )

private

Definition at line 1795 of file EventProcessor.cc.

                                                                                             {
    tbb::task::spawn(
        *make_functor_task(tbb::task::allocate_root(), [=]() { processEventAsyncImpl(iHolder, iStreamIndex); }));
  }

References edm::make_functor_task(), and processEventAsyncImpl().

Referenced by handleNextEventForStreamAsync().

processEventAsyncImpl()

void edm::EventProcessor::processEventAsyncImpl ( WaitingTaskHolder  iHolder, unsigned int  iStreamIndex  )

private

Definition at line 1800 of file EventProcessor.cc.

                                                                                                 {
    auto pep = &(principalCache_.eventPrincipal(iStreamIndex));
 
    ServiceRegistry::Operate operate(serviceToken_);
    Service<RandomNumberGenerator> rng;
    if (rng.isAvailable()) {
      Event ev(*pep, ModuleDescription(), nullptr);
      rng->postEventRead(ev);
    }
 
    WaitingTaskHolder finalizeEventTask(make_waiting_task(
        tbb::task::allocate_root(), [this, pep, iHolder, iStreamIndex](std::exception_ptr const* iPtr) mutable {
          //NOTE: If we have a looper we only have one Stream
          if (looper_) {
            ServiceRegistry::Operate operateLooper(serviceToken_);
            processEventWithLooper(*pep, iStreamIndex);
          }
 
          FDEBUG(1) << "\tprocessEvent\n";
          pep->clearEventPrincipal();
          if (iPtr) {
            iHolder.doneWaiting(*iPtr);
          } else {
            iHolder.doneWaiting(std::exception_ptr());
          }
        }));
    WaitingTaskHolder afterProcessTask;
    if (subProcesses_.empty()) {
      afterProcessTask = std::move(finalizeEventTask);
    } else {
      //Need to run SubProcesses after schedule has finished
      // with the event
      afterProcessTask = WaitingTaskHolder(make_waiting_task(
          tbb::task::allocate_root(),
          [this, pep, finalizeEventTask, iStreamIndex](std::exception_ptr const* iPtr) mutable {
            if (not iPtr) {
              //when run with 1 thread, we want to the order to be what
              // it was before. This requires reversing the order since
              // tasks are run last one in first one out
              for (auto& subProcess : boost::adaptors::reverse(subProcesses_)) {
                subProcess.doEventAsync(finalizeEventTask, *pep, &streamLumiStatus_[iStreamIndex]->eventSetupImpls());
              }
            } else {
              finalizeEventTask.doneWaiting(*iPtr);
            }
          }));
    }
 
    EventSetupImpl const& es = streamLumiStatus_[iStreamIndex]->eventSetupImpl(esp_->subProcessIndex());
    EventTransitionInfo info(*pep, es);
    schedule_->processOneEventAsync(std::move(afterProcessTask), iStreamIndex, info, serviceToken_);
  }

References edm::WaitingTaskHolder::doneWaiting(), esp_, ev, edm::PrincipalCache::eventPrincipal(), FDEBUG, info(), edm::Service< T >::isAvailable(), looper_, edm::make_waiting_task(), eostools::move(), principalCache_, processEventWithLooper(), groupFilesInBlocks::reverse, schedule_, serviceToken_, streamLumiStatus_, and subProcesses_.

Referenced by processEventAsync().

processEventWithLooper()

void edm::EventProcessor::processEventWithLooper ( EventPrincipal &  iPrincipal, unsigned int  iStreamIndex  )

private

Definition at line 1853 of file EventProcessor.cc.

                                                                                                   {
    bool randomAccess = input_->randomAccess();
    ProcessingController::ForwardState forwardState = input_->forwardState();
    ProcessingController::ReverseState reverseState = input_->reverseState();
    ProcessingController pc(forwardState, reverseState, randomAccess);
 
    EDLooperBase::Status status = EDLooperBase::kContinue;
    do {
      StreamContext streamContext(iPrincipal.streamID(), &processContext_);
      EventSetupImpl const& es = streamLumiStatus_[iStreamIndex]->eventSetupImpl(esp_->subProcessIndex());
      status = looper_->doDuringLoop(iPrincipal, es, pc, &streamContext);
 
      bool succeeded = true;
      if (randomAccess) {
        if (pc.requestedTransition() == ProcessingController::kToPreviousEvent) {
          input_->skipEvents(-2);
        } else if (pc.requestedTransition() == ProcessingController::kToSpecifiedEvent) {
          succeeded = input_->goToEvent(pc.specifiedEventTransition());
        }
      }
      pc.setLastOperationSucceeded(succeeded);
    } while (!pc.lastOperationSucceeded());
    if (status != EDLooperBase::kContinue) {
      shouldWeStop_ = true;
      lastSourceTransition_ = InputSource::IsStop;
    }
  }

References esp_, input_, edm::InputSource::IsStop, edm::EDLooperBase::kContinue, edm::ProcessingController::kToPreviousEvent, edm::ProcessingController::kToSpecifiedEvent, edm::ProcessingController::lastOperationSucceeded(), lastSourceTransition_, looper_, processContext_, edm::ProcessingController::requestedTransition(), edm::ProcessingController::setLastOperationSucceeded(), shouldWeStop_, edm::ProcessingController::specifiedEventTransition(), mps_update::status, edm::EventPrincipal::streamID(), streamLumiStatus_, and summarizeEdmComparisonLogfiles::succeeded.

Referenced by processEventAsyncImpl().

processLumis()

InputSource::ItemType edm::EventProcessor::processLumis

(

std::shared_ptr< void > const &

iRunResource

)

Definition at line 1133 of file EventProcessor.cc.

                                                                                          {
    auto waitTask = make_empty_waiting_task();
    waitTask->increment_ref_count();
 
    if (streamLumiActive_ > 0) {
      assert(streamLumiActive_ == preallocations_.numberOfStreams());
      // Continue after opening a new input file
      continueLumiAsync(WaitingTaskHolder{waitTask.get()});
    } else {
      beginLumiAsync(IOVSyncValue(EventID(input_->run(), input_->luminosityBlock(), 0),
                                  input_->luminosityBlockAuxiliary()->beginTime()),
                     iRunResource,
                     WaitingTaskHolder{waitTask.get()});
    }
    waitTask->wait_for_all();
 
    if (waitTask->exceptionPtr() != nullptr) {
      std::rethrow_exception(*(waitTask->exceptionPtr()));
    }
    return lastTransitionType();
  }

References cms::cuda::assert(), beginLumiAsync(), continueLumiAsync(), input_, lastTransitionType(), edm::make_empty_waiting_task(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, and streamLumiActive_.

readAndMergeLumi()

int edm::EventProcessor::readAndMergeLumi

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 1568 of file EventProcessor.cc.

                                                                               {
    auto& lumiPrincipal = *iStatus.lumiPrincipal();
    assert(lumiPrincipal.aux().sameIdentity(*input_->luminosityBlockAuxiliary()) or
           input_->processHistoryRegistry().reducedProcessHistoryID(lumiPrincipal.aux().processHistoryID()) ==
               input_->processHistoryRegistry().reducedProcessHistoryID(
                   input_->luminosityBlockAuxiliary()->processHistoryID()));
    bool lumiOK = lumiPrincipal.adjustToNewProductRegistry(*preg());
    assert(lumiOK);
    lumiPrincipal.mergeAuxiliary(*input_->luminosityBlockAuxiliary());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readAndMergeLumi(*iStatus.lumiPrincipal());
      sentry.completedSuccessfully();
    }
    return input_->luminosityBlock();
  }

References actReg_, cms::cuda::assert(), input_, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), or, and preg().

Referenced by readNextEventForStream().

readAndMergeRun()

std::pair< ProcessHistoryID, RunNumber_t > edm::EventProcessor::readAndMergeRun

(

)

Definition at line 1537 of file EventProcessor.cc.

                                                                         {
    principalCache_.merge(input_->runAuxiliary(), preg());
    auto runPrincipal = principalCache_.runPrincipalPtr();
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readAndMergeRun(*runPrincipal);
      sentry.completedSuccessfully();
    }
    assert(input_->reducedProcessHistoryID() == runPrincipal->reducedProcessHistoryID());
    return std::make_pair(runPrincipal->reducedProcessHistoryID(), input_->run());
  }

References actReg_, cms::cuda::assert(), input_, edm::PrincipalCache::merge(), preg(), principalCache_, and edm::PrincipalCache::runPrincipalPtr().

readEvent()

void edm::EventProcessor::readEvent ( unsigned int  iStreamIndex )

private

Definition at line 1781 of file EventProcessor.cc.

                                                          {
    //TODO this will have to become per stream
    auto& event = principalCache_.eventPrincipal(iStreamIndex);
    StreamContext streamContext(event.streamID(), &processContext_);
 
    SendSourceTerminationSignalIfException sentry(actReg_.get());
    input_->readEvent(event, streamContext);
 
    streamLumiStatus_[iStreamIndex]->updateLastTimestamp(input_->timestamp());
    sentry.completedSuccessfully();
 
    FDEBUG(1) << "\treadEvent\n";
  }

References actReg_, edm::PrincipalCache::eventPrincipal(), FDEBUG, input_, principalCache_, processContext_, and streamLumiStatus_.

Referenced by readNextEventForStream().

readFile()

void edm::EventProcessor::readFile

(

)

Definition at line 757 of file EventProcessor.cc.

                                {
    FDEBUG(1) << " \treadFile\n";
    size_t size = preg_->size();
    SendSourceTerminationSignalIfException sentry(actReg_.get());
 
    principalCache_.preReadFile();
 
    fb_ = input_->readFile();
    if (size < preg_->size()) {
      principalCache_.adjustIndexesAfterProductRegistryAddition();
    }
    principalCache_.adjustEventsToNewProductRegistry(preg());
    if (preallocations_.numberOfStreams() > 1 and preallocations_.numberOfThreads() > 1) {
      fb_->setNotFastClonable(FileBlock::ParallelProcesses);
    }
    sentry.completedSuccessfully();
  }

References actReg_, edm::PrincipalCache::adjustEventsToNewProductRegistry(), edm::PrincipalCache::adjustIndexesAfterProductRegistryAddition(), fb_, FDEBUG, input_, edm::PreallocationConfiguration::numberOfStreams(), edm::PreallocationConfiguration::numberOfThreads(), edm::FileBlock::ParallelProcesses, preallocations_, preg(), preg_, edm::PrincipalCache::preReadFile(), principalCache_, and findQualityFiles::size.

readLuminosityBlock()

void edm::EventProcessor::readLuminosityBlock

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 1549 of file EventProcessor.cc.

                                                                                   {
    if (!principalCache_.hasRunPrincipal()) {
      throw edm::Exception(edm::errors::LogicError) << "EventProcessor::readLuminosityBlock\n"
                                                    << "Illegal attempt to insert lumi into cache\n"
                                                    << "Run is invalid\n"
                                                    << "Contact a Framework Developer\n";
    }
    auto lbp = principalCache_.getAvailableLumiPrincipalPtr();
    assert(lbp);
    lbp->setAux(*input_->luminosityBlockAuxiliary());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readLuminosityBlock(*lbp, *historyAppender_);
      sentry.completedSuccessfully();
    }
    lbp->setRunPrincipal(principalCache_.runPrincipalPtr());
    iStatus.lumiPrincipal() = std::move(lbp);
  }

References actReg_, cms::cuda::assert(), Exception, edm::PrincipalCache::getAvailableLumiPrincipalPtr(), edm::PrincipalCache::hasRunPrincipal(), historyAppender_, input_, edm::errors::LogicError, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), eostools::move(), principalCache_, and edm::PrincipalCache::runPrincipalPtr().

Referenced by beginLumiAsync().

readNextEventForStream()

bool edm::EventProcessor::readNextEventForStream ( unsigned int  iStreamIndex, LuminosityBlockProcessingStatus &  iLumiStatus  )

private

Definition at line 1661 of file EventProcessor.cc.

                                                                                                                 {
    if (deferredExceptionPtrIsSet_.load(std::memory_order_acquire)) {
      iStatus.endLumi();
      return false;
    }
 
    if (iStatus.wasEventProcessingStopped()) {
      return false;
    }
 
    if (shouldWeStop()) {
      lastSourceTransition_ = InputSource::IsStop;
      iStatus.stopProcessingEvents();
      iStatus.endLumi();
      return false;
    }
 
    ServiceRegistry::Operate operate(serviceToken_);
    // Caught exception is propagated to EventProcessor::runToCompletion() via deferredExceptionPtr_
    CMS_SA_ALLOW try {
      //need to use lock in addition to the serial task queue because
      // of delayed provenance reading and reading data in response to
      // edm::Refs etc
      std::lock_guard<std::recursive_mutex> guard(*(sourceMutex_.get()));
 
      auto itemType = iStatus.continuingLumi() ? InputSource::IsLumi : nextTransitionType();
      if (InputSource::IsLumi == itemType) {
        iStatus.haveContinuedLumi();
        while (itemType == InputSource::IsLumi and iStatus.lumiPrincipal()->run() == input_->run() and
               iStatus.lumiPrincipal()->luminosityBlock() == nextLuminosityBlockID()) {
          readAndMergeLumi(iStatus);
          itemType = nextTransitionType();
        }
        if (InputSource::IsLumi == itemType) {
          iStatus.setNextSyncValue(IOVSyncValue(EventID(input_->run(), input_->luminosityBlock(), 0),
                                                input_->luminosityBlockAuxiliary()->beginTime()));
        }
      }
      if (InputSource::IsEvent != itemType) {
        iStatus.stopProcessingEvents();
 
        //IsFile may continue processing the lumi and
        // looper_ can cause the input source to declare a new IsRun which is actually
        // just a continuation of the previous run
        if (InputSource::IsStop == itemType or InputSource::IsLumi == itemType or
            (InputSource::IsRun == itemType and iStatus.lumiPrincipal()->run() != input_->run())) {
          iStatus.endLumi();
        }
        return false;
      }
      readEvent(iStreamIndex);
    } catch (...) {
      bool expected = false;
      if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
        deferredExceptionPtr_ = std::current_exception();
        iStatus.endLumi();
      }
      return false;
    }
    return true;
  }

References CMS_SA_ALLOW, edm::LuminosityBlockProcessingStatus::continuingLumi(), deferredExceptionPtr_, deferredExceptionPtrIsSet_, edm::LuminosityBlockProcessingStatus::endLumi(), edm::LuminosityBlockProcessingStatus::haveContinuedLumi(), input_, edm::InputSource::IsEvent, edm::InputSource::IsLumi, edm::InputSource::IsRun, edm::InputSource::IsStop, lastSourceTransition_, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), nextLuminosityBlockID(), nextTransitionType(), or, readAndMergeLumi(), readEvent(), serviceToken_, edm::LuminosityBlockProcessingStatus::setNextSyncValue(), shouldWeStop(), sourceMutex_, edm::LuminosityBlockProcessingStatus::stopProcessingEvents(), and edm::LuminosityBlockProcessingStatus::wasEventProcessingStopped().

Referenced by handleNextEventForStreamAsync().

readRun()

std::pair< ProcessHistoryID, RunNumber_t > edm::EventProcessor::readRun

(

)

Definition at line 1514 of file EventProcessor.cc.

                                                                 {
    if (principalCache_.hasRunPrincipal()) {
      throw edm::Exception(edm::errors::LogicError) << "EventProcessor::readRun\n"
                                                    << "Illegal attempt to insert run into cache\n"
                                                    << "Contact a Framework Developer\n";
    }
    auto rp = std::make_shared<RunPrincipal>(input_->runAuxiliary(),
                                             preg(),
                                             *processConfiguration_,
                                             historyAppender_.get(),
                                             0,
                                             true,
                                             &mergeableRunProductProcesses_);
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readRun(*rp, *historyAppender_);
      sentry.completedSuccessfully();
    }
    assert(input_->reducedProcessHistoryID() == rp->reducedProcessHistoryID());
    principalCache_.insert(rp);
    return std::make_pair(rp->reducedProcessHistoryID(), input_->run());
  }

References actReg_, cms::cuda::assert(), Exception, edm::PrincipalCache::hasRunPrincipal(), historyAppender_, input_, edm::PrincipalCache::insert(), edm::errors::LogicError, mergeableRunProductProcesses_, preg(), principalCache_, and processConfiguration_.

respondToCloseInputFile()

void edm::EventProcessor::respondToCloseInputFile

(

)

Definition at line 810 of file EventProcessor.cc.

                                               {
    if (fb_.get() != nullptr) {
      schedule_->respondToCloseInputFile(*fb_);
      for_all(subProcesses_, [this](auto& subProcess) { subProcess.respondToCloseInputFile(*fb_); });
    }
    FDEBUG(1) << "\trespondToCloseInputFile\n";
  }

References fb_, FDEBUG, edm::for_all(), schedule_, and subProcesses_.

respondToOpenInputFile()

void edm::EventProcessor::respondToOpenInputFile

(

)

Definition at line 800 of file EventProcessor.cc.

                                              {
    for_all(subProcesses_,
            [this](auto& subProcess) { subProcess.updateBranchIDListHelper(branchIDListHelper_->branchIDLists()); });
    if (fb_.get() != nullptr) {
      schedule_->respondToOpenInputFile(*fb_);
      for_all(subProcesses_, [this](auto& subProcess) { subProcess.respondToOpenInputFile(*fb_); });
    }
    FDEBUG(1) << "\trespondToOpenInputFile\n";
  }

References branchIDListHelper_, fb_, FDEBUG, edm::for_all(), schedule_, and subProcesses_.

rewindInput()

void edm::EventProcessor::rewindInput

(

)

Definition at line 843 of file EventProcessor.cc.

                                   {
    input_->repeat();
    input_->rewind();
    FDEBUG(1) << "\trewind\n";
  }

References FDEBUG, and input_.

Referenced by runToCompletion().

run()

EventProcessor::StatusCode edm::EventProcessor::run ( )

inline

Definition at line 371 of file EventProcessor.h.

{ return runToCompletion(); }

References runToCompletion().

Referenced by beginRun(), Types.EventID::cppID(), Types.LuminosityBlockID::cppID(), deleteRunFromCache(), endRun(), endUnfinishedRun(), and writeRunAsync().

runToCompletion()

EventProcessor::StatusCode edm::EventProcessor::runToCompletion

(

)

Definition at line 688 of file EventProcessor.cc.

                                                           {
    StatusCode returnCode = epSuccess;
    asyncStopStatusCodeFromProcessingEvents_ = epSuccess;
    {
      beginJob();  //make sure this was called
 
      // make the services available
      ServiceRegistry::Operate operate(serviceToken_);
 
      asyncStopRequestedWhileProcessingEvents_ = false;
      try {
        FilesProcessor fp(fileModeNoMerge_);
 
        convertException::wrap([&]() {
          bool firstTime = true;
          do {
            if (not firstTime) {
              prepareForNextLoop();
              rewindInput();
            } else {
              firstTime = false;
            }
            startingNewLoop();
 
            auto trans = fp.processFiles(*this);
 
            fp.normalEnd();
 
            if (deferredExceptionPtrIsSet_.load()) {
              std::rethrow_exception(deferredExceptionPtr_);
            }
            if (trans != InputSource::IsStop) {
              //problem with the source
              doErrorStuff();
 
              throw cms::Exception("BadTransition") << "Unexpected transition change " << trans;
            }
          } while (not endOfLoop());
        });  // convertException::wrap
 
      }  // Try block
      catch (cms::Exception& e) {
        if (exceptionMessageLumis_) {
          std::string message(
              "Another exception was caught while trying to clean up lumis after the primary fatal exception.");
          e.addAdditionalInfo(message);
          if (e.alreadyPrinted()) {
            LogAbsolute("Additional Exceptions") << message;
          }
        }
        if (!exceptionMessageRuns_.empty()) {
          e.addAdditionalInfo(exceptionMessageRuns_);
          if (e.alreadyPrinted()) {
            LogAbsolute("Additional Exceptions") << exceptionMessageRuns_;
          }
        }
        if (!exceptionMessageFiles_.empty()) {
          e.addAdditionalInfo(exceptionMessageFiles_);
          if (e.alreadyPrinted()) {
            LogAbsolute("Additional Exceptions") << exceptionMessageFiles_;
          }
        }
        throw;
      }
    }
 
    return returnCode;
  }

References asyncStopRequestedWhileProcessingEvents_, asyncStopStatusCodeFromProcessingEvents_, beginJob(), deferredExceptionPtr_, deferredExceptionPtrIsSet_, doErrorStuff(), MillePedeFileConverter_cfg::e, endOfLoop(), epSuccess, Exception, exceptionMessageFiles_, exceptionMessageLumis_, exceptionMessageRuns_, fileModeNoMerge_, personalPlayback::fp, edm::InputSource::IsStop, prepareForNextLoop(), runEdmFileComparison::returnCode, rewindInput(), serviceToken_, startingNewLoop(), AlCaHLTBitMon_QueryRunRegistry::string, and edm::convertException::wrap().

Referenced by PythonEventProcessor::run(), and run().

setDeferredException()

bool edm::EventProcessor::setDeferredException

(

std::exception_ptr

iException

)

Definition at line 1902 of file EventProcessor.cc.

                                                                       {
    bool expected = false;
    if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
      deferredExceptionPtr_ = iException;
      return true;
    }
    return false;
  }

References deferredExceptionPtr_, and deferredExceptionPtrIsSet_.

Referenced by handleEndLumiExceptions().

setExceptionMessageFiles()

void edm::EventProcessor::setExceptionMessageFiles

(

std::string &

message

)

Definition at line 1896 of file EventProcessor.cc.

{ exceptionMessageFiles_ = message; }

References exceptionMessageFiles_.

setExceptionMessageLumis()

void edm::EventProcessor::setExceptionMessageLumis

(

)

Definition at line 1900 of file EventProcessor.cc.

{ exceptionMessageLumis_ = true; }

References exceptionMessageLumis_.

Referenced by handleEndLumiExceptions().

setExceptionMessageRuns()

void edm::EventProcessor::setExceptionMessageRuns

(

std::string &

message

)

Definition at line 1898 of file EventProcessor.cc.

{ exceptionMessageRuns_ = message; }

References exceptionMessageRuns_.

shouldWeCloseOutput()

bool edm::EventProcessor::shouldWeCloseOutput

(

)

const

Definition at line 854 of file EventProcessor.cc.

                                                 {
    FDEBUG(1) << "\tshouldWeCloseOutput\n";
    if (!subProcesses_.empty()) {
      for (auto const& subProcess : subProcesses_) {
        if (subProcess.shouldWeCloseOutput()) {
          return true;
        }
      }
      return false;
    }
    return schedule_->shouldWeCloseOutput();
  }

References FDEBUG, schedule_, and subProcesses_.

shouldWeStop()

bool edm::EventProcessor::shouldWeStop

(

)

const

Definition at line 1881 of file EventProcessor.cc.

                                          {
    FDEBUG(1) << "\tshouldWeStop\n";
    if (shouldWeStop_)
      return true;
    if (!subProcesses_.empty()) {
      for (auto const& subProcess : subProcesses_) {
        if (subProcess.terminate()) {
          return true;
        }
      }
      return false;
    }
    return schedule_->terminate();
  }

References FDEBUG, schedule_, shouldWeStop_, and subProcesses_.

Referenced by readNextEventForStream().

startingNewLoop()

void edm::EventProcessor::startingNewLoop

(

)

Definition at line 818 of file EventProcessor.cc.

                                       {
    shouldWeStop_ = false;
    //NOTE: for first loop, need to delay calling 'doStartingNewLoop'
    // until after we've called beginOfJob
    if (looper_ && looperBeginJobRun_) {
      looper_->doStartingNewLoop();
    }
    FDEBUG(1) << "\tstartingNewLoop\n";
  }

References FDEBUG, looper_, looperBeginJobRun_, and shouldWeStop_.

Referenced by runToCompletion().

streamEndLumiAsync()

void edm::EventProcessor::streamEndLumiAsync	(	edm::WaitingTaskHolder	iTask,
		unsigned int	iStreamIndex
	)

Definition at line 1449 of file EventProcessor.cc.

                                                                                               {
    auto t = edm::make_waiting_task(tbb::task::allocate_root(),
                                    [this, iStreamIndex, iTask](std::exception_ptr const* iPtr) mutable {
                                      if (iPtr) {
                                        handleEndLumiExceptions(iPtr, iTask);
                                      }
                                      auto status = streamLumiStatus_[iStreamIndex];
                                      //reset status before releasing queue else get race condtion
                                      streamLumiStatus_[iStreamIndex].reset();
                                      --streamLumiActive_;
                                      streamQueues_[iStreamIndex].resume();
 
                                      //are we the last one?
                                      if (status->streamFinishedLumi()) {
                                        globalEndLumiAsync(iTask, std::move(status));
                                      }
                                    });
 
    edm::WaitingTaskHolder lumiDoneTask{t};
 
    //Need to be sure the lumi status is released before lumiDoneTask can every be called.
    // therefore we do not want to hold the shared_ptr
    auto lumiStatus = streamLumiStatus_[iStreamIndex].get();
    lumiStatus->setEndTime();
 
    EventSetupImpl const& es = lumiStatus->eventSetupImpl(esp_->subProcessIndex());
 
    bool cleaningUpAfterException = lumiStatus->cleaningUpAfterException();
    auto eventSetupImpls = &lumiStatus->eventSetupImpls();
 
    if (lumiStatus->didGlobalBeginSucceed()) {
      auto& lumiPrincipal = *lumiStatus->lumiPrincipal();
      IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), EventID::maxEventNumber()),
                      lumiPrincipal.endTime());
      using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd>;
      LumiTransitionInfo transitionInfo(lumiPrincipal, es, eventSetupImpls);
      endStreamTransitionAsync<Traits>(std::move(lumiDoneTask),
                                       *schedule_,
                                       iStreamIndex,
                                       transitionInfo,
                                       serviceToken_,
                                       subProcesses_,
                                       cleaningUpAfterException);
    }
  }

References esp_, globalEndLumiAsync(), handleEndLumiExceptions(), edm::make_waiting_task(), edm::EventID::maxEventNumber(), eostools::move(), schedule_, serviceToken_, mps_update::status, streamLumiActive_, streamLumiStatus_, streamQueues_, subProcesses_, and submitPVValidationJobs::t.

Referenced by beginLumiAsync(), endUnfinishedLumi(), and handleNextEventForStreamAsync().

taskCleanup()

void edm::EventProcessor::taskCleanup

(

)

Definition at line 533 of file EventProcessor.cc.

{ espController_->endIOVs(); }

References espController_.

thinnedAssociationsHelper() [1/2]

std::shared_ptr<ThinnedAssociationsHelper>& edm::EventProcessor::thinnedAssociationsHelper ( )

inlineprivate

Definition at line 298 of file EventProcessor.h.

                                                                          {
      return get_underlying_safe(thinnedAssociationsHelper_);
    }

References edm::get_underlying_safe(), and thinnedAssociationsHelper_.

thinnedAssociationsHelper() [2/2]

std::shared_ptr<ThinnedAssociationsHelper const> edm::EventProcessor::thinnedAssociationsHelper ( ) const

inlineprivate

Definition at line 295 of file EventProcessor.h.

                                                                                     {
      return get_underlying_safe(thinnedAssociationsHelper_);
    }

References edm::get_underlying_safe(), and thinnedAssociationsHelper_.

Referenced by init().

totalEvents()

int edm::EventProcessor::totalEvents

(

)

const

Return the number of events this EventProcessor has tried to process (inclues both successes and failures, including failures due to exceptions during processing).

Definition at line 629 of file EventProcessor.cc.

{ return schedule_->totalEvents(); }

References schedule_.

Referenced by PythonEventProcessor::totalEvents().

totalEventsFailed()

int edm::EventProcessor::totalEventsFailed

(

)

const

Return the number of events that have not passed any trigger. (N.B. totalEventsFailed() + totalEventsPassed() == totalEvents()

Definition at line 633 of file EventProcessor.cc.

{ return schedule_->totalEventsFailed(); }

References schedule_.

Referenced by PythonEventProcessor::totalEventsFailed().

totalEventsPassed()

int edm::EventProcessor::totalEventsPassed

(

)

const

Return the number of events processed by this EventProcessor which have been passed by one or more trigger paths.

Definition at line 631 of file EventProcessor.cc.

{ return schedule_->totalEventsPassed(); }

References schedule_.

Referenced by PythonEventProcessor::totalEventsPassed().

warnAboutModulesRequiringLuminosityBLockSynchronization()

void edm::EventProcessor::warnAboutModulesRequiringLuminosityBLockSynchronization ( ) const

private

Definition at line 1911 of file EventProcessor.cc.

                                                                                     {
    std::unique_ptr<LogSystem> s;
    for (auto worker : schedule_->allWorkers()) {
      if (worker->wantsGlobalLuminosityBlocks() and worker->globalLuminosityBlocksQueue()) {
        if (not s) {
          s = std::make_unique<LogSystem>("ModulesSynchingOnLumis");
          (*s) << "The following modules require synchronizing on LuminosityBlock boundaries:";
        }
        (*s) << "\n  " << worker->description().moduleName() << " " << worker->description().moduleLabel();
      }
    }
  }

References alignCSCRings::s, and schedule_.

Referenced by beginJob().

writeLumiAsync()

void edm::EventProcessor::writeLumiAsync	(	WaitingTaskHolder	task,
		LuminosityBlockPrincipal &	lumiPrincipal
	)

Definition at line 1634 of file EventProcessor.cc.

                                                                                                     {
    auto subsT = edm::make_waiting_task(tbb::task::allocate_root(),
                                        [this, task, &lumiPrincipal](std::exception_ptr const* iExcept) mutable {
                                          if (iExcept) {
                                            task.doneWaiting(*iExcept);
                                          } else {
                                            ServiceRegistry::Operate op(serviceToken_);
                                            for (auto& s : subProcesses_) {
                                              s.writeLumiAsync(task, lumiPrincipal);
                                            }
                                          }
                                        });
    ServiceRegistry::Operate op(serviceToken_);
 
    lumiPrincipal.runPrincipal().mergeableRunProductMetadata()->writeLumi(lumiPrincipal.luminosityBlock());
 
    schedule_->writeLumiAsync(WaitingTaskHolder{subsT}, lumiPrincipal, &processContext_, actReg_.get());
  }

References actReg_, edm::LuminosityBlockPrincipal::luminosityBlock(), edm::make_waiting_task(), edm::RunPrincipal::mergeableRunProductMetadata(), processContext_, edm::LuminosityBlockPrincipal::runPrincipal(), alignCSCRings::s, schedule_, serviceToken_, subProcesses_, TrackValidation_cff::task, and edm::MergeableRunProductMetadata::writeLumi().

Referenced by globalEndLumiAsync().

writeProcessBlockAsync()

void edm::EventProcessor::writeProcessBlockAsync	(	WaitingTaskHolder	task,
		ProcessBlockType	processBlockType
	)

Definition at line 1585 of file EventProcessor.cc.

                                                                                                       {
    auto subsT = edm::make_waiting_task(tbb::task::allocate_root(),
                                        [this, task, processBlockType](std::exception_ptr const* iExcept) mutable {
                                          if (iExcept) {
                                            task.doneWaiting(*iExcept);
                                          } else {
                                            ServiceRegistry::Operate op(serviceToken_);
                                            for (auto& s : subProcesses_) {
                                              s.writeProcessBlockAsync(task, processBlockType);
                                            }
                                          }
                                        });
    ServiceRegistry::Operate op(serviceToken_);
    schedule_->writeProcessBlockAsync(WaitingTaskHolder(subsT),
                                      principalCache_.processBlockPrincipal(processBlockType),
                                      &processContext_,
                                      actReg_.get());
  }

References actReg_, edm::make_waiting_task(), principalCache_, edm::PrincipalCache::processBlockPrincipal(), processContext_, alignCSCRings::s, schedule_, serviceToken_, subProcesses_, and TrackValidation_cff::task.

Referenced by endProcessBlock(), and inputProcessBlocks().

writeRunAsync()

void edm::EventProcessor::writeRunAsync	(	WaitingTaskHolder	task,
		ProcessHistoryID const &	phid,
		RunNumber_t	run,
		MergeableRunProductMetadata const *	mergeableRunProductMetadata
	)

Definition at line 1604 of file EventProcessor.cc.

                                                                                                     {
    auto subsT = edm::make_waiting_task(
        tbb::task::allocate_root(),
        [this, phid, run, task, mergeableRunProductMetadata](std::exception_ptr const* iExcept) mutable {
          if (iExcept) {
            task.doneWaiting(*iExcept);
          } else {
            ServiceRegistry::Operate op(serviceToken_);
            for (auto& s : subProcesses_) {
              s.writeRunAsync(task, phid, run, mergeableRunProductMetadata);
            }
          }
        });
    ServiceRegistry::Operate op(serviceToken_);
    schedule_->writeRunAsync(WaitingTaskHolder(subsT),
                             principalCache_.runPrincipal(phid, run),
                             &processContext_,
                             actReg_.get(),
                             mergeableRunProductMetadata);
  }

References actReg_, edm::make_waiting_task(), principalCache_, processContext_, run(), edm::PrincipalCache::runPrincipal(), alignCSCRings::s, schedule_, serviceToken_, subProcesses_, and TrackValidation_cff::task.

Referenced by endUnfinishedRun().

Member Data Documentation

act_table_

std::unique_ptr<ExceptionToActionTable const> edm::EventProcessor::act_table_

private

Definition at line 322 of file EventProcessor.h.

Referenced by init().

actReg_

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

private

Definition at line 312 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), closeInputFile(), endJob(), endRun(), init(), nextTransitionType(), readAndMergeLumi(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readRun(), writeLumiAsync(), writeProcessBlockAsync(), writeRunAsync(), and ~EventProcessor().

asyncStopRequestedWhileProcessingEvents_

bool edm::EventProcessor::asyncStopRequestedWhileProcessingEvents_

private

Definition at line 358 of file EventProcessor.h.

Referenced by runToCompletion().

asyncStopStatusCodeFromProcessingEvents_

StatusCode edm::EventProcessor::asyncStopStatusCodeFromProcessingEvents_

private

Definition at line 359 of file EventProcessor.h.

Referenced by runToCompletion().

beginJobCalled_

bool edm::EventProcessor::beginJobCalled_

private

Definition at line 346 of file EventProcessor.h.

Referenced by beginJob().

branchIDListHelper_

edm::propagate_const<std::shared_ptr<BranchIDListHelper> > edm::EventProcessor::branchIDListHelper_

private

Definition at line 314 of file EventProcessor.h.

Referenced by branchIDListHelper(), init(), and respondToOpenInputFile().

deferredExceptionPtr_

std::exception_ptr edm::EventProcessor::deferredExceptionPtr_

private

Definition at line 341 of file EventProcessor.h.

Referenced by handleNextEventForStreamAsync(), readNextEventForStream(), runToCompletion(), and setDeferredException().

deferredExceptionPtrIsSet_

std::atomic<bool> edm::EventProcessor::deferredExceptionPtrIsSet_

private

Definition at line 340 of file EventProcessor.h.

Referenced by handleNextEventForStreamAsync(), lastTransitionType(), nextTransitionType(), readNextEventForStream(), runToCompletion(), and setDeferredException().

esp_

edm::propagate_const<std::shared_ptr<eventsetup::EventSetupProvider> > edm::EventProcessor::esp_

private

Definition at line 320 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), endOfLoop(), endRun(), globalEndLumiAsync(), init(), prepareForNextLoop(), processEventAsyncImpl(), processEventWithLooper(), streamEndLumiAsync(), and ~EventProcessor().

espController_

edm::propagate_const<std::unique_ptr<eventsetup::EventSetupsController> > edm::EventProcessor::espController_

private

Definition at line 319 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), endRun(), init(), taskCleanup(), and ~EventProcessor().

eventSetupDataToExcludeFromPrefetching_

ExcludedDataMap edm::EventProcessor::eventSetupDataToExcludeFromPrefetching_

private

Definition at line 364 of file EventProcessor.h.

exceptionMessageFiles_

std::string edm::EventProcessor::exceptionMessageFiles_

private

Definition at line 349 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageFiles().

exceptionMessageLumis_

std::atomic<bool> edm::EventProcessor::exceptionMessageLumis_

private

Definition at line 351 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageLumis().

exceptionMessageRuns_

std::string edm::EventProcessor::exceptionMessageRuns_

private

Definition at line 350 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageRuns().

fb_

edm::propagate_const<std::unique_ptr<FileBlock> > edm::EventProcessor::fb_

private

Definition at line 336 of file EventProcessor.h.

Referenced by closeInputFile(), closeOutputFiles(), openOutputFiles(), readFile(), respondToCloseInputFile(), and respondToOpenInputFile().

fileModeNoMerge_

bool edm::EventProcessor::fileModeNoMerge_

private

Definition at line 348 of file EventProcessor.h.

Referenced by init(), and runToCompletion().

firstEventInBlock_

bool edm::EventProcessor::firstEventInBlock_ = true

private

Definition at line 360 of file EventProcessor.h.

forceESCacheClearOnNewRun_

bool edm::EventProcessor::forceESCacheClearOnNewRun_

private

Definition at line 354 of file EventProcessor.h.

Referenced by beginRun(), and init().

forceLooperToEnd_

bool edm::EventProcessor::forceLooperToEnd_

private

Definition at line 352 of file EventProcessor.h.

Referenced by endOfLoop().

historyAppender_

edm::propagate_const<std::unique_ptr<HistoryAppender> > edm::EventProcessor::historyAppender_

private

Definition at line 334 of file EventProcessor.h.

Referenced by init(), readLuminosityBlock(), and readRun().

input_

edm::propagate_const<std::unique_ptr<InputSource> > edm::EventProcessor::input_

private

Definition at line 317 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), closeInputFile(), endJob(), endRun(), init(), inputProcessBlocks(), nextLuminosityBlockID(), nextRunID(), nextTransitionType(), processEventWithLooper(), processLumis(), readAndMergeLumi(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readNextEventForStream(), readRun(), rewindInput(), and ~EventProcessor().

lastSourceTransition_

InputSource::ItemType edm::EventProcessor::lastSourceTransition_

private

Definition at line 318 of file EventProcessor.h.

Referenced by lastTransitionType(), nextTransitionType(), processEventWithLooper(), and readNextEventForStream().

looper_

edm::propagate_const<std::shared_ptr<EDLooperBase> > edm::EventProcessor::looper_

private

Definition at line 337 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginRun(), Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), endOfLoop(), endRun(), globalEndLumiAsync(), init(), looper(), prepareForNextLoop(), processEventAsyncImpl(), processEventWithLooper(), Config.Process::splitPython(), startingNewLoop(), and ~EventProcessor().

looperBeginJobRun_

bool edm::EventProcessor::looperBeginJobRun_

private

Definition at line 353 of file EventProcessor.h.

Referenced by beginRun(), and startingNewLoop().

lumiQueue_

std::unique_ptr<edm::LimitedTaskQueue> edm::EventProcessor::lumiQueue_

private

Definition at line 329 of file EventProcessor.h.

Referenced by beginLumiAsync(), and init().

mergeableRunProductProcesses_

MergeableRunProductProcesses edm::EventProcessor::mergeableRunProductProcesses_

private

Definition at line 326 of file EventProcessor.h.

Referenced by init(), and readRun().

pathsAndConsumesOfModules_

PathsAndConsumesOfModules edm::EventProcessor::pathsAndConsumesOfModules_

private

Definition at line 325 of file EventProcessor.h.

Referenced by beginJob().

preallocations_

PreallocationConfiguration edm::EventProcessor::preallocations_

private

Definition at line 356 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), continueLumiAsync(), endJob(), endRun(), endUnfinishedLumi(), init(), processLumis(), and readFile().

preg_

edm::propagate_const<std::shared_ptr<ProductRegistry> > edm::EventProcessor::preg_

private

Definition at line 313 of file EventProcessor.h.

Referenced by beginJob(), endOfLoop(), init(), preg(), and readFile().

principalCache_

PrincipalCache edm::EventProcessor::principalCache_

private

Definition at line 345 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginProcessBlock(), beginRun(), deleteRunFromCache(), endProcessBlock(), endRun(), endUnfinishedRun(), init(), inputProcessBlocks(), processEventAsyncImpl(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readRun(), writeProcessBlockAsync(), and writeRunAsync().

printDependencies_

bool edm::EventProcessor::printDependencies_ = false

private

Definition at line 366 of file EventProcessor.h.

Referenced by beginJob(), and init().

processConfiguration_

std::shared_ptr<ProcessConfiguration const> edm::EventProcessor::processConfiguration_

private

Definition at line 323 of file EventProcessor.h.

Referenced by beginJob(), beginProcessBlock(), init(), inputProcessBlocks(), processConfiguration(), and readRun().

processContext_

ProcessContext edm::EventProcessor::processContext_

private

Definition at line 324 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), endRun(), globalEndLumiAsync(), init(), processEventWithLooper(), readEvent(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

queueWhichWaitsForIOVsToFinish_

edm::SerialTaskQueue edm::EventProcessor::queueWhichWaitsForIOVsToFinish_

private

Definition at line 321 of file EventProcessor.h.

Referenced by beginLumiAsync(), and globalEndLumiAsync().

schedule_

edm::propagate_const<std::unique_ptr<Schedule> > edm::EventProcessor::schedule_

private

Definition at line 327 of file EventProcessor.h.

Referenced by Config.Process::_insertPaths(), Config.Process::_replaceInSchedule(), beginJob(), beginLumiAsync(), beginProcessBlock(), beginRun(), clearCounters(), closeOutputFiles(), Config.Process::dumpPython(), enableEndPaths(), endJob(), endOfLoop(), endPathsEnabled(), endProcessBlock(), endRun(), getAllModuleDescriptions(), globalEndLumiAsync(), init(), inputProcessBlocks(), openOutputFiles(), processEventAsyncImpl(), Config.Process::prune(), Config.Process::resolve(), respondToCloseInputFile(), respondToOpenInputFile(), shouldWeCloseOutput(), shouldWeStop(), Config.Process::splitPython(), streamEndLumiAsync(), totalEvents(), totalEventsFailed(), totalEventsPassed(), warnAboutModulesRequiringLuminosityBLockSynchronization(), writeLumiAsync(), writeProcessBlockAsync(), writeRunAsync(), and ~EventProcessor().

serviceToken_

ServiceToken edm::EventProcessor::serviceToken_

private

Definition at line 316 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginProcessBlock(), beginRun(), endJob(), endProcessBlock(), endRun(), getToken(), globalEndLumiAsync(), handleNextEventForStreamAsync(), init(), inputProcessBlocks(), processEventAsyncImpl(), readNextEventForStream(), runToCompletion(), streamEndLumiAsync(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

shouldWeStop_

bool edm::EventProcessor::shouldWeStop_

private

Definition at line 347 of file EventProcessor.h.

Referenced by processEventWithLooper(), shouldWeStop(), and startingNewLoop().

sourceMutex_

std::shared_ptr<std::recursive_mutex> edm::EventProcessor::sourceMutex_

private

Definition at line 344 of file EventProcessor.h.

Referenced by readNextEventForStream().

sourceResourcesAcquirer_

SharedResourcesAcquirer edm::EventProcessor::sourceResourcesAcquirer_

private

Definition at line 343 of file EventProcessor.h.

Referenced by beginLumiAsync(), and handleNextEventForStreamAsync().

streamLumiActive_

std::atomic<unsigned int> edm::EventProcessor::streamLumiActive_ {0}

private

Definition at line 331 of file EventProcessor.h.

Referenced by beginLumiAsync(), endUnfinishedLumi(), processLumis(), and streamEndLumiAsync().

streamLumiStatus_

std::vector<std::shared_ptr<LuminosityBlockProcessingStatus> > edm::EventProcessor::streamLumiStatus_

private

Definition at line 330 of file EventProcessor.h.

Referenced by beginLumiAsync(), continueLumiAsync(), endUnfinishedLumi(), handleNextEventForStreamAsync(), init(), processEventAsyncImpl(), processEventWithLooper(), readEvent(), and streamEndLumiAsync().

streamQueues_

std::vector<edm::SerialTaskQueue> edm::EventProcessor::streamQueues_

private

Definition at line 328 of file EventProcessor.h.

Referenced by beginLumiAsync(), init(), and streamEndLumiAsync().

subProcesses_

std::vector<SubProcess> edm::EventProcessor::subProcesses_

private

Definition at line 333 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginProcessBlock(), beginRun(), closeOutputFiles(), deleteLumiFromCache(), deleteRunFromCache(), Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), endProcessBlock(), endRun(), globalEndLumiAsync(), init(), inputProcessBlocks(), openOutputFiles(), processEventAsyncImpl(), respondToCloseInputFile(), respondToOpenInputFile(), shouldWeCloseOutput(), shouldWeStop(), Config.Process::splitPython(), streamEndLumiAsync(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

thinnedAssociationsHelper_

edm::propagate_const<std::shared_ptr<ThinnedAssociationsHelper> > edm::EventProcessor::thinnedAssociationsHelper_

private

Definition at line 315 of file EventProcessor.h.

Referenced by init(), and thinnedAssociationsHelper().