EventProcessor Class Reference

provides the core event processing loop. More...

#include <EventProcessor.h>

Inheritance diagram for EventProcessor:

Collaboration diagram for EventProcessor:

Classes
class	StoppedBySignalException
	Exception thrown when execution is stopped by a signal. More...

Public Member Functions
	EventProcessor ()
	Constructor.
virtual	~EventProcessor ()
	Destructor.
void	process (const PathPtr &startPath, long maxEvent=0)
	Processes the full module chain, starting with the first module in the given path.
void	setProfileModuleName (const std::string &name)
	Set the name of the module we want to profile.

Static Public Member Functions
static void	writeToStdErr (const char msg[])
	async-safe method to write something to STDERR.
static void	installSignalHandler (int sig, void(*fn)(int))
	Install a signal handler 'fn' for given signal.
static void	installMainSignalHandlers (void(*fn)(int)=nullptr)
	Install signal handler for INT, TERM and QUIT signals.

Protected Member Functions
void	processInitialize (const ModulePtrList &modulePathList, bool setEventInfo=true)
	Initializes the modules.
void	processCore (const PathPtr &startPath, const ModulePtrList &modulePathList, long maxEvent=0, bool isInputProcess=true)
	Processes the full module chain consisting of an arbitrary number of connected paths, starting with the first module in the specified path.
bool	processEvent (PathIterator moduleIter, bool skipMasterModule)
	Calls event() functions on all modules for the current event.
void	callEvent (Module *module)
	Calls event() on one single module, setting up logging and statistics as needed.
void	processTerminate (const ModulePtrList &modulePathList)
	Terminates the modules.
void	processBeginRun (bool skipDB=false)
	Calls the begin run methods of all modules.
void	processEndRun ()
	Calls the end run methods of all modules.
long	getMaximumEventNumber (long maxEvent) const
	Calculate the maximum event number out of the argument from command line and the environment.

Protected Attributes
const Module *	m_master
	The master module that determines the experiment/run/event number.
ModulePtrList	m_moduleList
	List of all modules in order initialized.
std::string	m_profileModuleName
	Name of the module which should be profiled, empty if no profiling is requested.
Module *	m_profileModule = nullptr
	Adress of the module which we want to profile, nullptr if no profiling is requested.
StoreObjPtr< EventMetaData >	m_eventMetaDataPtr
	EventMetaData is used by processEvent()/processCore().
EventMetaData	m_previousEventMetaData
	Stores state of EventMetaData before it was last changed.
StoreObjPtr< ProcessStatistics >	m_processStatisticsPtr
	Also used in a number of places.
bool	m_inRun
	Are we currently in a run? If yes, processEndRun() needs to do something.
double	m_lastMetadataUpdate
	Time in seconds of last call for metadata update in event loop.
double	m_metadataUpdateInterval
	Minimal time difference in seconds for metadata updates in event loop.
bool	m_steerRootInputModuleOn = false
	True if the SteerRootInputModule is in charge for event processing.

Detailed Description

provides the core event processing loop.

Definition at line 29 of file EventProcessor.h.

Constructor & Destructor Documentation

EventProcessor()

EventProcessor

(

)

Constructor.

Definition at line 85 of file EventProcessor.cc.

                               : m_master(nullptr), m_processStatisticsPtr("", DataStore::c_Persistent),
  m_inRun(false), m_lastMetadataUpdate(0), m_metadataUpdateInterval(1.0)
{
 
}

Member Function Documentation

callEvent()

void callEvent ( Module *  module )

protected

Calls event() on one single module, setting up logging and statistics as needed.

Parameters

module

Module to call the event() function

Definition at line 226 of file EventProcessor.cc.

{
  LogSystem& logSystem = LogSystem::Instance();
  const bool collectStats = !Environment::Instance().getNoStats();
  // set up logging
  logSystem.updateModule(&(module->getLogConfig()), module->getName());
  // set up statistics is requested
  if (collectStats) m_processStatisticsPtr->startModule();
  // call module
  CALL_MODULE(module, event);
  // stop timing
  if (collectStats) m_processStatisticsPtr->stopModule(module, ModuleStatistics::c_Event);
  // reset logging
  logSystem.updateModule(nullptr);
};

getMaximumEventNumber()

long getMaximumEventNumber ( long  maxEvent ) const

protected

Calculate the maximum event number out of the argument from command line and the environment.

Definition at line 113 of file EventProcessor.cc.

{
  //Check whether the number of events was set via command line argument
  unsigned int numEventsArgument = Environment::Instance().getNumberEventsOverride();
  if ((numEventsArgument > 0) && ((maxEvent == 0) || (maxEvent > numEventsArgument))) {
    return numEventsArgument;
  }
  return maxEvent;
}

installMainSignalHandlers()

void installMainSignalHandlers ( void(*)(int)  fn = nullptr )

static

Install signal handler for INT, TERM and QUIT signals.

If argument is NULL, EventProcessor's own signal handler will be installed.

Definition at line 312 of file EventProcessor.cc.

{
  if (!fn)
    fn = signalHandler;
  installSignalHandler(SIGINT, fn);
  installSignalHandler(SIGTERM, fn);
  installSignalHandler(SIGQUIT, fn);
}

installSignalHandler()

void installSignalHandler ( int  sig, void(*)(int)  fn  )

static

Install a signal handler 'fn' for given signal.

Definition at line 297 of file EventProcessor.cc.

{
  struct sigaction s;
  memset(&s, '\0', sizeof(s));
 
  s.sa_handler = fn;
  sigemptyset(&s.sa_mask);
  if (sig == SIGCHLD)
    s.sa_flags |= SA_NOCLDSTOP; //don't produce signal when children are stopped
 
  if (sigaction(sig, &s, nullptr) != 0) {
    B2FATAL("Cannot setup signal handler for signal " << sig);
  }
}

process()

void process	(	const PathPtr &	startPath,
		long	maxEvent = 0
	)

Processes the full module chain, starting with the first module in the given path.

Processes all events for the given run number and for events from 0 to maxEvent. If maxEvent is smaller or equal 0 the maximum number check is disabled and all events are processed. If runNumber is smaller than 0, the run number has to be set externally by a module and not the given number is used.

Parameters

startPath	The processing starts with the first module of this path.
maxEvent	Optional: The maximum number of events that will be processed. If the number is smaller or equal 0, all events will be processed.

Definition at line 123 of file EventProcessor.cc.

{
  maxEvent = getMaximumEventNumber(maxEvent);
  // Make sure the NumberEventsOverride reflects the actual number if
  // process(path, N) was used instead of -n and that it's reset to what it was
  // after we're done with processing()
  NumberEventsOverrideGuard numberOfEventsOverrideGuard(maxEvent);
 
  //Get list of modules which could be executed during the data processing.
  ModulePtrList moduleList = startPath->buildModulePathList();
 
  //Find the adress of the module we want to profile
  if (!m_profileModuleName.empty()) {
    for (const auto& module : moduleList) {
      if (module->getName() == m_profileModuleName) {
        m_profileModule = module.get();
        break;
      }
    }
    if (!m_profileModule)
      B2FATAL("Module profiling was requested via --profile, but no module '" << m_profileModuleName << "' was found!");
  }
 
  //Initialize modules
  processInitialize(moduleList);
 
  // SteerRootInputModule might have changed the number of events to be processed
  for (const auto& module : moduleList) {
    if (module->getType() == "SteerRootInput") {
      if (maxEvent != Environment::Instance().getNumberEventsOverride()) {
        B2INFO("Module 'SteerRootInputModule' is controlling the number of processed events.");
        maxEvent = Environment::Instance().getNumberEventsOverride();
        m_steerRootInputModuleOn = true;
      }
      break;
    }
  }
 
  //do we want to visualize DataStore input/ouput?
  if (Environment::Instance().getVisualizeDataFlow()) {
    DataFlowVisualization v(&DataStore::Instance().getDependencyMap());
    v.visualizePath("dataflow.dot", *startPath);
  }
 
  //Don't start processing in case of no master module
  if (!m_master) {
    B2ERROR("There is no module that provides event and run numbers (EventMetaData). You must add either the EventInfoSetter or an input module (e.g. RootInput) to the beginning of your path.");
  }
 
  //Check if errors appeared. If yes, don't start the event processing.
  int numLogError = LogSystem::Instance().getMessageCounter(LogConfig::c_Error);
  if ((numLogError == 0) && m_master) {
    installMainSignalHandlers();
    try {
      processCore(startPath, moduleList, maxEvent); //Do the event processing
    } catch (StoppedBySignalException& e) {
      if (e.signal != SIGINT) {
        // close all open ROOT files, ROOT's exit handler will crash otherwise
        gROOT->GetListOfFiles()->Delete();
 
        B2FATAL(e.what());
      }
      //in case of SIGINT, we move on to processTerminate() to shut down safely
    } catch (...) {
      if (m_eventMetaDataPtr)
        B2ERROR("Exception occured in exp/run/evt: "
                << m_eventMetaDataPtr->getExperiment() << " / "
                << m_eventMetaDataPtr->getRun() << " / "
                << m_eventMetaDataPtr->getEvent());
      throw;
    }
 
  } else {
    B2FATAL(numLogError << " ERROR(S) occurred! The processing of events will not be started.");
  }
 
  //Terminate modules
  processTerminate(moduleList);
 
  LogSystem::Instance().printErrorSummary();
 
  if (gSignalReceived == SIGINT) {
    const auto msg = R"(Processing aborted via SIGINT, terminating.
    Output files have been closed safely and should be readable. However
    processing was NOT COMPLETE. The output files do contain only events
    processed until this point.)";
    if (m_eventMetaDataPtr)
      B2ERROR(msg
              << LogVar("last_experiment", m_eventMetaDataPtr->getExperiment())
              << LogVar("last_run", m_eventMetaDataPtr->getRun())
              << LogVar("last_event", m_eventMetaDataPtr->getEvent()));
    else
      B2ERROR(msg);
    installSignalHandler(SIGINT, SIG_DFL);
    raise(SIGINT);
  }
}

processBeginRun()

void processBeginRun ( bool  skipDB = false )

protected

Calls the begin run methods of all modules.

Loops over all module instances specified in a list and calls their beginRun() method. Please note: the beginRun() method of the module which triggered the beginRun() loop will also be called.

Definition at line 467 of file EventProcessor.cc.

{
  MetadataService::Instance().addBasf2Status("beginning run");
 
  m_inRun = true;
  auto dbsession = Database::Instance().createScopedUpdateSession(); // cppcheck-suppress unreadVariable
 
  LogSystem& logSystem = LogSystem::Instance();
  m_processStatisticsPtr->startGlobal();
 
  if (!skipDB) DBStore::Instance().update();
 
  //initialize random generator for end run
  RandomNumbers::initializeBeginRun();
 
  for (const ModulePtr& modPtr : m_moduleList) {
    Module* module = modPtr.get();
 
    //Set the module dependent log level
    logSystem.updateModule(&(module->getLogConfig()), module->getName());
 
    //Do beginRun() call
    m_processStatisticsPtr->startModule();
    CALL_MODULE(module, beginRun);
    m_processStatisticsPtr->stopModule(module, ModuleStatistics::c_BeginRun);
 
    //Set the global log level
    logSystem.updateModule(nullptr);
  }
 
  m_processStatisticsPtr->stopGlobal(ModuleStatistics::c_BeginRun);
}

processCore()

void processCore ( const PathPtr &  startPath, const ModulePtrList &  modulePathList, long  maxEvent = 0, bool  isInputProcess = true  )

protected

Processes the full module chain consisting of an arbitrary number of connected paths, starting with the first module in the specified path.

Parameters

startPath	The processing starts with the first module of this path.
modulePathList	A list of all modules which could be executed during the data processing (used for calling the beginRun() and endRun() method).
maxEvent	The maximum number of events that will be processed. If the number is smaller or equal 0, all events are processed.
isInputProcess	true when this is either the only or the input process

Definition at line 406 of file EventProcessor.cc.

{
  DataStore::Instance().setInitializeActive(false);
  m_moduleList = modulePathList;
  //Remember the previous event meta data, and identify end of data meta data
  m_previousEventMetaData.setEndOfData(); //invalid start state
 
  const bool collectStats = !Environment::Instance().getNoStats();
 
  //Loop over the events
  long currEvent = 0;
  bool endProcess = false;
  while (!endProcess) {
    if (collectStats)
      m_processStatisticsPtr->startGlobal();
 
    PathIterator moduleIter(startPath);
    endProcess = processEvent(moduleIter, isInputProcess && currEvent == 0);
 
    //Delete event related data in DataStore
    DataStore::Instance().invalidateData(DataStore::c_Event);
 
    currEvent++;
    if ((maxEvent > 0) && (currEvent >= maxEvent)) endProcess = true;
    if (collectStats)
      m_processStatisticsPtr->stopGlobal(ModuleStatistics::c_Event);
  } //end event loop
 
  //End last run
  m_eventMetaDataPtr.create();
  processEndRun();
}

processEndRun()

void processEndRun ( )

protected

Calls the end run methods of all modules.

Loops over all module instances specified in a list and calls their endRun() method. Please note: the endRun() method of the module which triggered the endRun() loop will also be called.

Definition at line 501 of file EventProcessor.cc.

{
  MetadataService::Instance().addBasf2Status("ending run");
 
  if (!m_inRun)
    return;
  m_inRun = false;
 
  LogSystem& logSystem = LogSystem::Instance();
  m_processStatisticsPtr->startGlobal();
 
  const EventMetaData newEventMetaData = *m_eventMetaDataPtr;
  *m_eventMetaDataPtr = m_previousEventMetaData;
 
  //initialize random generator for end run
  RandomNumbers::initializeEndRun();
 
  for (const ModulePtr& modPtr : m_moduleList) {
    Module* module = modPtr.get();
 
    //Set the module dependent log level
    logSystem.updateModule(&(module->getLogConfig()), module->getName());
 
    //Do endRun() call
    m_processStatisticsPtr->startModule();
    CALL_MODULE(module, endRun);
    m_processStatisticsPtr->stopModule(module, ModuleStatistics::c_EndRun);
 
    //Set the global log level
    logSystem.updateModule(nullptr);
  }
  *m_eventMetaDataPtr = newEventMetaData;
 
  m_processStatisticsPtr->stopGlobal(ModuleStatistics::c_EndRun);
}

processEvent()

bool processEvent ( PathIterator  moduleIter, bool  skipMasterModule  )

protected

Calls event() functions on all modules for the current event.

Used by processCore.

Parameters

moduleIter	iterator of the path containing all the modules
skipMasterModule	skip the execution of the master module, presumably because this is the first event and it's already been done in initialize()

Returns

true if execution should stop.

Definition at line 321 of file EventProcessor.cc.

{
  double time = Utils::getClock() / Unit::s;
  if (time > m_lastMetadataUpdate + m_metadataUpdateInterval) {
    MetadataService::Instance().addBasf2Status("running event loop");
    m_lastMetadataUpdate = time;
  }
 
  const bool collectStats = !Environment::Instance().getNoStats();
 
  while (!moduleIter.isDone()) {
    Module* module = moduleIter.get();
 
    // run the module ... unless we don't want to
    if (!(skipMasterModule && module == m_master)) {
      callEvent(module);
    }
 
    //Check for end of data
    if ((m_eventMetaDataPtr && (m_eventMetaDataPtr->isEndOfData())) ||
        ((module == m_master) && !m_eventMetaDataPtr)) {
      if ((module != m_master) && !m_steerRootInputModuleOn) {
        B2WARNING("Event processing stopped by module '" << module->getName() <<
                  "', which is not in control of event processing (does not provide EventMetaData)");
      }
      return true;
    }
 
    //Handle EventMetaData changes by master module
    if (module == m_master) {
 
      //initialize random number state for the event
      RandomNumbers::initializeEvent();
 
      //Check for a change of the run
      if ((m_eventMetaDataPtr->getExperiment() != m_previousEventMetaData.getExperiment()) ||
          (m_eventMetaDataPtr->getRun() != m_previousEventMetaData.getRun())) {
 
        if (collectStats)
          m_processStatisticsPtr->suspendGlobal();
 
        processEndRun();
        processBeginRun(skipMasterModule);
 
        if (collectStats)
          m_processStatisticsPtr->resumeGlobal();
      }
 
      m_previousEventMetaData = *m_eventMetaDataPtr;
 
      //make sure we use the event dependent generator again
      RandomNumbers::useEventDependent();
 
      DBStore::Instance().updateEvent();
 
    } else {
      //Check for a second master module. Cannot do this if we skipped the
      //master module as the EventMetaData is probably set before we call this
      //function
      if (!skipMasterModule && m_eventMetaDataPtr &&
          (*m_eventMetaDataPtr != m_previousEventMetaData)) {
        B2FATAL("Two modules setting EventMetaData were discovered: " << m_master->getName() << " and " << module->getName());
      }
    }
 
    if (gSignalReceived != 0) {
      throw StoppedBySignalException(gSignalReceived);
    }
 
    //Check for the module conditions, evaluate them and if one is true switch to the new path
    if (module->evalCondition()) {
      PathPtr condPath = module->getConditionPath();
      //continue with parent Path after condition path is executed?
      if (module->getAfterConditionPath() == Module::EAfterConditionPath::c_Continue) {
        moduleIter = PathIterator(condPath, moduleIter);
      } else {
        moduleIter = PathIterator(condPath);
      }
    } else {
      moduleIter.next();
    }
  } //end module loop
  return false;
}

processInitialize()

void processInitialize ( const ModulePtrList &  modulePathList, bool  setEventInfo = true  )

protected

Initializes the modules.

Loops over all module instances specified in a list and calls their initialize() method.

Parameters

modulePathList	A list of all modules which could be executed during the data processing.
setEventInfo	if true the first event call of the master module will be called immidiately to load the event info right away so that it's available for subsequent modules

Definition at line 242 of file EventProcessor.cc.

{
  LogSystem& logSystem = LogSystem::Instance();
  auto dbsession = Database::Instance().createScopedUpdateSession();
 
  m_processStatisticsPtr.registerInDataStore();
  //TODO I might want to overwrite it in initialize (e.g. if read from file)
  //     For parallel processing or subevents, I don't want that, though.
  //     Maybe make this a function argument?
  if (!m_processStatisticsPtr)
    m_processStatisticsPtr.create();
  m_processStatisticsPtr->startGlobal();
 
  MetadataService::Instance().addBasf2Status("initializing");
 
  for (const ModulePtr& modPtr : modulePathList) {
    Module* module = modPtr.get();
 
    if (module->hasUnsetForcedParams()) {
      //error message was printed by module
      continue;
    }
 
    //Set the module dependent log level
    logSystem.updateModule(&(module->getLogConfig()), module->getName());
    DataStore::Instance().getDependencyMap().setModule(*module);
 
    //Do initialization
    m_processStatisticsPtr->initModule(module);
    m_processStatisticsPtr->startModule();
    CALL_MODULE(module, initialize);
    m_processStatisticsPtr->stopModule(module, ModuleStatistics::c_Init);
 
    //Set the global log level
    logSystem.updateModule(nullptr);
 
    //Check whether this is the master module
    if (!m_master && DataStore::Instance().getEntry(m_eventMetaDataPtr) != nullptr) {
      B2DEBUG(100, "Found module providing EventMetaData: " << module->getName());
      m_master = module;
      if (setEventInfo) {
        callEvent(module);
        // update Database payloads: we now have valid event meta data unless
        // we don't process any events
        if (m_eventMetaDataPtr) DBStore::Instance().update();
      }
    }
 
    if (gSignalReceived != 0) {
      throw StoppedBySignalException(gSignalReceived);
    }
  }
  m_processStatisticsPtr->stopGlobal(ModuleStatistics::c_Init);
}

processTerminate()

void processTerminate ( const ModulePtrList &  modulePathList )

protected

Terminates the modules.

Loops over all module instances in reverse order specified in a list and calls their terminate() method.

Parameters

modulePathList

A list of all modules which could be executed during the data processing.

Definition at line 440 of file EventProcessor.cc.

{
  MetadataService::Instance().addBasf2Status("terminating");
 
  LogSystem& logSystem = LogSystem::Instance();
  ModulePtrList::const_reverse_iterator listIter;
  m_processStatisticsPtr->startGlobal();
 
  for (listIter = modulePathList.rbegin(); listIter != modulePathList.rend(); ++listIter) {
    Module* module = listIter->get();
 
    //Set the module dependent log level
    logSystem.updateModule(&(module->getLogConfig()), module->getName());
 
    //Do termination
    m_processStatisticsPtr->startModule();
    CALL_MODULE(module, terminate);
    m_processStatisticsPtr->stopModule(module, ModuleStatistics::c_Term);
 
    //Set the global log level
    logSystem.updateModule(nullptr);
  }
 
  m_processStatisticsPtr->stopGlobal(ModuleStatistics::c_Term);
}

setProfileModuleName()

void setProfileModuleName ( const std::string &  name )

inline

Set the name of the module we want to profile.

Parameters

name	Name of the module as returned by getName()

Definition at line 57 of file EventProcessor.h.

{ m_profileModuleName = name; }

writeToStdErr()

void writeToStdErr ( const char  msg[] )

static

async-safe method to write something to STDERR.

Definition at line 72 of file EventProcessor.cc.

{
  //signal handlers are called asynchronously, making many standard functions (including output) dangerous
  //write() is, however, safe, so we'll use that to write to stderr.
 
  //strlen() not explicitly in safe list, but doesn't have any error handling routines that might alter global state
  const int len = strlen(msg);
 
  int rc = write(STDERR_FILENO, msg, len);
  (void) rc; //ignore return value (there's nothing we can do about a failed write)
 
}

Member Data Documentation

m_eventMetaDataPtr

StoreObjPtr<EventMetaData> m_eventMetaDataPtr

protected

EventMetaData is used by processEvent()/processCore().

Definition at line 163 of file EventProcessor.h.

m_inRun

bool m_inRun

protected

Are we currently in a run? If yes, processEndRun() needs to do something.

Definition at line 172 of file EventProcessor.h.

m_lastMetadataUpdate

double m_lastMetadataUpdate

protected

Time in seconds of last call for metadata update in event loop.

Definition at line 175 of file EventProcessor.h.

m_master

const Module* m_master

protected

The master module that determines the experiment/run/event number.

Definition at line 153 of file EventProcessor.h.

m_metadataUpdateInterval

double m_metadataUpdateInterval

protected

Minimal time difference in seconds for metadata updates in event loop.

Definition at line 178 of file EventProcessor.h.

m_moduleList

ModulePtrList m_moduleList

protected

List of all modules in order initialized.

Definition at line 154 of file EventProcessor.h.

m_previousEventMetaData

EventMetaData m_previousEventMetaData

protected

Stores state of EventMetaData before it was last changed.

Useful since processEndRun() needs info about which run it needs to end.

Definition at line 166 of file EventProcessor.h.

m_processStatisticsPtr

StoreObjPtr<ProcessStatistics> m_processStatisticsPtr

protected

Also used in a number of places.

Definition at line 169 of file EventProcessor.h.

m_profileModule

Module* m_profileModule = nullptr

protected

Adress of the module which we want to profile, nullptr if no profiling is requested.

Definition at line 160 of file EventProcessor.h.

m_profileModuleName

std::string m_profileModuleName

protected

Name of the module which should be profiled, empty if no profiling is requested.

Definition at line 157 of file EventProcessor.h.

m_steerRootInputModuleOn

bool m_steerRootInputModuleOn = false

protected

True if the SteerRootInputModule is in charge for event processing.

Definition at line 181 of file EventProcessor.h.

---

The documentation for this class was generated from the following files:

• framework/core/include/EventProcessor.h
• framework/core/src/EventProcessor.cc