Belle II Software light-2406-ragdoll
EventProcessor.cc
1/**************************************************************************
2 * basf2 (Belle II Analysis Software Framework) *
3 * Author: The Belle II Collaboration *
4 * *
5 * See git log for contributors and copyright holders. *
6 * This file is licensed under LGPL-3.0, see LICENSE.md. *
7 **************************************************************************/
8
9//first because of python include
10#include <framework/core/Module.h>
11
12#include <framework/core/EventProcessor.h>
13
14#include <framework/core/PathIterator.h>
15#include <framework/datastore/DataStore.h>
16#include <framework/database/DBStore.h>
17#include <framework/database/Database.h>
18#include <framework/logging/Logger.h>
19#include <framework/core/Environment.h>
20#include <framework/core/DataFlowVisualization.h>
21#include <framework/core/RandomNumbers.h>
22#include <framework/core/MetadataService.h>
23#include <framework/gearbox/Unit.h>
24#include <framework/utilities/Utils.h>
25
26#ifdef HAS_CALLGRIND
27#include <valgrind/callgrind.h>
34#define CALL_MODULE(module,x) \
35 if(m_profileModule && m_profileModule==module && RUNNING_ON_VALGRIND){\
36 CALLGRIND_START_INSTRUMENTATION;\
37 module->x();\
38 CALLGRIND_STOP_INSTRUMENTATION;\
39 }else{\
40 module->x();\
41 }
42#else
43#define CALL_MODULE(module, x) module->x()
44#endif
45
46#include <TROOT.h>
47
48#include <csignal>
49#include <unistd.h>
50#include <cstring>
51
52using namespace std;
53using namespace Belle2;
54
55namespace {
56 static int gSignalReceived = 0;
57 static void signalHandler(int signal)
58 {
59 gSignalReceived = signal;
60
61 if (signal == SIGINT) {
62 EventProcessor::writeToStdErr("Received Ctrl+C, basf2 will exit safely. (Press Ctrl+\\ (SIGQUIT) to abort immediately - this will break output files.)\n");
63 }
64 }
65}
67 runtime_error("Execution stopped by signal " + to_string(signal_) + "!"),
68 signal(signal_)
69{
70}
71
72void EventProcessor::writeToStdErr(const char msg[])
73{
74 //signal handlers are called asynchronously, making many standard functions (including output) dangerous
75 //write() is, however, safe, so we'll use that to write to stderr.
76
77 //strlen() not explicitly in safe list, but doesn't have any error handling routines that might alter global state
78 const int len = strlen(msg);
79
80 int rc = write(STDERR_FILENO, msg, len);
81 (void) rc; //ignore return value (there's nothing we can do about a failed write)
82
83}
84
87{
88
89}
90
92
93namespace {
96 struct NumberEventsOverrideGuard {
98 explicit NumberEventsOverrideGuard(unsigned int newValue)
99 {
102 }
104 ~NumberEventsOverrideGuard()
105 {
107 }
109 unsigned int m_maxEvent;
110 };
111}
112
114{
115 //Check whether the number of events was set via command line argument
116 unsigned int numEventsArgument = Environment::Instance().getNumberEventsOverride();
117 if ((numEventsArgument > 0) && ((maxEvent == 0) || (maxEvent > numEventsArgument))) {
118 return numEventsArgument;
119 }
120 return maxEvent;
121}
122
123void EventProcessor::process(const PathPtr& startPath, long maxEvent)
124{
125 maxEvent = getMaximumEventNumber(maxEvent);
126 // Make sure the NumberEventsOverride reflects the actual number if
127 // process(path, N) was used instead of -n and that it's reset to what it was
128 // after we're done with processing()
129 NumberEventsOverrideGuard numberOfEventsOverrideGuard(maxEvent);
130
131 //Get list of modules which could be executed during the data processing.
132 ModulePtrList moduleList = startPath->buildModulePathList();
133
134 //Find the adress of the module we want to profile
135 if (!m_profileModuleName.empty()) {
136 for (const auto& module : moduleList) {
137 if (module->getName() == m_profileModuleName) {
138 m_profileModule = module.get();
139 break;
140 }
141 }
142 if (!m_profileModule)
143 B2FATAL("Module profiling was requested via --profile, but no module '" << m_profileModuleName << "' was found!");
144 }
145
146 //Initialize modules
147 processInitialize(moduleList);
148
149 // SteerRootInputModule might have changed the number of events to be processed
150 for (const auto& module : moduleList) {
151 if (module->getType() == "SteerRootInput") {
152 if (maxEvent != Environment::Instance().getNumberEventsOverride()) {
153 B2INFO("Module 'SteerRootInputModule' is controlling the number of processed events.");
156 }
157 break;
158 }
159 }
160
161 //do we want to visualize DataStore input/ouput?
162 if (Environment::Instance().getVisualizeDataFlow()) {
163 DataFlowVisualization v(&DataStore::Instance().getDependencyMap());
164 v.visualizePath("dataflow.dot", *startPath);
165 }
166
167 //Don't start processing in case of no master module
168 if (!m_master) {
169 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.");
170 }
171
172 //Check if errors appeared. If yes, don't start the event processing.
174 if ((numLogError == 0) && m_master) {
176 try {
177 processCore(startPath, moduleList, maxEvent); //Do the event processing
178 } catch (StoppedBySignalException& e) {
179 if (e.signal != SIGINT) {
180 // close all open ROOT files, ROOT's exit handler will crash otherwise
181 gROOT->GetListOfFiles()->Delete();
182
183 B2FATAL(e.what());
184 }
185 //in case of SIGINT, we move on to processTerminate() to shut down safely
186 } catch (...) {
188 B2ERROR("Exception occured in exp/run/evt: "
189 << m_eventMetaDataPtr->getExperiment() << " / "
190 << m_eventMetaDataPtr->getRun() << " / "
191 << m_eventMetaDataPtr->getEvent());
192 throw;
193 }
194
195 } else {
196 B2FATAL(numLogError << " ERROR(S) occurred! The processing of events will not be started.");
197 }
198
199 //Terminate modules
200 processTerminate(moduleList);
201
203
204 if (gSignalReceived == SIGINT) {
205 const auto msg = R"(Processing aborted via SIGINT, terminating.
206 Output files have been closed safely and should be readable. However
207 processing was NOT COMPLETE. The output files do contain only events
208 processed until this point.)";
210 B2ERROR(msg
211 << LogVar("last_experiment", m_eventMetaDataPtr->getExperiment())
212 << LogVar("last_run", m_eventMetaDataPtr->getRun())
213 << LogVar("last_event", m_eventMetaDataPtr->getEvent()));
214 else
215 B2ERROR(msg);
216 installSignalHandler(SIGINT, SIG_DFL);
217 raise(SIGINT);
218 }
219}
220
221
222//============================================================================
223// Protected methods
224//============================================================================
225
227{
228 LogSystem& logSystem = LogSystem::Instance();
229 const bool collectStats = !Environment::Instance().getNoStats();
230 // set up logging
231 logSystem.updateModule(&(module->getLogConfig()), module->getName());
232 // set up statistics is requested
233 if (collectStats) m_processStatisticsPtr->startModule();
234 // call module
235 CALL_MODULE(module, event);
236 // stop timing
237 if (collectStats) m_processStatisticsPtr->stopModule(module, ModuleStatistics::c_Event);
238 // reset logging
239 logSystem.updateModule(nullptr);
240};
241
242void EventProcessor::processInitialize(const ModulePtrList& modulePathList, bool setEventInfo)
243{
244 LogSystem& logSystem = LogSystem::Instance();
246
247 m_processStatisticsPtr.registerInDataStore();
248 //TODO I might want to overwrite it in initialize (e.g. if read from file)
249 // For parallel processing or subevents, I don't want that, though.
250 // Maybe make this a function argument?
252 m_processStatisticsPtr.create();
253 m_processStatisticsPtr->startGlobal();
254
256
257 for (const ModulePtr& modPtr : modulePathList) {
258 Module* module = modPtr.get();
259
260 if (module->hasUnsetForcedParams()) {
261 //error message was printed by module
262 continue;
263 }
264
265 //Set the module dependent log level
266 logSystem.updateModule(&(module->getLogConfig()), module->getName());
268
269 //Do initialization
270 m_processStatisticsPtr->initModule(module);
271 m_processStatisticsPtr->startModule();
272 CALL_MODULE(module, initialize);
274
275 //Set the global log level
276 logSystem.updateModule(nullptr);
277
278 //Check whether this is the master module
279 if (!m_master && DataStore::Instance().getEntry(m_eventMetaDataPtr) != nullptr) {
280 B2DEBUG(100, "Found module providing EventMetaData: " << module->getName());
281 m_master = module;
282 if (setEventInfo) {
283 callEvent(module);
284 // update Database payloads: we now have valid event meta data unless
285 // we don't process any events
287 }
288 }
289
290 if (gSignalReceived != 0) {
291 throw StoppedBySignalException(gSignalReceived);
292 }
293 }
295}
296
297void EventProcessor::installSignalHandler(int sig, void (*fn)(int))
298{
299 struct sigaction s;
300 memset(&s, '\0', sizeof(s));
301
302 s.sa_handler = fn;
303 sigemptyset(&s.sa_mask);
304 if (sig == SIGCHLD)
305 s.sa_flags |= SA_NOCLDSTOP; //don't produce signal when children are stopped
306
307 if (sigaction(sig, &s, nullptr) != 0) {
308 B2FATAL("Cannot setup signal handler for signal " << sig);
309 }
310}
311
313{
314 if (!fn)
315 fn = signalHandler;
316 installSignalHandler(SIGINT, fn);
317 installSignalHandler(SIGTERM, fn);
318 installSignalHandler(SIGQUIT, fn);
319}
320
321bool EventProcessor::processEvent(PathIterator moduleIter, bool skipMasterModule)
322{
323 double time = Utils::getClock() / Unit::s;
325 MetadataService::Instance().addBasf2Status("running event loop");
327 }
328
329 const bool collectStats = !Environment::Instance().getNoStats();
330
331 while (!moduleIter.isDone()) {
332 Module* module = moduleIter.get();
333
334 // run the module ... unless we don't want to
335 if (!(skipMasterModule && module == m_master)) {
336 callEvent(module);
337 }
338
339 //Check for end of data
340 if ((m_eventMetaDataPtr && (m_eventMetaDataPtr->isEndOfData())) ||
341 ((module == m_master) && !m_eventMetaDataPtr)) {
342 if ((module != m_master) && !m_steerRootInputModuleOn) {
343 B2WARNING("Event processing stopped by module '" << module->getName() <<
344 "', which is not in control of event processing (does not provide EventMetaData)");
345 }
346 return true;
347 }
348
349 //Handle EventMetaData changes by master module
350 if (module == m_master) {
351
352 //initialize random number state for the event
354
355 //Check for a change of the run
356 if ((m_eventMetaDataPtr->getExperiment() != m_previousEventMetaData.getExperiment()) ||
358
359 if (collectStats)
360 m_processStatisticsPtr->suspendGlobal();
361
363 processBeginRun(skipMasterModule);
364
365 if (collectStats)
366 m_processStatisticsPtr->resumeGlobal();
367 }
368
370
371 //make sure we use the event dependent generator again
373
375
376 } else {
377 //Check for a second master module. Cannot do this if we skipped the
378 //master module as the EventMetaData is probably set before we call this
379 //function
380 if (!skipMasterModule && m_eventMetaDataPtr &&
382 B2FATAL("Two modules setting EventMetaData were discovered: " << m_master->getName() << " and " << module->getName());
383 }
384 }
385
386 if (gSignalReceived != 0) {
387 throw StoppedBySignalException(gSignalReceived);
388 }
389
390 //Check for the module conditions, evaluate them and if one is true switch to the new path
391 if (module->evalCondition()) {
392 PathPtr condPath = module->getConditionPath();
393 //continue with parent Path after condition path is executed?
394 if (module->getAfterConditionPath() == Module::EAfterConditionPath::c_Continue) {
395 moduleIter = PathIterator(condPath, moduleIter);
396 } else {
397 moduleIter = PathIterator(condPath);
398 }
399 } else {
400 moduleIter.next();
401 }
402 } //end module loop
403 return false;
404}
405
406void EventProcessor::processCore(const PathPtr& startPath, const ModulePtrList& modulePathList, long maxEvent, bool isInputProcess)
407{
409 m_moduleList = modulePathList;
410 //Remember the previous event meta data, and identify end of data meta data
411 m_previousEventMetaData.setEndOfData(); //invalid start state
412
413 const bool collectStats = !Environment::Instance().getNoStats();
414
415 //Loop over the events
416 long currEvent = 0;
417 bool endProcess = false;
418 while (!endProcess) {
419 if (collectStats)
420 m_processStatisticsPtr->startGlobal();
421
422 PathIterator moduleIter(startPath);
423 endProcess = processEvent(moduleIter, isInputProcess && currEvent == 0);
424
425 //Delete event related data in DataStore
427
428 currEvent++;
429 if ((maxEvent > 0) && (currEvent >= maxEvent)) endProcess = true;
430 if (collectStats)
432 } //end event loop
433
434 //End last run
435 m_eventMetaDataPtr.create();
437}
438
439
441{
443
444 LogSystem& logSystem = LogSystem::Instance();
445 ModulePtrList::const_reverse_iterator listIter;
446 m_processStatisticsPtr->startGlobal();
447
448 for (listIter = modulePathList.rbegin(); listIter != modulePathList.rend(); ++listIter) {
449 Module* module = listIter->get();
450
451 //Set the module dependent log level
452 logSystem.updateModule(&(module->getLogConfig()), module->getName());
453
454 //Do termination
455 m_processStatisticsPtr->startModule();
456 CALL_MODULE(module, terminate);
458
459 //Set the global log level
460 logSystem.updateModule(nullptr);
461 }
462
464}
465
466
468{
469 MetadataService::Instance().addBasf2Status("beginning run");
470
471 m_inRun = true;
472 auto dbsession = Database::Instance().createScopedUpdateSession(); // cppcheck-suppress unreadVariable
473
474 LogSystem& logSystem = LogSystem::Instance();
475 m_processStatisticsPtr->startGlobal();
476
477 if (!skipDB) DBStore::Instance().update();
478
479 //initialize random generator for end run
481
482 for (const ModulePtr& modPtr : m_moduleList) {
483 Module* module = modPtr.get();
484
485 //Set the module dependent log level
486 logSystem.updateModule(&(module->getLogConfig()), module->getName());
487
488 //Do beginRun() call
489 m_processStatisticsPtr->startModule();
490 CALL_MODULE(module, beginRun);
492
493 //Set the global log level
494 logSystem.updateModule(nullptr);
495 }
496
498}
499
500
502{
504
505 if (!m_inRun)
506 return;
507 m_inRun = false;
508
509 LogSystem& logSystem = LogSystem::Instance();
510 m_processStatisticsPtr->startGlobal();
511
512 const EventMetaData newEventMetaData = *m_eventMetaDataPtr;
514
515 //initialize random generator for end run
517
518 for (const ModulePtr& modPtr : m_moduleList) {
519 Module* module = modPtr.get();
520
521 //Set the module dependent log level
522 logSystem.updateModule(&(module->getLogConfig()), module->getName());
523
524 //Do endRun() call
525 m_processStatisticsPtr->startModule();
526 CALL_MODULE(module, endRun);
528
529 //Set the global log level
530 logSystem.updateModule(nullptr);
531 }
532 *m_eventMetaDataPtr = newEventMetaData;
533
535}
class to visualize data flow between modules.
void visualizePath(const std::string &filename, const Path &path)
Create graphs with datastore inputs/outputs of each module in path.
In the store you can park objects that have to be accessed by various modules.
Definition: DataStore.h:51
@ c_Event
Different object in each event, all objects/arrays are invalidated after event() function has been ca...
Definition: DataStore.h:59
static DataStore & Instance()
Instance of singleton Store.
Definition: DataStore.cc:54
void setInitializeActive(bool active)
Setter for m_initializeActive.
Definition: DataStore.cc:94
void invalidateData(EDurability durability)
Clears all registered StoreEntry objects of a specified durability, invalidating all objects.
Definition: DataStore.cc:715
DependencyMap & getDependencyMap()
Return map of depedencies between modules.
Definition: DataStore.h:524
void setModule(const Module &mod)
Set the current module (for getCurrentModuleInfo())
Definition: DependencyMap.h:60
void setNumberEventsOverride(unsigned int nevents)
Override the number of events in run 1 for EventInfoSetter module.
Definition: Environment.h:64
bool getNoStats() const
Disable collection of statistics during event processing.
Definition: Environment.h:199
static Environment & Instance()
Static method to get a reference to the Environment instance.
Definition: Environment.cc:28
unsigned int getNumberEventsOverride() const
Returns number of events in run 1 for EventInfoSetter module, or 0 for no override.
Definition: Environment.h:67
Store event, run, and experiment numbers.
Definition: EventMetaData.h:33
void setEndOfData()
Marks the end of the data processing.
int getRun() const
Run Getter.
int getExperiment() const
Experiment Getter.
Exception thrown when execution is stopped by a signal.
void processEndRun()
Calls the end run methods of all modules.
void processInitialize(const ModulePtrList &modulePathList, bool setEventInfo=true)
Initializes the modules.
bool m_inRun
Are we currently in a run? If yes, processEndRun() needs to do something.
std::string m_profileModuleName
Name of the module which should be profiled, empty if no profiling is requested.
void processBeginRun(bool skipDB=false)
Calls the begin run methods of all 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,...
double m_lastMetadataUpdate
Time in seconds of last call for metadata update in event loop.
static void installSignalHandler(int sig, void(*fn)(int))
Install a signal handler 'fn' for given signal.
StoreObjPtr< ProcessStatistics > m_processStatisticsPtr
Also used in a number of places.
void process(const PathPtr &startPath, long maxEvent=0)
Processes the full module chain, starting with the first module in the given 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.
Module * m_profileModule
Adress of the module which we want to profile, nullptr if no profiling is requested.
static void writeToStdErr(const char msg[])
async-safe method to write something to STDERR.
bool processEvent(PathIterator moduleIter, bool skipMasterModule)
Calls event() functions on all modules for the current event.
virtual ~EventProcessor()
Destructor.
const Module * m_master
The master module that determines the experiment/run/event number.
EventProcessor()
Constructor.
StoreObjPtr< EventMetaData > m_eventMetaDataPtr
EventMetaData is used by processEvent()/processCore().
ModulePtrList m_moduleList
List of all modules in order initialized.
long getMaximumEventNumber(long maxEvent) const
Calculate the maximum event number out of the argument from command line and the environment.
EventMetaData m_previousEventMetaData
Stores state of EventMetaData before it was last changed.
bool m_steerRootInputModuleOn
True if the SteerRootInputModule is in charge for event processing.
double m_metadataUpdateInterval
Minimal time difference in seconds for metadata updates in event loop.
static void installMainSignalHandlers(void(*fn)(int)=nullptr)
Install signal handler for INT, TERM and QUIT signals.
@ c_Error
Error: for things that went wrong and have to be fixed.
Definition: LogConfig.h:30
Class for logging debug, info and error messages.
Definition: LogSystem.h:46
void updateModule(const LogConfig *moduleLogConfig=nullptr, const std::string &moduleName="")
Sets the log configuration to the given module log configuration and sets the module name This method...
Definition: LogSystem.h:191
void printErrorSummary()
Print error/warning summary at end of execution.
Definition: LogSystem.cc:206
int getMessageCounter(LogConfig::ELogLevel logLevel) const
Returns the number of logging calls per log level.
Definition: LogSystem.cc:161
static LogSystem & Instance()
Static method to get a reference to the LogSystem instance.
Definition: LogSystem.cc:31
void addBasf2Status(const std::string &message="")
Add metadata of basf2 status.
static MetadataService & Instance()
Static method to get a reference to the MetadataService instance.
@ c_Init
Counting time/calls in initialize()
@ c_EndRun
Counting time/calls in endRun()
@ c_Term
Counting time/calls in terminate()
@ c_BeginRun
Counting time/calls in beginRun()
@ c_Event
Counting time/calls in event()
Base class for Modules.
Definition: Module.h:72
const std::string & getName() const
Returns the name of the module.
Definition: Module.h:187
Iterator over a Path (returning Module pointers).
Definition: PathIterator.h:26
void next()
increment.
Definition: PathIterator.h:49
bool isDone() const
Are we finished iterating?
Definition: PathIterator.h:72
Module * get() const
dereference.
Definition: PathIterator.h:75
static void initializeEndRun()
Initialize run independent random generator for end run.
static void initializeBeginRun()
Initialize run independent random generator for begin run.
static void useEventDependent()
Set Event dependent Random Generator as current one.
static void initializeEvent(bool force=false)
Initialize event information.
static const double s
[second]
Definition: Unit.h:95
Class to store variables with their name which were sent to the logging service.
std::shared_ptr< Path > PathPtr
Defines a pointer to a path object as a boost shared pointer.
Definition: Path.h:35
static Database & Instance()
Instance of a singleton Database.
Definition: Database.cc:42
static DBStore & Instance()
Instance of a singleton DBStore.
Definition: DBStore.cc:28
std::shared_ptr< Module > ModulePtr
Defines a pointer to a module object as a boost shared pointer.
Definition: Module.h:43
std::list< ModulePtr > ModulePtrList
Defines a std::list of shared module pointers.
Definition: Module.h:584
void updateEvent()
Updates all intra-run dependent objects.
Definition: DBStore.cc:142
void update()
Updates all objects that are outside their interval of validity.
Definition: DBStore.cc:79
ScopeGuard createScopedUpdateSession()
Make sure we have efficient http pipelinging during initialize/beginRun but don't keep session alive ...
Definition: Database.cc:62
double getClock()
Return current value of the real-time clock.
Definition: Utils.cc:66
Abstract base class for different kinds of events.
Definition: ClusterUtils.h:24
STL namespace.