LDMX Software
framework::Process Class Reference

Class which represents the process under execution. More...

#include <Process.h>

Public Member Functions

 Process (const framework::config::Parameters &configuration)
 Class constructor.
 
 ~Process ()
 Class Destructor.
 
const std::string & getPassName () const
 Get the processing pass label.
 
int getRunNumber () const
 Get the current run number or the run number to be used when initiating new events from the job.
 
const ldmx::EventHeadergetEventHeader () const
 Get the pointer to the current event header, if defined.
 
const ldmx::RunHeadergetRunHeader () const
 Get the pointer to the current run header, if defined.
 
ConditionsgetConditions ()
 Get a reference to the conditions system.
 
int getLogFrequency () const
 Get the frequency with which the event information is printed.
 
void run ()
 Run the process.
 
void requestFinish ()
 Request that the processing finish with this event.
 
TDirectory * makeHistoDirectory (const std::string &dirName)
 Construct a TDirectory* for the given module.
 
TDirectory * openHistoFile ()
 Open a ROOT TFile to write histograms and TTrees.
 
StorageControlgetStorageController ()
 Access the storage control unit for this process.
 
void setEventHeader (ldmx::EventHeader *h)
 Set the pointer to the current event header, used only for tests.
 

Private Member Functions

bool process (int n, int n_tries, Event &event) const
 Process the input event through the sequence of processors.
 
void newRun (ldmx::RunHeader &header)
 Run through the processors and let them know that we are starting a new run.
 
void onFileOpen (EventFile &file) const
 File is being opened.
 
void onFileClose (EventFile &file) const
 File is begin closed.
 
 enableLogging ("Process")
 Turn on logging for our process.
 

Private Attributes

framework::config::Parameters config_
 The parameters used to configure this class.
 
std::string pass_name_
 Processing pass name.
 
int event_limit_
 Limit on events to process.
 
int min_events_
 When reading a file in, what's the first event to read.
 
int total_events_
 Number of events we'd like to produce independetly of the number of tries it would take.
 
int log_frequency_
 The frequency with which event info is printed.
 
int max_tries_
 Maximum number of attempts to make before giving up on an event.
 
bool skip_corrupted_input_files_
 allow the Process to skip input files that are corrupted
 
StorageControl storage_controller_
 Storage controller.
 
std::vector< EventProcessor * > sequence_
 Ordered list of EventProcessors to execute.
 
Conditions conditions_
 Set of ConditionsProviders.
 
std::vector< std::string > input_files_
 List of input files to process.
 
std::vector< std::string > output_files_
 List of output file names.
 
int compression_setting_
 Compression setting to pass to output files.
 
std::vector< std::string > drop_keep_rules_
 Set of drop/keep rules.
 
int run_for_generation_ {1}
 Run number to use if generating events.
 
std::string histo_filename_
 Filename for histograms and other user products.
 
const ldmx::EventHeaderevent_header_ {0}
 Pointer to the current EventHeader, used for Conditions information.
 
ldmx::RunHeaderrun_header_ {0}
 Pointer to the current RunHeader, used for Conditions information.
 
TFile * histo_t_file_ {0}
 TFile for histograms and other user products.
 
performance::Trackerperformance_ {0}
 class with calls backs to track performance measurements of software
 

Detailed Description

Class which represents the process under execution.

Definition at line 37 of file Process.h.

Constructor & Destructor Documentation

◆ Process()

framework::Process::Process ( const framework::config::Parameters & configuration)

Class constructor.

Parameters
configurationParameters to configure process with

Definition at line 29 of file Process.cxx.

30 : conditions_{*this} {
31 config_ = configuration;
32
33 pass_name_ = configuration.get<std::string>("passName", "");
34 histo_filename_ = configuration.get<std::string>("histogramFile", "");
35
36 max_tries_ = configuration.get<int>("maxTriesPerEvent", 1);
37 event_limit_ = configuration.get<int>("maxEvents", -1);
38 min_events_ = configuration.get<int>("minEvents", -1);
39 total_events_ = configuration.get<int>("totalEvents", -1);
40 log_frequency_ = configuration.get<int>("logFrequency", -1);
41 compression_setting_ = configuration.get<int>("compressionSetting", 9);
43 configuration.get<bool>("skipCorruptedInputFiles", false);
44
45 input_files_ = configuration.get<std::vector<std::string>>("inputFiles", {});
47 configuration.get<std::vector<std::string>>("outputFiles", {});
48 drop_keep_rules_ = configuration.get<std::vector<std::string>>("keep", {});
49
50 event_header_ = 0;
51
52 // set up the logging for this run
53 logging::open(configuration.get<framework::config::Parameters>("logger", {}));
54
55 auto run{configuration.get<int>("run", -1)};
56 if (run > 0) run_for_generation_ = run;
57
58 auto libs{configuration.get<std::vector<std::string>>("libraries", {})};
59 std::set<std::string> libraries_loaded;
60 for (const auto &lib : libs) {
61 if (libraries_loaded.find(lib) != libraries_loaded.end()) {
62 continue;
63 }
64
65 void *handle = dlopen(lib.c_str(), RTLD_NOW);
66 if (handle == nullptr) {
67 EXCEPTION_RAISE("LibraryLoadFailure",
68 "Error loading library '" + lib + "':" + dlerror());
69 }
70
71 libraries_loaded.insert(lib);
72 }
73
75 configuration.get<bool>("skimDefaultIsKeep", true));
76 auto skim_rules{configuration.get<std::vector<std::string>>("skimRules", {})};
77 for (size_t i = 0; i < skim_rules.size(); i += 2) {
78 storage_controller_.addRule(skim_rules[i], skim_rules[i + 1]);
79 }
80
81 auto sequence{configuration.get<std::vector<framework::config::Parameters>>(
82 "sequence", {})};
83 if (sequence.empty() && configuration.get<bool>("testingMode", false)) {
84 EXCEPTION_RAISE(
85 "NoSeq",
86 "No sequence has been defined. What should I be doing?\nUse "
87 "p.sequence to tell me what processors to run.");
88 }
89 for (auto proc : sequence) {
90 auto class_name{proc.get<std::string>("className")};
91 auto instance_name{proc.get<std::string>("instanceName")};
92 auto ep{
93 EventProcessor::Factory::get().make(class_name, instance_name, *this)};
94 if (not ep) {
95 EXCEPTION_RAISE("UnableToCreate",
96 "The EventProcessor Factory was unable to create " +
97 instance_name + " of type " + class_name +
98 ". Did you inherit from framework::Producer or "
99 "framework::Analyzer? "
100 "Did you DECLARE_PRODUCER or DECLARE_ANALYZER in the "
101 "implementation (.cxx) file? "
102 "Did you use the class's full name (including "
103 "namespaces) in the Python configuration class? "
104 "Does the Python configuration class reference the "
105 "correct library it is a part of?");
106 }
107 auto histograms{
108 proc.get<std::vector<framework::config::Parameters>>("histograms", {})};
109 if (!histograms.empty()) {
110 ep.value()->getHistoDirectory();
111 ep.value()->createHistograms(histograms);
112 }
113 ep.value()->configure(proc);
114 sequence_.push_back(ep.value());
115 }
116
117 auto conditions_object_providers{
118 configuration.get<std::vector<framework::config::Parameters>>(
119 "conditionsObjectProviders", {})};
120 for (auto cop : conditions_object_providers) {
121 auto class_name{cop.get<std::string>("className")};
122 auto object_name{cop.get<std::string>("objectName")};
123 auto tag_name{cop.get<std::string>("tagName")};
124 conditions_.createConditionsObjectProvider(class_name, object_name,
125 tag_name, cop);
126 }
127
128 bool log_performance = configuration.get<bool>("logPerformance", false);
129 if (log_performance) {
130 std::vector<std::string> names{sequence_.size()};
131 for (std::size_t i{0}; i < sequence_.size(); i++) {
132 names[i] = sequence_[i]->getName();
133 }
135 new performance::Tracker(makeHistoDirectory("performance"), names);
136 }
137}
void createConditionsObjectProvider(const std::string &classname, const std::string &instancename, const std::string &tagname, const framework::config::Parameters &params)
Create a ConditionsObjectProvider given the information.
int log_frequency_
The frequency with which event info is printed.
Definition Process.h:169
std::vector< EventProcessor * > sequence_
Ordered list of EventProcessors to execute.
Definition Process.h:183
bool skip_corrupted_input_files_
allow the Process to skip input files that are corrupted
Definition Process.h:177
std::string histo_filename_
Filename for histograms and other user products.
Definition Process.h:211
int max_tries_
Maximum number of attempts to make before giving up on an event.
Definition Process.h:172
int run_for_generation_
Run number to use if generating events.
Definition Process.h:208
int compression_setting_
Compression setting to pass to output files.
Definition Process.h:202
void run()
Run the process.
Definition Process.cxx:153
int event_limit_
Limit on events to process.
Definition Process.h:158
std::string pass_name_
Processing pass name.
Definition Process.h:155
int total_events_
Number of events we'd like to produce independetly of the number of tries it would take.
Definition Process.h:166
std::vector< std::string > drop_keep_rules_
Set of drop/keep rules.
Definition Process.h:205
TDirectory * makeHistoDirectory(const std::string &dirName)
Construct a TDirectory* for the given module.
Definition Process.cxx:455
performance::Tracker * performance_
class with calls backs to track performance measurements of software
Definition Process.h:223
int min_events_
When reading a file in, what's the first event to read.
Definition Process.h:161
StorageControl storage_controller_
Storage controller.
Definition Process.h:180
std::vector< std::string > output_files_
List of output file names.
Definition Process.h:193
std::vector< std::string > input_files_
List of input files to process.
Definition Process.h:190
const ldmx::EventHeader * event_header_
Pointer to the current EventHeader, used for Conditions information.
Definition Process.h:214
Conditions conditions_
Set of ConditionsProviders.
Definition Process.h:186
framework::config::Parameters config_
The parameters used to configure this class.
Definition Process.h:152
void setDefaultKeep(bool keep)
Set the default state.
void addRule(const std::string &processor_pat, const std::string &purpose_pat)
Add a listening rule.
Class encapsulating parameters for configuring a processor.
Definition Parameters.h:29
const T & get(const std::string &name) const
Retrieve the parameter of the given name.
Definition Parameters.h:78

References framework::StorageControl::addRule(), compression_setting_, conditions_, config_, framework::Conditions::createConditionsObjectProvider(), drop_keep_rules_, event_header_, event_limit_, framework::config::Parameters::get(), histo_filename_, input_files_, log_frequency_, makeHistoDirectory(), max_tries_, min_events_, output_files_, pass_name_, performance_, run(), run_for_generation_, sequence_, framework::StorageControl::setDefaultKeep(), skip_corrupted_input_files_, storage_controller_, and total_events_.

◆ ~Process()

framework::Process::~Process ( )

Class Destructor.

Cleans up sequence of EventProcessors. These processors were created by ConfigurePython and should be deleted.

Definition at line 139 of file Process.cxx.

139 {
140 // need to delete the performance object so that it is
141 // written before we close the histogram file below
142 if (performance_) delete performance_;
143 for (EventProcessor *ep : sequence_) {
144 delete ep;
145 }
146 if (histo_t_file_) {
147 histo_t_file_->Write();
148 delete histo_t_file_;
149 histo_t_file_ = 0;
150 }
151}
TFile * histo_t_file_
TFile for histograms and other user products.
Definition Process.h:220

References histo_t_file_, performance_, and sequence_.

Member Function Documentation

◆ getConditions()

Conditions & framework::Process::getConditions ( )
inline

Get a reference to the conditions system.

Definition at line 79 of file Process.h.

79{ return conditions_; }

References conditions_.

Referenced by framework::ConditionsObjectProvider::requestParentCondition().

◆ getEventHeader()

const ldmx::EventHeader * framework::Process::getEventHeader ( ) const
inline

Get the pointer to the current event header, if defined.

Definition at line 69 of file Process.h.

69{ return event_header_; }

References event_header_.

Referenced by framework::Conditions::getConditionPtr().

◆ getLogFrequency()

int framework::Process::getLogFrequency ( ) const
inline

Get the frequency with which the event information is printed.

Returns
integer log frequency (negative if turned off)

Definition at line 85 of file Process.h.

85{ return log_frequency_; }

References log_frequency_.

◆ getPassName()

const std::string & framework::Process::getPassName ( ) const
inline

Get the processing pass label.

Returns
The processing pass label.

Definition at line 57 of file Process.h.

57{ return pass_name_; }

References pass_name_.

Referenced by framework::RandomNumberSeedService::onNewRun().

◆ getRunHeader()

const ldmx::RunHeader * framework::Process::getRunHeader ( ) const
inline

Get the pointer to the current run header, if defined.

Definition at line 74 of file Process.h.

74{ return run_header_; }
ldmx::RunHeader * run_header_
Pointer to the current RunHeader, used for Conditions information.
Definition Process.h:217

References run_header_.

◆ getRunNumber()

int framework::Process::getRunNumber ( ) const

Get the current run number or the run number to be used when initiating new events from the job.

Returns
int Run number

Definition at line 451 of file Process.cxx.

451 {
453}
int getRun() const
Return the run number.
Definition EventHeader.h:84

References event_header_, ldmx::EventHeader::getRun(), and run_for_generation_.

◆ getStorageController()

StorageControl & framework::Process::getStorageController ( )
inline

Access the storage control unit for this process.

Definition at line 110 of file Process.h.

110{ return storage_controller_; }

References storage_controller_.

◆ makeHistoDirectory()

TDirectory * framework::Process::makeHistoDirectory ( const std::string & dirName)

Construct a TDirectory* for the given module.

Definition at line 455 of file Process.cxx.

455 {
456 auto owner{openHistoFile()};
457 TDirectory *child = owner->mkdir((char *)dirName.c_str());
458 if (child) child->cd();
459 return child;
460}
TDirectory * openHistoFile()
Open a ROOT TFile to write histograms and TTrees.
Definition Process.cxx:462

References openHistoFile().

Referenced by Process().

◆ newRun()

void framework::Process::newRun ( ldmx::RunHeader & header)
private

Run through the processors and let them know that we are starting a new run.

Parameters
[in]headerRunHeader for the new run

Definition at line 484 of file Process.cxx.

484 {
485 // Producers are allowed to put parameters into
486 // the run header through 'beforeNewRun' method
487
488 // Put the version into the rh string param
489 header.setStringParameter("Pass = " + pass_name_ + ", version",
490 LDMXSW_VERSION);
491 if (performance_) performance_->start(performance::Callback::beforeNewRun, 0);
492 std::size_t i_proc{0};
493 for (auto proc : sequence_) {
494 i_proc++;
495 if (performance_)
496 performance_->start(performance::Callback::beforeNewRun, i_proc);
497 proc->beforeNewRun(header);
498 if (performance_)
499 performance_->stop(performance::Callback::beforeNewRun, i_proc);
500 }
501 if (performance_) performance_->stop(performance::Callback::beforeNewRun, 0);
502 // now run header has been modified by Producers,
503 // it is valid to read from for everyone else in 'onNewRun'
504 if (performance_) performance_->start(performance::Callback::onNewRun, 0);
505 conditions_.onNewRun(header);
506 i_proc = 0;
507 for (auto proc : sequence_) {
508 i_proc++;
509 if (performance_)
510 performance_->start(performance::Callback::onNewRun, i_proc);
511 proc->onNewRun(header);
512 if (performance_)
513 performance_->stop(performance::Callback::onNewRun, i_proc);
514 }
515 if (performance_) performance_->stop(performance::Callback::onNewRun, 0);
516 ldmx_log(info) << header;
517}
void onNewRun(ldmx::RunHeader &)
Calls onNewRun for all ConditionsObjectProviders.
void start(Callback cb, std::size_t i_proc)
start the timer for a specific callback and specific processor
Definition Tracker.cxx:90
void stop(Callback cb, std::size_t i_proc)
stop the timer for a specific callback and specific processor
Definition Tracker.cxx:94
void setStringParameter(const std::string &name, std::string value)
Set a string parameter value.
Definition RunHeader.h:222

References conditions_, framework::Conditions::onNewRun(), pass_name_, performance_, sequence_, ldmx::RunHeader::setStringParameter(), framework::performance::Tracker::start(), and framework::performance::Tracker::stop().

Referenced by run().

◆ onFileClose()

void framework::Process::onFileClose ( EventFile & file) const
private

File is begin closed.

Definition at line 571 of file Process.cxx.

571 {
572 if (performance_) performance_->start(performance::Callback::onFileClose, 0);
573 std::size_t i_proc{0};
574 for (auto proc : sequence_) {
575 i_proc++;
576 if (performance_)
577 performance_->start(performance::Callback::onFileClose, i_proc);
578 proc->onFileClose(file);
579 if (performance_)
580 performance_->stop(performance::Callback::onFileClose, i_proc);
581 }
582 if (performance_) performance_->stop(performance::Callback::onFileClose, 0);
583}

References performance_, sequence_, framework::performance::Tracker::start(), and framework::performance::Tracker::stop().

Referenced by run().

◆ onFileOpen()

void framework::Process::onFileOpen ( EventFile & file) const
private

File is being opened.

Definition at line 557 of file Process.cxx.

557 {
558 if (performance_) performance_->start(performance::Callback::onFileOpen, 0);
559 std::size_t i_proc{0};
560 for (auto proc : sequence_) {
561 i_proc++;
562 if (performance_)
563 performance_->start(performance::Callback::onFileOpen, i_proc);
564 proc->onFileOpen(file);
565 if (performance_)
566 performance_->stop(performance::Callback::onFileOpen, i_proc);
567 }
568 if (performance_) performance_->stop(performance::Callback::onFileOpen, 0);
569}

References performance_, sequence_, framework::performance::Tracker::start(), and framework::performance::Tracker::stop().

Referenced by run().

◆ openHistoFile()

TDirectory * framework::Process::openHistoFile ( )

Open a ROOT TFile to write histograms and TTrees.

Definition at line 462 of file Process.cxx.

462 {
463 TDirectory *owner{nullptr};
464
465 if (histo_filename_.empty()) {
466 // trying to write histograms/ntuples but no file defined
467 EXCEPTION_RAISE(
468 "NoHistFileName",
469 "You did not provide the necessary histogram file name to "
470 "put your histograms (or performance data) in.\n Provide this "
471 "name in the python configuration with 'p.histogramFile = "
472 "\"myHistFile.root\"' where p is the Process object.");
473 } else if (histo_t_file_ == nullptr) {
474 histo_t_file_ = new TFile(histo_filename_.c_str(), "RECREATE");
475 owner = histo_t_file_;
476 } else {
477 owner = histo_t_file_;
478 }
479 owner->cd();
480
481 return owner;
482}

References histo_filename_, and histo_t_file_.

Referenced by makeHistoDirectory().

◆ process()

bool framework::Process::process ( int n,
int n_tries,
Event & event ) const
private

Process the input event through the sequence of processors.

The input counters (for events and tries) are only used to print the status.

Parameters
[in]ncounter for number of events processed
[in]n_triescounter for number of tries on current event
[in,out]eventreference to event we are going to process
Returns
true if event was full processed (false if aborted)

Definition at line 519 of file Process.cxx.

519 {
520 if ((log_frequency_ != -1) && ((n + 1) % log_frequency_ == 0) &&
521 (n_try < 2)) {
522 // only printout event counter if we've enabled log frequency, the event
523 // matches the frequency and we are on the first try
524 TTimeStamp t;
525 ldmx_log(info) << "Processing " << n + 1 << " Run "
526 << event.getEventHeader().getRun() << " Event "
527 << event.getEventHeader().getEventNumber() << " ("
528 << t.AsString("lc") << ")";
529 }
530
531 if (performance_) performance_->start(performance::Callback::process, 0);
532 std::size_t i_proc{0};
533 try {
534 for (auto proc : sequence_) {
535 i_proc++;
536 if (performance_)
537 performance_->start(performance::Callback::process, i_proc);
538 proc->process(event);
539 if (performance_)
540 performance_->stop(performance::Callback::process, i_proc);
541 }
542 } catch (AbortEventException &) {
543 if (performance_) {
544 performance_->stop(performance::Callback::process, i_proc);
545 performance_->stop(performance::Callback::process, 0);
546 performance_->endEvent(false);
547 }
548 return false;
549 }
550 if (performance_) {
551 performance_->stop(performance::Callback::process, 0);
552 performance_->endEvent(true);
553 }
554 return true;
555}
void endEvent(bool completed)
inform us that we finished an event (and whether it was completed or not)
Definition Tracker.cxx:98

References framework::performance::Tracker::endEvent(), log_frequency_, performance_, sequence_, framework::performance::Tracker::start(), and framework::performance::Tracker::stop().

Referenced by run().

◆ requestFinish()

void framework::Process::requestFinish ( )
inline

Request that the processing finish with this event.

Definition at line 95 of file Process.h.

95{ event_limit_ = 0; }

References event_limit_.

◆ run()

void framework::Process::run ( )

Run the process.

Definition at line 153 of file Process.cxx.

153 {
155
156 // Counter to keep track of the number of events that have been
157 // procesed
158 auto n_events_processed{0};
159
160 // make sure the ntuple manager is in a blank state
162
163 // event bus for this process
164 Event the_event(pass_name_);
165 // the EventHeader object is created with the event bus as
166 // one of its members, we obtain a pointer for the header
167 // here so we can share it with the conditions system
168 event_header_ = the_event.getEventHeaderPtr();
169 the_event.getEventHeader().setRun(run_for_generation_);
170
171 // Start by notifying everyone that modules processing is beginning
172 std::size_t i_proc{0};
173 if (performance_)
174 performance_->start(performance::Callback::onProcessStart, 0);
176 for (auto proc : sequence_) {
177 i_proc++;
178 if (performance_)
179 performance_->start(performance::Callback::onProcessStart, i_proc);
180 proc->onProcessStart();
181 if (performance_)
182 performance_->stop(performance::Callback::onProcessStart, i_proc);
183 }
184 if (performance_)
185 performance_->stop(performance::Callback::onProcessStart, 0);
186
187 // If we have no input files, but do have an event number, run for
188 // that number of events and generate an output file.
189 if (input_files_.empty() && event_limit_ > 0) {
190 if (output_files_.empty()) {
191 EXCEPTION_RAISE("InvalidConfig",
192 "No input files or output files were given.");
193 } else if (output_files_.size() > 1) {
194 ldmx_log(warn) << "Several output files given with no input files. "
195 << "Only the first output file '" << output_files_.at(0)
196 << "' will be used.";
197 }
198 std::string output_file_name = output_files_.at(0);
199
200 // Configure the event file to create an output file with no parent. This
201 // requires setting the parameters isOutputFile and isSingleOutput to true.
202 EventFile out_file(config_, output_file_name, nullptr, true, true, false);
203 onFileOpen(out_file);
204 out_file.setupEvent(&the_event);
205
206 for (auto rule : drop_keep_rules_) out_file.addDrop(rule);
207
209 run_header.setRunStart(std::time(nullptr)); // set run starting
210 run_header_ = &run_header; // give handle to run header to process
211 out_file.writeRunHeader(run_header); // add run header to file
212
213 newRun(run_header);
214
215 int total_tries = 0; // total number of tries for entire run
216 int num_tries = 0; // number of tries for the current event number
217 int event_limit = event_limit_;
218 if (total_events_ > 0) {
219 // Have a warning at the first event
220 if (num_tries == 0)
221 ldmx_log(warn) << "The totalEvents was set, so maxEvents and "
222 "maxTriesPerEvent will be ignored!";
223 event_limit = total_events_;
224 }
225 while (n_events_processed < event_limit) {
226 // Check for preemption before processing each event
227 if (preemption_received_) {
228 ldmx_log(fatal)
229 << "Preemption signal received, stopping event generation";
230 break;
231 }
232
233 total_tries++;
234 num_tries++;
235
236 ldmx::EventHeader &eh = the_event.getEventHeader();
238 eh.setEventNumber(n_events_processed + 1);
239 eh.setTimestamp(TTimeStamp());
240
241 // reset the storage controller state
243 logging::Formatter::set(the_event.getEventNumber());
244
245 bool completed = process(n_events_processed, num_tries, the_event);
246
247 out_file.nextEvent(storage_controller_.keepEvent(completed));
248
249 // reset try counter only on successfully completed events
250 if (completed) num_tries = 0;
251
252 // we use modulo here insetad of >= because we want to carry
253 // the number of tries across the number of events processed boundary
254 // total_events_ is set let's not exit until that's reached
255 if (completed or (total_events_ < 0 and num_tries % max_tries_ == 0)) {
256 n_events_processed++; // increment events made
257 NtupleManager::getInstance().fill(); // fill ntuples
258 }
259
261 }
262
263 onFileClose(out_file);
264
265 run_header.setRunEnd(std::time(nullptr));
266 run_header.setNumTries(total_tries);
267 out_file.writeRunTree();
268
269 // Give a warning that this filter has very low efficiency
270 if (n_events_processed < total_tries / 10000) { // integer division is okay
271 ldmx_log(warn)
272 << "Less than 1 event out of every 10k events tried was accepted!";
273 ldmx_log(warn)
274 << "This could be an issue with your filtering and biasing procedure "
275 "since this is incredibly inefficient.";
276 }
277
278 } else {
279 // there are input files
280
281 EventFile *out_file(0);
282
283 bool single_output = false;
284 if (output_files_.size() == 1) {
285 single_output = true;
286 } else if (!output_files_.empty() and
287 output_files_.size() != input_files_.size()) {
288 EXCEPTION_RAISE("Process",
289 "Unable to handle case of different number of input and "
290 "output files (other than zero/one ouput file).");
291 }
292
293 // next, loop through the files
294 int ifile = 0;
295 int was_run = -1;
296 for (auto infilename : input_files_) {
297 EventFile in_file(config_, infilename);
298 if (in_file.isCorrupted()) {
300 ldmx_log(warn) << "Input file '" << infilename
301 << "' was found to be corrupted. Skipping.";
302 continue;
303 } else {
304 EXCEPTION_RAISE(
305 "BadCode",
306 "We should never get here. "
307 "EventFile is corrupted but we aren't skipping corrupted inputs. "
308 "EventFile should be throwing its own exceptions in this case.");
309 }
310 }
311
312 ldmx_log(info) << "Opening file " << infilename;
313 onFileOpen(in_file);
314
315 // configure event file that will be iterated over
316 EventFile *master_file;
317 if (!output_files_.empty()) {
318 // setup new output file if either
319 // 1) we are not in single output mode
320 // 2) this is the first input file
321 if (!single_output or ifile == 0) {
322 // setup new output file
323 out_file = new EventFile(config_, output_files_[ifile], &in_file,
324 single_output);
325 ifile++;
326
327 // setup theEvent we will iterate over
328 if (out_file) {
329 out_file->setupEvent(&the_event);
330 master_file = out_file;
331 } else {
332 EXCEPTION_RAISE("Process", "Unable to construct output file for " +
333 output_files_[ifile]);
334 }
335
336 for (auto rule : drop_keep_rules_) out_file->addDrop(rule);
337
338 } else {
339 // all other input files
340 out_file->updateParent(&in_file);
341 master_file = out_file;
342
343 } // check if in singleOutput mode
344
345 } else {
346 // empty output file list, use inputFile as master file
347 in_file.setupEvent(&the_event);
348 master_file = &in_file;
349 }
350
351 // In case we'd like to skip up to the event of min_events_
352 while (n_events_processed < (min_events_ - 1) &&
353 master_file->nextEvent(false)) {
354 n_events_processed++;
355 }
356
357 bool event_completed = true;
358 while (!preemption_received_ &&
359 master_file->nextEvent(
360 storage_controller_.keepEvent(event_completed)) &&
361 ((event_limit_ < 0) || (n_events_processed < event_limit_))) {
362 // clean up for storage control calculation
364 logging::Formatter::set(the_event.getEventNumber());
365
366 // notify for new run if necessary
367 if (the_event.getEventHeader().getRun() != was_run) {
368 was_run = the_event.getEventHeader().getRun();
369 ldmx::RunHeader *rh{master_file->getRunHeaderPtr(was_run)};
370 if (rh != nullptr) {
371 run_header_ = rh;
372 ldmx_log(info) << "Got new run header from '"
373 << master_file->getFileName() << "'";
375 } else {
376 ldmx_log(warn) << "Run header for run " << was_run
377 << " was not found!";
378 }
379 }
380
381 event_completed = process(n_events_processed, 1, the_event);
382
383 if (event_completed) NtupleManager::getInstance().fill();
385
386 n_events_processed++;
387 } // loop through events
388
389 if (preemption_received_) {
390 ldmx_log(fatal) << "Preemption signal received, stopping event "
391 "processing and closing files";
392 }
393
394 bool leave_early{false};
395 if (event_limit_ > 0 && n_events_processed == event_limit_) {
396 ldmx_log(info) << "Reached event limit of " << event_limit_
397 << " events";
398 leave_early = true;
399 }
400
401 if (event_limit_ == 0 && n_events_processed > event_limit_) {
402 ldmx_log(warn) << "Processing interrupted";
403 leave_early = true;
404 }
405
406 ldmx_log(info) << "Closing file " << infilename;
407 onFileClose(in_file);
408
409 // Reset the event in case of multiple input files
410 the_event.onEndOfFile();
411
412 if (out_file and !single_output) {
413 out_file->writeRunTree();
414 delete out_file;
415 out_file = nullptr;
416 }
417
418 if (leave_early) {
419 break;
420 }
421 } // loop through input files
422
423 if (out_file) {
424 // close outFile
425 // outFile would survive to here in single output mode
426 out_file->writeRunTree();
427 delete out_file;
428 out_file = nullptr;
429 }
430
431 } // are there input files? if-else tree
432
433 // finally, notify everyone that we are stopping
434 if (performance_) performance_->start(performance::Callback::onProcessEnd, 0);
435 i_proc = 0;
436 for (auto proc : sequence_) {
437 i_proc++;
438 if (performance_)
439 performance_->start(performance::Callback::onProcessEnd, i_proc);
440 proc->onProcessEnd();
441 if (performance_)
442 performance_->stop(performance::Callback::onProcessEnd, i_proc);
443 }
444 if (performance_) performance_->stop(performance::Callback::onProcessEnd, 0);
445
446 // we're done so let's close up the logging
447 logging::close();
449}
void onProcessStart()
Calls onProcessStart for all ConditionsObjectProviders.
void clear()
Reset all of the variables to their limits.
static NtupleManager & getInstance()
void reset()
Reset NtupleManager to blank state.
void newRun(ldmx::RunHeader &header)
Run through the processors and let them know that we are starting a new run.
Definition Process.cxx:484
void onFileClose(EventFile &file) const
File is begin closed.
Definition Process.cxx:571
bool process(int n, int n_tries, Event &event) const
Process the input event through the sequence of processors.
Definition Process.cxx:519
void onFileOpen(EventFile &file) const
File is being opened.
Definition Process.cxx:557
bool keepEvent(bool event_completed) const
Determine if the current event should be kept, based on the defined rules.
void resetEventState()
Reset the event-by-event state.
static void set(int n)
set the event number in the current Formatter
Definition Logger.cxx:158
void absoluteStop()
literally last line of Process::run (if run compeletes without error)
Definition Tracker.cxx:88
void absoluteStart()
literally first line of Process::run
Definition Tracker.cxx:86
Provides header information an event such as event number and timestamp.
Definition EventHeader.h:44
void setEventNumber(int eventNumber)
Set the event number.
void setRun(int run)
Set the run number.
void setTimestamp(const TTimeStamp &timestamp)
Set the timestamp.
Run-specific configuration and data stored in its own output TTree alongside the event TTree in the o...
Definition RunHeader.h:57

References framework::performance::Tracker::absoluteStart(), framework::performance::Tracker::absoluteStop(), framework::EventFile::addDrop(), framework::NtupleManager::clear(), conditions_, config_, drop_keep_rules_, event_header_, event_limit_, framework::Event::getEventHeader(), framework::Event::getEventHeaderPtr(), framework::Event::getEventNumber(), framework::EventFile::getFileName(), framework::NtupleManager::getInstance(), ldmx::EventHeader::getRun(), framework::EventFile::getRunHeaderPtr(), input_files_, framework::EventFile::isCorrupted(), framework::StorageControl::keepEvent(), max_tries_, min_events_, newRun(), framework::EventFile::nextEvent(), framework::Event::onEndOfFile(), onFileClose(), onFileOpen(), framework::Conditions::onProcessStart(), output_files_, pass_name_, performance_, process(), framework::NtupleManager::reset(), framework::StorageControl::resetEventState(), run_for_generation_, run_header_, sequence_, framework::logging::Formatter::set(), ldmx::EventHeader::setEventNumber(), ldmx::RunHeader::setNumTries(), ldmx::EventHeader::setRun(), ldmx::RunHeader::setRunEnd(), ldmx::RunHeader::setRunStart(), ldmx::EventHeader::setTimestamp(), framework::EventFile::setupEvent(), skip_corrupted_input_files_, framework::performance::Tracker::start(), framework::performance::Tracker::stop(), storage_controller_, total_events_, framework::EventFile::updateParent(), framework::EventFile::writeRunHeader(), and framework::EventFile::writeRunTree().

Referenced by Process().

◆ setEventHeader()

void framework::Process::setEventHeader ( ldmx::EventHeader * h)
inline

Set the pointer to the current event header, used only for tests.

Definition at line 115 of file Process.h.

115{ event_header_ = h; }

References event_header_.

Member Data Documentation

◆ compression_setting_

int framework::Process::compression_setting_
private

Compression setting to pass to output files.

Look at the documentation for the TFile constructor if you want to learn more details. Essentially, setting = 100*algo + level with algo = 0 being the global default.

Definition at line 202 of file Process.h.

Referenced by Process().

◆ conditions_

Conditions framework::Process::conditions_
private

Set of ConditionsProviders.

Definition at line 186 of file Process.h.

Referenced by getConditions(), newRun(), Process(), and run().

◆ config_

framework::config::Parameters framework::Process::config_
private

The parameters used to configure this class.

Definition at line 152 of file Process.h.

Referenced by Process(), and run().

◆ drop_keep_rules_

std::vector<std::string> framework::Process::drop_keep_rules_
private

Set of drop/keep rules.

Definition at line 205 of file Process.h.

Referenced by Process(), and run().

◆ event_header_

const ldmx::EventHeader* framework::Process::event_header_ {0}
private

Pointer to the current EventHeader, used for Conditions information.

Definition at line 214 of file Process.h.

214{0};

Referenced by getEventHeader(), getRunNumber(), Process(), run(), and setEventHeader().

◆ event_limit_

int framework::Process::event_limit_
private

Limit on events to process.

Definition at line 158 of file Process.h.

Referenced by Process(), requestFinish(), and run().

◆ histo_filename_

std::string framework::Process::histo_filename_
private

Filename for histograms and other user products.

Definition at line 211 of file Process.h.

Referenced by openHistoFile(), and Process().

◆ histo_t_file_

TFile* framework::Process::histo_t_file_ {0}
private

TFile for histograms and other user products.

Definition at line 220 of file Process.h.

220{0};

Referenced by openHistoFile(), and ~Process().

◆ input_files_

std::vector<std::string> framework::Process::input_files_
private

List of input files to process.

May be empty if this Process will generate new events.

Definition at line 190 of file Process.h.

Referenced by Process(), and run().

◆ log_frequency_

int framework::Process::log_frequency_
private

The frequency with which event info is printed.

Definition at line 169 of file Process.h.

Referenced by getLogFrequency(), Process(), and process().

◆ max_tries_

int framework::Process::max_tries_
private

Maximum number of attempts to make before giving up on an event.

Definition at line 172 of file Process.h.

Referenced by Process(), and run().

◆ min_events_

int framework::Process::min_events_
private

When reading a file in, what's the first event to read.

Definition at line 161 of file Process.h.

Referenced by Process(), and run().

◆ output_files_

std::vector<std::string> framework::Process::output_files_
private

List of output file names.

If empty, no output file will be created.

Definition at line 193 of file Process.h.

Referenced by Process(), and run().

◆ pass_name_

std::string framework::Process::pass_name_
private

Processing pass name.

Definition at line 155 of file Process.h.

Referenced by getPassName(), newRun(), Process(), and run().

◆ performance_

performance::Tracker* framework::Process::performance_ {0}
private

class with calls backs to track performance measurements of software

Definition at line 223 of file Process.h.

223{0};

Referenced by newRun(), onFileClose(), onFileOpen(), Process(), process(), run(), and ~Process().

◆ run_for_generation_

int framework::Process::run_for_generation_ {1}
private

Run number to use if generating events.

Definition at line 208 of file Process.h.

208{1};

Referenced by getRunNumber(), Process(), and run().

◆ run_header_

ldmx::RunHeader* framework::Process::run_header_ {0}
private

Pointer to the current RunHeader, used for Conditions information.

Definition at line 217 of file Process.h.

217{0};

Referenced by getRunHeader(), and run().

◆ sequence_

std::vector<EventProcessor *> framework::Process::sequence_
private

Ordered list of EventProcessors to execute.

Definition at line 183 of file Process.h.

Referenced by newRun(), onFileClose(), onFileOpen(), Process(), process(), run(), and ~Process().

◆ skip_corrupted_input_files_

bool framework::Process::skip_corrupted_input_files_
private

allow the Process to skip input files that are corrupted

Definition at line 177 of file Process.h.

Referenced by Process(), and run().

◆ storage_controller_

StorageControl framework::Process::storage_controller_
private

Storage controller.

Definition at line 180 of file Process.h.

Referenced by getStorageController(), Process(), and run().

◆ total_events_

int framework::Process::total_events_
private

Number of events we'd like to produce independetly of the number of tries it would take.

Be warned about infinite loops!

Definition at line 166 of file Process.h.

Referenced by Process(), and run().


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