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::EventHeader *	getEventHeader () const
	Get the pointer to the current event header, if defined.
const ldmx::RunHeader *	getRunHeader () const
	Get the pointer to the current run header, if defined.
Conditions &	getConditions ()
	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.
StorageControl &	getStorageController ()
	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, EventFile *out)
	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
bool	allow_incomplete_input_files_
	allow the Process to read input files whose runs never finished
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::EventHeader *	event_header_ {0}
	Pointer to the current EventHeader, used for Conditions information.
ldmx::RunHeader *	run_header_ {0}
	Pointer to the current RunHeader, used for Conditions information.
TFile *	histo_t_file_ {0}
	TFile for histograms and other user products.
performance::Tracker *	performance_ {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

configuration

Parameters to configure process with

Definition at line 30 of file Process.cxx.

    : conditions_{*this} {
  config_ = configuration;
 
  pass_name_ = configuration.get<std::string>("pass_name", "");
  histo_filename_ = configuration.get<std::string>("histogram_file", "");
 
  max_tries_ = configuration.get<int>("max_tries_per_event", 1);
  event_limit_ = configuration.get<int>("max_events", -1);
  min_events_ = configuration.get<int>("min_events", -1);
  total_events_ = configuration.get<int>("total_events", -1);
  log_frequency_ = configuration.get<int>("log_frequency", -1);
  compression_setting_ = configuration.get<int>("compression_setting", 9);
  skip_corrupted_input_files_ =
      configuration.get<bool>("skip_corrupted_input_files", false);
  allow_incomplete_input_files_ =
      configuration.get<bool>("allow_incomplete_input_files", false);
 
  input_files_ = configuration.get<std::vector<std::string>>("input_files", {});
  output_files_ =
      configuration.get<std::vector<std::string>>("output_files", {});
  drop_keep_rules_ = configuration.get<std::vector<std::string>>("keep", {});
 
  event_header_ = 0;
 
  // set up the logging for this run
  logging::open(configuration.get<framework::config::Parameters>("logger", {}));
 
  auto run{configuration.get<int>("run", -1)};
  if (run > 0) run_for_generation_ = run;
 
  auto libs{configuration.get<std::vector<std::string>>("libraries", {})};
  std::set<std::string> libraries_loaded;
  for (const auto& lib : libs) {
    if (libraries_loaded.find(lib) != libraries_loaded.end()) {
      continue;
    }
 
    void* handle = dlopen(lib.c_str(), RTLD_NOW);
    if (handle == nullptr) {
      EXCEPTION_RAISE("LibraryLoadFailure",
                      "Error loading library '" + lib + "':" + dlerror());
    }
 
    libraries_loaded.insert(lib);
  }
 
  storage_controller_.setDefaultKeep(
      configuration.get<bool>("skim_default_is_keep", true));
  auto skim_rules{
      configuration.get<std::vector<std::string>>("skim_rules", {})};
  for (size_t i = 0; i < skim_rules.size(); i += 2) {
    storage_controller_.addRule(skim_rules[i], skim_rules[i + 1]);
  }
 
  auto sequence{configuration.get<std::vector<framework::config::Parameters>>(
      "sequence", {})};
  if (sequence.empty() && configuration.get<bool>("testing_mode", false)) {
    EXCEPTION_RAISE(
        "NoSeq",
        "No sequence has been defined. What should I be doing?\nUse "
        "p.sequence to tell me what processors to run.");
  }
  for (auto proc : sequence) {
    auto class_name{proc.get<std::string>("class_name")};
    auto instance_name{proc.get<std::string>("instance_name")};
    auto ep{
        EventProcessor::Factory::get().make(class_name, instance_name, *this)};
    if (not ep) {
      EXCEPTION_RAISE("UnableToCreate",
                      "The EventProcessor Factory was unable to create " +
                          instance_name + " of type " + class_name +
                          ". Did you inherit from framework::Producer or "
                          "framework::Analyzer? "
                          "Did you DECLARE_PRODUCER or DECLARE_ANALYZER in the "
                          "implementation (.cxx) file? "
                          "Did you use the class's full name (including "
                          "namespaces) in the Python configuration class? "
                          "Does the Python configuration class reference the "
                          "correct library it is a part of?");
    }
    auto histograms{
        proc.get<std::vector<framework::config::Parameters>>("histograms", {})};
    if (!histograms.empty()) {
      ep.value()->getHistoDirectory();
      ep.value()->createHistograms(histograms);
    }
    ep.value()->configure(proc);
    sequence_.push_back(ep.value());
  }
 
  auto conditions_object_providers{
      configuration.get<std::vector<framework::config::Parameters>>(
          "conditions_object_providers", {})};
  for (auto cop : conditions_object_providers) {
    auto class_name{cop.get<std::string>("class_name")};
    auto object_name{cop.get<std::string>("object_name")};
    auto tag_name{cop.get<std::string>("tag_name")};
    conditions_.createConditionsObjectProvider(class_name, object_name,
                                               tag_name, cop);
  }
 
  bool log_performance = configuration.get<bool>("log_performance", false);
  if (log_performance) {
    std::vector<std::string> names{sequence_.size()};
    for (std::size_t i{0}; i < sequence_.size(); i++) {
      names[i] = sequence_[i]->getName();
    }
    performance_ =
        new performance::Tracker(makeHistoDirectory("performance"), names);
  }
}

References framework::StorageControl::addRule(), allow_incomplete_input_files_, 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 143 of file Process.cxx.

                  {
  // need to delete the performance object so that it is
  // written before we close the histogram file below
  if (performance_) delete performance_;
  for (EventProcessor* ep : sequence_) {
    delete ep;
  }
  if (histo_t_file_) {
    histo_t_file_->Write();
    delete histo_t_file_;
    histo_t_file_ = 0;
  }
}

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.

{ 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.

{ 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.

{ 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.

{ 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.

{ return run_header_; }

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 483 of file Process.cxx.

                                {
  return (event_header_) ? (event_header_->getRun()) : (run_for_generation_);
}

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.

{ 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 487 of file Process.cxx.

                                                                {
  auto owner{openHistoFile()};
  TDirectory* child = owner->mkdir((char*)dirName.c_str());
  if (child) child->cd();
  return child;
}

References openHistoFile().

Referenced by Process().

newRun()

void framework::Process::newRun ( ldmx::RunHeader & header, EventFile * out )

private

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

The header is written to the output file once the producers have filled it, so a killed job still leaves it behind.

Parameters

[in]	header	RunHeader for the new run
[in]	out	output file to write the header to, if any

Definition at line 516 of file Process.cxx.

                                                          {
  // Producers are allowed to put parameters into
  // the run header through 'beforeNewRun' method
 
  // Put the version into the rh string param
  header.setStringParameter("Pass = " + pass_name_ + ", version",
                            LDMXSW_VERSION);
  if (performance_) performance_->start(performance::Callback::beforeNewRun, 0);
  std::size_t i_proc{0};
  for (auto proc : sequence_) {
    i_proc++;
    if (performance_)
      performance_->start(performance::Callback::beforeNewRun, i_proc);
    proc->beforeNewRun(header);
    if (performance_)
      performance_->stop(performance::Callback::beforeNewRun, i_proc);
  }
  if (performance_) performance_->stop(performance::Callback::beforeNewRun, 0);
  // now run header has been modified by Producers,
  // it is valid to read from for everyone else in 'onNewRun'
  // header is final, so write it now in case we are killed
  if (out) out->writeRunTree();
  if (performance_) performance_->start(performance::Callback::onNewRun, 0);
  conditions_.onNewRun(header);
  i_proc = 0;
  for (auto proc : sequence_) {
    i_proc++;
    if (performance_)
      performance_->start(performance::Callback::onNewRun, i_proc);
    proc->onNewRun(header);
    if (performance_)
      performance_->stop(performance::Callback::onNewRun, i_proc);
  }
  if (performance_) performance_->stop(performance::Callback::onNewRun, 0);
  ldmx_log(info) << header;
}

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

Referenced by run().

onFileClose()

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

private

File is begin closed.

Definition at line 605 of file Process.cxx.

                                               {
  if (performance_) performance_->start(performance::Callback::onFileClose, 0);
  std::size_t i_proc{0};
  for (auto proc : sequence_) {
    i_proc++;
    if (performance_)
      performance_->start(performance::Callback::onFileClose, i_proc);
    proc->onFileClose(file);
    if (performance_)
      performance_->stop(performance::Callback::onFileClose, i_proc);
  }
  if (performance_) performance_->stop(performance::Callback::onFileClose, 0);
}

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 591 of file Process.cxx.

                                              {
  if (performance_) performance_->start(performance::Callback::onFileOpen, 0);
  std::size_t i_proc{0};
  for (auto proc : sequence_) {
    i_proc++;
    if (performance_)
      performance_->start(performance::Callback::onFileOpen, i_proc);
    proc->onFileOpen(file);
    if (performance_)
      performance_->stop(performance::Callback::onFileOpen, i_proc);
  }
  if (performance_) performance_->stop(performance::Callback::onFileOpen, 0);
}

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 494 of file Process.cxx.

                                   {
  TDirectory* owner{nullptr};
 
  if (histo_filename_.empty()) {
    // trying to write histograms/ntuples but no file defined
    EXCEPTION_RAISE(
        "NoHistFileName",
        "You did not provide the necessary histogram file name to "
        "put your histograms (or performance data) in.\n    Provide this "
        "name in the python configuration with 'p.histogramFile = "
        "\"myHistFile.root\"' where p is the Process object.");
  } else if (histo_t_file_ == nullptr) {
    histo_t_file_ = new TFile(histo_filename_.c_str(), "RECREATE");
    owner = histo_t_file_;
  } else {
    owner = histo_t_file_;
  }
  owner->cd();
 
  return owner;
}

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]	n	counter for number of events processed
[in]	n_tries	counter for number of tries on current event
[in,out]	event	reference to event we are going to process

Returns

true if event was full processed (false if aborted)

Definition at line 553 of file Process.cxx.

                                                          {
  if ((log_frequency_ != -1) && ((n + 1) % log_frequency_ == 0) &&
      (n_try < 2)) {
    // only printout event counter if we've enabled log frequency, the event
    // matches the frequency and we are on the first try
    TTimeStamp t;
    ldmx_log(info) << "Processing " << n + 1 << " Run "
                   << event.getEventHeader().getRun() << " Event "
                   << event.getEventHeader().getEventNumber() << "  ("
                   << t.AsString("lc") << ")";
  }
 
  if (performance_) performance_->start(performance::Callback::process, 0);
  std::size_t i_proc{0};
  try {
    for (auto proc : sequence_) {
      i_proc++;
      if (performance_)
        performance_->start(performance::Callback::process, i_proc);
      proc->process(event);
      if (performance_)
        performance_->stop(performance::Callback::process, i_proc);
    }
  } catch (AbortEventException&) {
    if (performance_) {
      performance_->stop(performance::Callback::process, i_proc);
      performance_->stop(performance::Callback::process, 0);
      performance_->endEvent(false);
    }
    return false;
  }
  if (performance_) {
    performance_->stop(performance::Callback::process, 0);
    performance_->endEvent(true);
  }
  return true;
}

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.

{ event_limit_ = 0; }

References event_limit_.

run()

void framework::Process::run

(

)

Run the process.

Definition at line 157 of file Process.cxx.

                  {
  if (performance_) performance_->absoluteStart();
 
  // Counter to keep track of the number of events that have been
  // procesed
  auto n_events_processed{0};
 
  // make sure the ntuple manager is in a blank state
  NtupleManager::getInstance().reset();
 
  // event bus for this process
  Event the_event(pass_name_);
  // the EventHeader object is created with the event bus as
  // one of its members, we obtain a pointer for the header
  // here so we can share it with the conditions system
  event_header_ = the_event.getEventHeaderPtr();
  the_event.getEventHeader().setRun(run_for_generation_);
 
  // Start by notifying everyone that modules processing is beginning
  std::size_t i_proc{0};
  if (performance_)
    performance_->start(performance::Callback::onProcessStart, 0);
  conditions_.onProcessStart();
  for (auto proc : sequence_) {
    i_proc++;
    if (performance_)
      performance_->start(performance::Callback::onProcessStart, i_proc);
    proc->onProcessStart();
    if (performance_)
      performance_->stop(performance::Callback::onProcessStart, i_proc);
  }
  if (performance_)
    performance_->stop(performance::Callback::onProcessStart, 0);
 
  // If we have no input files, but do have an event number, run for
  // that number of events and generate an output file.
  if (input_files_.empty() && event_limit_ > 0) {
    if (output_files_.empty()) {
      EXCEPTION_RAISE("InvalidConfig",
                      "No input files or output files were given.");
    } else if (output_files_.size() > 1) {
      ldmx_log(warn) << "Several output files given with no input files. "
                     << "Only the first output file '" << output_files_.at(0)
                     << "' will be used.";
    }
    std::string output_file_name = output_files_.at(0);
 
    // Configure the event file to create an output file with no parent. This
    // requires setting the parameters isOutputFile and isSingleOutput to true.
    EventFile out_file(config_, output_file_name, nullptr, true, true, false);
    onFileOpen(out_file);
    out_file.setupEvent(&the_event);
 
    for (auto rule : drop_keep_rules_) out_file.addDrop(rule);
 
    auto run_header = std::make_shared<ldmx::RunHeader>(run_for_generation_);
    run_header->setRunStart(std::time(nullptr));  // set run starting
    run_header_ = run_header.get();  // give handle to run header to process
    out_file.writeRunHeader(run_header);  // add run header to file
 
    newRun(*run_header, &out_file);
 
    int total_tries = 0;  // total number of tries for entire run
    int num_tries = 0;    // number of tries for the current event number
    int event_limit = event_limit_;
    if (total_events_ > 0) {
      // Have a warning at the first event
      if (num_tries == 0)
        ldmx_log(warn) << "The total_events was set, so max_events and "
                          "max_tries_per_event will be ignored!";
      event_limit = total_events_;
    }
    while (n_events_processed < event_limit) {
      // Check for preemption before processing each event
      if (preemption_received_) {
        ldmx_log(fatal)
            << "Preemption signal received, stopping event generation";
        break;
      }
 
      total_tries++;
      num_tries++;
 
      ldmx::EventHeader& eh = the_event.getEventHeader();
      eh.setRun(run_for_generation_);
      eh.setEventNumber(n_events_processed + 1);
      eh.setTimestamp(TTimeStamp());
 
      // reset the storage controller state
      storage_controller_.resetEventState();
      logging::Formatter::set(the_event.getEventNumber());
 
      bool completed = process(n_events_processed, num_tries, the_event);
 
      out_file.nextEvent(storage_controller_.keepEvent(completed));
 
      // reset try counter only on successfully completed events
      if (completed) num_tries = 0;
 
      // we use modulo here insetad of >= because we want to carry
      // the number of tries across the number of events processed boundary
      // total_events_ is set let's not exit until that's reached
      if (completed or (total_events_ < 0 and num_tries % max_tries_ == 0)) {
        n_events_processed++;                 // increment events made
        NtupleManager::getInstance().fill();  // fill ntuples
      }
 
      NtupleManager::getInstance().clear();
    }
 
    onFileClose(out_file);
 
    run_header->setRunEnd(std::time(nullptr));
    run_header->setNumTries(total_tries);
    out_file.writeRunTree(not preemption_received_);
 
    // Give a warning that this filter has very low efficiency
    if (n_events_processed < total_tries / 10000) {  // integer division is okay
      ldmx_log(warn)
          << "Less than 1 event out of every 10k events tried was accepted!";
      ldmx_log(warn)
          << "This could be an issue with your filtering and biasing procedure "
             "since this is incredibly inefficient.";
    }
 
  } else {
    // there are input files
 
    EventFile* out_file(0);
 
    bool single_output = false;
    if (output_files_.size() == 1) {
      single_output = true;
    } else if (!output_files_.empty() and
               output_files_.size() != input_files_.size()) {
      EXCEPTION_RAISE("Process",
                      "Unable to handle case of different number of input and "
                      "output files (other than zero/one ouput file).");
    }
 
    // next, loop through the files
    int ifile = 0;
    int was_run = -1;
    for (auto infilename : input_files_) {
      EventFile in_file(config_, infilename);
      if (in_file.isCorrupted()) {
        if (skip_corrupted_input_files_) {
          ldmx_log(warn) << "Input file '" << infilename
                         << "' was found to be corrupted. Skipping.";
          continue;
        } else {
          EXCEPTION_RAISE(
              "BadCode",
              "We should never get here. "
              "EventFile is corrupted but we aren't skipping corrupted inputs. "
              "EventFile should be throwing its own exceptions in this case.");
        }
      }
 
      auto incomplete_runs{in_file.getIncompleteRuns()};
      if (not incomplete_runs.empty()) {
        std::string run_list;
        for (int run : incomplete_runs) {
          if (not run_list.empty()) run_list += ", ";
          run_list += std::to_string(run);
        }
        if (allow_incomplete_input_files_) {
          ldmx_log(warn) << "Input file '" << infilename << "' holds run(s) "
                         << run_list
                         << " that were never finished, so events are missing. "
                            "Processing it anyway because "
                            "'allow_incomplete_input_files' is set.";
        } else {
          EXCEPTION_RAISE(
              "IncompleteInput",
              "Input file '" + infilename + "' holds run(s) " + run_list +
                  " that were never finished, so events are missing from the "
                  "end of them. Set 'allow_incomplete_input_files' on the "
                  "Process to read it anyway, but know that the sample it "
                  "gives you is biased.");
        }
      }
 
      ldmx_log(info) << "Opening file " << infilename;
      onFileOpen(in_file);
 
      // configure event file that will be iterated over
      EventFile* master_file;
      if (!output_files_.empty()) {
        // setup new output file if either
        // 1) we are not in single output mode
        // 2) this is the first input file
        if (!single_output or ifile == 0) {
          // setup new output file
          out_file = new EventFile(config_, output_files_[ifile], &in_file,
                                   single_output);
          ifile++;
 
          // setup theEvent we will iterate over
          if (out_file) {
            out_file->setupEvent(&the_event);
            master_file = out_file;
          } else {
            EXCEPTION_RAISE("Process", "Unable to construct output file for " +
                                           output_files_[ifile]);
          }
 
          for (auto rule : drop_keep_rules_) out_file->addDrop(rule);
 
        } else {
          // all other input files
          out_file->updateParent(&in_file);
          master_file = out_file;
 
        }  // check if in singleOutput mode
 
      } else {
        // empty output file list, use inputFile as master file
        in_file.setupEvent(&the_event);
        master_file = &in_file;
      }
 
      // In case we'd like to skip up to the event of min_events_
      while (n_events_processed < (min_events_ - 1) &&
             master_file->nextEvent(false)) {
        n_events_processed++;
      }
 
      bool event_completed = true;
      while (!preemption_received_ &&
             master_file->nextEvent(
                 storage_controller_.keepEvent(event_completed)) &&
             ((event_limit_ < 0) || (n_events_processed < event_limit_))) {
        // clean up for storage control calculation
        storage_controller_.resetEventState();
        logging::Formatter::set(the_event.getEventNumber());
 
        // notify for new run if necessary
        if (the_event.getEventHeader().getRun() != was_run) {
          was_run = the_event.getEventHeader().getRun();
          ldmx::RunHeader* rh{master_file->getRunHeaderPtr(was_run)};
          if (rh != nullptr) {
            run_header_ = rh;
            ldmx_log(info) << "Got new run header from '"
                           << master_file->getFileName() << "'";
            newRun(*run_header_, out_file);
          } else {
            // no header means no newRun, so conditions stay uninitialised
            EXCEPTION_RAISE(
                "MissingRunHeader",
                "Run header for run " + std::to_string(was_run) +
                    " was not found in '" + master_file->getFileName() +
                    "'. Conditions cannot be initialised without it.");
          }
        }
 
        event_completed = process(n_events_processed, 1, the_event);
 
        if (event_completed) NtupleManager::getInstance().fill();
        NtupleManager::getInstance().clear();
 
        n_events_processed++;
      }  // loop through events
 
      if (preemption_received_) {
        ldmx_log(fatal) << "Preemption signal received, stopping event "
                           "processing and closing files";
      }
 
      bool leave_early{false};
      if (event_limit_ > 0 && n_events_processed == event_limit_) {
        ldmx_log(info) << "Reached event limit of " << event_limit_
                       << " events";
        leave_early = true;
      }
 
      if (event_limit_ == 0 && n_events_processed > event_limit_) {
        ldmx_log(warn) << "Processing interrupted";
        leave_early = true;
      }
 
      ldmx_log(info) << "Closing file " << infilename;
      onFileClose(in_file);
 
      // Reset the event in case of multiple input files
      the_event.onEndOfFile();
 
      if (out_file and !single_output) {
        out_file->writeRunTree(not preemption_received_);
        delete out_file;
        out_file = nullptr;
      }
 
      if (leave_early) {
        break;
      }
    }  // loop through input files
 
    if (out_file) {
      // close outFile
      //  outFile would survive to here in single output mode
      out_file->writeRunTree(not preemption_received_);
      delete out_file;
      out_file = nullptr;
    }
 
  }  // are there input files? if-else tree
 
  // finally, notify everyone that we are stopping
  if (performance_) performance_->start(performance::Callback::onProcessEnd, 0);
  i_proc = 0;
  for (auto proc : sequence_) {
    i_proc++;
    if (performance_)
      performance_->start(performance::Callback::onProcessEnd, i_proc);
    proc->onProcessEnd();
    if (performance_)
      performance_->stop(performance::Callback::onProcessEnd, i_proc);
  }
  if (performance_) performance_->stop(performance::Callback::onProcessEnd, 0);
 
  // we're done so let's close up the logging
  logging::close();
  if (performance_) performance_->absoluteStop();
}

References framework::performance::Tracker::absoluteStart(), framework::performance::Tracker::absoluteStop(), framework::EventFile::addDrop(), allow_incomplete_input_files_, 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::EventFile::getIncompleteRuns(), 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(), run_for_generation_, run_header_, sequence_, framework::logging::Formatter::set(), ldmx::EventHeader::setEventNumber(), ldmx::EventHeader::setRun(), 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(), and run().

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.

{ event_header_ = h; }

References event_header_.

Member Data Documentation

allow_incomplete_input_files_

bool framework::Process::allow_incomplete_input_files_

private

allow the Process to read input files whose runs never finished

For experts. Such a file is missing events off the end of a run, so the sample it gives is biased.

Definition at line 189 of file Process.h.

Referenced by Process(), and run().

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 214 of file Process.h.

Referenced by Process().

conditions_

Conditions framework::Process::conditions_

private

Set of ConditionsProviders.

Definition at line 198 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 156 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 217 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 226 of file Process.h.

{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 162 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 223 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 232 of file Process.h.

{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 202 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 173 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 176 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 165 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 205 of file Process.h.

Referenced by Process(), and run().

pass_name_

std::string framework::Process::pass_name_

private

Processing pass name.

Definition at line 159 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 235 of file Process.h.

{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 220 of file Process.h.

{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 229 of file Process.h.

{0};

Referenced by getRunHeader(), and run().

sequence_

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

private

Ordered list of EventProcessors to execute.

Definition at line 195 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 181 of file Process.h.

Referenced by Process(), and run().

storage_controller_

StorageControl framework::Process::storage_controller_

private

Storage controller.

Definition at line 192 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 170 of file Process.h.

Referenced by Process(), and run().

---

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

• Framework/include/Framework/Process.h
• Framework/src/Framework/Process.cxx