edm::EventProcessor Class Reference

#include <EventProcessor.h>

Inheritance diagram for edm::EventProcessor:

Public Member Functions
virtual bool	alreadyHandlingException () const override
void	beginJob ()
virtual void	beginLumi (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi) override
virtual void	beginRun (statemachine::Run const &run) override
void	clearCounters ()
	Clears counters used by trigger report. More...
virtual void	closeInputFile (bool cleaningUpAfterException) override
virtual void	closeOutputFiles () override
virtual void	deleteLumiFromCache (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi) override
virtual void	deleteRunFromCache (statemachine::Run const &run) override
virtual void	doErrorStuff () override
void	enableEndPaths (bool active)
void	endJob ()
virtual void	endLumi (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi, bool cleaningUpAfterException) override
virtual bool	endOfLoop () override
bool	endPathsEnabled () const
virtual void	endRun (statemachine::Run const &run, bool cleaningUpAfterException) override
	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
bool	forkProcess (std::string const &jobReportFile)
std::vector< ModuleDescription const * >	getAllModuleDescriptions () const
ServiceToken	getToken ()
void	getTriggerReport (TriggerReport &rep) const
virtual void	openOutputFiles () override
EventProcessor &	operator= (EventProcessor const &)=delete
virtual void	prepareForNextLoop () override
ProcessConfiguration const &	processConfiguration () const
virtual int	readAndMergeLumi () override
virtual statemachine::Run	readAndMergeRun () override
virtual void	readAndProcessEvent () override
virtual void	readFile () override
virtual int	readLuminosityBlock () override
virtual statemachine::Run	readRun () override
virtual void	respondToCloseInputFile () override
virtual void	respondToOpenInputFile () override
virtual void	rewindInput () override
StatusCode	run ()
virtual StatusCode	runToCompletion () override
virtual void	setExceptionMessageFiles (std::string &message) override
virtual void	setExceptionMessageLumis (std::string &message) override
virtual void	setExceptionMessageRuns (std::string &message) override
virtual bool	shouldWeCloseOutput () const override
virtual bool	shouldWeStop () const override
virtual void	startingNewLoop () override
int	totalEvents () const
int	totalEventsFailed () const
int	totalEventsPassed () const
virtual void	writeLumi (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi) override
virtual void	writeRun (statemachine::Run const &run) override
	~EventProcessor ()
Public Member Functions inherited from edm::IEventProcessor
virtual	~IEventProcessor ()

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 &)
std::unique_ptr< statemachine::Machine >	createStateMachine ()
void	handleNextEventForStreamAsync (WaitingTask *iTask, unsigned int iStreamIndex, std::atomic< bool > *finishedProcessingEvents)
void	init (std::shared_ptr< ProcessDesc > &processDesc, ServiceToken const &token, serviceregistry::ServiceLegacy)
std::shared_ptr< EDLooperBase const >	looper () const
std::shared_ptr< EDLooperBase > &	looper ()
void	possiblyContinueAfterForkChildFailure ()
std::shared_ptr< ProductRegistry const >	preg () const
std::shared_ptr< ProductRegistry > &	preg ()
void	processEventAsync (WaitingTaskHolder iHolder, unsigned int iStreamIndex)
void	processEventWithLooper (EventPrincipal &)
void	readEvent (unsigned int iStreamIndex)
bool	readNextEventForStream (unsigned int iStreamIndex, std::atomic< bool > *finishedProcessingEvents)
void	setupSignal ()
void	terminateMachine (std::unique_ptr< statemachine::Machine >)
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	alreadyHandlingException_
bool	asyncStopRequestedWhileProcessingEvents_
StatusCode	asyncStopStatusCodeFromProcessingEvents_
bool	beginJobCalled_
edm::propagate_const< std::shared_ptr< BranchIDListHelper > >	branchIDListHelper_
bool	continueAfterChildFailure_
std::exception_ptr	deferredExceptionPtr_
std::atomic< bool >	deferredExceptionPtrIsSet_
std::string	emptyRunLumiMode_
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_
std::string	fileMode_
bool	firstEventInBlock_ =true
bool	forceESCacheClearOnNewRun_
bool	forceLooperToEnd_
edm::propagate_const< std::unique_ptr< HistoryAppender > >	historyAppender_
edm::propagate_const< std::unique_ptr< InputSource > >	input_
edm::propagate_const< std::shared_ptr< EDLooperBase > >	looper_
bool	looperBeginJobRun_
InputSource::ItemType	nextItemTypeFromProcessingEvents_
int	numberOfForkedChildren_
unsigned int	numberOfSequentialEventsPerChild_
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	setCpuAffinity_
bool	shouldWeStop_
std::shared_ptr< std::recursive_mutex >	sourceMutex_
SharedResourcesAcquirer	sourceResourcesAcquirer_
bool	stateMachineWasInErrorState_
std::vector< SubProcess >	subProcesses_
edm::propagate_const< std::shared_ptr< ThinnedAssociationsHelper > >	thinnedAssociationsHelper_

Additional Inherited Members
Public Types inherited from edm::IEventProcessor
enum	Status {   epSuccess =0, epException =1, epOther =2, epSignal =3,   epInputComplete =4, epTimedOut =5, epCountComplete =6 }
typedef Status	StatusCode

Detailed Description

Definition at line 65 of file EventProcessor.h.

Member Typedef Documentation

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

private

Definition at line 326 of file EventProcessor.h.

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

private

Definition at line 327 of file EventProcessor.h.

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 227 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),
    stateMachineWasInErrorState_(false),
    fileMode_(),
    emptyRunLumiMode_(),
    exceptionMessageFiles_(),
    exceptionMessageRuns_(),
    exceptionMessageLumis_(),
    alreadyHandlingException_(false),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    numberOfForkedChildren_(0),
    numberOfSequentialEventsPerChild_(1),
    setCpuAffinity_(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 272 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),
    stateMachineWasInErrorState_(false),
    fileMode_(),
    emptyRunLumiMode_(),
    exceptionMessageFiles_(),
    exceptionMessageRuns_(),
    exceptionMessageLumis_(),
    alreadyHandlingException_(false),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    numberOfForkedChildren_(0),
    numberOfSequentialEventsPerChild_(1),
    setCpuAffinity_(false),
    asyncStopRequestedWhileProcessingEvents_(false),
    nextItemTypeFromProcessingEvents_(InputSource::IsEvent),
    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 318 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),
    stateMachineWasInErrorState_(false),
    fileMode_(),
    emptyRunLumiMode_(),
    exceptionMessageFiles_(),
    exceptionMessageRuns_(),
    exceptionMessageLumis_(),
    alreadyHandlingException_(false),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    numberOfForkedChildren_(0),
    numberOfSequentialEventsPerChild_(1),
    setCpuAffinity_(false),
    asyncStopRequestedWhileProcessingEvents_(false),
    nextItemTypeFromProcessingEvents_(InputSource::IsEvent),
    eventSetupDataToExcludeFromPrefetching_()
  {
    init(processDesc, token, legacy);
  }

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

meant for unit tests

Definition at line 362 of file EventProcessor.cc.

References mps_alisetup::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),
    stateMachineWasInErrorState_(false),
    fileMode_(),
    emptyRunLumiMode_(),
    exceptionMessageFiles_(),
    exceptionMessageRuns_(),
    exceptionMessageLumis_(),
    alreadyHandlingException_(false),
    forceLooperToEnd_(false),
    looperBeginJobRun_(false),
    forceESCacheClearOnNewRun_(false),
    numberOfForkedChildren_(0),
    numberOfSequentialEventsPerChild_(1),
    setCpuAffinity_(false),
    asyncStopRequestedWhileProcessingEvents_(false),
    nextItemTypeFromProcessingEvents_(InputSource::IsEvent),
    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 588 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

bool edm::EventProcessor::alreadyHandlingException ( ) const

overridevirtual

Implements edm::IEventProcessor.

Definition at line 2220 of file EventProcessor.cc.

References alreadyHandlingException_.

                                                      {
    return alreadyHandlingException_;
  }

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 607 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 forkProcess(), and 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::beginLumi ( ProcessHistoryID const &  phid, RunNumber_t  run, LuminosityBlockNumber_t  lumi  )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1741 of file EventProcessor.cc.

References actReg_, edm::LuminosityBlockPrincipal::beginTime(), esp_, espController_, FDEBUG, input_, edm::Service< T >::isAvailable(), looper_, edm::LuminosityBlockPrincipal::luminosityBlock(), edm::PrincipalCache::lumiPrincipal(), edm::make_empty_waiting_task(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, edm::LuminosityBlockPrincipal::run(), schedule_, and subProcesses_.

                                                                                                            {
    LuminosityBlockPrincipal& lumiPrincipal = principalCache_.lumiPrincipal(phid, run, lumi);
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());

      input_->doBeginLumi(lumiPrincipal, &processContext_);
      sentry.completedSuccessfully();
    }

    Service<RandomNumberGenerator> rng;
    if(rng.isAvailable()) {
      LuminosityBlock lb(lumiPrincipal, ModuleDescription(), nullptr);
      rng->preBeginLumi(lb);
    }

    // NOTE: Using 0 as the event number for the begin of a lumi block is a bad idea
    // lumi blocks know their start and end times why not also start and end events?
    IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), 0), lumiPrincipal.beginTime());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      espController_->eventSetupForInstance(ts);
      sentry.completedSuccessfully();
    }
    EventSetup const& es = esp_->eventSetup();
    {
      typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin> Traits;
      auto globalWaitTask = make_empty_waiting_task();
      globalWaitTask->increment_ref_count();
      beginGlobalTransitionAsync<Traits>(WaitingTaskHolder(globalWaitTask.get()),
                                         *schedule_,
                                         lumiPrincipal,
                                         ts,
                                         es,
                                         subProcesses_);
      globalWaitTask->wait_for_all();
      if(globalWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(* (globalWaitTask->exceptionPtr()) );
      }
    }
    FDEBUG(1) << "\tbeginLumi " << run << "/" << lumi << "\n";
    if(looper_) {
      looper_->doBeginLuminosityBlock(lumiPrincipal, es, &processContext_);
    }
    {
      //To wait, the ref count has to b 1+#streams
      auto streamLoopWaitTask = make_empty_waiting_task();
      streamLoopWaitTask->increment_ref_count();

      typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin> Traits;

      beginStreamsTransitionAsync<Traits>(streamLoopWaitTask.get(),
                                         *schedule_,
                                         preallocations_.numberOfStreams(),
                                         lumiPrincipal,
                                         ts,
                                         es,
                                         subProcesses_);
      streamLoopWaitTask->wait_for_all();
      if(streamLoopWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(* (streamLoopWaitTask->exceptionPtr()) );
      }
    }
    
    FDEBUG(1) << "\tstreamBeginLumi " << run << "/" << lumi << "\n";
    if(looper_) {
      //looper_->doStreamBeginLuminosityBlock(schedule_->streamID(),lumiPrincipal, es);
    }
  }

void edm::EventProcessor::beginRun ( statemachine::Run const &  run )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1614 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_, statemachine::Run::processHistoryID(), edm::RunPrincipal::run(), statemachine::Run::runNumber(), edm::PrincipalCache::runPrincipal(), schedule_, and subProcesses_.

                                                          {
    RunPrincipal& runPrincipal = principalCache_.runPrincipal(run.processHistoryID(), run.runNumber());
    {
      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,
                                         subProcesses_);
      globalWaitTask->wait_for_all();
      if(globalWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(* (globalWaitTask->exceptionPtr()) );
      }
    }
    FDEBUG(1) << "\tbeginRun " << run.runNumber() << "\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,
                                         subProcesses_);

      streamLoopWaitTask->wait_for_all();
      if(streamLoopWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(* (streamLoopWaitTask->exceptionPtr()) );
      }
    }
    FDEBUG(1) << "\tstreamBeginRun " << run.runNumber() << "\n";
    if(looper_) {
      //looper_->doStreamBeginRun(schedule_->streamID(),runPrincipal, es);
    }
  }

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

inlineprivate

Definition at line 262 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 263 of file EventProcessor.h.

References edm::get_underlying_safe().

{return get_underlying_safe(branchIDListHelper_);}

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

private

Definition at line 1295 of file EventProcessor.cc.

References edm::IEventProcessor::epSignal, and edm::shutdown_flag.

Referenced by readNextEventForStream(), and runToCompletion().

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

References schedule_.

                                {
    schedule_->clearCounters();
  }

void edm::EventProcessor::closeInputFile ( bool  cleaningUpAfterException )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1514 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 ( )

overridevirtual

Implements edm::IEventProcessor.

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

std::unique_ptr< statemachine::Machine > edm::EventProcessor::createStateMachine ( )

private

Definition at line 1263 of file EventProcessor.cc.

References edm::errors::Configuration, statemachine::doNotHandleEmptyRunsAndLumis, emptyRunLumiMode_, Exception, fileMode_, statemachine::FULLMERGE, statemachine::handleEmptyRuns, statemachine::handleEmptyRunsAndLumis, statemachine::NOMERGE, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by runToCompletion().

                                     {
    statemachine::FileMode fileMode;
    if(fileMode_.empty()) fileMode = statemachine::FULLMERGE;
    else if(fileMode_ == std::string("NOMERGE")) fileMode = statemachine::NOMERGE;
    else if(fileMode_ == std::string("FULLMERGE")) fileMode = statemachine::FULLMERGE;
    else {
      throw Exception(errors::Configuration, "Illegal fileMode parameter value: ")
        << fileMode_ << ".\n"
        << "Legal values are 'NOMERGE' and 'FULLMERGE'.\n";
    }

    statemachine::EmptyRunLumiMode emptyRunLumiMode;
    if(emptyRunLumiMode_.empty()) emptyRunLumiMode = statemachine::handleEmptyRunsAndLumis;
    else if(emptyRunLumiMode_ == std::string("handleEmptyRunsAndLumis")) emptyRunLumiMode = statemachine::handleEmptyRunsAndLumis;
    else if(emptyRunLumiMode_ == std::string("handleEmptyRuns")) emptyRunLumiMode = statemachine::handleEmptyRuns;
    else if(emptyRunLumiMode_ == std::string("doNotHandleEmptyRunsAndLumis")) emptyRunLumiMode = statemachine::doNotHandleEmptyRunsAndLumis;
    else {
      throw Exception(errors::Configuration, "Illegal emptyMode parameter value: ")
        << emptyRunLumiMode_ << ".\n"
        << "Legal values are 'handleEmptyRunsAndLumis', 'handleEmptyRuns', and 'doNotHandleEmptyRunsAndLumis'.\n";
    }

    auto machine = std::make_unique<statemachine::Machine>(
                                                           this,
                                                           fileMode,
                                                           emptyRunLumiMode);

    machine->initiate();
    return machine;
  }

void edm::EventProcessor::deleteLumiFromCache ( ProcessHistoryID const &  phid, RunNumber_t  run, LuminosityBlockNumber_t  lumi  )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1949 of file EventProcessor.cc.

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

                                                                                                                      {
    principalCache_.deleteLumi(phid, run, lumi);
    for_all(subProcesses_, [&phid, run, lumi](auto& subProcess){ subProcess.deleteLumiFromCache(phid, run, lumi); });
    FDEBUG(1) << "\tdeleteLumiFromCache " << run << "/" << lumi << "\n";
  }

void edm::EventProcessor::deleteRunFromCache ( statemachine::Run const &  run )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1937 of file EventProcessor.cc.

References edm::PrincipalCache::deleteRun(), FDEBUG, edm::for_all(), principalCache_, statemachine::Run::processHistoryID(), statemachine::Run::runNumber(), and subProcesses_.

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

void edm::EventProcessor::doErrorStuff ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1603 of file EventProcessor.cc.

References FDEBUG, and stateMachineWasInErrorState_.

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

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

void edm::EventProcessor::endLumi ( ProcessHistoryID const &  phid, RunNumber_t  run, LuminosityBlockNumber_t  lumi, bool  cleaningUpAfterException  )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1810 of file EventProcessor.cc.

References actReg_, edm::LuminosityBlockPrincipal::endTime(), esp_, espController_, FDEBUG, edm::for_all(), input_, looper_, edm::LuminosityBlockPrincipal::luminosityBlock(), edm::PrincipalCache::lumiPrincipal(), edm::make_empty_waiting_task(), edm::EventID::maxEventNumber(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, edm::LuminosityBlockPrincipal::run(), schedule_, edm::Principal::setAtEndTransition(), edm::LuminosityBlockPrincipal::setComplete(), edm::LuminosityBlockPrincipal::setEndTime(), and subProcesses_.

Referenced by Types.EventRange::cppID().

                                                                                                                                         {
    LuminosityBlockPrincipal& lumiPrincipal = principalCache_.lumiPrincipal(phid, run, lumi);
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());

      lumiPrincipal.setEndTime(input_->timestamp());
      lumiPrincipal.setComplete();
      input_->doEndLumi(lumiPrincipal, cleaningUpAfterException, &processContext_);
      sentry.completedSuccessfully();
    }
    //NOTE: Using the max event number for the end of a lumi block is a bad idea
    // lumi blocks know their start and end times why not also start and end events?
    IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), EventID::maxEventNumber()),
                    lumiPrincipal.endTime());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      espController_->eventSetupForInstance(ts);
      sentry.completedSuccessfully();
    }
    EventSetup const& es = esp_->eventSetup();
    {
      //To wait, the ref count has to b 1+#streams
      auto streamLoopWaitTask = make_empty_waiting_task();
      streamLoopWaitTask->increment_ref_count();
      
      typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd> Traits;
      
      endStreamsTransitionAsync<Traits>(streamLoopWaitTask.get(),
                                       *schedule_,
                                       preallocations_.numberOfStreams(),
                                       lumiPrincipal,
                                       ts,
                                       es,
                                       subProcesses_,
                                       cleaningUpAfterException);
      streamLoopWaitTask->wait_for_all();
      if(streamLoopWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(* (streamLoopWaitTask->exceptionPtr()) );
      }
    }
    FDEBUG(1) << "\tendLumi " << run << "/" << lumi << "\n";
    if(looper_) {
      //looper_->doStreamEndLuminosityBlock(schedule_->streamID(),lumiPrincipal, es);
    }
    {
      lumiPrincipal.setAtEndTransition(true);
      typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd> Traits;
      schedule_->processOneGlobal<Traits>(lumiPrincipal, es, cleaningUpAfterException);
      for_all(subProcesses_, [&lumiPrincipal, &ts, cleaningUpAfterException](auto& subProcess){ subProcess.doEndLuminosityBlock(lumiPrincipal, ts, cleaningUpAfterException); });
    }
    FDEBUG(1) << "\tendLumi " << run << "/" << lumi << "\n";
    if(looper_) {
      looper_->doEndLuminosityBlock(lumiPrincipal, es, &processContext_);
    }
  }

bool edm::EventProcessor::endOfLoop ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1566 of file EventProcessor.cc.

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

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

References schedule_.

                                        {
    return schedule_->endPathsEnabled();
  }

void edm::EventProcessor::endRun ( statemachine::Run const &  run, bool  cleaningUpAfterException  )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1685 of file EventProcessor.cc.

References actReg_, edm::RunPrincipal::endTime(), esp_, espController_, FDEBUG, edm::for_all(), input_, looper_, edm::make_empty_waiting_task(), edm::EventID::maxEventNumber(), edm::LuminosityBlockID::maxLuminosityBlockNumber(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, statemachine::Run::processHistoryID(), edm::RunPrincipal::run(), statemachine::Run::runNumber(), edm::PrincipalCache::runPrincipal(), schedule_, edm::Principal::setAtEndTransition(), edm::RunPrincipal::setComplete(), edm::RunPrincipal::setEndTime(), and subProcesses_.

                                                                                       {
    RunPrincipal& runPrincipal = principalCache_.runPrincipal(run.processHistoryID(), run.runNumber());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());

      runPrincipal.setEndTime(input_->timestamp());
      runPrincipal.setComplete();
      input_->doEndRun(runPrincipal, cleaningUpAfterException, &processContext_);
      sentry.completedSuccessfully();
    }

    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();
    {
      //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>(streamLoopWaitTask.get(),
                                       *schedule_,
                                       preallocations_.numberOfStreams(),
                                       runPrincipal,
                                       ts,
                                       es,
                                       subProcesses_,
                                       cleaningUpAfterException);
      
      streamLoopWaitTask->wait_for_all();
      if(streamLoopWaitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(* (streamLoopWaitTask->exceptionPtr()) );
      }
    }
    FDEBUG(1) << "\tstreamEndRun " << run.runNumber() << "\n";
    if(looper_) {
      //looper_->doStreamEndRun(schedule_->streamID(),runPrincipal, es);
    }
    {
      runPrincipal.setAtEndTransition(true);
      typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd> Traits;
      schedule_->processOneGlobal<Traits>(runPrincipal, es, cleaningUpAfterException);
      for_all(subProcesses_, [&runPrincipal, &ts, cleaningUpAfterException](auto& subProcess){subProcess.doEndRun(runPrincipal, ts, cleaningUpAfterException); });
    }
    FDEBUG(1) << "\tendRun " << run.runNumber() << "\n";
    if(looper_) {
      looper_->doEndRun(runPrincipal, es, &processContext_);
    }
  }

bool edm::EventProcessor::forkProcess

(

std::string const &

jobReportFile

)

Definition at line 921 of file EventProcessor.cc.

References actReg_, beginJob(), continueAfterChildFailure_, edm::eventsetup::EventSetupRecord::doGet(), MillePedeFileConverter_cfg::e, esp_, espController_, eventSetupDataToExcludeFromPrefetching_, Exception, cmsRelvalreport::exit, edm::EventSetup::fillAvailableRecordKeys(), edm::eventsetup::EventSetupRecord::fillRegisteredDataKeys(), edm::EventSetup::find(), input_, edm::installCustomHandler(), edm::InputSource::IsFile, edm::InputSource::IsRun, RecoTauDiscriminantConfiguration::mask, NULL, numberOfForkedChildren_, numberOfSequentialEventsPerChild_, O_NONBLOCK, cmsPerfStripChart::operate(), or, pipe::pipe(), possiblyContinueAfterForkChildFailure(), readFile(), schedule_, serviceToken_, setCpuAffinity_, edm::shutdown_flag, and cms::Exception::what().

                                                            {

    if(0 == numberOfForkedChildren_) {return true;}
    assert(0<numberOfForkedChildren_);
    //do what we want done in common
    {
      beginJob(); //make sure this was run
      // make the services available
      ServiceRegistry::Operate operate(serviceToken_);

      InputSource::ItemType itemType;
      itemType = input_->nextItemType();

      assert(itemType == InputSource::IsFile);
      {
        readFile();
      }
      itemType = input_->nextItemType();
      assert(itemType == InputSource::IsRun);

      LogSystem("ForkingEventSetupPreFetching") << " prefetching for run " << input_->runAuxiliary()->run();
      IOVSyncValue ts(EventID(input_->runAuxiliary()->run(), 0, 0),
                      input_->runAuxiliary()->beginTime());
      espController_->eventSetupForInstance(ts);
      EventSetup const& es = esp_->eventSetup();

      //now get all the data available in the EventSetup
      std::vector<eventsetup::EventSetupRecordKey> recordKeys;
      es.fillAvailableRecordKeys(recordKeys);
      std::vector<eventsetup::DataKey> dataKeys;
      for(std::vector<eventsetup::EventSetupRecordKey>::const_iterator itKey = recordKeys.begin(), itEnd = recordKeys.end();
          itKey != itEnd;
          ++itKey) {
        eventsetup::EventSetupRecord const* recordPtr = es.find(*itKey);
        //see if this is on our exclusion list
        ExcludedDataMap::const_iterator itExcludeRec = eventSetupDataToExcludeFromPrefetching_.find(itKey->type().name());
        ExcludedData const* excludedData(nullptr);
        if(itExcludeRec != eventSetupDataToExcludeFromPrefetching_.end()) {
          excludedData = &(itExcludeRec->second);
          if(excludedData->size() == 0 || excludedData->begin()->first == "*") {
            //skip all items in this record
            continue;
          }
        }
        if(0 != recordPtr) {
          dataKeys.clear();
          recordPtr->fillRegisteredDataKeys(dataKeys);
          for(std::vector<eventsetup::DataKey>::const_iterator itDataKey = dataKeys.begin(), itDataKeyEnd = dataKeys.end();
              itDataKey != itDataKeyEnd;
              ++itDataKey) {
            //std::cout << "  " << itDataKey->type().name() << " " << itDataKey->name().value() << std::endl;
            if(0 != excludedData && excludedData->find(std::make_pair(itDataKey->type().name(), itDataKey->name().value())) != excludedData->end()) {
              LogInfo("ForkingEventSetupPreFetching") << "   excluding:" << itDataKey->type().name() << " " << itDataKey->name().value() << std::endl;
              continue;
            }
            try {
              recordPtr->doGet(*itDataKey);
            } catch(cms::Exception& e) {
              LogWarning("ForkingEventSetupPreFetching") << e.what();
            }
          }
        }
      }
    }
    LogSystem("ForkingEventSetupPreFetching") <<"  done prefetching";
    {
      // make the services available
      ServiceRegistry::Operate operate(serviceToken_);
      Service<JobReport> jobReport;
      jobReport->parentBeforeFork(jobReportFile, numberOfForkedChildren_);

      //Now actually do the forking
      actReg_->preForkReleaseResourcesSignal_();
      input_->doPreForkReleaseResources();
      schedule_->preForkReleaseResources();
    }
    installCustomHandler(SIGCHLD, ep_sigchld);


    unsigned int childIndex = 0;
    unsigned int const kMaxChildren = numberOfForkedChildren_;
    unsigned int const numberOfDigitsInIndex = numberOfDigitsInChildIndex(kMaxChildren);
    std::vector<pid_t> childrenIds;
    childrenIds.reserve(kMaxChildren);
    std::vector<int> childrenSockets;
    childrenSockets.reserve(kMaxChildren);
    std::vector<int> childrenPipes;
    childrenPipes.reserve(kMaxChildren);
    std::vector<int> childrenSocketsCopy;
    childrenSocketsCopy.reserve(kMaxChildren);
    std::vector<int> childrenPipesCopy;
    childrenPipesCopy.reserve(kMaxChildren);
    int pipes[] {0, 0};

    {
      // make the services available
      ServiceRegistry::Operate operate(serviceToken_);
      Service<JobReport> jobReport;
      int sockets[2], fd_flags;
      for(; childIndex < kMaxChildren; ++childIndex) {
        // Create a UNIX_DGRAM socket pair
        if (socketpair(AF_UNIX, SOCK_DGRAM, 0, sockets)) {
          printf("Error creating communication socket (errno=%d, %s)\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }
        if (pipe(pipes)) {
          printf("Error creating communication pipes (errno=%d, %s)\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }
        // set CLOEXEC so the socket/pipe doesn't get leaked if the child exec's.
        if ((fd_flags = fcntl(sockets[1], F_GETFD, NULL)) == -1) {
          printf("Failed to get fd flags: %d %s\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }
        // Mark socket as non-block.  Child must be careful to do select prior
        // to reading from socket.
        if (fcntl(sockets[1], F_SETFD, fd_flags | FD_CLOEXEC | O_NONBLOCK) == -1) {
          printf("Failed to set new fd flags: %d %s\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }
        if ((fd_flags = fcntl(pipes[1], F_GETFD, NULL)) == -1) {
          printf("Failed to get fd flags: %d %s\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }
        if (fcntl(pipes[1], F_SETFD, fd_flags | FD_CLOEXEC) == -1) {
          printf("Failed to set new fd flags: %d %s\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }
        // Linux man page notes there are some edge cases where reading from a
        // fd can block, even after a select.
        if ((fd_flags = fcntl(pipes[0], F_GETFD, NULL)) == -1) {
          printf("Failed to get fd flags: %d %s\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }
        if (fcntl(pipes[0], F_SETFD, fd_flags | O_NONBLOCK) == -1) {
          printf("Failed to set new fd flags: %d %s\n", errno, strerror(errno));
          exit(EXIT_FAILURE);
        }

        childrenPipesCopy = childrenPipes;
        childrenSocketsCopy = childrenSockets;

        pid_t value = fork();
        if(value == 0) {
          // Close the parent's side of the socket and pipe which will talk to us.
          close(pipes[0]);
          close(sockets[0]);
          // Close our copies of the parent's other communication pipes.
          for(std::vector<int>::const_iterator it=childrenPipesCopy.begin(); it != childrenPipesCopy.end(); it++) {
            close(*it);
          }
          for(std::vector<int>::const_iterator it=childrenSocketsCopy.begin(); it != childrenSocketsCopy.end(); it++) {
            close(*it);
          }

          // this is the child process, redirect stdout and stderr to a log file
          fflush(stdout);
          fflush(stderr);
          std::stringstream stout;
          stout << "redirectout_" << getpgrp() << "_" << std::setw(numberOfDigitsInIndex) << std::setfill('0') << childIndex << ".log";
          if(0 == freopen(stout.str().c_str(), "w", stdout)) {
            LogError("ForkingStdOutRedirect") << "Error during freopen of child process "<< childIndex;
          }
          if(dup2(fileno(stdout), fileno(stderr)) < 0) {
            LogError("ForkingStdOutRedirect") << "Error during dup2 of child process"<< childIndex;
          }

          LogInfo("ForkingChild") << "I am child " << childIndex << " with pgid " << getpgrp();
          if(setCpuAffinity_) {
            // CPU affinity is handled differently on macosx.
            // We disable it and print a message until someone reads:
            //
            // http://developer.apple.com/mac/library/releasenotes/Performance/RN-AffinityAPI/index.html
            //
            // and implements it.
#ifdef __APPLE__
            LogInfo("ForkingChildAffinity") << "Architecture support for CPU affinity not implemented.";
#else
            LogInfo("ForkingChildAffinity") << "Setting CPU affinity, setting this child to cpu " << childIndex;
            cpu_set_t mask;
            CPU_ZERO(&mask);
            CPU_SET(childIndex, &mask);
            if(sched_setaffinity(0, sizeof(mask), &mask) != 0) {
              LogError("ForkingChildAffinity") << "Failed to set the cpu affinity, errno " << errno;
              exit(-1);
            }
#endif
          }
          break;
        } else {
          //this is the parent
          close(pipes[1]);
          close(sockets[1]);
        }
        if(value < 0) {
          LogError("ForkingChild") << "failed to create a child";
          exit(-1);
        }
        childrenIds.push_back(value);
        childrenSockets.push_back(sockets[0]);
        childrenPipes.push_back(pipes[0]);
      }

      if(childIndex < kMaxChildren) {
        jobReport->childAfterFork(jobReportFile, childIndex, kMaxChildren);
        actReg_->postForkReacquireResourcesSignal_(childIndex, kMaxChildren);

        auto receiver = std::make_shared<multicore::MessageReceiverForSource>(sockets[1], pipes[1]);
        input_->doPostForkReacquireResources(receiver);
        schedule_->postForkReacquireResources(childIndex, kMaxChildren);
        //NOTE: sources have to reset themselves by listening to the post fork message
        //rewindInput();
        return true;
      }
      jobReport->parentAfterFork(jobReportFile);
    }

    //this is the original, which is now the master for all the children

    //Need to wait for signals from the children or externally
    // To wait we must
    // 1) block the signals we want to wait on so we do not have a race condition
    // 2) check that we haven't already meet our ending criteria
    // 3) call sigsuspend, which unblocks the signals and waits until a signal is caught
    sigset_t blockingSigSet;
    sigset_t unblockingSigSet;
    sigset_t oldSigSet;
    pthread_sigmask(SIG_SETMASK, NULL, &unblockingSigSet);
    pthread_sigmask(SIG_SETMASK, NULL, &blockingSigSet);
    sigaddset(&blockingSigSet, SIGCHLD);
    sigaddset(&blockingSigSet, SIGUSR2);
    sigaddset(&blockingSigSet, SIGINT);
    sigdelset(&unblockingSigSet, SIGCHLD);
    sigdelset(&unblockingSigSet, SIGUSR2);
    sigdelset(&unblockingSigSet, SIGINT);
    pthread_sigmask(SIG_BLOCK, &blockingSigSet, &oldSigSet);

    // If there are too many fd's (unlikely, but possible) for select, denote this
    // because the sender will fail.
    bool too_many_fds = false;
    if (pipes[1]+1 > FD_SETSIZE) {
      LogError("ForkingFileDescriptors") << "too many file descriptors for multicore job";
      too_many_fds = true;
    }

    //create a thread that sends the units of work to workers
    // we create it after all signals were blocked so that this
    // thread is never interupted by a signal
    MessageSenderToSource sender(childrenSockets, childrenPipes, numberOfSequentialEventsPerChild_);
    boost::thread senderThread(sender);

    if(not too_many_fds) {
      //NOTE: a child could have failed before we got here and even after this call
      // which is why the 'if' is conditional on continueAfterChildFailure_
      possiblyContinueAfterForkChildFailure();
      while(!shutdown_flag && (!child_failed or continueAfterChildFailure_) && (childrenIds.size() != num_children_done)) {
        sigsuspend(&unblockingSigSet);
        possiblyContinueAfterForkChildFailure();
        LogInfo("ForkingAwake") << "woke from sigwait" << std::endl;
      }
    }
    pthread_sigmask(SIG_SETMASK, &oldSigSet, NULL);

    LogInfo("ForkingStopping") << "num children who have already stopped " << num_children_done;
    if(child_failed) {
      LogError("ForkingStopping") << "child failed";
    }
    if(shutdown_flag) {
      LogSystem("ForkingStopping") << "asked to shutdown";
    }

    if(too_many_fds || shutdown_flag || (child_failed && (num_children_done != childrenIds.size()))) {
      LogInfo("ForkingStopping") << "must stop children" << std::endl;
      for(std::vector<pid_t>::iterator it = childrenIds.begin(), itEnd = childrenIds.end();
          it != itEnd; ++it) {
        /* int result = */ kill(*it, SIGUSR2);
      }
      pthread_sigmask(SIG_BLOCK, &blockingSigSet, &oldSigSet);
      while(num_children_done != kMaxChildren) {
        sigsuspend(&unblockingSigSet);
      }
      pthread_sigmask(SIG_SETMASK, &oldSigSet, NULL);
    }
    // The senderThread will notice the pipes die off, one by one.  Once all children are gone, it will exit.
    senderThread.join();
    if(child_failed && !continueAfterChildFailure_) {
      if (child_fail_signal) {
        throw cms::Exception("ForkedChildFailed") << "child process ended abnormally with signal " << child_fail_signal;
      } else if (child_fail_exit_status) {
        throw cms::Exception("ForkedChildFailed") << "child process ended abnormally with exit code " << child_fail_exit_status;
      } else {
        throw cms::Exception("ForkedChildFailed") << "child process ended abnormally for unknown reason";
      }
    }
    if(too_many_fds) {
      throw cms::Exception("ForkedParentFailed") << "hit select limit for number of fds";
    }
    return false;
  }

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

References schedule_.

                                                 {
    return schedule_->getAllModuleDescriptions();
  }

ServiceToken edm::EventProcessor::getToken

(

)

Definition at line 694 of file EventProcessor.cc.

References AlCaHLTBitMon_ParallelJobs::p, and 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 1252 of file EventProcessor.cc.

References schedule_.

                                                           {
    schedule_->getTriggerReport(rep);
  }

void edm::EventProcessor::handleNextEventForStreamAsync ( WaitingTask *  iTask, unsigned int  iStreamIndex, std::atomic< bool > *  finishedProcessingEvents  )

private

Definition at line 2006 of file EventProcessor.cc.

References deferredExceptionPtr_, deferredExceptionPtrIsSet_, pyrootRender::destroy(), edm::WaitingTaskHolder::doneWaiting(), h, edm::make_waiting_task(), cmsPerfStripChart::operate(), processEventAsync(), edm::SerialTaskQueueChain::push(), readNextEventForStream(), edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, and sourceResourcesAcquirer_.

Referenced by readAndProcessEvent().

  {
    auto recursionTask = make_waiting_task(tbb::task::allocate_root(), [this,iTask,iStreamIndex,finishedProcessingEvents](std::exception_ptr const* iPtr) {
      if(iPtr) {
        bool expected = false;
        if(deferredExceptionPtrIsSet_.compare_exchange_strong(expected,true)) {
          deferredExceptionPtr_ = *iPtr;
          {
            WaitingTaskHolder h(iTask);
            h.doneWaiting(*iPtr);
          }
        }
        //the stream will stop now
        iTask->decrement_ref_count();
        return;
      }

      handleNextEventForStreamAsync(iTask, iStreamIndex,finishedProcessingEvents);
    });
      
    sourceResourcesAcquirer_.serialQueueChain().push([this,finishedProcessingEvents,recursionTask,iTask,iStreamIndex]() {
           ServiceRegistry::Operate operate(serviceToken_);

           try {
             if(readNextEventForStream(iStreamIndex, finishedProcessingEvents) ) {
               processEventAsync( WaitingTaskHolder(recursionTask), iStreamIndex);
             } else {
               //the stream will stop now
               tbb::task::destroy(*recursionTask);
               iTask->decrement_ref_count();
             }
           } catch(...) {
             WaitingTaskHolder h(recursionTask);
             h.doneWaiting(std::current_exception());
           }
    });
  }

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

private

Definition at line 412 of file EventProcessor.cc.

References edm::ScheduleItems::act_table_, act_table_, edm::ScheduleItems::actReg_, actReg_, edm::ScheduleItems::addCPRandTNS(), edm::ScheduleItems::branchIDListHelper(), branchIDListHelper(), branchIDListHelper_, trackingPlots::common, continueAfterChildFailure_, emptyRunLumiMode_, esp_, espController_, eventSetupDataToExcludeFromPrefetching_, FDEBUG, fileMode_, edm::fillLooper(), forceESCacheClearOnNewRun_, edm::ParameterSet::getUntrackedParameter(), edm::ParameterSet::getUntrackedParameterSet(), edm::ParameterSet::getUntrackedParameterSetVector(), historyAppender_, diffTreeTool::index, edm::ScheduleItems::initMisc(), edm::ScheduleItems::initSchedule(), edm::ScheduleItems::initServices(), input_, edm::PrincipalCache::insert(), edm::ParentageRegistry::insertMapped(), edm::ParentageRegistry::instance(), mps_monitormerge::items, edm::serviceregistry::kConfigurationOverrides, looper_, edm::makeInput(), eostools::move(), numberOfForkedChildren_, numberOfSequentialEventsPerChild_, edm::PreallocationConfiguration::numberOfStreams(), cmsPerfStripChart::operate(), edm::parameterSet(), edm::popSubProcessVParameterSet(), preallocations_, edm::ScheduleItems::preg(), preg(), preg_, principalCache_, printDependencies_, edm::ScheduleItems::processConfiguration(), processConfiguration_, processContext_, edm::ParameterSet::registerIt(), schedule_, serviceToken_, setCpuAffinity_, edm::PrincipalCache::setNumberOfConcurrentPrincipals(), edm::ProcessContext::setProcessConfiguration(), edm::PrincipalCache::setProcessHistoryRegistry(), edm::IllegalParameters::setThrowAnException(), AlCaHLTBitMon_QueryRunRegistry::string, subProcesses_, edm::ScheduleItems::thinnedAssociationsHelper(), thinnedAssociationsHelper(), and thinnedAssociationsHelper_.

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

    // Now set some parameters specific to the main process.
    ParameterSet const& optionsPset(parameterSet->getUntrackedParameterSet("options", ParameterSet()));
    fileMode_ = optionsPset.getUntrackedParameter<std::string>("fileMode", "");
    emptyRunLumiMode_ = optionsPset.getUntrackedParameter<std::string>("emptyRunLumiMode", "");
    forceESCacheClearOnNewRun_ = optionsPset.getUntrackedParameter<bool>("forceEventSetupCacheClearOnNewRun", false);
    //threading
    unsigned int nThreads=1;
    if(optionsPset.existsAs<unsigned int>("numberOfThreads",false)) {
      nThreads = optionsPset.getUntrackedParameter<unsigned int>("numberOfThreads");
      if(nThreads == 0) {
        nThreads = 1;
      }
    }
    /* TODO: when we support having each stream run in a different thread use this default
       unsigned int nStreams =nThreads;
    */
    unsigned int nStreams =1;
    if(optionsPset.existsAs<unsigned int>("numberOfStreams",false)) {
      nStreams = optionsPset.getUntrackedParameter<unsigned int>("numberOfStreams");
      if(nStreams==0) {
        nStreams = nThreads;
      }
    }
    if(nThreads >1) {
      edm::LogInfo("ThreadStreamSetup") <<"setting # threads "<<nThreads<<"\nsetting # streams "<<nStreams;
    }

    /*
      bool nRunsSet = false;
    */
    unsigned int nConcurrentRuns =1;
    /*
      if(nRunsSet = optionsPset.existsAs<unsigned int>("numberOfConcurrentRuns",false)) {
      nConcurrentRuns = optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentRuns");
      }
    */
    unsigned int nConcurrentLumis =1;
    /*
      if(optionsPset.existsAs<unsigned int>("numberOfConcurrentLuminosityBlocks",false)) {
      nConcurrentLumis = optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentLuminosityBlocks");
      } else {
      nConcurrentLumis = nConcurrentRuns;
      }
    */
    //Check that relationships between threading parameters makes sense
    /*
      if(nThreads<nStreams) {
      //bad
      }
      if(nConcurrentRuns>nStreams) {
      //bad
      }
      if(nConcurrentRuns>nConcurrentLumis) {
      //bad
      }
    */
    //forking
    ParameterSet const& forking = optionsPset.getUntrackedParameterSet("multiProcesses", ParameterSet());
    numberOfForkedChildren_ = forking.getUntrackedParameter<int>("maxChildProcesses", 0);
    numberOfSequentialEventsPerChild_ = forking.getUntrackedParameter<unsigned int>("maxSequentialEventsPerChild", 1);
    setCpuAffinity_ = forking.getUntrackedParameter<bool>("setCpuAffinity", false);
    continueAfterChildFailure_ = forking.getUntrackedParameter<bool>("continueAfterChildFailure",false);
    std::vector<ParameterSet> const& excluded = forking.getUntrackedParameterSetVector("eventSetupDataToExcludeFromPrefetching", std::vector<ParameterSet>());
    for(auto const& ps : excluded) {
      eventSetupDataToExcludeFromPrefetching_[ps.getUntrackedParameter<std::string>("record")].emplace(ps.getUntrackedParameter<std::string>("type", "*"),
                                                                                                       ps.getUntrackedParameter<std::string>("label", ""));
    }
    IllegalParameters::setThrowAnException(optionsPset.getUntrackedParameter<bool>("throwIfIllegalParameter", true));

    printDependencies_ =  optionsPset.getUntrackedParameter("printDependencies", false);

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

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

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

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

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

inlineprivate

Definition at line 266 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 267 of file EventProcessor.h.

References edm::get_underlying_safe().

{return get_underlying_safe(looper_);}

void edm::EventProcessor::openOutputFiles ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1523 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

void edm::EventProcessor::possiblyContinueAfterForkChildFailure ( )

private

Definition at line 905 of file EventProcessor.cc.

References continueAfterChildFailure_.

Referenced by forkProcess().

                                                             {
    if(child_failed && continueAfterChildFailure_) {
      if (child_fail_signal) {
        LogSystem("ForkedChildFailed") << "child process ended abnormally with signal " << child_fail_signal;
        child_fail_signal=0;
      } else if (child_fail_exit_status) {
        LogSystem("ForkedChildFailed") << "child process ended abnormally with exit code " << child_fail_exit_status;
        child_fail_exit_status=0;
      } else {
        LogSystem("ForkedChildFailed") << "child process ended abnormally for unknown reason";
      }
      child_failed =false;
    }
  }

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

inlineprivate

Definition at line 260 of file EventProcessor.h.

References edm::get_underlying_safe().

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

{return get_underlying_safe(preg_);}

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

inlineprivate

Definition at line 261 of file EventProcessor.h.

References edm::get_underlying_safe().

{return get_underlying_safe(preg_);}

void edm::EventProcessor::prepareForNextLoop ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1585 of file EventProcessor.cc.

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

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

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

inline

Definition at line 124 of file EventProcessor.h.

References Ecal2004TBTDCRanges_v1_cff::endRun, EcalCondTools::getToken(), init, python.rootplot.argparse::message, cuy::rep, and AlCaHLTBitMon_QueryRunRegistry::string.

{ return *processConfiguration_; }

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

private

Definition at line 2103 of file EventProcessor.cc.

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

Referenced by handleNextEventForStreamAsync(), and readAndProcessEvent().

                                                                    {
    auto pep = &(principalCache_.eventPrincipal(iStreamIndex));
    pep->setLuminosityBlockPrincipal(principalCache_.lumiPrincipalPtr());
    Service<RandomNumberGenerator> rng;
    if(rng.isAvailable()) {
      Event ev(*pep, ModuleDescription(), nullptr);
      rng->postEventRead(ev);
    }
    assert(pep->luminosityBlockPrincipalPtrValid());
    assert(principalCache_.lumiPrincipalPtr()->run() == pep->run());
    assert(principalCache_.lumiPrincipalPtr()->luminosityBlock() == pep->luminosityBlock());
    
    WaitingTaskHolder finalizeEventTask( make_waiting_task(
                    tbb::task::allocate_root(),
                    [this,pep,iHolder](std::exception_ptr const* iPtr) mutable
             {
               ServiceRegistry::Operate operate(serviceToken_);

               //NOTE: If we have a looper we only have one Stream
               if(looper_) {
                 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) {
           ServiceRegistry::Operate operate(serviceToken_);

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

  }

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

private

Definition at line 2168 of file EventProcessor.cc.

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

Referenced by processEventAsync().

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

int edm::EventProcessor::readAndMergeLumi ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1921 of file EventProcessor.cc.

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

                                       {
    principalCache_.merge(input_->luminosityBlockAuxiliary(), preg());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readAndMergeLumi(*principalCache_.lumiPrincipalPtr());
      sentry.completedSuccessfully();
    }
    return input_->luminosityBlock();
  }

statemachine::Run edm::EventProcessor::readAndMergeRun ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1884 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 statemachine::Run(runPrincipal->reducedProcessHistoryID(), input_->run());
  }

void edm::EventProcessor::readAndProcessEvent ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 2046 of file EventProcessor.cc.

References asyncStopRequestedWhileProcessingEvents_, deferredExceptionPtr_, deferredExceptionPtrIsSet_, firstEventInBlock_, handleNextEventForStreamAsync(), edm::InputSource::IsEvent, edm::make_empty_waiting_task(), edm::make_waiting_task(), nextItemTypeFromProcessingEvents_, numberOfForkedChildren_, edm::PreallocationConfiguration::numberOfStreams(), preallocations_, processEventAsync(), and readEvent().

                                           {
    if(numberOfForkedChildren_>0) {
      //Have to do something special for forking since
      // after each event the system may have to skip
      // some transitions. This is handled in runToCompletion
      readEvent(0);
      auto eventLoopWaitTask = make_empty_waiting_task();
      eventLoopWaitTask->increment_ref_count();
      processEventAsync(WaitingTaskHolder(eventLoopWaitTask.get()),0);
      eventLoopWaitTask->wait_for_all();
      return;
    }
    nextItemTypeFromProcessingEvents_ = InputSource::IsEvent; //needed for looper
    asyncStopRequestedWhileProcessingEvents_ = false;

    std::atomic<bool> finishedProcessingEvents{false};
    auto finishedProcessingEventsPtr = &finishedProcessingEvents;

    //The state machine already found the event so
    // we have to avoid looking again
    firstEventInBlock_ = true;

    //To wait, the ref count has to b 1+#streams
    auto eventLoopWaitTask = make_empty_waiting_task();
    auto eventLoopWaitTaskPtr = eventLoopWaitTask.get();
    eventLoopWaitTask->increment_ref_count();

    const unsigned int kNumStreams = preallocations_.numberOfStreams();
    unsigned int iStreamIndex = 0;
    for(; iStreamIndex<kNumStreams-1; ++iStreamIndex) {
      eventLoopWaitTask->increment_ref_count();
      tbb::task::enqueue( *make_waiting_task(tbb::task::allocate_root(),[this,iStreamIndex,finishedProcessingEventsPtr,eventLoopWaitTaskPtr](std::exception_ptr const*){
        handleNextEventForStreamAsync(eventLoopWaitTaskPtr,iStreamIndex,finishedProcessingEventsPtr);
      }) );
    }
    eventLoopWaitTask->increment_ref_count();
    eventLoopWaitTask->spawn_and_wait_for_all( *make_waiting_task(tbb::task::allocate_root(),[this,iStreamIndex,finishedProcessingEventsPtr,eventLoopWaitTaskPtr](std::exception_ptr const*){
      handleNextEventForStreamAsync(eventLoopWaitTaskPtr,iStreamIndex,finishedProcessingEventsPtr);
    }));

    //One of the processing threads saw an exception
    if(deferredExceptionPtrIsSet_) {
      std::rethrow_exception(deferredExceptionPtr_);
    }
  }

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

private

Definition at line 2091 of file EventProcessor.cc.

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

Referenced by readAndProcessEvent(), and 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);
    sentry.completedSuccessfully();

    FDEBUG(1) << "\treadEvent\n";
  }

void edm::EventProcessor::readFile ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1496 of file EventProcessor.cc.

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

Referenced by forkProcess(), 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((numberOfForkedChildren_ > 0) or
       (preallocations_.numberOfStreams()>1 and
        preallocations_.numberOfThreads()>1)) {
      fb_->setNotFastClonable(FileBlock::ParallelProcesses);
    }
    sentry.completedSuccessfully();
  }

int edm::EventProcessor::readLuminosityBlock ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1896 of file EventProcessor.cc.

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

                                          {
    if (principalCache_.hasLumiPrincipal()) {
      throw edm::Exception(edm::errors::LogicError)
        << "EventProcessor::readRun\n"
        << "Illegal attempt to insert lumi into cache\n"
        << "Contact a Framework Developer\n";
    }
    if (!principalCache_.hasRunPrincipal()) {
      throw edm::Exception(edm::errors::LogicError)
        << "EventProcessor::readRun\n"
        << "Illegal attempt to insert lumi into cache\n"
        << "Run is invalid\n"
        << "Contact a Framework Developer\n";
    }
    auto lbp = std::make_shared<LuminosityBlockPrincipal>(input_->luminosityBlockAuxiliary(), preg(), *processConfiguration_, historyAppender_.get(), 0);
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readLuminosityBlock(*lbp, *historyAppender_);
      sentry.completedSuccessfully();
    }
    lbp->setRunPrincipal(principalCache_.runPrincipalPtr());
    principalCache_.insert(lbp);
    return input_->luminosityBlock();
  }

bool edm::EventProcessor::readNextEventForStream ( unsigned int  iStreamIndex, std::atomic< bool > *  finishedProcessingEvents  )

private

Definition at line 1955 of file EventProcessor.cc.

References actReg_, asyncStopRequestedWhileProcessingEvents_, asyncStopStatusCodeFromProcessingEvents_, checkForAsyncStopRequest(), deferredExceptionPtr_, deferredExceptionPtrIsSet_, edm::ExternalSignal, firstEventInBlock_, input_, edm::InputSource::IsEvent, nextItemTypeFromProcessingEvents_, cmsPerfStripChart::operate(), readEvent(), serviceToken_, shouldWeStop(), and sourceMutex_.

Referenced by handleNextEventForStreamAsync().

                                                                                         {
    if(shouldWeStop()) {
      return false;
    }
    
    if(deferredExceptionPtrIsSet_.load(std::memory_order_acquire)) {
      return false;
    }
    
    if(finishedProcessingEvents->load(std::memory_order_acquire)) {
      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()));
      if(not firstEventInBlock_) {
        //The state machine already called input_->nextItemType
        // and found an event. We can't call input_->nextItemType
        // again since it would move to the next transition
        InputSource::ItemType itemType = input_->nextItemType();
        if (InputSource::IsEvent !=itemType) {
          nextItemTypeFromProcessingEvents_ = itemType;
          finishedProcessingEvents->store(true,std::memory_order_release);
          //std::cerr<<"next item type "<<itemType<<"\n";
          return false;
        }
        if((asyncStopRequestedWhileProcessingEvents_=checkForAsyncStopRequest(asyncStopStatusCodeFromProcessingEvents_))) {
          //std::cerr<<"task told to async stop\n";
          actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
          return false;
        }
      } else {
        firstEventInBlock_ = false;
      }
      readEvent(iStreamIndex);
    } catch (...) {
      bool expected =false;
      if(deferredExceptionPtrIsSet_.compare_exchange_strong(expected,true)) {
        deferredExceptionPtr_ = std::current_exception();

      }
      return false;
    }
    return true;
  }

statemachine::Run edm::EventProcessor::readRun ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1866 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 statemachine::Run(rp->reducedProcessHistoryID(), input_->run());
  }

void edm::EventProcessor::respondToCloseInputFile ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1548 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 ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1539 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 ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1579 of file EventProcessor.cc.

References FDEBUG, and input_.

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

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

inline

Definition at line 337 of file EventProcessor.h.

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

                      {
    return runToCompletion();
  }

EventProcessor::StatusCode edm::EventProcessor::runToCompletion ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1308 of file EventProcessor.cc.

References actReg_, cms::Exception::addAdditionalInfo(), edm::PrincipalCache::adjustEventsToNewProductRegistry(), edm::PrincipalCache::adjustIndexesAfterProductRegistryAddition(), alreadyHandlingException_, cms::Exception::alreadyPrinted(), asyncStopRequestedWhileProcessingEvents_, asyncStopStatusCodeFromProcessingEvents_, beginJob(), checkForAsyncStopRequest(), createStateMachine(), MillePedeFileConverter_cfg::e, edm::IEventProcessor::epSuccess, Exception, exceptionMessageFiles_, exceptionMessageLumis_, exceptionMessageRuns_, edm::ExternalSignal, FDEBUG, forceLooperToEnd_, input_, edm::InputSource::IsEvent, edm::InputSource::IsFile, edm::InputSource::IsLumi, edm::InputSource::IsRun, edm::InputSource::IsStop, edm::InputSource::IsSynchronize, edm::errors::LogicError, eostools::move(), nextItemTypeFromProcessingEvents_, numberOfForkedChildren_, cmsPerfStripChart::operate(), preg(), preg_, principalCache_, runEdmFileComparison::returnCode, serviceToken_, findQualityFiles::size, stateMachineWasInErrorState_, terminateMachine(), and edm::convertException::wrap().

Referenced by PythonEventProcessor::run().

                                  {

    StatusCode returnCode=epSuccess;
    asyncStopStatusCodeFromProcessingEvents_=epSuccess;
    std::unique_ptr<statemachine::Machine> machine;
    {
      beginJob(); //make sure this was called

      //StatusCode returnCode = epSuccess;
      stateMachineWasInErrorState_ = false;

      // make the services available
      ServiceRegistry::Operate operate(serviceToken_);

      machine = createStateMachine();
      nextItemTypeFromProcessingEvents_=InputSource::IsEvent;
      asyncStopRequestedWhileProcessingEvents_=false;
      try {
        convertException::wrap([&]() {

            InputSource::ItemType itemType;

            while(true) {

              bool more = true;
              if(numberOfForkedChildren_ > 0) {
                size_t size = preg_->size();
                {
                  SendSourceTerminationSignalIfException sentry(actReg_.get());
                  more = input_->skipForForking();
                  sentry.completedSuccessfully();
                }
                if(more) {
                  if(size < preg_->size()) {
                    principalCache_.adjustIndexesAfterProductRegistryAddition();
                  }
                  principalCache_.adjustEventsToNewProductRegistry(preg());
                }
              }
              {
                SendSourceTerminationSignalIfException sentry(actReg_.get());
                itemType = (more ? input_->nextItemType() : InputSource::IsStop);
                sentry.completedSuccessfully();
              }

              FDEBUG(1) << "itemType = " << itemType << "\n";

              if(checkForAsyncStopRequest(returnCode)) {
                actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
                forceLooperToEnd_ = true;
                machine->process_event(statemachine::Stop());
                forceLooperToEnd_ = false;
                break;
              }

              if(itemType == InputSource::IsEvent) {
                machine->process_event(statemachine::Event());
                if(asyncStopRequestedWhileProcessingEvents_) {
                  forceLooperToEnd_ = true;
                  machine->process_event(statemachine::Stop());
                  forceLooperToEnd_ = false;
                  returnCode = asyncStopStatusCodeFromProcessingEvents_;
                  break;
                }
                itemType = nextItemTypeFromProcessingEvents_;
              }

              if(itemType == InputSource::IsEvent) {
              }
              else if(itemType == InputSource::IsStop) {
                machine->process_event(statemachine::Stop());
              }
              else if(itemType == InputSource::IsFile) {
                machine->process_event(statemachine::File());
              }
              else if(itemType == InputSource::IsRun) {
                machine->process_event(statemachine::Run(input_->reducedProcessHistoryID(), input_->run()));
              }
              else if(itemType == InputSource::IsLumi) {
                machine->process_event(statemachine::Lumi(input_->luminosityBlock()));
              }
              else if(itemType == InputSource::IsSynchronize) {
                //For now, we don't have to do anything
              }
              // This should be impossible
              else {
                throw Exception(errors::LogicError)
                  << "Unknown next item type passed to EventProcessor\n"
                  << "Please report this error to the Framework group\n";
              }
              if(machine->terminated()) {
                break;
              }
            }  // End of loop over state machine events
          }); // convertException::wrap
      } // Try block
      // Some comments on exception handling related to the boost state machine:
      //
      // Some states used in the machine are special because they
      // perform actions while the machine is being terminated, actions
      // such as close files, call endRun, call endLumi etc ...  Each of these
      // states has two functions that perform these actions.  The functions
      // are almost identical.  The major difference is that one version
      // catches all exceptions and the other lets exceptions pass through.
      // The destructor catches them and the other function named "exit" lets
      // them pass through.  On a normal termination, boost will always call
      // "exit" and then the state destructor.  In our state classes, the
      // the destructors do nothing if the exit function already took
      // care of things.  Here's the interesting part.  When boost is
      // handling an exception the "exit" function is not called (a boost
      // feature).
      //
      // If an exception occurs while the boost machine is in control
      // (which usually means inside a process_event call), then
      // the boost state machine destroys its states and "terminates" itself.
      // This already done before we hit the catch blocks below. In this case
      // the call to terminateMachine below only destroys an already
      // terminated state machine.  Because exit is not called, the state destructors
      // handle cleaning up lumis, runs, and files.  The destructors swallow
      // all exceptions and only pass through the exceptions messages, which
      // are tacked onto the original exception below.
      //
      // If an exception occurs when the boost state machine is not
      // in control (outside the process_event functions), then boost
      // cannot destroy its own states.  The terminateMachine function
      // below takes care of that.  The flag "alreadyHandlingException"
      // is set true so that the state exit functions do nothing (and
      // cannot throw more exceptions while handling the first).  Then the
      // state destructors take care of this because exit did nothing.
      //
      // In both cases above, the EventProcessor::endOfLoop function is
      // not called because it can throw exceptions.
      //
      // One tricky aspect of the state machine is that things that can
      // throw should not be invoked by the state machine while another
      // exception is being handled.
      // Another tricky aspect is that it appears to be important to
      // terminate the state machine before invoking its destructor.
      // We've seen crashes that are not understood when that is not
      // done.  Maintainers of this code should be careful about this.

      catch (cms::Exception & e) {
        alreadyHandlingException_ = true;
        terminateMachine(std::move(machine));
        alreadyHandlingException_ = false;
        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;
      }

      if(machine->terminated()) {
        FDEBUG(1) << "The state machine reports it has been terminated\n";
        machine.reset();
      }

      if(stateMachineWasInErrorState_) {
        throw cms::Exception("BadState")
          << "The boost state machine in the EventProcessor exited after\n"
          << "entering the Error state.\n";
      }

    }
    if(machine.get() != nullptr) {
      terminateMachine(std::move(machine));
      throw Exception(errors::LogicError)
        << "State machine not destroyed on exit from EventProcessor::runToCompletion\n"
        << "Please report this error to the Framework group\n";
    }

    return returnCode;
  }

void edm::EventProcessor::setExceptionMessageFiles ( std::string &  message )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 2208 of file EventProcessor.cc.

References exceptionMessageFiles_, and python.rootplot.argparse::message.

                                                                  {
    exceptionMessageFiles_ = message;
  }

void edm::EventProcessor::setExceptionMessageLumis ( std::string &  message )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 2216 of file EventProcessor.cc.

References exceptionMessageLumis_, and python.rootplot.argparse::message.

                                                                  {
    exceptionMessageLumis_ = message;
  }

void edm::EventProcessor::setExceptionMessageRuns ( std::string &  message )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 2212 of file EventProcessor.cc.

References exceptionMessageRuns_, and python.rootplot.argparse::message.

                                                                 {
    exceptionMessageRuns_ = message;
  }

void edm::EventProcessor::setupSignal ( )

private

bool edm::EventProcessor::shouldWeCloseOutput ( ) const

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1590 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

overridevirtual

Implements edm::IEventProcessor.

Definition at line 2194 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 ( )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1556 of file EventProcessor.cc.

References FDEBUG, looper_, looperBeginJobRun_, and shouldWeStop_.

                                       {
    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::terminateMachine ( std::unique_ptr< statemachine::Machine >  iMachine )

private

Definition at line 2224 of file EventProcessor.cc.

References FDEBUG, and forceLooperToEnd_.

Referenced by runToCompletion().

                                                                                   {
    if(iMachine.get() != nullptr) {
      if(!iMachine->terminated()) {
        forceLooperToEnd_ = true;
        iMachine->process_event(statemachine::Stop());
        forceLooperToEnd_ = false;
      }
      else {
        FDEBUG(1) << "EventProcess::terminateMachine  The state machine was already terminated \n";
      }
      if(iMachine->terminated()) {
        FDEBUG(1) << "The state machine reports it has been terminated (3)\n";
      }
    }
  }

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

inlineprivate

Definition at line 264 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 265 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 1227 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 1237 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 1232 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEventsPassed().

                                          {
    return schedule_->totalEventsPassed();
  }

void edm::EventProcessor::writeLumi ( ProcessHistoryID const &  phid, RunNumber_t  run, LuminosityBlockNumber_t  lumi  )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1943 of file EventProcessor.cc.

References FDEBUG, edm::for_all(), edm::PrincipalCache::lumiPrincipal(), principalCache_, processContext_, schedule_, and subProcesses_.

                                                                                                            {
    schedule_->writeLumi(principalCache_.lumiPrincipal(phid, run, lumi), &processContext_);
    for_all(subProcesses_, [&phid, run, lumi](auto& subProcess){ subProcess.writeLumi(phid, run, lumi); });
    FDEBUG(1) << "\twriteLumi " << run << "/" << lumi << "\n";
  }

void edm::EventProcessor::writeRun ( statemachine::Run const &  run )

overridevirtual

Implements edm::IEventProcessor.

Definition at line 1931 of file EventProcessor.cc.

References FDEBUG, edm::for_all(), principalCache_, processContext_, statemachine::Run::processHistoryID(), statemachine::Run::runNumber(), edm::PrincipalCache::runPrincipal(), schedule_, and subProcesses_.

                                                          {
    schedule_->writeRun(principalCache_.runPrincipal(run.processHistoryID(), run.runNumber()), &processContext_);
    for_all(subProcesses_, [&run](auto& subProcess){ subProcess.writeRun(run.processHistoryID(), run.runNumber()); });
    FDEBUG(1) << "\twriteRun " << run.runNumber() << "\n";
  }

Member Data Documentation

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

private

Definition at line 283 of file EventProcessor.h.

Referenced by init().

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

private

Definition at line 275 of file EventProcessor.h.

Referenced by beginJob(), beginLumi(), beginRun(), closeInputFile(), endJob(), endLumi(), endRun(), forkProcess(), init(), readAndMergeLumi(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readNextEventForStream(), readRun(), runToCompletion(), and ~EventProcessor().

bool edm::EventProcessor::alreadyHandlingException_

private

Definition at line 309 of file EventProcessor.h.

Referenced by alreadyHandlingException(), and runToCompletion().

bool edm::EventProcessor::asyncStopRequestedWhileProcessingEvents_

private

Definition at line 321 of file EventProcessor.h.

Referenced by readAndProcessEvent(), readNextEventForStream(), and runToCompletion().

StatusCode edm::EventProcessor::asyncStopStatusCodeFromProcessingEvents_

private

Definition at line 323 of file EventProcessor.h.

Referenced by readNextEventForStream(), and runToCompletion().

bool edm::EventProcessor::beginJobCalled_

private

Definition at line 301 of file EventProcessor.h.

Referenced by beginJob().

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

private

Definition at line 277 of file EventProcessor.h.

Referenced by init(), and respondToOpenInputFile().

bool edm::EventProcessor::continueAfterChildFailure_

private

Definition at line 317 of file EventProcessor.h.

Referenced by forkProcess(), init(), and possiblyContinueAfterForkChildFailure().

std::exception_ptr edm::EventProcessor::deferredExceptionPtr_

private

Definition at line 296 of file EventProcessor.h.

Referenced by handleNextEventForStreamAsync(), readAndProcessEvent(), and readNextEventForStream().

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

private

Definition at line 295 of file EventProcessor.h.

Referenced by handleNextEventForStreamAsync(), readAndProcessEvent(), and readNextEventForStream().

std::string edm::EventProcessor::emptyRunLumiMode_

private

Definition at line 305 of file EventProcessor.h.

Referenced by createStateMachine(), and init().

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

private

Definition at line 282 of file EventProcessor.h.

Referenced by beginLumi(), beginRun(), endLumi(), endOfLoop(), endRun(), forkProcess(), init(), prepareForNextLoop(), processEventAsync(), processEventWithLooper(), and ~EventProcessor().

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

private

Definition at line 281 of file EventProcessor.h.

Referenced by beginLumi(), beginRun(), endLumi(), endRun(), forkProcess(), init(), and ~EventProcessor().

ExcludedDataMap edm::EventProcessor::eventSetupDataToExcludeFromPrefetching_

private

Definition at line 328 of file EventProcessor.h.

Referenced by forkProcess(), and init().

std::string edm::EventProcessor::exceptionMessageFiles_

private

Definition at line 306 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageFiles().

std::string edm::EventProcessor::exceptionMessageLumis_

private

Definition at line 308 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageLumis().

std::string edm::EventProcessor::exceptionMessageRuns_

private

Definition at line 307 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageRuns().

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

private

Definition at line 291 of file EventProcessor.h.

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

std::string edm::EventProcessor::fileMode_

private

Definition at line 304 of file EventProcessor.h.

Referenced by createStateMachine(), and init().

bool edm::EventProcessor::firstEventInBlock_ =true

private

Definition at line 324 of file EventProcessor.h.

Referenced by readAndProcessEvent(), and readNextEventForStream().

bool edm::EventProcessor::forceESCacheClearOnNewRun_

private

Definition at line 312 of file EventProcessor.h.

Referenced by beginRun(), and init().

bool edm::EventProcessor::forceLooperToEnd_

private

Definition at line 310 of file EventProcessor.h.

Referenced by endOfLoop(), runToCompletion(), and terminateMachine().

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

private

Definition at line 289 of file EventProcessor.h.

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

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

private

Definition at line 280 of file EventProcessor.h.

Referenced by beginJob(), beginLumi(), beginRun(), closeInputFile(), endJob(), endLumi(), endRun(), forkProcess(), init(), processEventWithLooper(), readAndMergeLumi(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readNextEventForStream(), readRun(), rewindInput(), runToCompletion(), and ~EventProcessor().

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

private

Definition at line 292 of file EventProcessor.h.

Referenced by beginLumi(), beginRun(), Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), endLumi(), endOfLoop(), endRun(), init(), prepareForNextLoop(), processEventAsync(), processEventWithLooper(), startingNewLoop(), and ~EventProcessor().

bool edm::EventProcessor::looperBeginJobRun_

private

Definition at line 311 of file EventProcessor.h.

Referenced by beginRun(), and startingNewLoop().

InputSource::ItemType edm::EventProcessor::nextItemTypeFromProcessingEvents_

private

Definition at line 322 of file EventProcessor.h.

Referenced by readAndProcessEvent(), readNextEventForStream(), and runToCompletion().

int edm::EventProcessor::numberOfForkedChildren_

private

Definition at line 314 of file EventProcessor.h.

Referenced by forkProcess(), init(), readAndProcessEvent(), readFile(), and runToCompletion().

unsigned int edm::EventProcessor::numberOfSequentialEventsPerChild_

private

Definition at line 315 of file EventProcessor.h.

Referenced by forkProcess(), and init().

PathsAndConsumesOfModules edm::EventProcessor::pathsAndConsumesOfModules_

private

Definition at line 286 of file EventProcessor.h.

Referenced by beginJob().

PreallocationConfiguration edm::EventProcessor::preallocations_

private

Definition at line 319 of file EventProcessor.h.

Referenced by beginJob(), beginLumi(), beginRun(), endJob(), endLumi(), endRun(), init(), readAndProcessEvent(), and readFile().

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

private

Definition at line 276 of file EventProcessor.h.

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

PrincipalCache edm::EventProcessor::principalCache_

private

Definition at line 300 of file EventProcessor.h.

Referenced by beginLumi(), beginRun(), deleteLumiFromCache(), deleteRunFromCache(), endLumi(), endRun(), init(), processEventAsync(), readAndMergeLumi(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readRun(), runToCompletion(), writeLumi(), and writeRun().

bool edm::EventProcessor::printDependencies_ = false

private

Definition at line 330 of file EventProcessor.h.

Referenced by beginJob(), and init().

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

private

Definition at line 284 of file EventProcessor.h.

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

ProcessContext edm::EventProcessor::processContext_

private

Definition at line 285 of file EventProcessor.h.

Referenced by beginJob(), beginLumi(), beginRun(), endLumi(), endRun(), init(), processEventWithLooper(), readEvent(), writeLumi(), and writeRun().

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

private

Definition at line 287 of file EventProcessor.h.

Referenced by Config.Process::_insertPaths(), Config.Process::_replaceInSchedule(), beginJob(), beginLumi(), beginRun(), clearCounters(), closeOutputFiles(), Config.Process::dumpPython(), enableEndPaths(), endJob(), endLumi(), endOfLoop(), endPathsEnabled(), endRun(), forkProcess(), getAllModuleDescriptions(), getTriggerReport(), init(), openOutputFiles(), processEventAsync(), Config.Process::prune(), respondToCloseInputFile(), respondToOpenInputFile(), shouldWeCloseOutput(), shouldWeStop(), totalEvents(), totalEventsFailed(), totalEventsPassed(), writeLumi(), writeRun(), and ~EventProcessor().

ServiceToken edm::EventProcessor::serviceToken_

private

Definition at line 279 of file EventProcessor.h.

Referenced by beginJob(), endJob(), forkProcess(), getToken(), handleNextEventForStreamAsync(), init(), processEventAsync(), readNextEventForStream(), and runToCompletion().

bool edm::EventProcessor::setCpuAffinity_

private

Definition at line 316 of file EventProcessor.h.

Referenced by forkProcess(), and init().

bool edm::EventProcessor::shouldWeStop_

private

Definition at line 302 of file EventProcessor.h.

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

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

private

Definition at line 299 of file EventProcessor.h.

Referenced by readNextEventForStream().

SharedResourcesAcquirer edm::EventProcessor::sourceResourcesAcquirer_

private

Definition at line 298 of file EventProcessor.h.

Referenced by handleNextEventForStreamAsync().

bool edm::EventProcessor::stateMachineWasInErrorState_

private

Definition at line 303 of file EventProcessor.h.

Referenced by doErrorStuff(), and runToCompletion().

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

private

Definition at line 288 of file EventProcessor.h.

Referenced by beginJob(), beginLumi(), beginRun(), closeOutputFiles(), deleteLumiFromCache(), deleteRunFromCache(), Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), endLumi(), endRun(), init(), openOutputFiles(), processEventAsync(), respondToCloseInputFile(), respondToOpenInputFile(), shouldWeCloseOutput(), shouldWeStop(), writeLumi(), and writeRun().

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

private

Definition at line 278 of file EventProcessor.h.

Referenced by init().