edm::EventProcessor Class Reference

#include <EventProcessor.h>

Public Types
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	beginRun (ProcessHistoryID const &phid, RunNumber_t run, bool &globalBeginSucceeded)
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	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)
	EventProcessor (std::string const &config, 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::string const &config, std::vector< std::string > const &defaultServices, std::vector< std::string > const &forcedServices=std::vector< std::string >())
	EventProcessor (std::shared_ptr< ProcessDesc > processDesc, ServiceToken const &token, serviceregistry::ServiceLegacy legacy)
	EventProcessor (std::string const &config, bool isPython)
	meant for unit tests More...
	EventProcessor (EventProcessor const &)=delete
std::vector< ModuleDescription const * >	getAllModuleDescriptions () const
ServiceToken	getToken ()
void	getTriggerReport (TriggerReport &rep) const
void	globalEndLumiAsync (edm::WaitingTaskHolder iTask, std::shared_ptr< LuminosityBlockProcessingStatus > iLumiStatus)
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 (std::string &message)
void	setExceptionMessageRuns (std::string &message)
bool	shouldWeCloseOutput () const
bool	shouldWeStop () const
void	startingNewLoop ()
void	streamEndLumiAsync (edm::WaitingTaskHolder iTask, unsigned int iStreamIndex, std::shared_ptr< LuminosityBlockProcessingStatus > iLumiStatus)
int	totalEvents () const
int	totalEventsFailed () const
int	totalEventsPassed () const
void	writeLumiAsync (WaitingTaskHolder, std::shared_ptr< LuminosityBlockProcessingStatus >)
void	writeRunAsync (WaitingTaskHolder, ProcessHistoryID const &phid, RunNumber_t run)
	~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 const >	branchIDListHelper () const
std::shared_ptr< BranchIDListHelper > &	branchIDListHelper ()
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 const >	looper () const
std::shared_ptr< EDLooperBase > &	looper ()
std::shared_ptr< ProductRegistry const >	preg () const
std::shared_ptr< ProductRegistry > &	preg ()
void	processEventAsync (WaitingTaskHolder iHolder, unsigned int iStreamIndex)
void	processEventAsyncImpl (WaitingTaskHolder iHolder, unsigned int iStreamIndex)
void	processEventWithLooper (EventPrincipal &)
void	readEvent (unsigned int iStreamIndex)
bool	readNextEventForStream (unsigned int iStreamIndex, LuminosityBlockProcessingStatus &iLumiStatus)
std::shared_ptr< ThinnedAssociationsHelper const >	thinnedAssociationsHelper () const
std::shared_ptr< ThinnedAssociationsHelper > &	thinnedAssociationsHelper ()

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::string	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_
edm::SerialTaskQueue	iovQueue_
InputSource::ItemType	lastSourceTransition_
edm::propagate_const< std::shared_ptr< EDLooperBase > >	looper_
bool	looperBeginJobRun_
std::unique_ptr< edm::LimitedTaskQueue >	lumiQueue_
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::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 61 of file EventProcessor.h.

Member Typedef Documentation

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

private

Definition at line 338 of file EventProcessor.h.

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

private

Definition at line 339 of file EventProcessor.h.

Member Enumeration Documentation

enum edm::EventProcessor::StatusCode

Enumerator
epSuccess
epException
epOther
epSignal
epInputComplete
epTimedOut
epCountComplete

Definition at line 72 of file EventProcessor.h.

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

Constructor & Destructor Documentation

edm::EventProcessor::EventProcessor ( std::string const &  config, 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.

References init(), edm::parameterSet(), and PythonProcessDesc::parameterSet().

                                                                             :
    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_(),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    eventSetupDataToExcludeFromPrefetching_() {
    std::shared_ptr<ParameterSet> parameterSet = PythonProcessDesc(config).parameterSet();
    auto processDesc = std::make_shared<ProcessDesc>(parameterSet);
    processDesc->addServices(defaultServices, forcedServices);
    init(processDesc, iToken, iLegacy);
  }

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

Definition at line 257 of file EventProcessor.cc.

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

                                                                             :
    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_(),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    asyncStopRequestedWhileProcessingEvents_(false),
    eventSetupDataToExcludeFromPrefetching_()
  {
    std::shared_ptr<ParameterSet> parameterSet = PythonProcessDesc(config).parameterSet();
    auto processDesc = std::make_shared<ProcessDesc>(parameterSet);
    processDesc->addServices(defaultServices, forcedServices);
    init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
  }

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

Definition at line 296 of file EventProcessor.cc.

References init().

                                                                      :
    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_(),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    asyncStopRequestedWhileProcessingEvents_(false),
    eventSetupDataToExcludeFromPrefetching_()
  {
    init(processDesc, token, legacy);
  }

edm::EventProcessor::EventProcessor	(	std::string const &	config,
		bool	isPython
	)

meant for unit tests

Definition at line 333 of file EventProcessor.cc.

References looper::config, init(), edm::serviceregistry::kOverlapIsError, edm::parameterSet(), and PythonProcessDesc::parameterSet().

                                                                      :
    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_(),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    asyncStopRequestedWhileProcessingEvents_(false),
    eventSetupDataToExcludeFromPrefetching_()
  {
    if(isPython) {
      std::shared_ptr<ParameterSet> parameterSet = PythonProcessDesc(config).parameterSet();
      auto processDesc = std::make_shared<ProcessDesc>(parameterSet);
      init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
    }
    else {
      auto processDesc = std::make_shared<ProcessDesc>(config);
      init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
    }
  }

edm::EventProcessor::~EventProcessor

(

)

Definition at line 548 of file EventProcessor.cc.

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

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

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

delete

Member Function Documentation

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 567 of file EventProcessor.cc.

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

Referenced by runToCompletion().

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

    //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;
    }
    schedule_->beginJob(*preg_);
    // 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); });
    }
  }

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

Definition at line 1115 of file EventProcessor.cc.

References actReg_, edm::LuminosityBlockPrincipal::beginTime(), edm::WaitingTaskHolder::doneWaiting(), esp_, espController_, edm::PrincipalCache::eventPrincipal(), h, handleNextEventForStreamAsync(), mps_fire::i, input_, iovQueue_, edm::Service< T >::isAvailable(), looper_, edm::LuminosityBlockPrincipal::luminosityBlock(), lumiQueue_, edm::make_waiting_task(), eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), cmsPerfStripChart::operate(), edm::SerialTaskQueue::pause(), preallocations_, principalCache_, processContext_, edm::SerialTaskQueueChain::push(), edm::SerialTaskQueue::push(), readLuminosityBlock(), edm::LuminosityBlockPrincipal::run(), schedule_, edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, sourceResourcesAcquirer_, mps_update::status, streamLumiActive_, streamLumiStatus_, streamQueues_, subProcesses_, and edm::WaitingTaskHolder::taskHasFailed().

Referenced by handleNextEventForStreamAsync(), and processLumis().

                                                                                                        {
    if(iHolder.taskHasFailed()) { return; }
    
    auto status= std::make_shared<LuminosityBlockProcessingStatus>(this, preallocations_.numberOfStreams(), iRunResource) ;

    auto lumiWork = [this, iHolder, status](edm::LimitedTaskQueue::Resumer iResumer) mutable {
      if(iHolder.taskHasFailed()) { return; }

      status->setResumer(std::move(iResumer));
      
      sourceResourcesAcquirer_.serialQueueChain().push([this,iHolder,status]() mutable {
        //make the services available
        ServiceRegistry::Operate operate(serviceToken_);

        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);
          }
          
          IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), 0), lumiPrincipal.beginTime());

          //Task to start the stream beginLumis
          auto beginStreamsTask= make_waiting_task(tbb::task::allocate_root()
                                                   ,[this, holder = iHolder, status, ts] (std::exception_ptr const* iPtr) mutable {
            if (iPtr) {
              holder.doneWaiting(*iPtr);
            } else {

              status->globalBeginDidSucceed();
              EventSetup const& es = esp_->eventSetup();
              if(looper_) {
                try {
                  //make the services available
                  ServiceRegistry::Operate operate(serviceToken_);
                  looper_->doBeginLuminosityBlock(*(status->lumiPrincipal()), es, &processContext_);
                }catch(...) {
                  holder.doneWaiting(std::current_exception());
                  return;
                }
              }
              typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin> Traits;
              
              for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
                streamQueues_[i].push([this,i,status,holder,ts,&es] () {
                  streamQueues_[i].pause();
                  
                  auto eventTask = edm::make_waiting_task(tbb::task::allocate_root(),
                                                          [this,i,h = holder](std::exception_ptr const* iPtr) mutable
                  {
                    if(iPtr) {
                      h.doneWaiting(*iPtr);
                    } else {
                      handleNextEventForStreamAsync(std::move(h), i);
                    }
                  });
                  auto& event = principalCache_.eventPrincipal(i);
                  streamLumiStatus_[i] = status;
                  ++streamLumiActive_;
                  auto lp = status->lumiPrincipal();
                  event.setLuminosityBlockPrincipal(lp.get());
                  beginStreamTransitionAsync<Traits>(WaitingTaskHolder{eventTask},
                                                     *schedule_,i,*lp,ts,es,
                                                     serviceToken_,subProcesses_);
                });
              }
            }
          });
          
          //task to start the global begin lumi
          WaitingTaskHolder beginStreamsHolder{beginStreamsTask};
          EventSetup const& es = esp_->eventSetup();
          {
            typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin> Traits;
            beginGlobalTransitionAsync<Traits>(beginStreamsHolder,
                                               *schedule_,
                                               *(status->lumiPrincipal()),
                                               ts,
                                               es,
                                               serviceToken_,
                                               subProcesses_);
          }
        } catch(...) {
          iHolder.doneWaiting(std::current_exception());
        }
      });
    };
        
    //Safe to do check now since can not have multiple beginLumis at same time in this part of the code
    // because we do not attempt to read from the source again until we try to get the first event in a lumi
    if(espController_->isWithinValidityInterval(iSync)) {
      iovQueue_.pause();
      lumiQueue_->pushAndPause(std::move(lumiWork));
    } else {
      //If EventSetup fails, need beginStreamsHolder in order to pass back exception
      iovQueue_.push([this,iHolder,lumiWork,iSync]() mutable {
        try {
          SendSourceTerminationSignalIfException sentry(actReg_.get());
          espController_->eventSetupForInstance(iSync);
          sentry.completedSuccessfully();
        } catch(...) {
          iHolder.doneWaiting(std::current_exception());
          return;
        }
        iovQueue_.pause();
        lumiQueue_->pushAndPause(std::move(lumiWork));
      });
    }
  }

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

Definition at line 939 of file EventProcessor.cc.

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(), edm::PrincipalCache::runPrincipal(), schedule_, serviceToken_, and subProcesses_.

                                                                                                         {
    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);
      sentry.completedSuccessfully();
    }
    EventSetup const& es = esp_->eventSetup();
    if(looper_ && looperBeginJobRun_== false) {
      looper_->copyInfo(ScheduleInfo(schedule_.get()));
      looper_->beginOfJob(es);
      looperBeginJobRun_ = true;
      looper_->doStartingNewLoop();
    }
    {
      typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin> Traits;
      auto globalWaitTask = make_empty_waiting_task();
      globalWaitTask->increment_ref_count();
      beginGlobalTransitionAsync<Traits>(WaitingTaskHolder(globalWaitTask.get()),
                                         *schedule_,
                                         runPrincipal,
                                         ts,
                                         es,
                                         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();
      
      typedef OccurrenceTraits<RunPrincipal, BranchActionStreamBegin> Traits;
      
      beginStreamsTransitionAsync<Traits>(streamLoopWaitTask.get(),
                                         *schedule_,
                                         preallocations_.numberOfStreams(),
                                         runPrincipal,
                                         ts,
                                         es,
                                         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);
    }
  }

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

inlineprivate

Definition at line 276 of file EventProcessor.h.

References edm::get_underlying_safe().

Referenced by init().

{return get_underlying_safe(branchIDListHelper_);}

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

inlineprivate

Definition at line 277 of file EventProcessor.h.

References edm::get_underlying_safe().

{return get_underlying_safe(branchIDListHelper_);}

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

private

Definition at line 703 of file EventProcessor.cc.

References epSignal, and edm::shutdown_flag.

Referenced by nextTransitionType().

                                                                 {
    bool returnValue = false;

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

void edm::EventProcessor::clearCounters

(

)

Clears counters used by trigger report.

Definition at line 694 of file EventProcessor.cc.

References schedule_.

                                {
    schedule_->clearCounters();
  }

void edm::EventProcessor::closeInputFile

(

bool

cleaningUpAfterException

)

Definition at line 840 of file EventProcessor.cc.

References actReg_, fb_, FDEBUG, and input_.

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

void edm::EventProcessor::closeOutputFiles

(

)

Definition at line 857 of file EventProcessor.cc.

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

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

void edm::EventProcessor::continueLumiAsync

(

edm::WaitingTaskHolder

iHolder

)

Definition at line 1238 of file EventProcessor.cc.

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

Referenced by processLumis().

                                                                {
    {
      //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(std::move(h), streamIndex);
      }) );

    }
    tbb::task::spawn( *edm::make_functor_task(tbb::task::allocate_root(),[this,streamIndex,h=std::move(iHolder)](){
      handleNextEventForStreamAsync(std::move(h),streamIndex);
    }) );
  }

void edm::EventProcessor::deleteLumiFromCache

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 1501 of file EventProcessor.cc.

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

Referenced by globalEndLumiAsync().

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

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

Definition at line 1479 of file EventProcessor.cc.

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

Referenced by endUnfinishedRun().

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

void edm::EventProcessor::doErrorStuff

(

)

Definition at line 929 of file EventProcessor.cc.

References FDEBUG.

Referenced by runToCompletion().

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

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 679 of file EventProcessor.cc.

References schedule_.

                                            {
    schedule_->enableEndPaths(active);
  }

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 623 of file EventProcessor.cc.

References actReg_, EnergyCorrector::c, edm::ExceptionCollector::call(), edm::SubProcess::doEndJob(), edm::InputSource::doEndJob(), edm::EDLooperBase::endOfJob(), edm::ExceptionCollector::hasThrown(), mps_fire::i, input_, looper(), looper_, edm::PreallocationConfiguration::numberOfStreams(), cmsPerfStripChart::operate(), preallocations_, edm::ExceptionCollector::rethrow(), schedule_, serviceToken_, and subProcesses_.

Referenced by PythonEventProcessor::~PythonEventProcessor().

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

bool edm::EventProcessor::endOfLoop

(

)

Definition at line 892 of file EventProcessor.cc.

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

Referenced by runToCompletion().

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

bool edm::EventProcessor::endPathsEnabled

(

)

const

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

Definition at line 684 of file EventProcessor.cc.

References schedule_.

                                        {
    return schedule_->endPathsEnabled();
  }

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

Definition at line 1030 of file EventProcessor.cc.

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(), edm::PrincipalCache::runPrincipal(), schedule_, serviceToken_, edm::RunPrincipal::setEndTime(), and subProcesses_.

Referenced by endUnfinishedRun().

                                                                                                                                     {
    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();
    }
    EventSetup const& es = esp_->eventSetup();
    if(globalBeginSucceeded){
      //To wait, the ref count has to be 1+#streams
      auto streamLoopWaitTask = make_empty_waiting_task();
      streamLoopWaitTask->increment_ref_count();
      
      typedef OccurrenceTraits<RunPrincipal, BranchActionStreamEnd> Traits;
      
      endStreamsTransitionAsync<Traits>(WaitingTaskHolder(streamLoopWaitTask.get()),
                                       *schedule_,
                                       preallocations_.numberOfStreams(),
                                       runPrincipal,
                                       ts,
                                       es,
                                       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();

      typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd> Traits;
      endGlobalTransitionAsync<Traits>(WaitingTaskHolder(globalWaitTask.get()),
                                       *schedule_,
                                       runPrincipal,
                                       ts,
                                       es,
                                       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_);
    }
  }

void edm::EventProcessor::endUnfinishedLumi

(

)

Definition at line 1380 of file EventProcessor.cc.

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

                                         {
    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, streamLumiStatus_[i]);
          }
        }
      }
      globalWaitTask->wait_for_all();
      if(globalWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(* (globalWaitTask->exceptionPtr()) );
      }
    }
  }

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

Definition at line 1014 of file EventProcessor.cc.

References deleteRunFromCache(), endRun(), edm::make_empty_waiting_task(), run(), lumiQTWidget::t, and writeRunAsync().

                                                                                                                                               {
    //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();
      writeRunAsync(edm::WaitingTaskHolder{t.get()}, phid, run);
      t->wait_for_all();
      if(t->exceptionPtr()) {
        std::rethrow_exception(*t->exceptionPtr());
      }
    }
    deleteRunFromCache(phid, run);
  }

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 659 of file EventProcessor.cc.

References schedule_.

                                                 {
    return schedule_->getAllModuleDescriptions();
  }

ServiceToken edm::EventProcessor::getToken

(

)

Definition at line 654 of file EventProcessor.cc.

References serviceToken_.

Referenced by ~EventProcessor().

                           {
    return serviceToken_;
  }

void edm::EventProcessor::getTriggerReport

(

TriggerReport &

rep

)

const

Return the trigger report information on paths, modules-in-path, modules-in-endpath, and modules.

Definition at line 689 of file EventProcessor.cc.

References schedule_.

                                                           {
    schedule_->getTriggerReport(rep);
  }

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

Definition at line 1259 of file EventProcessor.cc.

References deleteLumiFromCache(), edm::WaitingTaskHolder::doneWaiting(), esp_, iovQueue_, mps_monitormerge::items, looper_, edm::make_waiting_task(), edm::EventID::maxEventNumber(), eostools::move(), cmsPerfStripChart::operate(), processContext_, edm::SerialTaskQueue::resume(), schedule_, serviceToken_, mps_update::status, subProcesses_, lumiQTWidget::t, tmp, and writeLumiAsync().

Referenced by streamEndLumiAsync().

                                                                                                                                {
    //Need to be sure iTask is always destroyed after iLumiStatus since iLumiStatus can cause endRun to start.
    auto t = edm::make_waiting_task(tbb::task::allocate_root(), [ items = std::make_pair(iLumiStatus,std::move(iTask)), this] (std::exception_ptr const* iPtr) mutable {
      std::exception_ptr ptr;
      //use an easier to remember variable name
      auto status = std::move(items.first);
      if(iPtr) {
        ptr = *iPtr;
        WaitingTaskHolder tmp(items.second);
        //set the exception early to prevent a beginLumi from running
        // we use a copy to keep t from resetting on doneWaiting call.
        tmp.doneWaiting(ptr);
      } else {
        try {
          ServiceRegistry::Operate operate(serviceToken_);
          if(looper_) {
            auto& lp = *(status->lumiPrincipal());
            EventSetup const& es = esp_->eventSetup();
            looper_->doEndLuminosityBlock(lp, es, &processContext_);
          }
        }catch(...) {
          if(not ptr) {
            ptr = std::current_exception();
          }
        }
      }
      ServiceRegistry::Operate operate(serviceToken_);
      try {
        deleteLumiFromCache(*status);
        //release our hold on the IOV
        iovQueue_.resume();
        status->resumeGlobalLumiQueue();
      } catch(...) {
        if( not ptr) {
          ptr = std::current_exception();
        }
      }
      try {
        status.reset();
      } catch(...) {
        if( not ptr) {
          ptr = std::current_exception();
        }
      }
      //have to wait until reset is called since that could call endRun
      items.second.doneWaiting(ptr);
    });

    auto writeT = edm::make_waiting_task(tbb::task::allocate_root(), [this,status =iLumiStatus, task = WaitingTaskHolder(t)] (std::exception_ptr const* iExcept) mutable {
      if(iExcept) {
        task.doneWaiting(*iExcept);
      } else {
        //Only call writeLumi if beginLumi succeeded
        if(status->didGlobalBeginSucceed()) {
          writeLumiAsync(std::move(task),status);
        }
      }
    });
    auto& lp = *(iLumiStatus->lumiPrincipal());

    IOVSyncValue ts(EventID(lp.run(), lp.luminosityBlock(), EventID::maxEventNumber()),
                    lp.beginTime());


    typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd> Traits;
    EventSetup const& es = esp_->eventSetup();

    endGlobalTransitionAsync<Traits>(WaitingTaskHolder(writeT),
                                     *schedule_,
                                     lp,
                                     ts,
                                     es,
                                     serviceToken_,
                                     subProcesses_,
                                     iLumiStatus->cleaningUpAfterException());
  }

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

private

Definition at line 1566 of file EventProcessor.cc.

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

Referenced by beginLumiAsync(), and continueLumiAsync().

  {
    sourceResourcesAcquirer_.serialQueueChain().push([this,iTask,iStreamIndex]() mutable {
           ServiceRegistry::Operate operate(serviceToken_);
           auto& status = streamLumiStatus_[iStreamIndex];
           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) {
                   bool expected = false;
                   if(deferredExceptionPtrIsSet_.compare_exchange_strong(expected,true)) {
                     deferredExceptionPtr_ = *iPtr;
                     iTask.doneWaiting(*iPtr);
                   }
                   //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, status);
               } else {
                 iTask.doneWaiting(std::exception_ptr{});
               }
             }
           } catch(...) {
             bool expected = false;
             if(deferredExceptionPtrIsSet_.compare_exchange_strong(expected,true)) {
               auto e =std::current_exception();
               deferredExceptionPtr_ = e;
               iTask.doneWaiting(e);
             }
           }
    });
  }

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

private

Definition at line 376 of file EventProcessor.cc.

References act_table_, actReg_, branchIDListHelper(), branchIDListHelper_, trackingPlots::common, edm::errors::Configuration, esp_, espController_, Exception, FDEBUG, fileModeNoMerge_, edm::fillLooper(), forceESCacheClearOnNewRun_, edm::ParameterSet::getUntrackedParameter(), historyAppender_, input_, edm::PrincipalCache::insert(), edm::ParentageRegistry::insertMapped(), edm::ParentageRegistry::instance(), mps_monitormerge::items, edm::serviceregistry::kConfigurationOverrides, looper_, lumiQueue_, edm::makeInput(), eostools::move(), jets_cff::nThreads, edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfStreams(), cmsPerfStripChart::operate(), edm::parameterSet(), edm::popSubProcessVParameterSet(), preallocations_, preg(), preg_, principalCache_, printDependencies_, processConfiguration_, processContext_, edm::ParameterSet::registerIt(), schedule_, serviceToken_, edm::PrincipalCache::setNumberOfConcurrentPrincipals(), edm::ProcessContext::setProcessConfiguration(), edm::PrincipalCache::setProcessHistoryRegistry(), edm::IllegalParameters::setThrowAnException(), streamLumiStatus_, streamQueues_, AlCaHLTBitMon_QueryRunRegistry::string, subProcesses_, thinnedAssociationsHelper(), thinnedAssociationsHelper_, and edm::validateTopLevelParameterSets().

Referenced by EventProcessor().

                                                             {

    //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) {
      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
    esp_ = espController_->makeProvider(*parameterSet, items.actReg_.get());

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

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

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

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

inline

Definition at line 189 of file EventProcessor.h.

References Ecal2004TBTDCRanges_v1_cff::endRun, init, edm::InputSource::IsStop, findQualityFiles::run, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by handleNextEventForStreamAsync(), and processLumis().

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

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

inlineprivate

Definition at line 280 of file EventProcessor.h.

References edm::get_underlying_safe().

Referenced by endJob().

{return get_underlying_safe(looper_);}

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

inlineprivate

Definition at line 281 of file EventProcessor.h.

References edm::get_underlying_safe().

{return get_underlying_safe(looper_);}

edm::LuminosityBlockNumber_t edm::EventProcessor::nextLuminosityBlockID

(

)

Definition at line 747 of file EventProcessor.cc.

References input_.

Referenced by readNextEventForStream().

                                        {
    return input_->luminosityBlock();
  }

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

(

)

Definition at line 742 of file EventProcessor.cc.

References input_.

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

InputSource::ItemType edm::EventProcessor::nextTransitionType

(

)

Definition at line 715 of file EventProcessor.cc.

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

Referenced by readNextEventForStream().

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

void edm::EventProcessor::openOutputFiles

(

)

Definition at line 849 of file EventProcessor.cc.

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

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

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

delete

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

inlineprivate

Definition at line 274 of file EventProcessor.h.

References edm::get_underlying_safe().

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

{return get_underlying_safe(preg_);}

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

inlineprivate

Definition at line 275 of file EventProcessor.h.

References edm::get_underlying_safe().

{return get_underlying_safe(preg_);}

void edm::EventProcessor::prepareForNextLoop

(

)

Definition at line 911 of file EventProcessor.cc.

References esp_, FDEBUG, edm::propagate_const< T >::get(), and looper_.

Referenced by runToCompletion().

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

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

inline

Definition at line 131 of file EventProcessor.h.

References EcalCondTools::getToken(), and cuy::rep.

{ return *processConfiguration_; }

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

private

Definition at line 1625 of file EventProcessor.cc.

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

Referenced by handleNextEventForStreamAsync().

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

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

private

Definition at line 1632 of file EventProcessor.cc.

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

Referenced by processEventAsync().

                                                                    {
    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](std::exception_ptr const* iPtr) mutable
             {

               //NOTE: If we have a looper we only have one Stream
               if(looper_) {
                 ServiceRegistry::Operate operate(serviceToken_);
                 processEventWithLooper(*pep);
               }
               
               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] (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);
           }
         } else {
           finalizeEventTask.doneWaiting(*iPtr);
         }
       })
                                           );
    }
    
    schedule_->processOneEventAsync(std::move(afterProcessTask),
                                    iStreamIndex,*pep, esp_->eventSetup(), serviceToken_);

  }

void edm::EventProcessor::processEventWithLooper ( EventPrincipal &  iPrincipal )

private

Definition at line 1693 of file EventProcessor.cc.

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(), and summarizeEdmComparisonLogfiles::succeeded.

Referenced by processEventAsyncImpl().

                                                                        {
    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_);
      status = looper_->doDuringLoop(iPrincipal, esp_->eventSetup(), 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;
    }
  }

InputSource::ItemType edm::EventProcessor::processLumis

(

std::shared_ptr< void > const &

iRunResource

)

Definition at line 1093 of file EventProcessor.cc.

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

                                                                      {
    auto waitTask = make_empty_waiting_task();
    waitTask->increment_ref_count();
    
    if(streamLumiActive_> 0) {
      assert(streamLumiActive_ == preallocations_.numberOfStreams());
      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();
  }

int edm::EventProcessor::readAndMergeLumi

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 1449 of file EventProcessor.cc.

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

Referenced by readNextEventForStream().

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

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

(

)

Definition at line 1417 of file EventProcessor.cc.

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

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

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

private

Definition at line 1611 of file EventProcessor.cc.

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

Referenced by readNextEventForStream().

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

void edm::EventProcessor::readFile

(

)

Definition at line 823 of file EventProcessor.cc.

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

Referenced by Vispa.Plugins.EventBrowser.EventBrowserTabController.EventBrowserTabController::navigate(), Vispa.Main.TabController.TabController::open(), and Vispa.Main.TabController.TabController::refresh().

                                {
    FDEBUG(1) << " \treadFile\n";
    size_t size = preg_->size();
    SendSourceTerminationSignalIfException sentry(actReg_.get());

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

void edm::EventProcessor::readLuminosityBlock

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 1429 of file EventProcessor.cc.

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

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

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

private

Definition at line 1507 of file EventProcessor.cc.

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

Referenced by handleNextEventForStreamAsync().

                                                                                        {
    if(shouldWeStop()) {
      return false;
    }
    
    if(deferredExceptionPtrIsSet_.load(std::memory_order_acquire)) {
      return false;
    }
    
    if(iStatus.wasEventProcessingStopped()) {
      return false;
    }

    ServiceRegistry::Operate operate(serviceToken_);
    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();
      }
      return false;
    }
    return true;
  }

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

(

)

Definition at line 1399 of file EventProcessor.cc.

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

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

void edm::EventProcessor::respondToCloseInputFile

(

)

Definition at line 874 of file EventProcessor.cc.

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

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

void edm::EventProcessor::respondToOpenInputFile

(

)

Definition at line 865 of file EventProcessor.cc.

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

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

void edm::EventProcessor::rewindInput

(

)

Definition at line 905 of file EventProcessor.cc.

References FDEBUG, and input_.

Referenced by runToCompletion().

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

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

inline

Definition at line 349 of file EventProcessor.h.

Referenced by Types.EventID::cppID(), Types.LuminosityBlockID::cppID(), and endUnfinishedRun().

                      {
    return runToCompletion();
  }

EventProcessor::StatusCode edm::EventProcessor::runToCompletion

(

)

Definition at line 752 of file EventProcessor.cc.

References cms::Exception::addAdditionalInfo(), cms::Exception::alreadyPrinted(), asyncStopRequestedWhileProcessingEvents_, asyncStopStatusCodeFromProcessingEvents_, beginJob(), deferredExceptionPtr_, deferredExceptionPtrIsSet_, doErrorStuff(), MillePedeFileConverter_cfg::e, endOfLoop(), epSuccess, Exception, exceptionMessageFiles_, exceptionMessageLumis_, exceptionMessageRuns_, fileModeNoMerge_, edm::InputSource::IsStop, cmsPerfStripChart::operate(), prepareForNextLoop(), runEdmFileComparison::returnCode, rewindInput(), serviceToken_, startingNewLoop(), and edm::convertException::wrap().

Referenced by PythonEventProcessor::run().

                                  {
    
    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_.empty()) {
          e.addAdditionalInfo(exceptionMessageLumis_);
          if (e.alreadyPrinted()) {
            LogAbsolute("Additional Exceptions") << exceptionMessageLumis_;
          }
        }
        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;
  }

bool edm::EventProcessor::setDeferredException

(

std::exception_ptr

iException

)

Definition at line 1748 of file EventProcessor.cc.

References deferredExceptionPtr_, and deferredExceptionPtrIsSet_.

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

void edm::EventProcessor::setExceptionMessageFiles

(

std::string &

message

)

Definition at line 1736 of file EventProcessor.cc.

References exceptionMessageFiles_.

                                                                  {
    exceptionMessageFiles_ = message;
  }

void edm::EventProcessor::setExceptionMessageLumis

(

std::string &

message

)

Definition at line 1744 of file EventProcessor.cc.

References exceptionMessageLumis_.

                                                                  {
    exceptionMessageLumis_ = message;
  }

void edm::EventProcessor::setExceptionMessageRuns

(

std::string &

message

)

Definition at line 1740 of file EventProcessor.cc.

References exceptionMessageRuns_.

                                                                 {
    exceptionMessageRuns_ = message;
  }

bool edm::EventProcessor::shouldWeCloseOutput

(

)

const

Definition at line 916 of file EventProcessor.cc.

References FDEBUG, schedule_, and subProcesses_.

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

bool edm::EventProcessor::shouldWeStop

(

)

const

Definition at line 1722 of file EventProcessor.cc.

References FDEBUG, schedule_, shouldWeStop_, and subProcesses_.

Referenced by readNextEventForStream().

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

void edm::EventProcessor::startingNewLoop

(

)

Definition at line 882 of file EventProcessor.cc.

References FDEBUG, looper_, looperBeginJobRun_, and shouldWeStop_.

Referenced by runToCompletion().

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

void edm::EventProcessor::streamEndLumiAsync	(	edm::WaitingTaskHolder	iTask,
		unsigned int	iStreamIndex,
		std::shared_ptr< LuminosityBlockProcessingStatus >	iLumiStatus
	)

Definition at line 1336 of file EventProcessor.cc.

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

Referenced by endUnfinishedLumi(), and handleNextEventForStreamAsync().

                                                                                                      {
    
    auto t =edm::make_waiting_task(tbb::task::allocate_root(), [this, iStreamIndex, iTask](std::exception_ptr const* iPtr) mutable {
      std::exception_ptr ptr;
      if(iPtr) {
        ptr = *iPtr;
      }
      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));
      }
      iTask.doneWaiting(ptr);
    });
    
    edm::WaitingTaskHolder lumiDoneTask{t};
    
    iLumiStatus->setEndTime();
    
    if(iLumiStatus->didGlobalBeginSucceed()) {
      auto & lumiPrincipal = *iLumiStatus->lumiPrincipal();
      IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), EventID::maxEventNumber()),
                      lumiPrincipal.endTime());
      EventSetup const& es = esp_->eventSetup();

      bool cleaningUpAfterException = iLumiStatus->cleaningUpAfterException();
      
      using Traits =  OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd>;
      endStreamTransitionAsync<Traits>(std::move(lumiDoneTask),
                                       *schedule_,iStreamIndex,
                                       lumiPrincipal,ts,es,
                                       serviceToken_,
                                       subProcesses_,cleaningUpAfterException);
    }
  }

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

inlineprivate

Definition at line 278 of file EventProcessor.h.

References edm::get_underlying_safe().

Referenced by init().

{return get_underlying_safe(thinnedAssociationsHelper_);}

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

inlineprivate

Definition at line 279 of file EventProcessor.h.

References edm::get_underlying_safe().

{return get_underlying_safe(thinnedAssociationsHelper_);}

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 664 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEvents().

                                    {
    return schedule_->totalEvents();
  }

int edm::EventProcessor::totalEventsFailed

(

)

const

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

Definition at line 674 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEventsFailed().

                                          {
    return schedule_->totalEventsFailed();
  }

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 669 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEventsPassed().

                                          {
    return schedule_->totalEventsPassed();
  }

void edm::EventProcessor::writeLumiAsync	(	WaitingTaskHolder	task,
		std::shared_ptr< LuminosityBlockProcessingStatus >	iStatus
	)

Definition at line 1485 of file EventProcessor.cc.

References actReg_, edm::WaitingTaskHolder::doneWaiting(), edm::make_waiting_task(), processContext_, alignCSCRings::s, schedule_, serviceToken_, and subProcesses_.

Referenced by globalEndLumiAsync().

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

    schedule_->writeLumiAsync(WaitingTaskHolder{subsT}, *(iStatus->lumiPrincipal()), &processContext_, actReg_.get());
  }

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

Definition at line 1464 of file EventProcessor.cc.

References actReg_, edm::WaitingTaskHolder::doneWaiting(), edm::make_waiting_task(), principalCache_, processContext_, edm::PrincipalCache::runPrincipal(), alignCSCRings::s, schedule_, serviceToken_, and subProcesses_.

Referenced by endUnfinishedRun().

                                                                                                          {
    auto subsT = edm::make_waiting_task(tbb::task::allocate_root(), [this,phid,run,task](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);
        }
      }
    });
    ServiceRegistry::Operate op(serviceToken_);
    schedule_->writeRunAsync(WaitingTaskHolder(subsT), principalCache_.runPrincipal(phid, run), &processContext_, actReg_.get());
  }

Member Data Documentation

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

private

Definition at line 299 of file EventProcessor.h.

Referenced by init().

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

private

Definition at line 289 of file EventProcessor.h.

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

bool edm::EventProcessor::asyncStopRequestedWhileProcessingEvents_

private

Definition at line 334 of file EventProcessor.h.

Referenced by runToCompletion().

StatusCode edm::EventProcessor::asyncStopStatusCodeFromProcessingEvents_

private

Definition at line 335 of file EventProcessor.h.

Referenced by runToCompletion().

bool edm::EventProcessor::beginJobCalled_

private

Definition at line 322 of file EventProcessor.h.

Referenced by beginJob().

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

private

Definition at line 291 of file EventProcessor.h.

Referenced by init(), and respondToOpenInputFile().

std::exception_ptr edm::EventProcessor::deferredExceptionPtr_

private

Definition at line 317 of file EventProcessor.h.

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

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

private

Definition at line 316 of file EventProcessor.h.

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

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

private

Definition at line 297 of file EventProcessor.h.

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

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

private

Definition at line 296 of file EventProcessor.h.

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

ExcludedDataMap edm::EventProcessor::eventSetupDataToExcludeFromPrefetching_

private

Definition at line 340 of file EventProcessor.h.

std::string edm::EventProcessor::exceptionMessageFiles_

private

Definition at line 325 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageFiles().

std::string edm::EventProcessor::exceptionMessageLumis_

private

Definition at line 327 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageLumis().

std::string edm::EventProcessor::exceptionMessageRuns_

private

Definition at line 326 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageRuns().

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

private

Definition at line 312 of file EventProcessor.h.

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

bool edm::EventProcessor::fileModeNoMerge_

private

Definition at line 324 of file EventProcessor.h.

Referenced by init(), and runToCompletion().

bool edm::EventProcessor::firstEventInBlock_ =true

private

Definition at line 336 of file EventProcessor.h.

bool edm::EventProcessor::forceESCacheClearOnNewRun_

private

Definition at line 330 of file EventProcessor.h.

Referenced by beginRun(), and init().

bool edm::EventProcessor::forceLooperToEnd_

private

Definition at line 328 of file EventProcessor.h.

Referenced by endOfLoop().

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

private

Definition at line 310 of file EventProcessor.h.

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

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

private

Definition at line 294 of file EventProcessor.h.

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

edm::SerialTaskQueue edm::EventProcessor::iovQueue_

private

Definition at line 298 of file EventProcessor.h.

Referenced by beginLumiAsync(), and globalEndLumiAsync().

InputSource::ItemType edm::EventProcessor::lastSourceTransition_

private

Definition at line 295 of file EventProcessor.h.

Referenced by nextTransitionType(), and processEventWithLooper().

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

private

Definition at line 313 of file EventProcessor.h.

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

bool edm::EventProcessor::looperBeginJobRun_

private

Definition at line 329 of file EventProcessor.h.

Referenced by beginRun(), and startingNewLoop().

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

private

Definition at line 305 of file EventProcessor.h.

Referenced by beginLumiAsync(), and init().

PathsAndConsumesOfModules edm::EventProcessor::pathsAndConsumesOfModules_

private

Definition at line 302 of file EventProcessor.h.

Referenced by beginJob().

PreallocationConfiguration edm::EventProcessor::preallocations_

private

Definition at line 332 of file EventProcessor.h.

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

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

private

Definition at line 290 of file EventProcessor.h.

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

PrincipalCache edm::EventProcessor::principalCache_

private

Definition at line 321 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginRun(), deleteRunFromCache(), endRun(), init(), processEventAsyncImpl(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readRun(), and writeRunAsync().

bool edm::EventProcessor::printDependencies_ = false

private

Definition at line 342 of file EventProcessor.h.

Referenced by beginJob(), and init().

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

private

Definition at line 300 of file EventProcessor.h.

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

ProcessContext edm::EventProcessor::processContext_

private

Definition at line 301 of file EventProcessor.h.

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

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

private

Definition at line 303 of file EventProcessor.h.

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

ServiceToken edm::EventProcessor::serviceToken_

private

Definition at line 293 of file EventProcessor.h.

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

bool edm::EventProcessor::shouldWeStop_

private

Definition at line 323 of file EventProcessor.h.

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

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

private

Definition at line 320 of file EventProcessor.h.

Referenced by readNextEventForStream().

SharedResourcesAcquirer edm::EventProcessor::sourceResourcesAcquirer_

private

Definition at line 319 of file EventProcessor.h.

Referenced by beginLumiAsync(), and handleNextEventForStreamAsync().

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

private

Definition at line 307 of file EventProcessor.h.

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

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

private

Definition at line 306 of file EventProcessor.h.

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

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

private

Definition at line 304 of file EventProcessor.h.

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

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

private

Definition at line 309 of file EventProcessor.h.

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

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

private

Definition at line 292 of file EventProcessor.h.

Referenced by init().