HLTEventProcessor Class Reference

EventProcessor to be used on the HLT with all specialities of the HLT processing: More...

#include <HLTEventProcessor.h>

Inheritance diagram for HLTEventProcessor:

Public Member Functions
	HLTEventProcessor (const std::vector< std::string > &outputAddresses)
	Create a new event processor and store the ZMQ addresses where to unregister workers.
void	process (PathPtr spath, bool restartFailedWorkers, bool appendProcessNumberToModuleName=false)
	Process the given path.
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.
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.

Private Member Functions
void	sendTerminatedMessage (unsigned int pid, bool waitForConfirmation)
	Send an unregister message to all sockets if the given PID died. Wait at max 10s for the confirmation (if requested)
void	runWorkers (PathPtr path, unsigned int numProcesses, bool appendProcessNumberToModuleName=false)
	Fork out as much workers as requested and in each run the given path using processCore.
void	processCore (PathPtr path)
	Process the path by basically calling processEvent until a termination is requested.
bool	processEvent (PathIterator moduleIter, bool firstRound)
	Process a single event by iterating through the module path once.
std::pair< unsigned int, unsigned int >	checkChildProcesses ()
	Check if one of the started processes has died.
void	release ()
	Release the parent resource, which is needed after forking to not close it twice.
bool	forkOut ()
	Helper function to fork out. Sets the Python state correctly and adds the process to the internal state.

Private Attributes
ZMQParent	m_parent
	An instance of a ZMQParent to create sockets for unregistering workers.
std::vector< int >	m_processList
	The current list of running processes (with their PIDs)
std::vector< std::unique_ptr< zmq::socket_t > >	m_sockets
	The created sockets for unregistering workers. TODO: use connections.

Detailed Description

EventProcessor to be used on the HLT with all specialities of the HLT processing:

• no input or output path - just workers
• multiprocessing forked out after initialization, which is before the first real event is processed
• restart of terminated workers (configurable) and unregistration and DQM server and collector
• special run start/end handling: please check the HLTZMQ2DsModule for details
• check of the child processes in an additional monitoring process (happens to be the parent process) by keeping track of the current state

This event processor is specialies to the HLT and should not be used apart from that. The event processor is exported as a python module. It can be called with

import hbasf2 hbasf2.process(path, [address, ...], True)

or if you want to put number on each folked process,

hbasf2.processNumbered(path, [address, ...], True(, True))

Definition at line 37 of file HLTEventProcessor.h.

Constructor & Destructor Documentation

HLTEventProcessor()

HLTEventProcessor

(

const std::vector< std::string > &

outputAddresses

)

Create a new event processor and store the ZMQ addresses where to unregister workers.

Definition at line 76 of file HLTEventProcessor.cc.

{
  m_sockets.reserve(outputAddresses.size());
  for (const auto& address : outputAddresses) {
    m_sockets.push_back(m_parent.createSocket<ZMQ_DEALER>(address, false));
  }
}

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

checkChildProcesses()

std::pair< unsigned int, unsigned int > checkChildProcesses ( )

private

Check if one of the started processes has died.

If it has died with a non-zero exit code, increase the counter of workers to restart.

Returns the number of still alive workers and the number of workers needed to restart (aka the number of workers that died with a non-zero exit code) as a pair.

Definition at line 396 of file HLTEventProcessor.cc.

{
  unsigned int needToRestart = 0;
 
  // Check for processes, which where there last time but are gone now (so they died)
  for (auto iter = m_processList.begin(); iter != m_processList.end();) {
    const auto& pid = *iter;
 
    // check the status of this process pid
    int status;
    const int result = waitpid(pid, &status, WNOHANG);
    if (result == -1) {
      if (errno == EINTR) {
        // interrupted, try again next time
        ++iter;
        continue;
      } else {
        B2FATAL("waitpid() failed.");
      }
    } else if (result == 0) {
      // No change, so lets continue with the next worker
      ++iter;
      continue;
    }
 
    B2ASSERT("Do not understand the result of waitpid()", result == pid);
 
    // state has changed, which means it is dead!
    const auto exitCode = WEXITSTATUS(status);
 
    // we only need to restart unexpected deads
    if (exitCode != 0) {
      B2WARNING("A worker process has died unexpected!");
      needToRestart += 1;
 
      sendTerminatedMessage(pid, true);
    }
 
    // once a process is gone from the global list, remove them from our own, too.
    iter = m_processList.erase(iter);
  }
 
  return {m_processList.size(), needToRestart};
}

forkOut()

bool forkOut ( )

private

Helper function to fork out. Sets the Python state correctly and adds the process to the internal state.

Definition at line 449 of file HLTEventProcessor.cc.

{
  fflush(stdout);
  fflush(stderr);
 
  pid_t pid = fork();
 
  if (pid > 0) {
    g_processNumber++;
    m_processList.push_back(pid);
    return false;
  } else if (pid < 0) {
    B2FATAL("fork() failed: " << strerror(errno));
  } else {
    // Child process
    // Reset some python state: signals, threads, gil in the child
    PyOS_AfterFork_Child();
    // InputController becomes useless in child process
    InputController::resetForChildProcess();
    // die when parent dies
    prctl(PR_SET_PDEATHSIG, SIGHUP);
 
    ProcHandler::setProcessID(getpid());
    return true;
  }
}

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

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

staticinherited

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() [1/2]

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

inherited

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

process() [2/2]

void process	(	PathPtr	spath,
		bool	restartFailedWorkers,
		bool	appendProcessNumberToModuleName = false
	)

Process the given path.

If requested, restart failed workers (or not) For this,

• first check if all modules have a parallel flag
• then call initialize in the main process
• fork out all workers and process the path
• while monitoring their status
• in the end kill remaining processes (if needed) and re-raise any collected signals

Definition at line 84 of file HLTEventProcessor.cc.

{
  using namespace std::chrono_literals;
 
  m_moduleList = path->buildModulePathList();
 
  // Assert path is what we want: fully parallel certified, not empty. Set first module to master module
  B2ASSERT("You try to process an empty path!", not m_moduleList.empty());
  for (const auto& module : m_moduleList) {
    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;
        }
      }
    }
    B2ASSERT("Module with name " << module->getName() << " does not have parallel flag!", hasParallelFlag);
  }
 
  // Initialize of all modules (including event() of master module)
  installMainSignalHandlers();
  processInitialize(m_moduleList);
 
  // Don't start processing in case of no master module
  if (not m_master) {
    B2ERROR("There is no module that provides event and run numbers (EventMetaData). "
            "You must add the specific HLT module as first module to the path.");
  }
 
  // Check if errors appeared. If yes, don't start the event processing.
  const int numLogError = LogSystem::Instance().getMessageCounter(LogConfig::c_Error);
  if (numLogError != 0) {
    B2FATAL(numLogError << " ERROR(S) occurred! The processing of events will not be started.");
  }
 
  // Start the workers, which call the main loop
  const int numProcesses = Environment::Instance().getNumberProcesses();
  runWorkers(path, numProcesses, appendProcessNumberToModuleName);
 
  installMainSignalHandlers(storeSignal);
  // Back in the main process: wait for the processes and monitor them
  int numberOfRestartedWorkers = 0;
  while (true) {
    // check if we have received any signal from the user or OS.
    // Killing of the remaining processes happens after the loop.
    if (g_signalReceived > 0) {
      B2WARNING("Received a signal to go down.");
      break;
    }
 
    // Test if we need more workers and if one has died
    unsigned int presentWorkers;
    unsigned int neededWorkers;
 
    std::tie(presentWorkers, neededWorkers) = checkChildProcesses();
    if (neededWorkers > 0) {
      if (restartFailedWorkers) {
        runWorkers(path, neededWorkers);
        numberOfRestartedWorkers += neededWorkers;
      } else {
        B2ERROR("A worker failed. Will try to end the process smoothly now.");
        break;
      }
    } else if (presentWorkers == 0) {
      B2DEBUG(10, "All workers have cleanly exited. Will now also exit");
      break;
    }
 
    if (numberOfRestartedWorkers > numProcesses) {
      B2ERROR("I needed to restart on total " << numberOfRestartedWorkers << ", which I think is abnormal. "
              "Will terminate the process now!");
      break;
    }
 
    std::this_thread::sleep_for(10ms);
  }
 
  if (appendProcessNumberToModuleName) {
    for (const int& pid : m_processList) {
      B2INFO(g_processNumber << ": Send SIGINT to " << pid);
      kill(pid, SIGINT);
    }
    for (const int& pid : m_processList) {
      int count = 0;
      while (true) {
        // Do not allow internal SIGKILL to prevent data loss in case of a numbered process
        if (kill(pid, 0) != 0) {
          break;
        }
        B2DEBUG(10, g_processNumber << ": Checking process termination, count = " << count);
        std::this_thread::sleep_for(1000ms);
        if (count % 5 == 1) kill(pid, SIGINT);
        ++count;
      }
    }
  }
 
  checkChildProcesses();
 
  // if we still have/had processes, we should unregister them
  std::this_thread::sleep_for(500ms);
 
  for (const int& pid : m_processList) {
    if (kill(pid, SIGKILL) >= 0) {
      B2WARNING("Needed to hard kill process " << pid);
    } else {
      B2DEBUG(100, "no process " << pid << " found, already gone?");
    }
    sendTerminatedMessage(pid, false);
  }
  m_processList.clear();
 
  B2DEBUG(10, "Done here");
 
  // Normally, we would call terminate here, but not on HLT!
  // Normally, we would print the error summary here, but not on HLT!
  if (g_signalReceived == SIGINT) {
    installSignalHandler(SIGINT, SIG_DFL);
    raise(SIGINT);
  }
}

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

processCore() [2/2]

void processCore ( PathPtr  path )

private

Process the path by basically calling processEvent until a termination is requested.

# Will not any initialization - it is assumed this has already happened before. In the end, terminate is called.

Definition at line 249 of file HLTEventProcessor.cc.

{
  bool terminationRequested = false;
  bool firstRound = true;
 
  // Initialisation is done
  DataStore::Instance().setInitializeActive(false);
 
  // Set the previous event meta data to something invalid
  m_previousEventMetaData.setEndOfData();
 
  while (not terminationRequested) {
    B2DEBUG(100, "Processing new event");
 
    // Start the measurement
    m_processStatisticsPtr->startGlobal();
 
    // Main call to event() of the modules, and maybe beginRun() and endRun()
    PathIterator moduleIter(path);
    terminationRequested = processEvent(moduleIter, firstRound);
 
    // Delete event related data in DataStore
    DataStore::Instance().invalidateData(DataStore::c_Event);
 
    // Stop the measurement
    m_processStatisticsPtr->stopGlobal(ModuleStatistics::c_Event);
 
    // We are surely not in the first round the next time
    firstRound = false;
  }
 
  // End last run with a terminate. Yes, we are not calling a endRun() here and yes, we are calling this in the worker
  B2DEBUG(10, "Calling terminate");
  m_eventMetaDataPtr.create();
  processTerminate(m_moduleList);
}

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

private

Process a single event by iterating through the module path once.

In principle very similar to the EventProcessor::processEvent function, but has different assumptions for the run changes happening induced by the master module (which is always the ZMQ2Ds module in the HLT case).

The logic happening after the master module is the following:

• if an EndOfData is set in the event meta data, just break out
• if a HLT-specific EndOfRun is set, call the end of run methods of all modules (without begin run)
• if the previous event meta data has the EndOfRun/Data set, call the begin run functions
• if the run change was induced due to a changing run number, call begin and end run functions

The rest (e.g. module conditions, db store) is the same as the EventProcessor case.

Definition at line 286 of file HLTEventProcessor.cc.

{
  while (not moduleIter.isDone()) {
    Module* module = moduleIter.get();
    B2DEBUG(10, "Starting event of " << module->getName());
 
    // The actual call of the event function
    if (module != m_master) {
      // If this is not the master module it is quite simple: just call the event function
      callEvent(module);
 
      // Check for a second master module. Cannot do this if we are in the first round after initialize
      // (as previous event meta data is not set properly here)
      if (not m_previousEventMetaData.isEndOfData() and m_eventMetaDataPtr and
          (*m_eventMetaDataPtr != m_previousEventMetaData)) {
        B2FATAL("Two modules setting EventMetaData were discovered: " << m_master->getName() << " and "
                << module->getName());
      }
    } else {
      if (not firstRound) {
        // Also call the event function for the master, but not the first time
        callEvent(module);
      }
      // initialize random number state for the event handling after we have
      // recieved the event information from the master module.
      RandomNumbers::initializeEvent(true);
    }
 
    if (g_signalReceived != 0) {
      if (g_signalReceived != SIGINT) {
        throw StoppedBySignalException(g_signalReceived);
      } else {
        B2DEBUG(10, "Received a SIGINT in the worker process...");
        return true;
      }
    }
 
    B2ASSERT("The event meta data must always be valid at this stage!", m_eventMetaDataPtr and m_eventMetaDataPtr.isValid());
 
    if (m_eventMetaDataPtr->isEndOfData()) {
      // Immediately leave the loop and terminate (true)
      return true;
    }
 
    if (module == m_master and not firstRound) {
      if (m_eventMetaDataPtr->isEndOfRun()) {
        B2DEBUG(10, "Calling endRun()");
        // call endRun() of all modules (internally uses the previous event meta data) and skip to the next event
        m_processStatisticsPtr->suspendGlobal();
        m_inRun = true;
        processEndRun();
        m_processStatisticsPtr->resumeGlobal();
 
        // 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() or m_previousEventMetaData.isEndOfRun()) {
        // The run has changes (or we never had one), so call beginRun() before going on
        // The run number should not be 0
        if (m_eventMetaDataPtr->getRun() != 0) {
          m_processStatisticsPtr->suspendGlobal();
          processBeginRun();
          m_processStatisticsPtr->resumeGlobal();
        } 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) {
        m_processStatisticsPtr->suspendGlobal();
 
        m_inRun = true;
        processEndRun();
        processBeginRun();
 
        m_processStatisticsPtr->resumeGlobal();
      }
 
      // make sure we use the event dependent generator again
      RandomNumbers::useEventDependent();
 
      // and the correct database
      DBStore::Instance().updateEvent();
 
      // Store the current event meta data for the next round
      m_previousEventMetaData = *m_eventMetaDataPtr;
    }
 
    // 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();
    }
  }
  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 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 )

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

release()

void release ( )

private

Release the parent resource, which is needed after forking to not close it twice.

Definition at line 441 of file HLTEventProcessor.cc.

{
  for (auto& socket : m_sockets) {
    socket.release();
  }
  m_parent.reset();
}

runWorkers()

void runWorkers ( PathPtr  path, unsigned int  numProcesses, bool  appendProcessNumberToModuleName = false  )

private

Fork out as much workers as requested and in each run the given path using processCore.

Definition at line 208 of file HLTEventProcessor.cc.

{
  for (unsigned int i = 0; i < numProcesses; i++) {
    if (forkOut()) {
      // Do only run in forked out worker process:
      B2DEBUG(10, "Starting a new worker process");
      // Reset the parent and sockets
      release();
 
      // Start the main loop with our signal handling and error catching
      installMainSignalHandlers(storeSignal);
      try {
        if (appendProcessNumberToModuleName) {
          for (const auto& module : m_moduleList) {
            module->setName(std::to_string(g_processNumber) + std::string("_") + module->getName());
            B2INFO("New worker name is " << module->getName());
          }
        }
        processCore(path);
      } catch (StoppedBySignalException& e) {
        // close all open ROOT files, ROOT's exit handler will crash otherwise
        gROOT->GetListOfFiles()->Delete();
 
        B2ERROR(e.what());
        exit(1);
      } catch (...) {
        if (m_eventMetaDataPtr)
          B2ERROR("Exception occured in exp/run/evt: "
                  << m_eventMetaDataPtr->getExperiment() << " / "
                  << m_eventMetaDataPtr->getRun() << " / "
                  << m_eventMetaDataPtr->getEvent());
        throw;
      }
 
      B2DEBUG(10, "Ending a worker process here.");
      // Ok, we are done here!
      exit(0);
    }
  }
}

sendTerminatedMessage()

void sendTerminatedMessage ( unsigned int  pid, bool  waitForConfirmation  )

private

Send an unregister message to all sockets if the given PID died. Wait at max 10s for the confirmation (if requested)

Definition at line 57 of file HLTEventProcessor.cc.

{
  for (auto& socket : m_sockets) {
    auto message = ZMQMessageFactory::createMessage(EMessageTypes::c_deleteWorkerMessage,
                                                    ZMQParent::createIdentity(pid));
    ZMQParent::send(socket, std::move(message));
 
    if (not waitForConfirmation) {
      continue;
    }
    if (ZMQParent::poll({socket.get()}, 10 * 1000)) {
      auto acceptMessage = ZMQMessageFactory::fromSocket<ZMQNoIdMessage>(socket);
      B2ASSERT("Should be an accept message", acceptMessage->isMessage(EMessageTypes::c_confirmMessage));
    } else {
      B2FATAL("Did not receive a confirmation message! waitForConfirmation is " << waitForConfirmation);
    }
  }
}

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 57 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_eventMetaDataPtr

StoreObjPtr<EventMetaData> m_eventMetaDataPtr

protectedinherited

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

Definition at line 163 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 172 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 175 of file EventProcessor.h.

m_master

const Module* m_master

protectedinherited

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

Definition at line 153 of file EventProcessor.h.

m_metadataUpdateInterval

double m_metadataUpdateInterval

protectedinherited

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

Definition at line 178 of file EventProcessor.h.

m_moduleList

ModulePtrList m_moduleList

protectedinherited

List of all modules in order initialized.

Definition at line 154 of file EventProcessor.h.

m_parent

ZMQParent m_parent

private

An instance of a ZMQParent to create sockets for unregistering workers.

Definition at line 55 of file HLTEventProcessor.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 166 of file EventProcessor.h.

m_processList

std::vector<int> m_processList

private

The current list of running processes (with their PIDs)

Definition at line 57 of file HLTEventProcessor.h.

m_processStatisticsPtr

StoreObjPtr<ProcessStatistics> m_processStatisticsPtr

protectedinherited

Also used in a number of places.

Definition at line 169 of file EventProcessor.h.

m_profileModule

Module* m_profileModule = nullptr

protectedinherited

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

protectedinherited

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

Definition at line 157 of file EventProcessor.h.

m_sockets

std::vector<std::unique_ptr<zmq::socket_t> > m_sockets

private

The created sockets for unregistering workers. TODO: use connections.

Definition at line 59 of file HLTEventProcessor.h.

m_steerRootInputModuleOn

bool m_steerRootInputModuleOn = false

protectedinherited

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:

• daq/hbasf2/utils/include/HLTEventProcessor.h
• daq/hbasf2/utils/src/HLTEventProcessor.cc