#include <EventProcessor.h>

Inheritance diagram for edm::EventProcessor:

Public Member Functions
virtual bool	alreadyHandlingException () const

void	beginJob ()

virtual void	beginLumi (ProcessHistoryID const &phid, int run, int 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 ()

char const *	currentStateName () const

void	declareRunNumber (RunNumber_t runNumber)

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

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

virtual void	doErrorStuff ()

void	enableEndPaths (bool active)

void	endJob ()

virtual void	endLumi (ProcessHistoryID const &phid, int run, int 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 (boost::shared_ptr< ProcessDesc > &processDesc, ServiceToken const &token, serviceregistry::ServiceLegacy legacy)

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

bool	forkProcess (std::string const &jobReportFile)

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

event_processor::State	getState () const

ServiceToken	getToken ()

void	getTriggerReport (TriggerReport &rep) const

char const *	msgName (event_processor::Msg m) const

virtual void	openOutputFiles ()

ActivityRegistry::PostProcessEvent &	postProcessEventSignal ()

virtual void	prepareForNextLoop ()

ActivityRegistry::PreProcessEvent &	preProcessEventSignal ()

virtual int	readAndCacheLumi (bool merge)

virtual statemachine::Run	readAndCacheRun (bool merge)

virtual void	readAndProcessEvent ()

virtual void	readFile ()

virtual void	respondToCloseInputFile ()

virtual void	respondToCloseOutputFiles ()

virtual void	respondToOpenInputFile ()

virtual void	respondToOpenOutputFiles ()

void	rewind ()

virtual void	rewindInput ()

StatusCode	run (int numberEventsToProcess, bool repeatable=true)

StatusCode	run ()

void	runAsync ()

virtual StatusCode	runEventCount (int numberOfEventsToProcess)

virtual StatusCode	runToCompletion (bool onlineStateTransitions)

virtual void	setExceptionMessageFiles (std::string &message)

virtual void	setExceptionMessageLumis (std::string &message)

virtual void	setExceptionMessageRuns (std::string &message)

void	setRunNumber (RunNumber_t runNumber)

virtual bool	shouldWeCloseOutput () const

virtual bool	shouldWeStop () const

StatusCode	shutdownAsync (unsigned int timeout_secs=60 *2)

StatusCode	skip (int numberToSkip)

virtual void	startingNewLoop ()

char const *	stateName (event_processor::State s) const

StatusCode	statusAsync () const

StatusCode	stopAsync (unsigned int timeout_secs=60 *2)

int	totalEvents () const

int	totalEventsFailed () const

int	totalEventsPassed () const

StatusCode	waitTillDoneAsync (unsigned int timeout_seconds=0)

virtual void	writeLumi (ProcessHistoryID const &phid, int run, int 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
void	changeState (event_processor::Msg)

void	connectSigs (EventProcessor *ep)

StatusCode	doneAsync (event_processor::Msg m)

void	errorState ()

bool	hasSubProcess () const

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

StatusCode	runCommon (bool onlineStateTransitions, int numberOfEventsToProcess)

void	setupSignal ()

void	terminateMachine ()

StatusCode	waitForAsyncCompletion (unsigned int timeout_seconds)

Static Private Member Functions
static void	asyncRun (EventProcessor *)

Private Attributes
boost::shared_ptr< ActionTable const >	act_table_

boost::shared_ptr < ActivityRegistry >	actReg_

bool	alreadyHandlingException_

std::string	emptyRunLumiMode_

boost::shared_ptr < eventsetup::EventSetupProvider >	esp_

boost::scoped_ptr < eventsetup::EventSetupsController >	espController_

boost::shared_ptr< boost::thread >	event_loop_

volatile pthread_t	event_loop_id_

ExcludedDataMap	eventSetupDataToExcludeFromPrefetching_

std::string	exceptionMessageFiles_

std::string	exceptionMessageLumis_

std::string	exceptionMessageRuns_

boost::shared_ptr< FileBlock >	fb_

std::string	fileMode_

bool	forceESCacheClearOnNewRun_

bool	forceLooperToEnd_

boost::scoped_ptr < HistoryAppender >	historyAppender_

volatile bool	id_set_

boost::shared_ptr< InputSource >	input_

std::string	last_error_text_

volatile Status	last_rc_

boost::shared_ptr< EDLooperBase >	looper_

bool	looperBeginJobRun_

std::auto_ptr < statemachine::Machine >	machine_

int	my_sig_num_

int	numberOfForkedChildren_

unsigned int	numberOfSequentialEventsPerChild_

ActivityRegistry::PostProcessEvent	postProcessEventSignal_

boost::shared_ptr < SignallingProductRegistry >	preg_

ActivityRegistry::PreProcessEvent	preProcessEventSignal_

PrincipalCache	principalCache_

boost::shared_ptr < ProcessConfiguration >	processConfiguration_

std::auto_ptr< Schedule >	schedule_

ServiceToken	serviceToken_

bool	setCpuAffinity_

bool	shouldWeStop_

boost::condition	starter_

volatile event_processor::State	state_

boost::mutex	state_lock_

bool	stateMachineWasInErrorState_

volatile int	stop_count_

boost::mutex	stop_lock_

boost::condition	stopper_

std::auto_ptr< SubProcess >	subProcess_

Friends
class	event_processor::StateSentry

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

Member Typedef Documentation

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

private

Definition at line 393 of file EventProcessor.h.

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

private

Definition at line 394 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 367 of file EventProcessor.cc.

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

                                                                             :
     preProcessEventSignal_(),
     postProcessEventSignal_(),
     actReg_(),
     preg_(),
     serviceToken_(),
     input_(),
     espController_(new eventsetup::EventSetupsController),
     esp_(),
     act_table_(),
     processConfiguration_(),
     schedule_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     state_(sInit),
     event_loop_(),
     state_lock_(),
     stop_lock_(),
     stopper_(),
     starter_(),
     stop_count_(-1),
     last_rc_(epSuccess),
     last_error_text_(),
     id_set_(false),
     event_loop_id_(),
     my_sig_num_(getSigNum()),
     fb_(),
     looper_(),
     machine_(),
     principalCache_(),
     shouldWeStop_(false),
     stateMachineWasInErrorState_(false),
     fileMode_(),
     emptyRunLumiMode_(),
     exceptionMessageFiles_(),
     exceptionMessageRuns_(),
     exceptionMessageLumis_(),
     alreadyHandlingException_(false),
     forceLooperToEnd_(false),
     looperBeginJobRun_(false),
     forceESCacheClearOnNewRun_(false),
     numberOfForkedChildren_(0),
     numberOfSequentialEventsPerChild_(1),
     setCpuAffinity_(false),
     eventSetupDataToExcludeFromPrefetching_() {
     boost::shared_ptr<ParameterSet> parameterSet = PythonProcessDesc(config).parameterSet();
     boost::shared_ptr<ProcessDesc> processDesc(new 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 422 of file EventProcessor.cc.

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

                                                                             :
     preProcessEventSignal_(),
     postProcessEventSignal_(),
     actReg_(),
     preg_(),
     serviceToken_(),
     input_(),
     espController_(new eventsetup::EventSetupsController),
     esp_(),
     act_table_(),
     processConfiguration_(),
     schedule_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     state_(sInit),
     event_loop_(),
     state_lock_(),
     stop_lock_(),
     stopper_(),
     starter_(),
     stop_count_(-1),
     last_rc_(epSuccess),
     last_error_text_(),
     id_set_(false),
     event_loop_id_(),
     my_sig_num_(getSigNum()),
     fb_(),
     looper_(),
     machine_(),
     principalCache_(),
     shouldWeStop_(false),
     stateMachineWasInErrorState_(false),
     fileMode_(),
     emptyRunLumiMode_(),
     exceptionMessageFiles_(),
     exceptionMessageRuns_(),
     exceptionMessageLumis_(),
     alreadyHandlingException_(false),
     forceLooperToEnd_(false),
     looperBeginJobRun_(false),
     forceESCacheClearOnNewRun_(false),
     numberOfForkedChildren_(0),
     numberOfSequentialEventsPerChild_(1),
     setCpuAffinity_(false),
     eventSetupDataToExcludeFromPrefetching_() {
     boost::shared_ptr<ParameterSet> parameterSet = PythonProcessDesc(config).parameterSet();
     boost::shared_ptr<ProcessDesc> processDesc(new ProcessDesc(parameterSet));
     processDesc->addServices(defaultServices, forcedServices);
     init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
   }

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

Definition at line 475 of file EventProcessor.cc.

References init().

                                                       :
     preProcessEventSignal_(),
     postProcessEventSignal_(),
     actReg_(),
     preg_(),
     serviceToken_(),
     input_(),
     espController_(new eventsetup::EventSetupsController),
     esp_(),
     act_table_(),
     processConfiguration_(),
     schedule_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     state_(sInit),
     event_loop_(),
     state_lock_(),
     stop_lock_(),
     stopper_(),
     starter_(),
     stop_count_(-1),
     last_rc_(epSuccess),
     last_error_text_(),
     id_set_(false),
     event_loop_id_(),
     my_sig_num_(getSigNum()),
     fb_(),
     looper_(),
     machine_(),
     principalCache_(),
     shouldWeStop_(false),
     stateMachineWasInErrorState_(false),
     fileMode_(),
     emptyRunLumiMode_(),
     exceptionMessageFiles_(),
     exceptionMessageRuns_(),
     exceptionMessageLumis_(),
     alreadyHandlingException_(false),
     forceLooperToEnd_(false),
     looperBeginJobRun_(false),
     forceESCacheClearOnNewRun_(false),
     numberOfForkedChildren_(0),
     numberOfSequentialEventsPerChild_(1),
     setCpuAffinity_(false),
     eventSetupDataToExcludeFromPrefetching_() {
     init(processDesc, token, legacy);
   }

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

meant for unit tests

Definition at line 526 of file EventProcessor.cc.

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

                                                                       :
     preProcessEventSignal_(),
     postProcessEventSignal_(),
     actReg_(),
     preg_(),
     serviceToken_(),
     input_(),
     espController_(new eventsetup::EventSetupsController),
     esp_(),
     act_table_(),
     processConfiguration_(),
     schedule_(),
     subProcess_(),
     historyAppender_(new HistoryAppender),
     state_(sInit),
     event_loop_(),
     state_lock_(),
     stop_lock_(),
     stopper_(),
     starter_(),
     stop_count_(-1),
     last_rc_(epSuccess),
     last_error_text_(),
     id_set_(false),
     event_loop_id_(),
     my_sig_num_(getSigNum()),
     fb_(),
     looper_(),
     machine_(),
     principalCache_(),
     shouldWeStop_(false),
     stateMachineWasInErrorState_(false),
     fileMode_(),
     emptyRunLumiMode_(),
     exceptionMessageFiles_(),
     exceptionMessageRuns_(),
     exceptionMessageLumis_(),
     alreadyHandlingException_(false),
     forceLooperToEnd_(false),
     looperBeginJobRun_(false),
     forceESCacheClearOnNewRun_(false),
     numberOfForkedChildren_(0),
     numberOfSequentialEventsPerChild_(1),
     setCpuAffinity_(false),
     eventSetupDataToExcludeFromPrefetching_() {
     if(isPython) {
       boost::shared_ptr<ParameterSet> parameterSet = PythonProcessDesc(config).parameterSet();
       boost::shared_ptr<ProcessDesc> processDesc(new ProcessDesc(parameterSet));
       init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
     }
     else {
       boost::shared_ptr<ProcessDesc> processDesc(new ProcessDesc(config));
       init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
     }
   }

edm::EventProcessor::~EventProcessor ( )

Definition at line 663 of file EventProcessor.cc.

References actReg_, changeState(), edm::BranchIDListHelper::clearRegistries(), edm::detail::ThreadSafeRegistry< KEY, T, E >::data(), alignCSCRings::e, esp_, espController_, cms::Exception::explainSelf(), edm::detail::ThreadSafeRegistry< KEY, T, E >::extra(), getToken(), input_, edm::detail::ThreadSafeRegistry< KEY, T, E >::instance(), looper_, edm::event_processor::mDtor, schedule_, subProcess_, and terminateMachine().

                                   {
     // Make the services available while everything is being deleted.
     ServiceToken token = getToken();
     ServiceRegistry::Operate op(token);
 
     // The state machine should have already been cleaned up
     // and destroyed at this point by a call to EndJob or
     // earlier when it completed processing events, but if it
     // has not been we'll take care of it here at the last moment.
     // This could cause problems if we are already handling an
     // exception and another one is thrown here ...  For a critical
     // executable the solution to this problem is for the code using
     // the EventProcessor to explicitly call EndJob or use runToCompletion,
     // then the next line of code is never executed.
     terminateMachine();
 
     try {
       changeState(mDtor);
     }
     catch(cms::Exception& e) {
       LogError("System")
         << e.explainSelf() << "\n";
     }
 
     // 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* psetRegistry = pset::Registry::instance();
     psetRegistry->data().clear();
     psetRegistry->extra().setID(ParameterSetID());
 
     EntryDescriptionRegistry::instance()->data().clear();
     ParentageRegistry::instance()->data().clear();
     ProcessConfigurationRegistry::instance()->data().clear();
     ProcessHistoryRegistry::instance()->data().clear();
     BranchIDListHelper::clearRegistries();
   }

Member Function Documentation

bool edm::EventProcessor::alreadyHandlingException ( ) const

virtual

Implements edm::IEventProcessor.

Definition at line 2203 of file EventProcessor.cc.

                                                       {
     return alreadyHandlingException_;
   }

void edm::EventProcessor::asyncRun ( EventProcessor * me )

staticprivate

Definition at line 1569 of file EventProcessor.cc.

References alignCSCRings::e, event_loop_id_, cppFunctionSkipper::exception, cms::Exception::explainSelf(), FDEBUG, id_set_, last_error_text_, last_rc_, runToCompletion(), starter_, stop_count_, stop_lock_, and stopper_.

                                                   {
     // set up signals to allow for interruptions
     // ignore all other signals
     // make sure no exceptions escape out
 
     // temporary hack until we modify the input source to allow
     // wakeup calls from other threads.  This mimics the solution
     // in EventFilter/Processor, which I do not like.
     // allowing cancels means that the thread just disappears at
     // certain points.  This is bad for C++ stack variables.
     pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, 0);
     //pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, 0);
     pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, 0);
     pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, 0);
 
     {
       boost::mutex::scoped_lock(me->stop_lock_);
       me->event_loop_id_ = pthread_self();
       me->id_set_ = true;
       me->starter_.notify_all();
     }
 
     Status rc = epException;
     FDEBUG(2) << "asyncRun starting ......................\n";
 
     try {
       bool onlineStateTransitions = true;
       rc = me->runToCompletion(onlineStateTransitions);
     }
     catch (cms::Exception& e) {
       LogError("FwkJob") << "cms::Exception caught in "
                          << "EventProcessor::asyncRun"
                          << "\n"
                          << e.explainSelf();
       me->last_error_text_ = e.explainSelf();
     }
     catch (std::exception& e) {
       LogError("FwkJob") << "Standard library exception caught in "
                          << "EventProcessor::asyncRun"
                          << "\n"
                          << e.what();
       me->last_error_text_ = e.what();
     }
     catch (...) {
       LogError("FwkJob") << "Unknown exception caught in "
                          << "EventProcessor::asyncRun"
                          << "\n";
       me->last_error_text_ = "Unknown exception caught";
       rc = epOther;
     }
 
     me->last_rc_ = rc;
 
     {
       // notify anyone waiting for exit that we are doing so now
       boost::mutex::scoped_lock sl(me->stop_lock_);
       ++me->stop_count_;
       me->stopper_.notify_all();
     }
     FDEBUG(2) << "asyncRun ending ......................\n";
   }

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

References actReg_, cms::Exception::addContext(), edm::convertException::badAllocToEDM(), bk::beginJob(), trackerHits::c, changeState(), edm::convertException::charPtrToEDM(), alignCSCRings::e, cppFunctionSkipper::exception, hasSubProcess(), input_, edm::event_processor::mBeginJob, cmsPerfStripChart::operate(), alignCSCRings::s, schedule_, serviceToken_, edm::event_processor::sInit, state_, edm::convertException::stdToEDM(), edm::convertException::stringToEDM(), subProcess_, and edm::convertException::unknownToEDM().

Referenced by evf::FUEventProcessor::configuring(), evf::FUEventProcessor::enabling(), rewind(), and skip().

                            {
     if(state_ != sInit) return;
     bk::beginJob();
     // can only be run if in the initial state
     changeState(mBeginJob);
 
     // StateSentry toerror(this); // should we add this ?
     //make the services available
     ServiceRegistry::Operate operate(serviceToken_);
 
     //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 {
       try {
         input_->doBeginJob();
       }
       catch (cms::Exception& e) { throw; }
       catch(std::bad_alloc& bda) { convertException::badAllocToEDM(); }
       catch (std::exception& e) { convertException::stdToEDM(e); }
       catch(std::string& s) { convertException::stringToEDM(s); }
       catch(char const* c) { convertException::charPtrToEDM(c); }
       catch (...) { convertException::unknownToEDM(); }
     }
     catch(cms::Exception& ex) {
       ex.addContext("Calling beginJob for the source");
       throw;
     }
     schedule_->beginJob();
     // toerror.succeeded(); // should we add this?
     if(hasSubProcess()) subProcess_->doBeginJob();
     actReg_->postBeginJobSignal_();
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2051 of file EventProcessor.cc.

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

                                                                                 {
     LuminosityBlockPrincipal& lumiPrincipal = principalCache_.lumiPrincipal(phid, run, lumi);
     input_->doBeginLumi(lumiPrincipal);
 
     Service<RandomNumberGenerator> rng;
     if(rng.isAvailable()) {
       LuminosityBlock lb(lumiPrincipal, ModuleDescription());
       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());
     espController_->eventSetupForInstance(ts);
     EventSetup const& es = esp_->eventSetup();
     {
       typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionBegin> Traits;
       ScheduleSignalSentry<Traits> sentry(actReg_.get(), &lumiPrincipal, &es);
       schedule_->processOneOccurrence<Traits>(lumiPrincipal, es);
       if(hasSubProcess()) {
         subProcess_->doBeginLuminosityBlock(lumiPrincipal, ts);
       }
     }
     FDEBUG(1) << "\tbeginLumi " << run << "/" << lumi << "\n";
     if(looper_) {
       looper_->doBeginLuminosityBlock(lumiPrincipal, es);
     }
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2000 of file EventProcessor.cc.

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

                                                           {
     RunPrincipal& runPrincipal = principalCache_.runPrincipal(run.processHistoryID(), run.runNumber());
     input_->doBeginRun(runPrincipal);
     IOVSyncValue ts(EventID(runPrincipal.run(), 0, 0),
                     runPrincipal.beginTime());
     if(forceESCacheClearOnNewRun_){
       espController_->forceCacheClear();
     }
     espController_->eventSetupForInstance(ts);
     EventSetup const& es = esp_->eventSetup();
     if(looper_ && looperBeginJobRun_== false) {
       looper_->copyInfo(ScheduleInfo(schedule_.get()));
       looper_->beginOfJob(es);
       looperBeginJobRun_ = true;
       looper_->doStartingNewLoop();
     }
     {
       typedef OccurrenceTraits<RunPrincipal, BranchActionBegin> Traits;
       ScheduleSignalSentry<Traits> sentry(actReg_.get(), &runPrincipal, &es);
       schedule_->processOneOccurrence<Traits>(runPrincipal, es);
       if(hasSubProcess()) {
         subProcess_->doBeginRun(runPrincipal, ts);
       }
     }
     FDEBUG(1) << "\tbeginRun " << run.runNumber() << "\n";
     if(looper_) {
       looper_->doBeginRun(runPrincipal, es);
     }
   }

void edm::EventProcessor::changeState ( event_processor::Msg msg )

private

Definition at line 1513 of file EventProcessor.cc.

References cond::rpcobimon::current, edm::hlt::Exception, FDEBUG, edm::event_processor::mAny, edm::event_processor::sInvalid, and asciidump::table.

Referenced by beginJob(), endJob(), rewind(), skip(), ~EventProcessor(), and edm::event_processor::StateSentry::~StateSentry().

                                           {
     // most likely need to serialize access to this routine
 
     boost::mutex::scoped_lock sl(state_lock_);
     State curr = state_;
     int rc;
     // found if(not end of table) and
     // (state == table.state && (msg == table.message || msg == any))
     for(rc = 0;
         table[rc].current != sInvalid &&
           (curr != table[rc].current ||
            (curr == table[rc].current &&
              msg != table[rc].message && table[rc].message != mAny));
         ++rc);
 
     if(table[rc].current == sInvalid)
       throw cms::Exception("BadState")
         << "A member function of EventProcessor has been called in an"
         << " inappropriate order.\n"
         << "Bad transition from " << stateName(curr) << " "
         << "using message " << msgName(msg) << "\n"
         << "No where to go from here.\n";
 
     FDEBUG(1) << "changeState: current=" << stateName(curr)
               << ", message=" << msgName(msg)
               << " -> new=" << stateName(table[rc].final) << "\n";
 
     state_ = table[rc].final;
   }

void edm::EventProcessor::clearCounters ( )

Clears counters used by trigger report.

Definition at line 1375 of file EventProcessor.cc.

Referenced by evf::FUEventProcessor::enableCommon(), and evf::FUEventProcessor::enableForkInEDM().

                                 {
     schedule_->clearCounters();
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 1895 of file EventProcessor.cc.

References FDEBUG.

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

void edm::EventProcessor::closeOutputFiles ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1910 of file EventProcessor.cc.

References FDEBUG.

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

void edm::EventProcessor::connectSigs ( EventProcessor * ep )

private

Definition at line 1331 of file EventProcessor.cc.

References postProcessEventSignal_, and preProcessEventSignal_.

Referenced by init().

                                                 {
     // When the FwkImpl signals are given, pass them to the
     // appropriate EventProcessor signals so that the outside world
     // can see the signal.
     actReg_->preProcessEventSignal_.connect(ep->preProcessEventSignal_);
     actReg_->postProcessEventSignal_.connect(ep->postProcessEventSignal_);
   }

char const * edm::EventProcessor::currentStateName ( ) const

Member functions to support asynchronous interface.

Definition at line 1379 of file EventProcessor.cc.

Referenced by evf::FWEPWrapper::microState(), and evf::FWEPWrapper::monitoring().

                                                      {
     return stateName(getState());
   }

void edm::EventProcessor::declareRunNumber ( RunNumber_t runNumber )

Definition at line 1419 of file EventProcessor.cc.

References bk::beginJob(), and edm::event_processor::mSetRun.

Referenced by evf::FUEventProcessor::enableCommon(), and evf::FUEventProcessor::enableForkInEDM().

                                                 {
     // inside of beginJob there is a check to see if it has been called before
     beginJob();
     changeState(mSetRun);
 
     // interface not correct yet - wait for Bill to be done with run/lumi loop stuff 21-Jun-2007
     //input_->declareRunNumber(runNumber);
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2133 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 2121 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 1989 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;
   }

EventProcessor::StatusCode edm::EventProcessor::doneAsync ( event_processor::Msg m )

private

Definition at line 1505 of file EventProcessor.cc.

                                                           {
     // make sure to include a timeout here so we don't wait forever
     // I suspect there are still timing issues with thread startup
     // and the setting of the various control variables (stop_count, id_set)
     changeState(m);
     return waitForAsyncCompletion(60*2);
   }

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

Referenced by evf::FUEventProcessor::actionPerformed().

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

References actReg_, trackerHits::c, edm::ExceptionCollector::call(), changeState(), edm::OutputModule::doEndJob(), edm::InputSource::doEndJob(), edm::EDLooperBase::endOfJob(), hasSubProcess(), edm::ExceptionCollector::hasThrown(), input_, looper_, edm::event_processor::mEndJob, cmsPerfStripChart::operate(), cppFunctionSkipper::operator, edm::ExceptionCollector::rethrow(), schedule_, serviceToken_, subProcess_, and terminateMachine().

Referenced by evf::FWEPWrapper::init(), and evf::FWEPWrapper::stopAndHalt().

                          {
     // 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");
 
     // only allowed to run if state is sIdle, sJobReady, sRunGiven
     c.call(boost::bind(&EventProcessor::changeState, this, mEndJob));
 
     //make the services available
     ServiceRegistry::Operate operate(serviceToken_);
 
     c.call(boost::bind(&EventProcessor::terminateMachine, this));
     schedule_->endJob(c);
     if(hasSubProcess()) {
       c.call(boost::bind(&SubProcess::doEndJob, subProcess_.get()));
     }
     c.call(boost::bind(&InputSource::doEndJob, input_));
     if(looper_) {
       c.call(boost::bind(&EDLooperBase::endOfJob, looper_));
     }
     c.call(boost::bind(&ActivityRegistry::PostEndJob::operator(), &actReg_->postEndJobSignal_));
     if(c.hasThrown()) {
       c.rethrow();
     }
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2080 of file EventProcessor.cc.

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

Referenced by Types.EventRange::cppID().

                                                                                                              {
     LuminosityBlockPrincipal& lumiPrincipal = principalCache_.lumiPrincipal(phid, run, lumi);
     input_->doEndLumi(lumiPrincipal, cleaningUpAfterException);
     //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());
     espController_->eventSetupForInstance(ts);
     EventSetup const& es = esp_->eventSetup();
     {
       typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionEnd> Traits;
       ScheduleSignalSentry<Traits> sentry(actReg_.get(), &lumiPrincipal, &es);
       schedule_->processOneOccurrence<Traits>(lumiPrincipal, es, cleaningUpAfterException);
       if(hasSubProcess()) {
         subProcess_->doEndLuminosityBlock(lumiPrincipal, ts, cleaningUpAfterException);
       }
     }
     FDEBUG(1) << "\tendLumi " << run << "/" << lumi << "\n";
     if(looper_) {
       looper_->doEndLuminosityBlock(lumiPrincipal, es);
     }
   }

bool edm::EventProcessor::endOfLoop ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1960 of file EventProcessor.cc.

References FDEBUG, and ntuplemaker::status.

                                  {
     if(looper_) {
       ModuleChanger changer(schedule_.get());
       looper_->setModuleChanger(&changer);
       EDLooperBase::Status status = looper_->doEndOfLoop(esp_->eventSetup());
       looper_->setModuleChanger(0);
       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 1365 of file EventProcessor.cc.

                                         {
     return schedule_->endPathsEnabled();
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2030 of file EventProcessor.cc.

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

                                                                                        {
     RunPrincipal& runPrincipal = principalCache_.runPrincipal(run.processHistoryID(), run.runNumber());
     input_->doEndRun(runPrincipal, cleaningUpAfterException);
     IOVSyncValue ts(EventID(runPrincipal.run(), LuminosityBlockID::maxLuminosityBlockNumber(), EventID::maxEventNumber()),
                     runPrincipal.endTime());
     espController_->eventSetupForInstance(ts);
     EventSetup const& es = esp_->eventSetup();
     {
       typedef OccurrenceTraits<RunPrincipal, BranchActionEnd> Traits;
       ScheduleSignalSentry<Traits> sentry(actReg_.get(), &runPrincipal, &es);
       schedule_->processOneOccurrence<Traits>(runPrincipal, es, cleaningUpAfterException);
       if(hasSubProcess()) {
         subProcess_->doEndRun(runPrincipal, ts, cleaningUpAfterException);
       }
     }
     FDEBUG(1) << "\tendRun " << run.runNumber() << "\n";
     if(looper_) {
       looper_->doEndRun(runPrincipal, es);
     }
   }

void edm::EventProcessor::errorState ( )

private

Definition at line 1500 of file EventProcessor.cc.

References edm::event_processor::sError.

                                   {
     state_ = sError;
   }

bool edm::EventProcessor::forkProcess ( std::string const & jobReportFile )

Definition at line 1036 of file EventProcessor.cc.

References 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(), pipe::pipe(), RooFit::readFile(), 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(0);
         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[2];
 
     {
       // 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);
 
         boost::shared_ptr<multicore::MessageReceiverForSource> receiver(new 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);
 
     while(!too_many_fds && !shutdown_flag && !child_failed && (childrenIds.size() != num_children_done)) {
       sigsuspend(&unblockingSigSet);
       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) {
       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 1340 of file EventProcessor.cc.

Referenced by evf::FWEPWrapper::init().

                                                  {
     return schedule_->getAllModuleDescriptions();
   }

State edm::EventProcessor::getState ( ) const

Definition at line 1391 of file EventProcessor.cc.

Referenced by evf::FUEventProcessor::doEndRunInEDM(), evf::FUEventProcessor::enableClassic(), evf::FUEventProcessor::enabling(), evf::FWEPWrapper::init(), evf::FWEPWrapper::microState(), evf::FWEPWrapper::monitoring(), evf::FWEPWrapper::stop(), and evf::FWEPWrapper::stopAndHalt().

                                        {
     return state_;
   }

ServiceToken edm::EventProcessor::getToken ( )

Definition at line 826 of file EventProcessor.cc.

References serviceToken_.

Referenced by evf::FUEventProcessor::attachDqmToShm(), evf::FUEventProcessor::detachDqmFromShm(), evf::FWEPWrapper::moduleWeb(), evf::FWEPWrapper::serviceWeb(), evf::FUEventProcessor::setAttachDqmToShm(), evf::FWEPWrapper::taskWebPage(), and ~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 1370 of file EventProcessor.cc.

Referenced by evf::FWEPWrapper::getTriggerReport(), evf::FWEPWrapper::init(), and evf::FWEPWrapper::taskWebPage().

                                                            {
     schedule_->getTriggerReport(rep);
   }

bool edm::EventProcessor::hasSubProcess ( ) const

inlineprivate

Definition at line 335 of file EventProcessor.h.

References subProcess_.

Referenced by beginJob(), and endJob().

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

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

private

Definition at line 583 of file EventProcessor.cc.

Referenced by EventProcessor().

                                                               {
 
     //std::cerr << processDesc->dump() << std::endl;
     // The BranchIDListRegistry and ProductIDListRegistry are indexed registries, and are singletons.
     //  They must be cleared here because some processes run multiple EventProcessors in succession.
     BranchIDListHelper::clearRegistries();
 
     boost::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
     boost::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);
     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);
     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
     boost::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_);
 
     // intialize miscellaneous items
     boost::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_);
     }
 
     // initialize the input source
     input_ = makeInput(*parameterSet, *common, *items.preg_, principalCache_, items.actReg_, items.processConfiguration_);
 
     // intialize the Schedule
     schedule_ = items.initSchedule(*parameterSet,subProcessParameterSet.get());
 
     // set the data members
     act_table_ = items.act_table_;
     actReg_ = items.actReg_;
     preg_ = items.preg_;
     processConfiguration_ = items.processConfiguration_;
 
     FDEBUG(2) << parameterSet << std::endl;
     connectSigs(this);
 
     // initialize the subprocess, if there is one
     if(subProcessParameterSet) {
       subProcess_.reset(new SubProcess(*subProcessParameterSet, *parameterSet, preg_, *espController_, *actReg_, token, serviceregistry::kConfigurationOverrides));
     }
     espController_->clearComponents();
   }

char const * edm::EventProcessor::msgName ( event_processor::Msg m ) const

Definition at line 1387 of file EventProcessor.cc.

References m.

                                                  {
     return msgNames[m];
   }

void edm::EventProcessor::openOutputFiles ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1902 of file EventProcessor.cc.

References FDEBUG.

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

ActivityRegistry::PostProcessEvent& edm::EventProcessor::postProcessEventSignal ( )

inline

signal is emitted after all modules have finished processing the Event

Definition at line 214 of file EventProcessor.h.

References postProcessEventSignal_.

214 {return postProcessEventSignal_;}

edm::EventProcessor::postProcessEventSignal_

ActivityRegistry::PostProcessEvent postProcessEventSignal_

Definition: EventProcessor.h:347

void edm::EventProcessor::prepareForNextLoop ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1979 of file EventProcessor.cc.

References FDEBUG.

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

ActivityRegistry::PreProcessEvent& edm::EventProcessor::preProcessEventSignal ( )

inline

signal is emitted after the Event has been created by the InputSource but before any modules have seen the Event

Definition at line 209 of file EventProcessor.h.

References preProcessEventSignal_.

209 {return preProcessEventSignal_;}

edm::EventProcessor::preProcessEventSignal_

ActivityRegistry::PreProcessEvent preProcessEventSignal_

Definition: EventProcessor.h:346

int edm::EventProcessor::readAndCacheLumi ( bool merge )

virtual

Implements edm::IEventProcessor.

Definition at line 2109 of file EventProcessor.cc.

                                                  {
     input_->readAndCacheLumi(merge, *historyAppender_);
     input_->markLumi();
     return input_->luminosityBlock();
   }

statemachine::Run edm::EventProcessor::readAndCacheRun ( bool merge )

virtual

Implements edm::IEventProcessor.

Definition at line 2103 of file EventProcessor.cc.

References PDRates::Run.

                                                             {
     input_->readAndCacheRun(merge, *historyAppender_);
     input_->markRun();
     return statemachine::Run(input_->reducedProcessHistoryID(), input_->run());
   }

void edm::EventProcessor::readAndProcessEvent ( )

virtual

Implements edm::IEventProcessor.

Definition at line 2139 of file EventProcessor.cc.

References edm::EventPrincipal::clearEventPrincipal(), FDEBUG, edm::EventPrincipal::id(), edm::ProcessingController::lastOperationSucceeded(), edm::ProcessingController::requestedTransition(), edm::ProcessingController::setLastOperationSucceeded(), edm::ProcessingController::specifiedEventTransition(), ntuplemaker::status, summarizeEdmComparisonLogfiles::succeeded, and edm::EventPrincipal::time().

                                            {
     EventPrincipal *pep = input_->readEvent(principalCache_.lumiPrincipalPtr());
     FDEBUG(1) << "\treadEvent\n";
     assert(pep != 0);
 
     IOVSyncValue ts(pep->id(), pep->time());
     espController_->eventSetupForInstance(ts);
     EventSetup const& es = esp_->eventSetup();
     {
       typedef OccurrenceTraits<EventPrincipal, BranchActionBegin> Traits;
       ScheduleSignalSentry<Traits> sentry(actReg_.get(), pep, &es);
       schedule_->processOneOccurrence<Traits>(*pep, es);
       if(hasSubProcess()) {
         subProcess_->doEvent(*pep, ts);
       }
     }
 
     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 {
         status = looper_->doDuringLoop(*pep, esp_->eventSetup(), pc);
 
         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::readFile ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1887 of file EventProcessor.cc.

References FDEBUG.

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

                                 {
     FDEBUG(1) << " \treadFile\n";
     fb_ = input_->readFile();
     if(numberOfForkedChildren_ > 0) {
         fb_->setNotFastClonable(FileBlock::ParallelProcesses);
     }
   }

void edm::EventProcessor::respondToCloseInputFile ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1926 of file EventProcessor.cc.

References FDEBUG.

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

void edm::EventProcessor::respondToCloseOutputFiles ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1942 of file EventProcessor.cc.

References FDEBUG.

                                                  {
     if (fb_.get() != 0) {
       schedule_->respondToCloseOutputFiles(*fb_);
       if(hasSubProcess()) subProcess_->respondToCloseOutputFiles(*fb_);
     }
     FDEBUG(1) << "\trespondToCloseOutputFiles\n";
   }

void edm::EventProcessor::respondToOpenInputFile ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1918 of file EventProcessor.cc.

References FDEBUG.

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

void edm::EventProcessor::respondToOpenOutputFiles ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1934 of file EventProcessor.cc.

References FDEBUG.

                                                 {
     if (fb_.get() != 0) {
       schedule_->respondToOpenOutputFiles(*fb_);
       if(hasSubProcess()) subProcess_->respondToOpenOutputFiles(*fb_);
     }
     FDEBUG(1) << "\trespondToOpenOutputFiles\n";
   }

void edm::EventProcessor::rewind ( )

Definition at line 708 of file EventProcessor.cc.

References beginJob(), changeState(), input_, edm::event_processor::mCountComplete, edm::event_processor::mFinished, edm::event_processor::mInputRewind, edm::event_processor::mStopAsync, cmsPerfStripChart::operate(), serviceToken_, and edm::event_processor::StateSentry::succeeded().

                          {
     beginJob(); //make sure this was called
     changeState(mStopAsync);
     changeState(mInputRewind);
     {
       StateSentry toerror(this);
 
       //make the services available
       ServiceRegistry::Operate operate(serviceToken_);
 
       {
         input_->repeat();
         input_->rewind();
       }
       changeState(mCountComplete);
       toerror.succeeded();
     }
     changeState(mFinished);
   }

void edm::EventProcessor::rewindInput ( )

virtual

Implements edm::IEventProcessor.

Definition at line 1973 of file EventProcessor.cc.

References FDEBUG.

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

EventProcessor::StatusCode edm::EventProcessor::run	(	int	numberEventsToProcess,
		bool	repeatable = `true`
	)

Definition at line 729 of file EventProcessor.cc.

References runEventCount().

Referenced by Types.LuminosityBlockID::cppID().

                                                      {
     return runEventCount(numberEventsToProcess);
   }

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

inline

Definition at line 403 of file EventProcessor.h.

Referenced by Types.LuminosityBlockID::cppID().

                       {
     return run(-1, false);
   }

void edm::EventProcessor::runAsync ( )

Definition at line 1543 of file EventProcessor.cc.

References bk::beginJob(), edm::hlt::Exception, and edm::event_processor::mRunAsync.

Referenced by evf::FUEventProcessor::enableCommon(), and evf::FUEventProcessor::enableForkInEDM().

                                 {
     beginJob();
     {
       boost::mutex::scoped_lock sl(stop_lock_);
       if(id_set_ == true) {
           std::string err("runAsync called while async event loop already running\n");
           LogError("FwkJob") << err;
           throw cms::Exception("BadState") << err;
       }
 
       changeState(mRunAsync);
 
       stop_count_ = 0;
       last_rc_ = epSuccess; // forget the last value!
       event_loop_.reset(new boost::thread(boost::bind(EventProcessor::asyncRun, this)));
       boost::xtime timeout;
       boost::xtime_get(&timeout, boost::TIME_UTC);
       timeout.sec += 60; // 60 seconds to start!!!!
       if(starter_.timed_wait(sl, timeout) == false) {
           // yikes - the thread did not start
           throw cms::Exception("BadState")
             << "Async run thread did not start in 60 seconds\n";
       }
     }
   }

EventProcessor::StatusCode edm::EventProcessor::runCommon	(	bool	onlineStateTransitions,
		int	numberOfEventsToProcess
	)

private

Definition at line 1665 of file EventProcessor.cc.

References cms::Exception::addAdditionalInfo(), cms::Exception::alreadyPrinted(), edm::convertException::badAllocToEDM(), bk::beginJob(), trackerHits::c, edm::convertException::charPtrToEDM(), edm::errors::Configuration, statemachine::doNotHandleEmptyRunsAndLumis, alignCSCRings::e, cppFunctionSkipper::exception, edm::hlt::Exception, FDEBUG, dtDQMClient_cfg::fileMode, statemachine::FULLMERGE, statemachine::handleEmptyRuns, statemachine::handleEmptyRunsAndLumis, edm::errors::LogicError, edm::event_processor::mFinished, edm::event_processor::mInputExhausted, edm::event_processor::mRunCount, edm::event_processor::mShutdownSignal, statemachine::NOMERGE, cmsPerfStripChart::operate(), runEdmFileComparison::returnCode, alignCSCRings::s, edm::shutdown_flag, edm::event_processor::sShuttingDown, edm::event_processor::sStopping, edm::convertException::stdToEDM(), edm::convertException::stringToEDM(), edm::convertException::unknownToEDM(), and edm::usr2_lock.

                                                                                     {
 
     // Reusable event principal
     boost::shared_ptr<EventPrincipal> ep(new EventPrincipal(preg_, *processConfiguration_, historyAppender_.get()));
     principalCache_.insert(ep);
 
     beginJob(); //make sure this was called
 
     if(!onlineStateTransitions) changeState(mRunCount);
 
     StatusCode returnCode = epSuccess;
     stateMachineWasInErrorState_ = false;
 
     // make the services available
     ServiceRegistry::Operate operate(serviceToken_);
 
     if(machine_.get() == 0) {
 
       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";
       }
 
       machine_.reset(new statemachine::Machine(this,
                                                fileMode,
                                                emptyRunLumiMode));
 
       machine_->initiate();
     }
 
     try {
       try {
 
         InputSource::ItemType itemType;
 
         int iEvents = 0;
 
         while(true) {
 
           itemType = input_->nextItemType();
 
           FDEBUG(1) << "itemType = " << itemType << "\n";
 
           // These are used for asynchronous running only and
           // and are checking to see if stopAsync or shutdownAsync
           // were called from another thread.  In the future, we
           // may need to do something better than polling the state.
           // With the current code this is the simplest thing and
           // it should always work.  If the interaction between
           // threads becomes more complex this may cause problems.
           if(state_ == sStopping) {
             FDEBUG(1) << "In main processing loop, encountered sStopping state\n";
             forceLooperToEnd_ = true;
             machine_->process_event(statemachine::Stop());
             forceLooperToEnd_ = false;
             break;
           }
           else if(state_ == sShuttingDown) {
             FDEBUG(1) << "In main processing loop, encountered sShuttingDown state\n";
             forceLooperToEnd_ = true;
             machine_->process_event(statemachine::Stop());
             forceLooperToEnd_ = false;
             break;
           }
 
           // Look for a shutdown signal
           {
             boost::mutex::scoped_lock sl(usr2_lock);
             if(shutdown_flag) {
               changeState(mShutdownSignal);
               returnCode = epSignal;
               forceLooperToEnd_ = true;
               machine_->process_event(statemachine::Stop());
               forceLooperToEnd_ = false;
               break;
             }
           }
 
           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::IsEvent) {
             machine_->process_event(statemachine::Event());
             ++iEvents;
             if(numberOfEventsToProcess > 0 && iEvents >= numberOfEventsToProcess) {
               returnCode = epCountComplete;
               changeState(mInputExhausted);
               FDEBUG(1) << "Event count complete, pausing event loop\n";
               break;
             }
           }
           // 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()) {
             changeState(mInputExhausted);
             break;
           }
         }  // End of loop over state machine events
       } // Try block
       catch (cms::Exception& e) { throw; }
       catch(std::bad_alloc& bda) { convertException::badAllocToEDM(); }
       catch (std::exception& e) { convertException::stdToEDM(e); }
       catch(std::string& s) { convertException::stringToEDM(s); }
       catch(char const* c) { convertException::charPtrToEDM(c); }
       catch (...) { convertException::unknownToEDM(); }
     } // 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();
       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(!onlineStateTransitions) changeState(mFinished);
 
     if(stateMachineWasInErrorState_) {
       throw cms::Exception("BadState")
         << "The boost state machine in the EventProcessor exited after\n"
         << "entering the Error state.\n";
     }
 
     return returnCode;
   }

EventProcessor::StatusCode edm::EventProcessor::runEventCount ( int numberOfEventsToProcess )

virtual

Implements edm::IEventProcessor.

Definition at line 1652 of file EventProcessor.cc.

References runEdmFileComparison::returnCode, and edm::event_processor::StateSentry::succeeded().

Referenced by run().

                                                            {
 
     StateSentry toerror(this);
 
     bool onlineStateTransitions = false;
     StatusCode returnCode = runCommon(onlineStateTransitions, numberOfEventsToProcess);
 
     toerror.succeeded();
 
     return returnCode;
   }

EventProcessor::StatusCode edm::EventProcessor::runToCompletion ( bool onlineStateTransitions )

virtual

Implements edm::IEventProcessor.

Definition at line 1633 of file EventProcessor.cc.

References edm::hlt::Exception, edm::errors::LogicError, runEdmFileComparison::returnCode, and edm::event_processor::StateSentry::succeeded().

Referenced by asyncRun().

                                                              {
 
     StateSentry toerror(this);
 
     int numberOfEventsToProcess = -1;
     StatusCode returnCode = runCommon(onlineStateTransitions, numberOfEventsToProcess);
 
     if(machine_.get() != 0) {
       throw Exception(errors::LogicError)
         << "State machine not destroyed on exit from EventProcessor::runToCompletion\n"
         << "Please report this error to the Framework group\n";
     }
 
     toerror.succeeded();
 
     return returnCode;
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2191 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 2199 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 2195 of file EventProcessor.cc.

References python.rootplot.argparse::message.

                                                                  {
     exceptionMessageRuns_ = message;
   }

void edm::EventProcessor::setRunNumber ( RunNumber_t runNumber )

Definition at line 1402 of file EventProcessor.cc.

References bk::beginJob(), and edm::event_processor::mSetRun.

Referenced by evf::FUEventProcessor::enableCommon(), and evf::FUEventProcessor::enableForkInEDM().

                                                     {
     if(runNumber == 0) {
       runNumber = 1;
       LogWarning("Invalid Run")
         << "EventProcessor::setRunNumber was called with an invalid run number (0)\n"
         << "Run number was set to 1 instead\n";
     }
 
     // inside of beginJob there is a check to see if it has been called before
     beginJob();
     changeState(mSetRun);
 
     // interface not correct yet
     input_->setRunNumber(runNumber);
   }

void edm::EventProcessor::setupSignal ( )

private

bool edm::EventProcessor::shouldWeCloseOutput ( ) const

virtual

Implements edm::IEventProcessor.

Definition at line 1984 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 2185 of file EventProcessor.cc.

References FDEBUG.

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

EventProcessor::StatusCode edm::EventProcessor::shutdownAsync ( unsigned int timeout_secs = 60 * 2 )

Definition at line 1492 of file EventProcessor.cc.

References edm::event_processor::mFinished, and edm::event_processor::mShutdownAsync.

                                                                           {
     changeState(mShutdownAsync);
     StatusCode rc = waitForAsyncCompletion(secs);
     if(rc != epTimedOut) changeState(mFinished);
     else errorState();
     return rc;
   }

EventProcessor::StatusCode edm::EventProcessor::skip ( int numberToSkip )

Definition at line 734 of file EventProcessor.cc.

References beginJob(), changeState(), edm::IEventProcessor::epSuccess, input_, edm::event_processor::mCountComplete, edm::event_processor::mFinished, edm::event_processor::mSkip, cmsPerfStripChart::operate(), serviceToken_, and edm::event_processor::StateSentry::succeeded().

                                        {
     beginJob(); //make sure this was called
     changeState(mSkip);
     {
       StateSentry toerror(this);
 
       //make the services available
       ServiceRegistry::Operate operate(serviceToken_);
 
       {
         input_->skipEvents(numberToSkip);
       }
       changeState(mCountComplete);
       toerror.succeeded();
     }
     changeState(mFinished);
     return epSuccess;
   }

void edm::EventProcessor::startingNewLoop ( )

virtual

Implements edm::IEventProcessor.

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

char const * edm::EventProcessor::stateName ( event_processor::State s ) const

Definition at line 1383 of file EventProcessor.cc.

References alignCSCRings::s.

Referenced by evf::FUEventProcessor::doEndRunInEDM(), evf::FUEventProcessor::enabling(), evf::FWEPWrapper::publishConfigAndMonitorItems(), and evf::FWEPWrapper::stop().

                                                      {
     return stateNames[s];
   }

EventProcessor::StatusCode edm::EventProcessor::statusAsync ( ) const

Definition at line 1395 of file EventProcessor.cc.

Referenced by evf::FUEventProcessor::enableCommon(), and evf::FUEventProcessor::enableForkInEDM().

                                                              {
     // the thread will record exception/error status in the event processor
     // for us to look at and report here
     return last_rc_;
   }

EventProcessor::StatusCode edm::EventProcessor::stopAsync ( unsigned int timeout_secs = 60 * 2 )

Definition at line 1484 of file EventProcessor.cc.

References edm::event_processor::mFinished, and edm::event_processor::mStopAsync.

                                                                       {
     changeState(mStopAsync);
     StatusCode rc = waitForAsyncCompletion(secs);
     if(rc != epTimedOut) changeState(mFinished);
     else errorState();
     return rc;
   }

void edm::EventProcessor::terminateMachine ( )

private

Definition at line 2207 of file EventProcessor.cc.

References FDEBUG.

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

                                         {
     if(machine_.get() != 0) {
       if(!machine_->terminated()) {
         forceLooperToEnd_ = true;
         machine_->process_event(statemachine::Stop());
         forceLooperToEnd_ = false;
       }
       else {
         FDEBUG(1) << "EventProcess::terminateMachine  The state machine was already terminated \n";
       }
       if(machine_->terminated()) {
         FDEBUG(1) << "The state machine reports it has been terminated (3)\n";
       }
       machine_.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 1345 of file EventProcessor.cc.

Referenced by evf::FWEPWrapper::getTriggerReport(), evf::FWEPWrapper::monitoring(), and evf::FUEventProcessor::receivingAndMonitor().

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

Referenced by evf::FWEPWrapper::getTriggerReport(), evf::FWEPWrapper::monitoring(), and evf::FUEventProcessor::receivingAndMonitor().

                                           {
     return schedule_->totalEventsPassed();
   }

EventProcessor::StatusCode edm::EventProcessor::waitForAsyncCompletion ( unsigned int timeout_seconds )

private

Definition at line 1429 of file EventProcessor.cc.

                                                                      {
     bool rc = true;
     boost::xtime timeout;
     boost::xtime_get(&timeout, boost::TIME_UTC);
     timeout.sec += timeout_seconds;
 
     // make sure to include a timeout here so we don't wait forever
     // I suspect there are still timing issues with thread startup
     // and the setting of the various control variables (stop_count, id_set)
     {
       boost::mutex::scoped_lock sl(stop_lock_);
 
       // look here - if runAsync not active, just return the last return code
       if(stop_count_ < 0) return last_rc_;
 
       if(timeout_seconds == 0) {
         while(stop_count_ == 0) stopper_.wait(sl);
       } else {
         while(stop_count_ == 0 && (rc = stopper_.timed_wait(sl, timeout)) == true);
       }
 
       if(rc == false) {
           // timeout occurred
           // if(id_set_) pthread_kill(event_loop_id_, my_sig_num_);
           // this is a temporary hack until we get the input source
           // upgraded to allow blocking input sources to be unblocked
 
           // the next line is dangerous and causes all sorts of trouble
           if(id_set_) pthread_cancel(event_loop_id_);
 
           // we will not do anything yet
           LogWarning("timeout")
             << "An asynchronous request was made to shut down "
             << "the event loop "
             << "and the event loop did not shutdown after "
             << timeout_seconds << " seconds\n";
       } else {
           event_loop_->join();
           event_loop_.reset();
           id_set_ = false;
           stop_count_ = -1;
       }
     }
     return rc == false ? epTimedOut : last_rc_;
   }

EventProcessor::StatusCode edm::EventProcessor::waitTillDoneAsync ( unsigned int timeout_seconds = 0 )

Definition at line 1476 of file EventProcessor.cc.

References edm::event_processor::mCountComplete.

Referenced by evf::FWEPWrapper::stop().

                                                                    {
     StatusCode rc = waitForAsyncCompletion(timeout_value_secs);
     if(rc != epTimedOut) changeState(mCountComplete);
     else errorState();
     return rc;
   }

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

virtual

Implements edm::IEventProcessor.

Definition at line 2127 of file EventProcessor.cc.

References FDEBUG.

                                                                                 {
     schedule_->writeLumi(principalCache_.lumiPrincipal(phid, run, lumi));
     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 2115 of file EventProcessor.cc.

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

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

Friends And Related Function Documentation

friend class event_processor::StateSentry

friend

Definition at line 396 of file EventProcessor.h.

Member Data Documentation

boost::shared_ptr<ActionTable const> edm::EventProcessor::act_table_

private

Definition at line 354 of file EventProcessor.h.

Referenced by init().

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

private

Definition at line 348 of file EventProcessor.h.

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

bool edm::EventProcessor::alreadyHandlingException_

private

Definition at line 385 of file EventProcessor.h.

std::string edm::EventProcessor::emptyRunLumiMode_

private

Definition at line 381 of file EventProcessor.h.

Referenced by init().

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

private

Definition at line 353 of file EventProcessor.h.

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

boost::scoped_ptr<eventsetup::EventSetupsController> edm::EventProcessor::espController_

private

Definition at line 352 of file EventProcessor.h.

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

boost::shared_ptr<boost::thread> edm::EventProcessor::event_loop_

private

Definition at line 361 of file EventProcessor.h.

volatile pthread_t edm::EventProcessor::event_loop_id_

private

Definition at line 371 of file EventProcessor.h.

Referenced by asyncRun().

ExcludedDataMap edm::EventProcessor::eventSetupDataToExcludeFromPrefetching_

private

Definition at line 395 of file EventProcessor.h.

Referenced by init().

std::string edm::EventProcessor::exceptionMessageFiles_

private

Definition at line 382 of file EventProcessor.h.

std::string edm::EventProcessor::exceptionMessageLumis_

private

Definition at line 384 of file EventProcessor.h.

std::string edm::EventProcessor::exceptionMessageRuns_

private

Definition at line 383 of file EventProcessor.h.

boost::shared_ptr<FileBlock> edm::EventProcessor::fb_

private

Definition at line 373 of file EventProcessor.h.

std::string edm::EventProcessor::fileMode_

private

Definition at line 380 of file EventProcessor.h.

Referenced by init().

bool edm::EventProcessor::forceESCacheClearOnNewRun_

private

Definition at line 388 of file EventProcessor.h.

Referenced by init().

bool edm::EventProcessor::forceLooperToEnd_

private

Definition at line 386 of file EventProcessor.h.

boost::scoped_ptr<HistoryAppender> edm::EventProcessor::historyAppender_

private

Definition at line 358 of file EventProcessor.h.

volatile bool edm::EventProcessor::id_set_

private

Definition at line 370 of file EventProcessor.h.

Referenced by asyncRun().

boost::shared_ptr<InputSource> edm::EventProcessor::input_

private

Definition at line 351 of file EventProcessor.h.

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

std::string edm::EventProcessor::last_error_text_

private

Definition at line 369 of file EventProcessor.h.

Referenced by asyncRun().

volatile Status edm::EventProcessor::last_rc_

private

Definition at line 368 of file EventProcessor.h.

Referenced by asyncRun().

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

private

Definition at line 374 of file EventProcessor.h.

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

bool edm::EventProcessor::looperBeginJobRun_

private

Definition at line 387 of file EventProcessor.h.

std::auto_ptr<statemachine::Machine> edm::EventProcessor::machine_

private

Definition at line 376 of file EventProcessor.h.

int edm::EventProcessor::my_sig_num_

private

Definition at line 372 of file EventProcessor.h.

int edm::EventProcessor::numberOfForkedChildren_

private

Definition at line 390 of file EventProcessor.h.

Referenced by init().

unsigned int edm::EventProcessor::numberOfSequentialEventsPerChild_

private

Definition at line 391 of file EventProcessor.h.

Referenced by init().

ActivityRegistry::PostProcessEvent edm::EventProcessor::postProcessEventSignal_

private

Definition at line 347 of file EventProcessor.h.

Referenced by connectSigs(), and postProcessEventSignal().

boost::shared_ptr<SignallingProductRegistry> edm::EventProcessor::preg_

private

Definition at line 349 of file EventProcessor.h.

Referenced by init().

ActivityRegistry::PreProcessEvent edm::EventProcessor::preProcessEventSignal_

private

Definition at line 346 of file EventProcessor.h.

Referenced by connectSigs(), and preProcessEventSignal().

PrincipalCache edm::EventProcessor::principalCache_

private

Definition at line 377 of file EventProcessor.h.

Referenced by init().

boost::shared_ptr<ProcessConfiguration> edm::EventProcessor::processConfiguration_

private

Definition at line 355 of file EventProcessor.h.

Referenced by init().

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

private

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

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

bool edm::EventProcessor::setCpuAffinity_

private

Definition at line 392 of file EventProcessor.h.

Referenced by init().

bool edm::EventProcessor::shouldWeStop_

private

Definition at line 378 of file EventProcessor.h.

boost::condition edm::EventProcessor::starter_

private

Definition at line 366 of file EventProcessor.h.

Referenced by asyncRun().

volatile event_processor::State edm::EventProcessor::state_

private

Definition at line 360 of file EventProcessor.h.

Referenced by beginJob().

boost::mutex edm::EventProcessor::state_lock_

private

Definition at line 363 of file EventProcessor.h.

bool edm::EventProcessor::stateMachineWasInErrorState_

private

Definition at line 379 of file EventProcessor.h.

volatile int edm::EventProcessor::stop_count_

private

Definition at line 367 of file EventProcessor.h.

Referenced by asyncRun().

boost::mutex edm::EventProcessor::stop_lock_

private

Definition at line 364 of file EventProcessor.h.

Referenced by asyncRun().

boost::condition edm::EventProcessor::stopper_

private

Definition at line 365 of file EventProcessor.h.

Referenced by asyncRun().

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

private

Definition at line 357 of file EventProcessor.h.

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

Public Member Functions

Private Types

Private Member Functions

Static 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