pEventProcessor Class Reference

This class provides the core event processing loop for parallel processing. More...

#include <pEventProcessor.h>

Inheritance diagram for pEventProcessor:

Public Member Functions
	pEventProcessor ()
	Constructor.
virtual	~pEventProcessor ()
	Destructor.
void	process (const PathPtr &spath, long maxEvent)
	Processes the full module chain, starting with the first module in the given path.
void	gotSigINT ()
	signal handler for Ctrl+C (async-safe)
void	killRingBuffers ()
	signal handler (async-safe)
void	cleanup ()
	clean up IPC resources (should only be called in one process).
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
	Address 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< EventExtraInfo >	m_eventExtraInfo
	event extra info object pointer
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.

Private Member Functions
void	analyzePath (const PathPtr &path)
	Analyze given path.
void	preparePaths ()
	Adds internal modules to paths, prepare RingBuffers.
RingBuffer *	connectViaRingBuffer (const char *name, const PathPtr &a, PathPtr &b)
	Create RingBuffer with name from given environment variable, add Tx and Rx modules to a and b.
void	dump_modules (const std::string &, const ModulePtrList &)
	Dump module names in the ModulePtrList.
void	clearFileList ()
	TFiles are stored in a global list and cleaned up by root since this will happen in all forked processes, these will be corrupted if we don't clean the list!

Static Private Member Functions
static ModulePtrList	getModulesWithFlag (const ModulePtrList &modules, Module::EModulePropFlags flag)
	Return only modules which have the given Module flag set.
static ModulePtrList	getModulesWithoutFlag (const ModulePtrList &modules, Module::EModulePropFlags flag)
	Return only modules which do not have the given Module flag set.
static void	prependModulesIfNotPresent (ModulePtrList *modules, const ModulePtrList &prependModules)
	Prepend given 'prependModules' to 'modules', if they're not already present.

Private Attributes
std::unique_ptr< ProcHandler >	m_procHandler
	handler to fork and manage processes.
PathPtr	m_inputPath
	Input path.
PathPtr	m_mainPath
	Main (parallel section) path.
PathPtr	m_outputPath
	Output path.
RingBuffer *	m_rbin = nullptr
	input RingBuffer
RingBuffer *	m_rbout = nullptr
	output RingBuffer
ModulePtr	m_histoman
	Pointer to HistoManagerModule, or nullptr if not found.

Detailed Description

This class provides the core event processing loop for parallel processing.

Definition at line 28 of file pEventProcessor.h.

Constructor & Destructor Documentation

pEventProcessor()

pEventProcessor

(

)

Constructor.

Definition at line 62 of file pEventProcessor.cc.

                                 : EventProcessor(),
  m_histoman(nullptr)
{
  g_pEventProcessor = this;
}

~pEventProcessor()

~pEventProcessor ( )

virtual

Destructor.

Definition at line 69 of file pEventProcessor.cc.

{
  cleanup();
  g_pEventProcessor = nullptr;
}

Member Function Documentation

analyzePath()

void analyzePath ( const PathPtr &  path )

private

Analyze given path.

Fills m_*path objects.

Definition at line 296 of file pEventProcessor.cc.

{
  //modules that can be parallelised, but should not go into a parallel section by themselves
  std::set<std::string> uselessParallelModules({"HistoManager", "Gearbox", "Geometry"});
 
  PathPtr inpath(new Path);
  PathPtr mainpath(new Path);
  PathPtr outpath(new Path);
 
  int stage = 0; //0: in, 1: event/main, 2: out
  for (const ModulePtr& module : path->getModules()) {
    bool hasParallelFlag = module->hasProperties(Module::c_ParallelProcessingCertified);
    //entire conditional path must also be compatible
    if (hasParallelFlag and module->hasCondition()) {
      for (const auto& conditionPath : module->getAllConditionPaths()) {
        if (!ModuleManager::allModulesHaveFlag(conditionPath->getModules(), Module::c_ParallelProcessingCertified)) {
          hasParallelFlag = false;
        }
      }
    }
 
    //update stage?
    if ((stage == 0 and hasParallelFlag) or (stage == 1 and !hasParallelFlag)) {
      stage++;
 
      if (stage == 2) {
        bool path_is_useful = false;
        for (const auto& parallelModule : mainpath->getModules()) {
          if (uselessParallelModules.count(parallelModule->getType()) == 0) {
            path_is_useful = true;
            break;
          }
        }
        if (not path_is_useful) {
          //merge mainpath back into input path
          inpath->addPath(mainpath);
          mainpath.reset(new Path);
          //and search for further parallel sections
          stage = 0;
        }
      }
    }
 
    if (stage == 0) { //fill input path
      inpath->addModule(module);
 
      if (module->hasProperties(Module::c_HistogramManager)) {
        // Initialize histogram manager if found in the path
        m_histoman = module;
 
        //add histoman to other paths
        mainpath->addModule(m_histoman);
        outpath->addModule(m_histoman);
      }
    }
 
    if (stage == 1)
      mainpath->addModule(module);
    if (stage == 2)
      outpath->addModule(module);
  }
 
  bool createAllPaths = false; //usually we might not need e.g. an output path
  for (const ModulePtr& module : path->getModules()) {
    if (module->hasProperties(Module::c_TerminateInAllProcesses))
      createAllPaths = true; //ensure there are all kinds of processes
  }
 
  //if main path is empty, createAllPaths doesn't really matter, since we'll fall back to single-core processing
  if (!mainpath->isEmpty())
    m_mainPath = mainpath;
  if (createAllPaths or !inpath->isEmpty())
    m_inputPath = inpath;
  if (createAllPaths or !outpath->isEmpty())
    m_outputPath = outpath;
}

callEvent()

void callEvent ( Module *  module )

protectedinherited

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

cleanup()

void cleanup

(

)

clean up IPC resources (should only be called in one process).

Definition at line 75 of file pEventProcessor.cc.

{
  if (!ProcHandler::parallelProcessingUsed() or ProcHandler::isOutputProcess()) {
    delete m_rbin;
    m_rbin = nullptr;
    delete m_rbout;
    m_rbout = nullptr;
  }
}

clearFileList()

void clearFileList ( )

private

TFiles are stored in a global list and cleaned up by root since this will happen in all forked processes, these will be corrupted if we don't clean the list!

needs to be called at the end of every process.

Definition at line 99 of file pEventProcessor.cc.

{
  //B2WARNING("list of files: " << gROOT->GetListOfFiles()->GetEntries());
  //clear list, but don't actually delete the objects
  gROOT->GetListOfFiles()->Clear("nodelete");
}

connectViaRingBuffer()

RingBuffer * connectViaRingBuffer ( const char *  name, const PathPtr &  a, PathPtr &  b  )

private

Create RingBuffer with name from given environment variable, add Tx and Rx modules to a and b.

Definition at line 373 of file pEventProcessor.cc.

{
  //create ringbuffers and add rx/tx where needed
  const char* inrbname = getenv(name);
  RingBuffer* rbuf;
  if (inrbname == nullptr) {
    rbuf = new RingBuffer();
  } else {
    string rbname(inrbname + to_string(0)); //currently at most one input, one output buffer
    rbuf = new RingBuffer(rbname.c_str(), RingBuffer::c_DefaultSize);
  }
 
  // Insert Tx at the end of current path
  ModulePtr txptr(new TxModule(rbuf));
  a->addModule(txptr);
  // Insert Rx at beginning of next path
  ModulePtr rxptr(new RxModule(rbuf));
  PathPtr newB(new Path());
  newB->addModule(rxptr);
  newB->addPath(b);
  b.swap(newB);
 
  return rbuf;
}

dump_modules()

void dump_modules ( const std::string &  title, const ModulePtrList &  modlist  )

private

Dump module names in the ModulePtrList.

Definition at line 413 of file pEventProcessor.cc.

{
  ModulePtrList::const_iterator it;
  std::ostringstream strbuf;
  strbuf << title << " : ";
  for (it = modlist.begin(); it != modlist.end(); ++it) {
    const Module* module = it->get();
    strbuf << module->getName();
    if (module->hasCondition()) {
      for (const auto& conditionPath : module->getAllConditionPaths()) {
        strbuf << "[->" << conditionPath.get() << "] ";
      }
    }
    if (*it != modlist.back())
      strbuf << " -> ";
  }
  B2INFO(strbuf.str());
}

getMaximumEventNumber()

long getMaximumEventNumber ( long  maxEvent ) const

protectedinherited

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

getModulesWithFlag()

ModulePtrList getModulesWithFlag ( const ModulePtrList &  modules, Module::EModulePropFlags  flag  )

staticprivate

Return only modules which have the given Module flag set.

Definition at line 433 of file pEventProcessor.cc.

{
  ModulePtrList tmpModuleList;
  for (const ModulePtr& m : modules) {
    if (m->hasProperties(flag))
      tmpModuleList.push_back(m);
  }
 
  return tmpModuleList;
}

getModulesWithoutFlag()

ModulePtrList getModulesWithoutFlag ( const ModulePtrList &  modules, Module::EModulePropFlags  flag  )

staticprivate

Return only modules which do not have the given Module flag set.

Definition at line 443 of file pEventProcessor.cc.

{
  ModulePtrList tmpModuleList;
  for (const ModulePtr& m : modules) {
    if (!m->hasProperties(flag))
      tmpModuleList.push_back(m);
  }
  return tmpModuleList;
}

gotSigINT()

void gotSigINT

(

)

signal handler for Ctrl+C (async-safe)

When called the first time, does nothing (input process handles SIGINT by itself). On subsequent calls, RingBuffers are cleared, discarding any events that have been partly produced (mostly equivalent to previous behaviour on Ctrl+C)

Definition at line 86 of file pEventProcessor.cc.

{
  EventProcessor::writeToStdErr("\nStopping basf2...\n");
  killRingBuffers();
}

installMainSignalHandlers()

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

staticinherited

Install signal handler for INT, TERM and QUIT signals.

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

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

staticinherited

Install a signal handler 'fn' for given signal.

Definition at line 300 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);
  }
}

killRingBuffers()

void killRingBuffers

(

)

signal handler (async-safe)

Fairly abrupt termination after the current event.

Definition at line 92 of file pEventProcessor.cc.

{
  m_rbin->kill();
  //these might be locked by _this_ process, so we cannot escape
  m_rbout->kill(); //atomic, so doesn't lock
}

preparePaths()

void preparePaths ( )

private

Adds internal modules to paths, prepare RingBuffers.

Definition at line 398 of file pEventProcessor.cc.

{
  if (m_histoman) {
    m_histoman->initialize();
  }
  if (not m_mainPath)
    return; //we'll fall back to single-core
 
  if (m_inputPath)
    m_rbin = connectViaRingBuffer("BASF2_RBIN", m_inputPath, m_mainPath);
  if (m_outputPath)
    m_rbout = connectViaRingBuffer("BASF2_RBOUT", m_mainPath, m_outputPath);
}

prependModulesIfNotPresent()

void prependModulesIfNotPresent ( ModulePtrList *  modules, const ModulePtrList &  prependModules  )

staticprivate

Prepend given 'prependModules' to 'modules', if they're not already present.

Definition at line 453 of file pEventProcessor.cc.

{
  for (const ModulePtr& m : prependModules) {
    if (std::find(modules->begin(), modules->end(), m) == modules->end()) { //not present
      modules->push_front(m);
    }
  }
}

process()

void process	(	const PathPtr &	spath,
		long	maxEvent
	)

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.

Parameters

spath	The processing starts with the first module of this path.
maxEvent	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 107 of file pEventProcessor.cc.

{
  if (spath->getModules().size() == 0) return;
 
  const int numProcesses = Environment::Instance().getNumberProcesses();
 
  //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))) {
    maxEvent = numEventsArgument;
  }
 
  if (numProcesses == 0)
    B2FATAL("pEventProcessor::process() called for serial processing! Most likely a bug in Framework.");
 
  // 1. Analyze start path and split into parallel paths
  analyzePath(spath);
 
 
  if (m_inputPath) {
    B2INFO("Input Path " << m_inputPath->getPathString());
  }
  if (m_mainPath) {
    if (m_mainPath->getModules().size() <= 5) {
      B2INFO("Main Path " << m_mainPath->getPathString());
    } else {
      B2INFO("Main Path [" << m_mainPath->getModules().front()->getName() << " -> ... (" << m_mainPath->getModules().size() - 2 <<
             " further modules) ... -> " << m_mainPath->getModules().back()->getName() << " ]");
    }
  }
  if (m_outputPath) {
    B2INFO("Output Path " << m_outputPath->getPathString());
  }
  if (not m_mainPath) {
    B2WARNING("Cannot run any modules in parallel (no c_ParallelProcessingCertified flag), falling back to single-core mode.");
    EventProcessor::process(spath, maxEvent);
    return;
  }
 
  installMainSignalHandlers(cleanupAndStop);
 
  //inserts Rx/Tx modules into path (sets up IPC structures)
  preparePaths();
 
  // ensure that we free the IPC resources!
  atexit(cleanupIPC);
 
 
  //init statistics
  {
    m_processStatisticsPtr.registerInDataStore();
    if (!m_processStatisticsPtr)
      m_processStatisticsPtr.create();
    Path mergedPath;
    if (m_inputPath)
      mergedPath.addPath(m_inputPath);
    mergedPath.addPath(m_mainPath);
    if (m_outputPath)
      mergedPath.addPath(m_outputPath);
    for (const ModulePtr& module : mergedPath.buildModulePathList())
      m_processStatisticsPtr->initModule(module.get());
  }
 
  // 2. Initialization
  ModulePtrList modulelist = spath->buildModulePathList();
  ModulePtrList initGlobally = getModulesWithoutFlag(modulelist, Module::c_InternalSerializer);
  //dump_modules("Initializing globally: ", initGlobally);
  processInitialize(initGlobally);
 
  ModulePtrList terminateGlobally = getModulesWithFlag(modulelist, Module::c_TerminateInAllProcesses);
 
  //Don't start processing in case of no master module
  if (!m_master) {
    B2ERROR("There is no module that provides event and run numbers. You must either add the EventInfoSetter module to your path, or, if using an input module, read EventMetaData objects from file.");
  }
 
  //Check if errors appeared. If yes, don't start the event processing.
  int numLogError = LogSystem::Instance().getMessageCounter(LogConfig::c_Error);
  if (numLogError != 0) {
    B2FATAL(numLogError << " ERROR(S) occurred! The processing of events will not be started.");
  }
 
  //disable ROOT's management of TFiles
  clearFileList();
 
 
  //install new signal handlers before forking
  installMainSignalHandlers(parentSignalHandler);
 
  //Path for current process
  PathPtr localPath;
 
  m_procHandler = std::make_unique<ProcHandler>(numProcesses);
 
  // 3. Fork input path
  m_procHandler->startInputProcess();
  if (m_procHandler->isInputProcess()) {   // In input process
    localPath = m_inputPath;
  } else {
    // This is not the input path, clean up datastore to not contain the first event
    DataStore::Instance().invalidateData(DataStore::c_Event);
  }
 
  if (localPath == nullptr) { //not forked yet
    // 5. Fork out worker path (parallel section)
    m_procHandler->startWorkerProcesses();
    if (m_procHandler->isWorkerProcess()) {
      localPath = m_mainPath;
      m_master = localPath->getModules().begin()->get(); //set Rx as master
    }
  }
 
  if (localPath == nullptr) { //not forked yet -> this process is the output process
    m_procHandler->startOutputProcess();
    if (m_outputPath) {
      localPath = m_outputPath;
      m_master = localPath->getModules().begin()->get(); //set Rx as master
    }
  }
 
 
  if (!m_procHandler->isOutputProcess()) {
    DataStoreStreamer::removeSideEffects();
  }
 
  bool gotSigINT = false;
  if (localPath != nullptr) {
    ModulePtrList localModules = localPath->buildModulePathList();
    ModulePtrList procinitmodules = getModulesWithFlag(localModules, Module::c_InternalSerializer);
    //dump_modules("processInitialize for ", procinitmodules);
    processInitialize(procinitmodules);
 
    //input: handle signals normally, will slowly cascade down
    if (m_procHandler->isInputProcess())
      installMainSignalHandlers();
    //workers will have to ignore the signals, there's no good way to do this safely
    if (m_procHandler->isWorkerProcess())
      installMainSignalHandlers(SIG_IGN);
 
    try {
      processCore(localPath, localModules, maxEvent, m_procHandler->isInputProcess());
    } catch (StoppedBySignalException& e) {
      if (e.signal != SIGINT) {
        B2FATAL(e.what());
      }
      //in case of SIGINT, we move on to processTerminate() to shut down safely
      gotSigINT = true;
    }
    prependModulesIfNotPresent(&localModules, terminateGlobally);
    processTerminate(localModules);
  }
 
  B2INFO(m_procHandler->getProcessName() << " process finished.");
 
  //all processes except output stop here
  if (!m_procHandler->isOutputProcess()) {
    if (gotSigINT) {
      installSignalHandler(SIGINT, SIG_DFL);
      raise(SIGINT);
    } else {
      exit(0);
    }
  }
 
  //output process: do final cleanup
  m_procHandler->waitForAllProcesses();
  B2INFO("All processes completed");
 
  //finished, disable handler again
  installSignalHandler(SIGINT, SIG_IGN);
 
  cleanup();
  B2INFO("Global process: completed");
 
  if (m_histoman) {
    B2INFO("HistoManager:: adding histogram files");
    RbTupleManager::Instance().hadd();
  }
 
  //did anything bad happen?
  if (gSignalReceived) {
    B2ERROR("Processing aborted via signal " << gSignalReceived <<
            ", terminating. Output files have been closed safely and should be readable.");
    installSignalHandler(gSignalReceived, SIG_DFL);
    raise(gSignalReceived);
  }
 
}

processBeginRun()

void processBeginRun ( bool  skipDB = false )

protectedinherited

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 470 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  )

protectedinherited

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

protectedinherited

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 504 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  )

protectedinherited

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 324 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  )

protectedinherited

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 immediately 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");
 
  // EventExtraInfo is needed in several modules so register it here
  m_eventExtraInfo.registerInDataStore();
 
  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 )

protectedinherited

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 443 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 )

inlineinherited

Set the name of the module we want to profile.

Parameters

name	Name of the module as returned by getName()

Definition at line 58 of file EventProcessor.h.

{ m_profileModuleName = name; }

writeToStdErr()

void writeToStdErr ( const char  msg[] )

staticinherited

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_eventExtraInfo

StoreObjPtr<EventExtraInfo> m_eventExtraInfo

protectedinherited

event extra info object pointer

Definition at line 170 of file EventProcessor.h.

m_eventMetaDataPtr

StoreObjPtr<EventMetaData> m_eventMetaDataPtr

protectedinherited

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

Definition at line 164 of file EventProcessor.h.

m_histoman

ModulePtr m_histoman

private

Pointer to HistoManagerModule, or nullptr if not found.

Definition at line 108 of file pEventProcessor.h.

m_inputPath

PathPtr m_inputPath

private

Input path.

Definition at line 96 of file pEventProcessor.h.

m_inRun

bool m_inRun

protectedinherited

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

Definition at line 176 of file EventProcessor.h.

m_lastMetadataUpdate

double m_lastMetadataUpdate

protectedinherited

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

Definition at line 179 of file EventProcessor.h.

m_mainPath

PathPtr m_mainPath

private

Main (parallel section) path.

Definition at line 98 of file pEventProcessor.h.

m_master

const Module* m_master

protectedinherited

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

Definition at line 154 of file EventProcessor.h.

m_metadataUpdateInterval

double m_metadataUpdateInterval

protectedinherited

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

Definition at line 182 of file EventProcessor.h.

m_moduleList

ModulePtrList m_moduleList

protectedinherited

List of all modules in order initialized.

Definition at line 155 of file EventProcessor.h.

m_outputPath

PathPtr m_outputPath

private

Output path.

Definition at line 100 of file pEventProcessor.h.

m_previousEventMetaData

EventMetaData m_previousEventMetaData

protectedinherited

Stores state of EventMetaData before it was last changed.

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

Definition at line 167 of file EventProcessor.h.

m_processStatisticsPtr

StoreObjPtr<ProcessStatistics> m_processStatisticsPtr

protectedinherited

Also used in a number of places.

Definition at line 173 of file EventProcessor.h.

m_procHandler

std::unique_ptr<ProcHandler> m_procHandler

private

handler to fork and manage processes.

Definition at line 93 of file pEventProcessor.h.

m_profileModule

Module* m_profileModule = nullptr

protectedinherited

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

Definition at line 161 of file EventProcessor.h.

m_profileModuleName

std::string m_profileModuleName

protectedinherited

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

Definition at line 158 of file EventProcessor.h.

m_rbin

RingBuffer* m_rbin = nullptr

private

input RingBuffer

Definition at line 103 of file pEventProcessor.h.

m_rbout

RingBuffer* m_rbout = nullptr

private

output RingBuffer

Definition at line 105 of file pEventProcessor.h.

m_steerRootInputModuleOn

bool m_steerRootInputModuleOn = false

protectedinherited

True if the SteerRootInputModule is in charge for event processing.

Definition at line 185 of file EventProcessor.h.

---

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

• framework/pcore/include/pEventProcessor.h
• framework/pcore/src/pEventProcessor.cc