ZMQEventProcessor Class Reference

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

#include <ZMQEventProcessor.h>

Inheritance diagram for ZMQEventProcessor:

Public Member Functions
	ZMQEventProcessor ()
	Init the socket cleaning at exit.
virtual	~ZMQEventProcessor ()
	Make sure we remove all sockets cleanly.
void	process (const PathPtr &spath, long maxEvent)
	Processes the full module chain using parallel processing, starting with the first module in the given path.
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.
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().
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	initialize (const ModulePtrList &moduleList, const ModulePtr &histogramManager)
	First step in the process: init the module in the list.
void	forkAndRun (long maxEvent, const PathPtr &inputPath, const PathPtr &mainPath, const PathPtr &outputPath, const ModulePtrList &terminateGlobally)
	Second step in the process: fork out the processes we need to have and call the event loop.
void	terminateAndCleanup (const ModulePtr &histogramManager)
	Last step in the process: run the termination and cleanup (kill all remaining processes)
void	runMonitoring (const PathPtr &inputPath, const PathPtr &mainPath, const ModulePtrList &terminateGlobally, long maxEvent)
	Start the monitoring (without forking)
void	runInput (const PathPtr &inputPath, const ModulePtrList &terminateGlobally, long maxEvent)
	Fork out the input process.
void	runOutput (const PathPtr &outputPath, const ModulePtrList &terminateGlobally, long maxEvent)
	Fork out the output process.
void	runWorker (unsigned int numProcesses, const PathPtr &inputPath, const PathPtr &mainPath, const ModulePtrList &terminateGlobally, long maxEvent)
	Fork out the N worker process.
void	processPath (const PathPtr &localPath, const ModulePtrList &terminateGlobally, long maxEvent)
	Basic function run in every process: process the event loop of the given path.
void	processCore (const PathPtr &startPath, const ModulePtrList &modulePathList, long maxEvent=0, bool isInputProcess=true, bool isWorkerProcess=false, bool isOutputProcess=false)
	Process modules in the path.
bool	processEvent (PathIterator moduleIter, bool skipMasterModule, bool Worker=false, bool output=false)
	Calls Event function.
void	processBeginRun (bool skipDB=false)
	Calls BeginRun function.
void	processEndRun ()
	Calls EndRun function.

Private Attributes
ProcessMonitor	m_processMonitor
	Instance of the process monitor.
EventMetaData	m_previousEventMetaData
	Stores previous eventMetaData.

Detailed Description

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

Definition at line 23 of file ZMQEventProcessor.h.

Constructor & Destructor Documentation

ZMQEventProcessor()

ZMQEventProcessor

(

)

Init the socket cleaning at exit.

Definition at line 101 of file ZMQEventProcessor.cc.

{
  B2ASSERT("You are having two instances of the ZMQEventProcessor running! This is not possible",
           not g_eventProcessorForSignalHandling);
  g_eventProcessorForSignalHandling = this;
 
  // Make sure to remove the sockets
  g_socketAddress = Environment::Instance().getZMQSocketAddress();
  std::atexit(deleteSocketFiles);
}

~ZMQEventProcessor()

~ZMQEventProcessor ( )

virtual

Make sure we remove all sockets cleanly.

Definition at line 112 of file ZMQEventProcessor.cc.

{
  cleanup();
  g_eventProcessorForSignalHandling = nullptr;
}

Member Function Documentation

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 422 of file ZMQEventProcessor.cc.

{
  if (not GlobalProcHandler::isProcess(ProcType::c_Monitor) and not GlobalProcHandler::isProcess(ProcType::c_Init)) {
    B2DEBUG(30, "Not running cleanup, as I am in process type " << GlobalProcHandler::getProcessName());
    return;
  }
  m_processMonitor.killProcesses(5000);
  m_processMonitor.terminate();
 
  deleteSocketFiles();
}

forkAndRun()

void forkAndRun ( long  maxEvent, const PathPtr &  inputPath, const PathPtr &  mainPath, const PathPtr &  outputPath, const ModulePtrList &  terminateGlobally  )

private

Second step in the process: fork out the processes we need to have and call the event loop.

Definition at line 400 of file ZMQEventProcessor.cc.

{
  const int numProcesses = Environment::Instance().getNumberProcesses();
  GlobalProcHandler::initialize(numProcesses);
 
  const auto& socketAddress = Environment::Instance().getZMQSocketAddress();
 
  const auto pubSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_pub));
  const auto subSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_sub));
  const auto controlSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_control));
 
  // We catch all signals and store them into a variable. This is used during the main loop then.
  // From now on, we have to make sure to clean up behind us
  installMainSignalHandlers(cleanupAndRaiseSignal);
  m_processMonitor.subscribe(pubSocketAddress, subSocketAddress, controlSocketAddress);
 
  runInput(inputPath, terminateGlobally, maxEvent);
  runOutput(outputPath, terminateGlobally, maxEvent);
  runMonitoring(inputPath, mainPath, terminateGlobally, maxEvent);
}

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

initialize()

void initialize ( const ModulePtrList &  moduleList, const ModulePtr &  histogramManager  )

private

First step in the process: init the module in the list.

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 170 of file ZMQEventProcessor.cc.

{
  if (histogramManager) {
    histogramManager->initialize();
  }
  // from now on the datastore is available
  processInitialize(moduleList, true);
 
  B2INFO("ZMQEventProcessor : processInitialize done");
 
  // 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.");
  }
 
  // TODO: I do not really understand what is going on here...
  // disable ROOT's management of TFiles
  // clear list, but don't actually delete the objects
  gROOT->GetListOfFiles()->Clear("nodelete");
}

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

process()

void process	(	const PathPtr &	spath,
		long	maxEvent
	)

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

Definition at line 118 of file ZMQEventProcessor.cc.

{
  // Concerning signal handling:
  // * During the initialization, we just raise the signal without doing any cleanup etc.
  // * During the event execution, we will not allow for any signal in all processes except the parent process.
  //   Here, we catch sigint and clean up the processes AND WHAT DO WE DO IN THE OTHER CASES?
  // * During cleanup, we will just ignore sigint, but the rest will be raised
 
  if (path->isEmpty()) {
    return;
  }
 
  const int numProcesses = Environment::Instance().getNumberProcesses();
  if (numProcesses == 0) {
    B2FATAL("ZMQEventProcessor::process() called for serial processing! Most likely a bug in Framework.");
  }
 
  // Split the path into input, main and output. A nullptr means, the path should not be used
  PathPtr inputPath, mainPath, outputPath;
  std::tie(inputPath, mainPath, outputPath) = PathUtils::splitPath(path);
  const ModulePtr& histogramManager = PathUtils::getHistogramManager(inputPath);
 
  // Check for existence of HLTZMQ2Ds module in input path to set DAQ environment
  for (const ModulePtr& module : inputPath->getModules()) {
    if (module->getName() == "HLTZMQ2Ds") {
      Environment::Instance().setZMQDAQEnvironment(true);
      B2INFO("ZMQEventProcessor : DAQ environment set");
      break;
    }
  }
 
  if (not mainPath or mainPath->isEmpty()) {
    B2WARNING("Cannot run any modules in parallel (no c_ParallelProcessingCertified flag), falling back to single-core mode.");
    EventProcessor::process(path, maxEvent);
    return;
  }
 
  // inserts Rx/Tx modules into path (sets up IPC structures)
  const ModulePtrList& moduleList = PathUtils::preparePaths(inputPath, mainPath, outputPath);
 
  // Run the initialization of the modules and the histogram manager
  initialize(moduleList, histogramManager);
 
  // The main part: fork into the different processes and run!
  const ModulePtrList& terminateGlobally = PathUtils::getTerminateGloballyModules(moduleList);
  forkAndRun(maxEvent, inputPath, mainPath, outputPath, terminateGlobally);
 
  installMainSignalHandlers(cleanupAndRaiseSignal);
  // Run the final termination and cleanup with error check
  terminateAndCleanup(histogramManager);
}

processBeginRun()

void processBeginRun ( bool  skipDB = false )

private

Calls BeginRun function.

Definition at line 622 of file ZMQEventProcessor.cc.

{
  MetadataService::Instance().addBasf2Status("beginning run");
 
  m_inRun = true;
 
  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();
    module->beginRun();
    m_processStatisticsPtr->stopModule(module, ModuleStatistics::c_BeginRun);
 
    //Set the global log level
    logSystem.updateModule(nullptr);
  }
 
  m_processStatisticsPtr->stopGlobal(ModuleStatistics::c_BeginRun);
}

processCore() [1/2]

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

processCore() [2/2]

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

private

Process modules in the path.

Definition at line 434 of file ZMQEventProcessor.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);
 
    if (isInputProcess) {
      endProcess = ZMQEventProcessor::processEvent(moduleIter, isInputProcess && currEvent == 0);
    } else if (isWorkerProcess) {
      endProcess = ZMQEventProcessor::processEvent(moduleIter, false,
                                                   isWorkerProcess && currEvent == 0 && Environment::Instance().getZMQDAQEnvironment());
    } else if (isOutputProcess) {
      endProcess = ZMQEventProcessor::processEvent(moduleIter, false, false,
                                                   isOutputProcess && currEvent == 0 && Environment::Instance().getZMQDAQEnvironment());
    } else {
      B2INFO("processCore : should not come here. Specified path is invalid");
      return;
    }
 
    //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();
  B2INFO("processCore : End Last Run. calling processEndRun()");
  processEndRun();
}

processEndRun()

void processEndRun ( )

private

Calls EndRun function.

Definition at line 655 of file ZMQEventProcessor.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;
 
  //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();
    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() [1/2]

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

processEvent() [2/2]

bool processEvent ( PathIterator  moduleIter, bool  skipMasterModule, bool  Worker = false, bool  output = false  )

private

Calls Event function.

Definition at line 483 of file ZMQEventProcessor.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 (module != m_master) {
      callEvent(module);
    } else if (!skipMasterModule) {
      callEvent(module);
    } else
      B2INFO("Skipping execution of module " << module->getName());
 
    if (!m_eventMetaDataPtr) {
      return false;
    }
 
    //Check for end of data
    if (m_eventMetaDataPtr->isEndOfData()) {
      // Immediately leave the loop and terminate (true)
      B2INFO("isEndOfData. Return");
      return true;
    }
 
    //Handle EventMetaData changes by master module
    if (module == m_master && !skipMasterModule) {
 
      //initialize random number state for the event
      RandomNumbers::initializeEvent();
 
      // Worker Path
      if (WorkerPath) {
        B2INFO("Worker Path and First Event!");
        if (Environment::Instance().isZMQDAQFirstEvent(m_eventMetaDataPtr->getExperiment(), m_eventMetaDataPtr->getRun())) {
          B2INFO("Worker path processing for ZMQDAQ first event.....Skip to the end of path");
          B2INFO("    --> exp = " << m_eventMetaDataPtr->getExperiment() << " run = " << m_eventMetaDataPtr->getRun());
          while (true) {
            module = moduleIter.get();
            if (module->getName() == "ZMQTxWorker") break;
            moduleIter.next();
          }
          continue;
        }
      }
 
      // Check for EndOfRun
      if (!WorkerPath && !OutputPath) {
        if (m_eventMetaDataPtr->isEndOfRun()) {
          B2INFO("===> EndOfRun : calling processEndRun(); isEndOfRun = " << m_eventMetaDataPtr->isEndOfRun());
          processEndRun();
          // Store the current event meta data for the next round
          m_previousEventMetaData = *m_eventMetaDataPtr;
          // Leave this event, but not the full processing (false)
          return false;
        } else if (m_previousEventMetaData.isEndOfData() || m_previousEventMetaData.isEndOfRun()) {
          if (m_eventMetaDataPtr->getRun() ==  m_previousEventMetaData.getRun()) {
            B2INFO("===> EndOfData : ----> Run change request to the same run!!! Skip this event.");
            return false;
          }
          B2INFO("===> EndOfData : calling processBeginRun(); isEndOfData = " << m_previousEventMetaData.isEndOfData() <<
                 " isEndOfRun = " << m_previousEventMetaData.isEndOfRun());
          B2INFO("--> cur run = " << m_eventMetaDataPtr->getRun() << " <- prev run = " << m_previousEventMetaData.getRun());
          B2INFO("--> cur evt = " << m_eventMetaDataPtr->getEvent() << " <- prev evt = " << m_previousEventMetaData.getEvent());
          // The run number should not be 0
          if (m_eventMetaDataPtr->getRun() != 0) {
            processBeginRun();
            m_previousEventMetaData = *m_eventMetaDataPtr;
          } else {
            return false;
          }
        }
 
        const bool runChanged = ((m_eventMetaDataPtr->getExperiment() != m_previousEventMetaData.getExperiment()) or
                                 (m_eventMetaDataPtr->getRun() != m_previousEventMetaData.getRun()));
        const bool runChangedWithoutNotice = runChanged and not m_previousEventMetaData.isEndOfData()
                                             and not m_previousEventMetaData.isEndOfRun();
        //  if (runChangedWithoutNotice && !g_first_round) {
        if (runChangedWithoutNotice) {
          if (collectStats)
            m_processStatisticsPtr->suspendGlobal();
 
          B2INFO("===> Run Change (possibly offline) : calling processEndRun() and processBeginRun()");
          B2INFO("--> cur run = " << m_eventMetaDataPtr->getRun() << " <- prev run = " << m_previousEventMetaData.getRun());
          B2INFO("--> cur evt = " << m_eventMetaDataPtr->getEvent() << " <- prev evt = " << m_previousEventMetaData.getEvent());
          B2INFO("--> runChanged = " << runChanged << " runChangedWithoutNotice = " << runChangedWithoutNotice);
 
          processEndRun();
          processBeginRun();
 
          if (collectStats)
            m_processStatisticsPtr->resumeGlobal();
        }
        m_previousEventMetaData = *m_eventMetaDataPtr;
      } else
        B2INFO("Skipping begin/end run processing");
 
      //make sure we use the event dependent generator again
      RandomNumbers::useEventDependent();
 
      DBStore::Instance().updateEvent();
 
    } else if (!WorkerPath && !OutputPath) {
      //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 (g_signalReceived != 0) {
      throw StoppedBySignalException(g_signalReceived);
    }
 
    //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);
}

processPath()

void processPath ( const PathPtr &  localPath, const ModulePtrList &  terminateGlobally, long  maxEvent  )

private

Basic function run in every process: process the event loop of the given path.

Definition at line 318 of file ZMQEventProcessor.cc.

{
  ModulePtrList localModules = localPath->buildModulePathList();
  maxEvent = getMaximumEventNumber(maxEvent);
  // we are not using the default signal handler, so the processCore can not throw any exception because if sigint...
  processCore(localPath, localModules, maxEvent, GlobalProcHandler::isProcess(ProcType::c_Input),
              GlobalProcHandler::isProcess(ProcType::c_Worker),
              GlobalProcHandler::isProcess(ProcType::c_Output));
 
  B2DEBUG(30, "terminate process...");
  PathUtils::prependModulesIfNotPresent(&localModules, terminateGlobally);
  processTerminate(localModules);
}

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

runInput()

void runInput ( const PathPtr &  inputPath, const ModulePtrList &  terminateGlobally, long  maxEvent  )

private

Fork out the input process.

Definition at line 226 of file ZMQEventProcessor.cc.

{
  if (not inputPath or inputPath->isEmpty()) {
    return;
  }
 
  if (not GlobalProcHandler::startInputProcess()) {
    // This is not the input process, clean up datastore to not contain the first event
    DataStore::Instance().invalidateData(DataStore::c_Event);
    return;
  }
 
  // The default will be to not do anything on signals...
  installMainSignalHandlers(SIG_IGN);
 
  m_processMonitor.reset();
  DataStoreStreamer::removeSideEffects();
 
  processPath(inputPath, terminateGlobally, maxEvent);
  B2DEBUG(30, "Finished an input process");
  exit(0);
}

runMonitoring()

void runMonitoring ( const PathPtr &  inputPath, const PathPtr &  mainPath, const ModulePtrList &  terminateGlobally, long  maxEvent  )

private

Start the monitoring (without forking)

Definition at line 333 of file ZMQEventProcessor.cc.

{
  if (not GlobalProcHandler::startMonitoringProcess()) {
    return;
  }
 
  const auto& environment = Environment::Instance();
 
  B2DEBUG(30, "Will now start process monitor...");
  const int numProcesses = environment.getNumberProcesses();
  m_processMonitor.initialize(numProcesses);
 
  // Make sure the input process is running until we go on
  m_processMonitor.waitForRunningInput(60 * 1000);
  if (m_processMonitor.hasEnded()) {
    return;
  }
  // Make sure the output process is running until we go on
  m_processMonitor.waitForRunningOutput(60 * 1000);
  if (m_processMonitor.hasEnded()) {
    return;
  }
 
  installMainSignalHandlers(storeSignal);
 
  // at least start the number of workers requested
  runWorker(m_processMonitor.needMoreWorkers(), inputPath, mainPath, terminateGlobally, maxEvent);
 
  const auto& restartFailedWorkers = environment.getZMQRestartFailedWorkers();
  const auto& failOnFailedWorkers = environment.getZMQFailOnFailedWorkers();
 
  B2DEBUG(30, "Will now start main loop...");
  while (true) {
    // check multicast for messages and kill workers if requested
    m_processMonitor.checkMulticast();
    // check the child processes, if one has died
    m_processMonitor.checkChildProcesses();
    // check if we have received any signal from the user or OS. Kill the processes if not SIGINT.
    m_processMonitor.checkSignals(g_signalReceived);
 
    // If we have received a SIGINT signal or the last process is gone, we can end smoothly
    if (m_processMonitor.hasEnded()) {
      break;
    }
 
    // Test if we need more workers
    const unsigned int neededWorkers = m_processMonitor.needMoreWorkers();
    if (neededWorkers > 0) {
      B2DEBUG(30, "restartFailedWorkers = " << restartFailedWorkers);
      if (restartFailedWorkers) {
        B2DEBUG(30, ".... Restarting a new worker");
        B2ERROR(".... Restarting a new worker process");
        runWorker(neededWorkers, inputPath, mainPath, terminateGlobally, maxEvent);
      } else if (failOnFailedWorkers) {
        B2ERROR("A worker failed. Will try to end the process smoothly now.");
        break;
      } else if (not m_processMonitor.hasWorkers()) {
        B2WARNING("All workers have died and you did not request to restart them. Going down now.");
        break;
      }
    }
  }
 
  B2DEBUG(30, "Finished the monitoring process");
}

runOutput()

void runOutput ( const PathPtr &  outputPath, const ModulePtrList &  terminateGlobally, long  maxEvent  )

private

Fork out the output process.

Definition at line 249 of file ZMQEventProcessor.cc.

{
  const auto& socketAddress = Environment::Instance().getZMQSocketAddress();
  const auto pubSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_pub));
  const auto subSocketAddress(ZMQAddressUtils::getSocketAddress(socketAddress, ZMQAddressType::c_sub));
 
  if (not outputPath or outputPath->isEmpty()) {
    return;
  }
 
  if (not GlobalProcHandler::startOutputProcess()) {
    return;
  }
 
  // The default will be to not do anything on signals...
  installMainSignalHandlers(SIG_IGN);
 
  m_processMonitor.reset();
 
  // Set the rx module as main module
  m_master = outputPath->getModules().begin()->get();
 
  processPath(outputPath, terminateGlobally, maxEvent);
 
  // Send the statistics to the process monitor
  StreamHelper streamer;
  ZMQClient zmqClient;
 
  // TODO: true?
  streamer.initialize(0, true);
  zmqClient.initialize(pubSocketAddress, subSocketAddress);
 
  // TODO: make sure to only send statistics!
  const auto& evtMessage = streamer.stream();
  auto message = ZMQMessageFactory::createMessage(EMessageTypes::c_statisticMessage, evtMessage);
  zmqClient.publish(std::move(message));
 
  B2DEBUG(30, "Finished an output process");
  exit(0);
}

runWorker()

void runWorker ( unsigned int  numProcesses, const PathPtr &  inputPath, const PathPtr &  mainPath, const ModulePtrList &  terminateGlobally, long  maxEvent  )

private

Fork out the N worker process.

Definition at line 289 of file ZMQEventProcessor.cc.

{
  if (numProcesses == 0) {
    return;
  }
 
  if (not GlobalProcHandler::startWorkerProcesses(numProcesses)) {
    // Make sure the worker process is running until we go on
    m_processMonitor.waitForRunningWorker(Environment::Instance().getZMQMaximalWaitingTime());
    return;
  }
 
  // The default will be to not do anything on signals...
  installMainSignalHandlers(SIG_IGN);
 
  if (inputPath and not inputPath->isEmpty()) {
    // set Rx as master
    m_master = mainPath->getModules().begin()->get();
  }
 
  m_processMonitor.reset();
  DataStoreStreamer::removeSideEffects();
 
  processPath(mainPath, terminateGlobally, maxEvent);
  B2DEBUG(30, "Finished a worker process");
  exit(0);
}

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

terminateAndCleanup()

void terminateAndCleanup ( const ModulePtr &  histogramManager )

private

Last step in the process: run the termination and cleanup (kill all remaining processes)

Definition at line 202 of file ZMQEventProcessor.cc.

{
  cleanup();
 
  if (histogramManager) {
    B2INFO("HistoManager:: adding histogram files");
    RbTupleManager::Instance().hadd();
  }
 
  // did anything bad happen?
  if (g_signalReceived) {
    if (g_signalReceived == SIGINT) {
      B2RESULT("Processing aborted via signal " << g_signalReceived <<
               ", terminating. Output files have been closed safely and should be readable.");
    } else {
      B2ERROR("Processing aborted via signal " << g_signalReceived <<
              ", terminating. Output files have been closed safely and should be readable.");
    }
    // re-raise the signal
    installSignalHandler(g_signalReceived, SIG_DFL);
    raise(g_signalReceived);
  }
}

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_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_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_previousEventMetaData

EventMetaData m_previousEventMetaData

private

Stores previous eventMetaData.

Definition at line 82 of file ZMQEventProcessor.h.

m_processMonitor

ProcessMonitor m_processMonitor

private

Instance of the process monitor.

Definition at line 79 of file ZMQEventProcessor.h.

m_processStatisticsPtr

StoreObjPtr<ProcessStatistics> m_processStatisticsPtr

protectedinherited

Also used in a number of places.

Definition at line 173 of file EventProcessor.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_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/ZMQEventProcessor.h
• framework/pcore/src/ZMQEventProcessor.cc