#include <EventProcessor.h>

Inheritance diagram for edm::EventProcessor:

Public Member Functions
virtual bool	alreadyHandlingException () const

void	beginJob ()

virtual void	beginLumi (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi)

virtual void	beginRun (statemachine::Run const &run)

void	clearCounters ()
	Clears counters used by trigger report. More...

virtual void	closeInputFile (bool cleaningUpAfterException)

virtual void	closeOutputFiles ()

virtual void	deleteLumiFromCache (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi)

virtual void	deleteRunFromCache (statemachine::Run const &run)

virtual void	doErrorStuff ()

void	enableEndPaths (bool active)

void	endJob ()

virtual void	endLumi (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi, bool cleaningUpAfterException)

virtual bool	endOfLoop ()

bool	endPathsEnabled () const

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

	EventProcessor (std::string const &config, ServiceToken const &token=ServiceToken(), serviceregistry::ServiceLegacy=serviceregistry::kOverlapIsError, std::vector< std::string > const &defaultServices=std::vector< std::string >(), std::vector< std::string > const &forcedServices=std::vector< std::string >())

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

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

	EventProcessor (std::string const &config, bool isPython)
	meant for unit tests More...

	EventProcessor (EventProcessor const &)=delete

bool	forkProcess (std::string const &jobReportFile)

std::vector< ModuleDescription const * >	getAllModuleDescriptions () const

ServiceToken	getToken ()

void	getTriggerReport (TriggerReport &rep) const

virtual void	openOutputFiles ()

EventProcessor &	operator= (EventProcessor const &)=delete

virtual void	prepareForNextLoop ()

virtual int	readAndMergeLumi ()

virtual statemachine::Run	readAndMergeRun ()

virtual void	readAndProcessEvent ()

virtual void	readFile ()

virtual int	readLuminosityBlock ()

virtual statemachine::Run	readRun ()

virtual void	respondToCloseInputFile ()

virtual void	respondToOpenInputFile ()

virtual void	rewindInput ()

StatusCode	run ()

virtual StatusCode	runToCompletion ()

virtual void	setExceptionMessageFiles (std::string &message)

virtual void	setExceptionMessageLumis (std::string &message)

virtual void	setExceptionMessageRuns (std::string &message)

virtual bool	shouldWeCloseOutput () const

virtual bool	shouldWeStop () const

virtual void	startingNewLoop ()

int	totalEvents () const

int	totalEventsFailed () const

int	totalEventsPassed () const

virtual void	writeLumi (ProcessHistoryID const &phid, RunNumber_t run, LuminosityBlockNumber_t lumi)

virtual void	writeRun (statemachine::Run const &run)

	~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
bool	checkForAsyncStopRequest (StatusCode &)

std::auto_ptr < statemachine::Machine >	createStateMachine ()

bool	hasSubProcess () const

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

void	possiblyContinueAfterForkChildFailure ()

void	processEvent (unsigned int iStreamIndex)

void	processEventsForStreamAsync (unsigned int iStreamIndex, std::atomic< bool > *finishedProcessingEvents)

void	readEvent (unsigned int iStreamIndex)

void	setupSignal ()

void	terminateMachine (std::auto_ptr< statemachine::Machine > &)

Private Attributes
std::unique_ptr < ExceptionToActionTable const >	act_table_

std::shared_ptr< ActivityRegistry >	actReg_

bool	alreadyHandlingException_

bool	asyncStopRequestedWhileProcessingEvents_

StatusCode	asyncStopStatusCodeFromProcessingEvents_

bool	beginJobCalled_

std::shared_ptr < BranchIDListHelper >	branchIDListHelper_

bool	continueAfterChildFailure_

std::exception_ptr	deferredExceptionPtr_

std::atomic< bool >	deferredExceptionPtrIsSet_

std::string	emptyRunLumiMode_

boost::shared_ptr < eventsetup::EventSetupProvider >	esp_

std::unique_ptr < eventsetup::EventSetupsController >	espController_

ExcludedDataMap	eventSetupDataToExcludeFromPrefetching_

std::string	exceptionMessageFiles_

std::string	exceptionMessageLumis_

std::string	exceptionMessageRuns_

std::unique_ptr< FileBlock >	fb_

std::string	fileMode_

bool	forceESCacheClearOnNewRun_

bool	forceLooperToEnd_

std::unique_ptr< HistoryAppender >	historyAppender_

std::unique_ptr< InputSource >	input_

boost::shared_ptr< EDLooperBase >	looper_

bool	looperBeginJobRun_

InputSource::ItemType	nextItemTypeFromProcessingEvents_

std::mutex	nextTransitionMutex_

int	numberOfForkedChildren_

unsigned int	numberOfSequentialEventsPerChild_

PathsAndConsumesOfModules	pathsAndConsumesOfModules_

PreallocationConfiguration	preallocations_

std::shared_ptr< ProductRegistry >	preg_

PrincipalCache	principalCache_

std::shared_ptr < ProcessConfiguration const >	processConfiguration_

ProcessContext	processContext_

std::auto_ptr< Schedule >	schedule_

ServiceToken	serviceToken_

bool	setCpuAffinity_

bool	shouldWeStop_

bool	stateMachineWasInErrorState_

std::auto_ptr< SubProcess >	subProcess_

std::shared_ptr < ThinnedAssociationsHelper >	thinnedAssociationsHelper_

Friends
class	StreamProcessingTask

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 60 of file EventProcessor.h.

Member Typedef Documentation

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

private

Definition at line 306 of file EventProcessor.h.

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

private

Definition at line 307 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 216 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_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     fb_(),
     looper_(),
     deferredExceptionPtrIsSet_(false),
     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 259 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_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     fb_(),
     looper_(),
     deferredExceptionPtrIsSet_(false),
     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 303 of file EventProcessor.cc.

References init().

                                                       :
     actReg_(),
     preg_(),
     branchIDListHelper_(),
     serviceToken_(),
     input_(),
     espController_(new eventsetup::EventSetupsController),
     esp_(),
     act_table_(),
     processConfiguration_(),
     schedule_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     fb_(),
     looper_(),
     deferredExceptionPtrIsSet_(false),
     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 345 of file EventProcessor.cc.

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

                                                                       :
     actReg_(),
     preg_(),
     branchIDListHelper_(),
     serviceToken_(),
     input_(),
     espController_(new eventsetup::EventSetupsController),
     esp_(),
     act_table_(),
     processConfiguration_(),
     schedule_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     fb_(),
     looper_(),
     deferredExceptionPtrIsSet_(false),
     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 557 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_, schedule_, and subProcess_.

                                   {
     // 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
     espController_.reset();
     subProcess_.reset();
     esp_.reset();
     schedule_.reset();
     input_.reset();
     looper_.reset();
     actReg_.reset();
 
     pset::Registry::instance()->clear();
     ParentageRegistry::instance()->clear();
   }

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

delete

Member Function Documentation

bool edm::EventProcessor::alreadyHandlingException ( ) const

virtual

Implements edm::IEventProcessor.

Definition at line 2067 of file EventProcessor.cc.

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

References actReg_, cms::Exception::addContext(), bk::beginJob(), beginJobCalled_, hasSubProcess(), i, edm::PathsAndConsumesOfModules::initialize(), input_, edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfRuns(), edm::PreallocationConfiguration::numberOfStreams(), edm::PreallocationConfiguration::numberOfThreads(), cmsPerfStripChart::operate(), pathsAndConsumesOfModules_, preallocations_, preg_, processContext_, schedule_, serviceToken_, subProcess_, and edm::convertException::wrap().

                            {
     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);
     pathsAndConsumesOfModules_.initialize(schedule_.get(), preg_);
     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?
     if(hasSubProcess()) subProcess_->doBeginJob();
     actReg_->postBeginJobSignal_();
     
     for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
       schedule_->beginStream(i);
       if(hasSubProcess()) subProcess_->doBeginStream(i);
     }
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1669 of file EventProcessor.cc.

References edm::LuminosityBlockPrincipal::beginTime(), FDEBUG, i, edm::Service< T >::isAvailable(), edm::LuminosityBlockPrincipal::luminosityBlock(), and edm::LuminosityBlockPrincipal::run().

                                                                                                             {
     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;
       schedule_->processOneGlobal<Traits>(lumiPrincipal, es);
       if(hasSubProcess()) {
         subProcess_->doBeginLuminosityBlock(lumiPrincipal, ts);
       }
     }
     FDEBUG(1) << "\tbeginLumi " << run << "/" << lumi << "\n";
     if(looper_) {
       looper_->doBeginLuminosityBlock(lumiPrincipal, es, &processContext_);
     }
     {
       for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
         typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin> Traits;
         schedule_->processOneStream<Traits>(i,lumiPrincipal, es);
         if(hasSubProcess()) {
           subProcess_->doStreamBeginLuminosityBlock(i,lumiPrincipal, ts);
         }
       }
     }
     FDEBUG(1) << "\tstreamBeginLumi " << run << "/" << lumi << "\n";
     if(looper_) {
       //looper_->doStreamBeginLuminosityBlock(schedule_->streamID(),lumiPrincipal, es);
     }
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1574 of file EventProcessor.cc.

References edm::RunPrincipal::beginTime(), FDEBUG, i, statemachine::Run::processHistoryID(), edm::RunPrincipal::run(), and statemachine::Run::runNumber().

                                                           {
     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;
       schedule_->processOneGlobal<Traits>(runPrincipal, es);
       if(hasSubProcess()) {
         subProcess_->doBeginRun(runPrincipal, ts);
       }
     }
     FDEBUG(1) << "\tbeginRun " << run.runNumber() << "\n";
     if(looper_) {
       looper_->doBeginRun(runPrincipal, es, &processContext_);
     }
     {
       typedef OccurrenceTraits<RunPrincipal, BranchActionStreamBegin> Traits;
       for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
         schedule_->processOneStream<Traits>(i,runPrincipal, es);
         if(hasSubProcess()) {
           subProcess_->doStreamBeginRun(i, runPrincipal, ts);
         }
       }
     }
     FDEBUG(1) << "\tstreamBeginRun " << run.runNumber() << "\n";
     if(looper_) {
       //looper_->doStreamBeginRun(schedule_->streamID(),runPrincipal, es);
     }
   }

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

private

Definition at line 1261 of file EventProcessor.cc.

References edm::shutdown_flag.

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

                                 {
     schedule_->clearCounters();
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1480 of file EventProcessor.cc.

References FDEBUG.

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

void edm::EventProcessor::closeOutputFiles ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1497 of file EventProcessor.cc.

References FDEBUG.

                                         {
     if (fb_.get() != nullptr) {
       schedule_->closeOutputFiles();
       if(hasSubProcess()) subProcess_->closeOutputFiles();
     }
     FDEBUG(1) << "\tcloseOutputFiles\n";
   }

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

private

Definition at line 1230 of file EventProcessor.cc.

References edm::errors::Configuration, statemachine::doNotHandleEmptyRunsAndLumis, edm::hlt::Exception, dtDQMClient_cfg::fileMode, statemachine::FULLMERGE, statemachine::handleEmptyRuns, statemachine::handleEmptyRunsAndLumis, statemachine::NOMERGE, and AlCaHLTBitMon_QueryRunRegistry::string.

                                      {
     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";
     }
     
     std::auto_ptr<statemachine::Machine> machine(new statemachine::Machine(this,
                                              fileMode,
                                              emptyRunLumiMode));
     
     machine->initiate();
     return machine;
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1846 of file EventProcessor.cc.

References FDEBUG.

                                                                                                                       {
     principalCache_.deleteLumi(phid, run, lumi);
     if(hasSubProcess()) subProcess_->deleteLumiFromCache(phid, run, lumi);
     FDEBUG(1) << "\tdeleteLumiFromCache " << run << "/" << lumi << "\n";
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1834 of file EventProcessor.cc.

References FDEBUG, statemachine::Run::processHistoryID(), and statemachine::Run::runNumber().

                                                                     {
     principalCache_.deleteRun(run.processHistoryID(), run.runNumber());
     if(hasSubProcess()) subProcess_->deleteRunFromCache(run.processHistoryID(), run.runNumber());
     FDEBUG(1) << "\tdeleteRunFromCache " << run.runNumber() << "\n";
   }

void edm::EventProcessor::doErrorStuff ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1563 of file EventProcessor.cc.

References FDEBUG.

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

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

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

                          {
     // 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);});
       if(hasSubProcess()) {
         c.call([this,i](){ this->subProcess_->doEndStream(i); } );
       }
     }
     auto actReg = actReg_.get();
     c.call([actReg](){actReg->preEndJobSignal_();});
     schedule_->endJob(c);
     if(hasSubProcess()) {
       c.call(std::bind(&SubProcess::doEndJob, subProcess_.get()));
     }
     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
	)

virtual

Implements edm::IEventProcessor.

Definition at line 1719 of file EventProcessor.cc.

References edm::LuminosityBlockPrincipal::endTime(), FDEBUG, i, edm::LuminosityBlockPrincipal::luminosityBlock(), and edm::LuminosityBlockPrincipal::run().

Referenced by Types.EventRange::cppID().

                                                                                                                                          {
     LuminosityBlockPrincipal& lumiPrincipal = principalCache_.lumiPrincipal(phid, run, lumi);
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
 
       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();
     {
       for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
         typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd> Traits;
         schedule_->processOneStream<Traits>(i,lumiPrincipal, es, cleaningUpAfterException);
         if(hasSubProcess()) {
           subProcess_->doStreamEndLuminosityBlock(i,lumiPrincipal, ts, cleaningUpAfterException);
         }
       }
     }
     FDEBUG(1) << "\tendLumi " << run << "/" << lumi << "\n";
     if(looper_) {
       //looper_->doStreamEndLuminosityBlock(schedule_->streamID(),lumiPrincipal, es);
     }
     {
       typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd> Traits;
       schedule_->processOneGlobal<Traits>(lumiPrincipal, es, cleaningUpAfterException);
       if(hasSubProcess()) {
         subProcess_->doEndLuminosityBlock(lumiPrincipal, ts, cleaningUpAfterException);
       }
     }
     FDEBUG(1) << "\tendLumi " << run << "/" << lumi << "\n";
     if(looper_) {
       looper_->doEndLuminosityBlock(lumiPrincipal, es, &processContext_);
     }
   }

bool edm::EventProcessor::endOfLoop ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1534 of file EventProcessor.cc.

References FDEBUG, and ntuplemaker::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 1214 of file EventProcessor.cc.

                                         {
     return schedule_->endPathsEnabled();
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1626 of file EventProcessor.cc.

References edm::RunPrincipal::endTime(), FDEBUG, i, statemachine::Run::processHistoryID(), edm::RunPrincipal::run(), and statemachine::Run::runNumber().

                                                                                        {
     RunPrincipal& runPrincipal = principalCache_.runPrincipal(run.processHistoryID(), run.runNumber());
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
 
       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();
     {
       for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
         typedef OccurrenceTraits<RunPrincipal, BranchActionStreamEnd> Traits;
         schedule_->processOneStream<Traits>(i,runPrincipal, es, cleaningUpAfterException);
         if(hasSubProcess()) {
           subProcess_->doStreamEndRun(i,runPrincipal, ts, cleaningUpAfterException);
         }
       }
     }
     FDEBUG(1) << "\tstreamEndRun " << run.runNumber() << "\n";
     if(looper_) {
       //looper_->doStreamEndRun(schedule_->streamID(),runPrincipal, es);
     }
     {
       typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd> Traits;
       schedule_->processOneGlobal<Traits>(runPrincipal, es, cleaningUpAfterException);
       if(hasSubProcess()) {
         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 888 of file EventProcessor.cc.

References assert(), bk::beginJob(), edm::eventsetup::EventSetupRecord::doGet(), alignCSCRings::e, edm::hlt::Exception, cmsRelvalreport::exit, edm::EventSetup::fillAvailableRecordKeys(), edm::eventsetup::EventSetupRecord::fillRegisteredDataKeys(), edm::EventSetup::find(), edm::eventsetup::EventSetupRecord::find(), edm::installCustomHandler(), NULL, O_NONBLOCK, cmsPerfStripChart::operate(), or, pipe::pipe(), edm::shutdown_flag, relativeConstraints::value, 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 1189 of file EventProcessor.cc.

                                                  {
     return schedule_->getAllModuleDescriptions();
   }

ServiceToken edm::EventProcessor::getToken ( )

Definition at line 659 of file EventProcessor.cc.

References serviceToken_.

Referenced by ~EventProcessor().

                            {
     return serviceToken_;
   }

void edm::EventProcessor::getTriggerReport ( TriggerReport & rep ) const

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

Definition at line 1219 of file EventProcessor.cc.

                                                            {
     schedule_->getTriggerReport(rep);
   }

bool edm::EventProcessor::hasSubProcess ( ) const

inlineprivate

Definition at line 231 of file EventProcessor.h.

References subProcess_.

Referenced by beginJob(), and endJob().

                                {
       return subProcess_.get() != 0;
     }

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

private

Definition at line 393 of file EventProcessor.cc.

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();
     //std::cerr << parameterSet->dump() << std::endl;
 
     // If there is a subprocess, pop the subprocess parameter set out of the process parameter set
     std::shared_ptr<ParameterSet> subProcessParameterSet(popSubProcessParameterSet(*parameterSet).release());
 
     // 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;
       }
     }
     /*
     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(std::vector<ParameterSet>::const_iterator itPS = excluded.begin(), itPSEnd = excluded.end();
         itPS != itPSEnd;
         ++itPS) {
       eventSetupDataToExcludeFromPrefetching_[itPS->getUntrackedParameter<std::string>("record")].insert(
                                                 std::make_pair(itPS->getUntrackedParameter<std::string>("type", "*"),
                                                                itPS->getUntrackedParameter<std::string>("label", "")));
     }
     IllegalParameters::setThrowAnException(optionsPset.getUntrackedParameter<bool>("throwIfIllegalParameter", true));
 
     // Now do general initialization
     ScheduleItems items;
 
     //initialize the services
     std::shared_ptr<std::vector<ParameterSet> > pServiceSets = processDesc->getServicesPSets();
     ServiceToken token = items.initServices(*pServiceSets, *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,subProcessParameterSet.get(),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);
       ep->preModuleDelayedGetSignal_.connect(std::cref(actReg_->preModuleEventDelayedGetSignal_));
       ep->postModuleDelayedGetSignal_.connect(std::cref(actReg_->postModuleEventDelayedGetSignal_));
       principalCache_.insert(ep);
     }
     // initialize the subprocess, if there is one
     if(subProcessParameterSet) {
       subProcess_.reset(new SubProcess(*subProcessParameterSet,
                                        *parameterSet,
                                        preg_,
                                        branchIDListHelper_,
                                        *thinnedAssociationsHelper_,
                                        *espController_,
                                        *actReg_,
                                        token,
                                        serviceregistry::kConfigurationOverrides,
                                        preallocations_,
                                        &processContext_));
     }
   }

void edm::EventProcessor::openOutputFiles ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1489 of file EventProcessor.cc.

References FDEBUG.

                                        {
     if (fb_.get() != nullptr) {
       schedule_->openOutputFiles(*fb_);
       if(hasSubProcess()) subProcess_->openOutputFiles(*fb_);
     }
     FDEBUG(1) << "\topenOutputFiles\n";
   }

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

delete

void edm::EventProcessor::possiblyContinueAfterForkChildFailure ( )

private

Definition at line 872 of file EventProcessor.cc.

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

void edm::EventProcessor::prepareForNextLoop ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1553 of file EventProcessor.cc.

References FDEBUG.

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

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

private

Definition at line 1993 of file EventProcessor.cc.

References assert(), ev, FDEBUG, edm::Service< T >::isAvailable(), edm::ProcessingController::lastOperationSucceeded(), edm::ProcessingController::requestedTransition(), edm::ProcessingController::setLastOperationSucceeded(), edm::ProcessingController::specifiedEventTransition(), ntuplemaker::status, and summarizeEdmComparisonLogfiles::succeeded.

                                                              {
     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());
 
     //We can only update IOVs on Lumi boundaries
     //IOVSyncValue ts(pep->id(), pep->time());
     //espController_->eventSetupForInstance(ts);
     EventSetup const& es = esp_->eventSetup();
     {
       typedef OccurrenceTraits<EventPrincipal, BranchActionStreamBegin> Traits;
       schedule_->processOneEvent<Traits>(iStreamIndex,*pep, es);
       if(hasSubProcess()) {
         subProcess_->doEvent(*pep);
       }
     }
 
     //NOTE: If we have a looper we only have one Stream
     if(looper_) {
       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(pep->streamID(), &processContext_);
         status = looper_->doDuringLoop(*pep, 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;
 
     }
 
     FDEBUG(1) << "\tprocessEvent\n";
     pep->clearEventPrincipal();
   }

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

private

Definition at line 1875 of file EventProcessor.cc.

References cmsPerfStripChart::operate(), and processEvent().

                                                                                               {
     try {
       // make the services available
       ServiceRegistry::Operate operate(serviceToken_);
       if(preallocations_.numberOfStreams()>1) {
         edm::Service<RootHandlers> handler;
         handler->initializeThisThreadForUse();
       }
       
       if(iStreamIndex==0) {
         processEvent(0);
       }
       do {
         if(shouldWeStop()) {
           break;
         }
         if(deferredExceptionPtrIsSet_.load(std::memory_order_acquire)) {
           //another thread hit an exception
           //std::cerr<<"another thread saw an exception\n";
           break;
         }
         {
           
           
           {
             //nextItemType and readEvent need to be in same critical section
             std::lock_guard<std::mutex> sourceGuard(nextTransitionMutex_);
             
             if(finishedProcessingEvents->load(std::memory_order_acquire)) {
               //std::cerr<<"finishedProcessingEvents\n";
               break;
             }
 
             //If source and DelayedReader share a resource we must serialize them
             auto sr = input_->resourceSharedWithDelayedReader();
             std::unique_lock<SharedResourcesAcquirer> delayedReaderGuard;
             if(sr) {
               delayedReaderGuard = std::unique_lock<SharedResourcesAcquirer>(*sr);
             }
             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";
               break;
             }
             if((asyncStopRequestedWhileProcessingEvents_=checkForAsyncStopRequest(asyncStopStatusCodeFromProcessingEvents_))) {
               //std::cerr<<"task told to async stop\n";
               actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
               break;
             }
             readEvent(iStreamIndex);
           }
         }
         if(deferredExceptionPtrIsSet_.load(std::memory_order_acquire)) {
           //another thread hit an exception
           //std::cerr<<"another thread saw an exception\n";
           actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExceptionFromAnotherContext);
 
           break;
         }
         processEvent(iStreamIndex);
       }while(true);
     } catch (...) {
       bool expected =false;
       if(deferredExceptionPtrIsSet_.compare_exchange_strong(expected,true)) {
         deferredExceptionPtr_ = std::current_exception();
       }
       //std::cerr<<"task caught exception\n";
     }
   }

int edm::EventProcessor::readAndMergeLumi ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1818 of file EventProcessor.cc.

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

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

virtual

Implements edm::IEventProcessor.

Definition at line 1781 of file EventProcessor.cc.

References assert(), and PDRates::Run.

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

virtual

Implements edm::IEventProcessor.

Definition at line 1948 of file EventProcessor.cc.

References pyrootRender::destroy(), and processEvent().

                                            {
     if(numberOfForkedChildren_>0) {
       readEvent(0);
       processEvent(0);
       return;
     }
     nextItemTypeFromProcessingEvents_ = InputSource::IsEvent; //needed for looper
     asyncStopRequestedWhileProcessingEvents_ = false;
 
     std::atomic<bool> finishedProcessingEvents{false};
 
     //Task assumes Stream 0 has already read the event that caused us to go here
     readEvent(0);
     
     //To wait, the ref count has to b 1+#streams
     tbb::task* eventLoopWaitTask{new (tbb::task::allocate_root()) tbb::empty_task{}};
     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( *(new (tbb::task::allocate_root()) StreamProcessingTask{this,iStreamIndex, &finishedProcessingEvents, eventLoopWaitTask}));
 
     }
     eventLoopWaitTask->increment_ref_count();
     eventLoopWaitTask->spawn_and_wait_for_all(*(new (tbb::task::allocate_root()) StreamProcessingTask{this,iStreamIndex,&finishedProcessingEvents,eventLoopWaitTask}));
     tbb::task::destroy(*eventLoopWaitTask);
     
     //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 1982 of file EventProcessor.cc.

References event(), and FDEBUG.

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

virtual

Implements edm::IEventProcessor.

Definition at line 1462 of file EventProcessor.cc.

References FDEBUG, or, and findQualityFiles::size.

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

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

int edm::EventProcessor::readLuminosityBlock ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1793 of file EventProcessor.cc.

References edm::hlt::Exception, and edm::errors::LogicError.

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

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

virtual

Implements edm::IEventProcessor.

Definition at line 1763 of file EventProcessor.cc.

References assert(), edm::hlt::Exception, edm::errors::LogicError, and PDRates::Run.

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

virtual

Implements edm::IEventProcessor.

Definition at line 1516 of file EventProcessor.cc.

References FDEBUG.

                                                {
     if (fb_.get() != nullptr) {
       schedule_->respondToCloseInputFile(*fb_);
       if(hasSubProcess()) subProcess_->respondToCloseInputFile(*fb_);
     }
     FDEBUG(1) << "\trespondToCloseInputFile\n";
   }

void edm::EventProcessor::respondToOpenInputFile ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1505 of file EventProcessor.cc.

References FDEBUG.

                                               {
     if(hasSubProcess()) {
       subProcess_->updateBranchIDListHelper(branchIDListHelper_->branchIDLists());
     }
     if (fb_.get() != nullptr) {
       schedule_->respondToOpenInputFile(*fb_);
       if(hasSubProcess()) subProcess_->respondToOpenInputFile(*fb_);
     }
     FDEBUG(1) << "\trespondToOpenInputFile\n";
   }

void edm::EventProcessor::rewindInput ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1547 of file EventProcessor.cc.

References FDEBUG.

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

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

inline

Definition at line 315 of file EventProcessor.h.

References runToCompletion().

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

                       {
     return runToCompletion();
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1274 of file EventProcessor.cc.

References cms::Exception::addAdditionalInfo(), cms::Exception::alreadyPrinted(), bk::beginJob(), alignCSCRings::e, edm::hlt::Exception, FDEBUG, edm::errors::LogicError, cmsPerfStripChart::operate(), runEdmFileComparison::returnCode, findQualityFiles::size, and edm::convertException::wrap().

Referenced by run().

                                   {
 
     StatusCode returnCode=epSuccess;
     asyncStopStatusCodeFromProcessingEvents_=epSuccess;
     std::auto_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(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() != 0) {
       terminateMachine(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 )

virtual

Implements edm::IEventProcessor.

Definition at line 2055 of file EventProcessor.cc.

References python.rootplot.argparse::message.

                                                                   {
     exceptionMessageFiles_ = message;
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2063 of file EventProcessor.cc.

References python.rootplot.argparse::message.

                                                                   {
     exceptionMessageLumis_ = message;
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2059 of file EventProcessor.cc.

References python.rootplot.argparse::message.

                                                                  {
     exceptionMessageRuns_ = message;
   }

void edm::EventProcessor::setupSignal ( )

private

bool edm::EventProcessor::shouldWeCloseOutput ( ) const

virtual

Implements edm::IEventProcessor.

Definition at line 1558 of file EventProcessor.cc.

References FDEBUG.

                                                  {
     FDEBUG(1) << "\tshouldWeCloseOutput\n";
     return hasSubProcess() ? subProcess_->shouldWeCloseOutput() : schedule_->shouldWeCloseOutput();
   }

bool edm::EventProcessor::shouldWeStop ( ) const

virtual

Implements edm::IEventProcessor.

Definition at line 2049 of file EventProcessor.cc.

References FDEBUG.

                                           {
     FDEBUG(1) << "\tshouldWeStop\n";
     if(shouldWeStop_) return true;
     return (schedule_->terminate() || (hasSubProcess() && subProcess_->terminate()));
   }

void edm::EventProcessor::startingNewLoop ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1524 of file EventProcessor.cc.

References FDEBUG.

                                        {
     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::auto_ptr< statemachine::Machine > & iMachine )

private

Definition at line 2071 of file EventProcessor.cc.

References FDEBUG.

                                                                                   {
     if(iMachine.get() != 0) {
       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";
       }
       iMachine.reset();
     }
   }

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

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

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

                                           {
     return schedule_->totalEventsPassed();
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1840 of file EventProcessor.cc.

References FDEBUG.

                                                                                                             {
     schedule_->writeLumi(principalCache_.lumiPrincipal(phid, run, lumi), &processContext_);
     if(hasSubProcess()) subProcess_->writeLumi(phid, run, lumi);
     FDEBUG(1) << "\twriteLumi " << run << "/" << lumi << "\n";
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1828 of file EventProcessor.cc.

References FDEBUG, statemachine::Run::processHistoryID(), and statemachine::Run::runNumber().

                                                           {
     schedule_->writeRun(principalCache_.runPrincipal(run.processHistoryID(), run.runNumber()), &processContext_);
     if(hasSubProcess()) subProcess_->writeRun(run.processHistoryID(), run.runNumber());
     FDEBUG(1) << "\twriteRun " << run.runNumber() << "\n";
   }

Friends And Related Function Documentation

friend class StreamProcessingTask

friend

Definition at line 237 of file EventProcessor.h.

Member Data Documentation

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

private

Definition at line 265 of file EventProcessor.h.

Referenced by init().

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

private

Definition at line 257 of file EventProcessor.h.

Referenced by beginJob(), endJob(), init(), and ~EventProcessor().

bool edm::EventProcessor::alreadyHandlingException_

private

Definition at line 290 of file EventProcessor.h.

bool edm::EventProcessor::asyncStopRequestedWhileProcessingEvents_

private

Definition at line 302 of file EventProcessor.h.

StatusCode edm::EventProcessor::asyncStopStatusCodeFromProcessingEvents_

private

Definition at line 304 of file EventProcessor.h.

bool edm::EventProcessor::beginJobCalled_

private

Definition at line 282 of file EventProcessor.h.

Referenced by beginJob().

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

private

Definition at line 259 of file EventProcessor.h.

Referenced by init().

bool edm::EventProcessor::continueAfterChildFailure_

private

Definition at line 298 of file EventProcessor.h.

Referenced by init().

std::exception_ptr edm::EventProcessor::deferredExceptionPtr_

private

Definition at line 278 of file EventProcessor.h.

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

private

Definition at line 277 of file EventProcessor.h.

std::string edm::EventProcessor::emptyRunLumiMode_

private

Definition at line 286 of file EventProcessor.h.

Referenced by init().

boost::shared_ptr<eventsetup::EventSetupProvider> edm::EventProcessor::esp_

private

Definition at line 264 of file EventProcessor.h.

Referenced by init(), and ~EventProcessor().

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

private

Definition at line 263 of file EventProcessor.h.

Referenced by init(), and ~EventProcessor().

ExcludedDataMap edm::EventProcessor::eventSetupDataToExcludeFromPrefetching_

private

Definition at line 308 of file EventProcessor.h.

Referenced by init().

std::string edm::EventProcessor::exceptionMessageFiles_

private

Definition at line 287 of file EventProcessor.h.

std::string edm::EventProcessor::exceptionMessageLumis_

private

Definition at line 289 of file EventProcessor.h.

std::string edm::EventProcessor::exceptionMessageRuns_

private

Definition at line 288 of file EventProcessor.h.

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

private

Definition at line 273 of file EventProcessor.h.

std::string edm::EventProcessor::fileMode_

private

Definition at line 285 of file EventProcessor.h.

Referenced by init().

bool edm::EventProcessor::forceESCacheClearOnNewRun_

private

Definition at line 293 of file EventProcessor.h.

Referenced by init().

bool edm::EventProcessor::forceLooperToEnd_

private

Definition at line 291 of file EventProcessor.h.

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

private

Definition at line 271 of file EventProcessor.h.

Referenced by init().

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

private

Definition at line 262 of file EventProcessor.h.

Referenced by beginJob(), endJob(), init(), and ~EventProcessor().

boost::shared_ptr<EDLooperBase> edm::EventProcessor::looper_

private

Definition at line 274 of file EventProcessor.h.

Referenced by Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), init(), and ~EventProcessor().

bool edm::EventProcessor::looperBeginJobRun_

private

Definition at line 292 of file EventProcessor.h.

InputSource::ItemType edm::EventProcessor::nextItemTypeFromProcessingEvents_

private

Definition at line 303 of file EventProcessor.h.

std::mutex edm::EventProcessor::nextTransitionMutex_

private

Definition at line 280 of file EventProcessor.h.

int edm::EventProcessor::numberOfForkedChildren_

private

Definition at line 295 of file EventProcessor.h.

Referenced by init().

unsigned int edm::EventProcessor::numberOfSequentialEventsPerChild_

private

Definition at line 296 of file EventProcessor.h.

Referenced by init().

PathsAndConsumesOfModules edm::EventProcessor::pathsAndConsumesOfModules_

private

Definition at line 268 of file EventProcessor.h.

Referenced by beginJob().

PreallocationConfiguration edm::EventProcessor::preallocations_

private

Definition at line 300 of file EventProcessor.h.

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

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

private

Definition at line 258 of file EventProcessor.h.

Referenced by beginJob(), and init().

PrincipalCache edm::EventProcessor::principalCache_

private

Definition at line 281 of file EventProcessor.h.

Referenced by init().

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

private

Definition at line 266 of file EventProcessor.h.

Referenced by init().

ProcessContext edm::EventProcessor::processContext_

private

Definition at line 267 of file EventProcessor.h.

Referenced by beginJob(), and init().

std::auto_ptr<Schedule> edm::EventProcessor::schedule_

private

Definition at line 269 of file EventProcessor.h.

Referenced by Config.Process::_insertPaths(), beginJob(), endJob(), init(), Config.Process::prune(), and ~EventProcessor().

ServiceToken edm::EventProcessor::serviceToken_

private

Definition at line 261 of file EventProcessor.h.

Referenced by beginJob(), endJob(), getToken(), and init().

bool edm::EventProcessor::setCpuAffinity_

private

Definition at line 297 of file EventProcessor.h.

Referenced by init().

bool edm::EventProcessor::shouldWeStop_

private

Definition at line 283 of file EventProcessor.h.

bool edm::EventProcessor::stateMachineWasInErrorState_

private

Definition at line 284 of file EventProcessor.h.

std::auto_ptr<SubProcess> edm::EventProcessor::subProcess_

private

Definition at line 270 of file EventProcessor.h.

Referenced by beginJob(), Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), hasSubProcess(), init(), and ~EventProcessor().

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

private

Definition at line 260 of file EventProcessor.h.

Referenced by init().

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation

Member Data Documentation