EL::Worker Class Reference

#include <Worker.h>

Inheritance diagram for EL::Worker:

Collaboration diagram for EL::Worker:

Public Member Functions
void	testInvariant () const
	effects: test the invariant of this object guarantee: no-fail More...
virtual	~Worker ()
	effects: standard destructor guarantee: no-fail More...
void	addOutput (TObject *output_swallow) final override
	effects: add an object to the output. More...
void	addOutputList (const std::string &name, TObject *output_swallow) override
	effects: add a given object to the output. More...
TObject *	getOutputHist (const std::string &name) const final override
	get the output histogram with the given name More...
TFile *	getOutputFile (const std::string &label) const override
	effects: get the output file that goes into the dataset with the given label. More...
TFile *	getOutputFileNull (const std::string &label) const override
	effects: get the output file that goes into the dataset with the given label. More...
::StatusCode	addTree (const TTree &tree, const std::string &stream) final override
	effects: adds a tree to an output file specified by the stream/label failures: Incorrect stream/label specified, called at the wrong time note: See getOutputFile for failure types... More...
TTree *	getOutputTree (const std::string &name, const std::string &stream) const final override
	effects: get the tree that was added to an output file earlier failures: Tree doesn't exist More...
const SH::MetaObject *	metaData () const override
	description: the sample meta-data we are working on guarantee: no-fail invariant: metaData != 0 rationale: this can be used for accessing sample meta-data More...
TTree *	tree () const override
	description: the tree we are running on guarantee: no-fail More...
Long64_t	treeEntry () const override
	description: the entry in the tree we are reading guarantee: no-fail More...
TFile *	inputFile () const override
	description: the file we are reading the current tree from guarantee: no-fail More...
std::string	inputFileName () const override
	the name of the file we are reading the current tree from, without the path component More...
TTree *	triggerConfig () const override
	description: the trigger config tree from the input file, or NULL if we did not find it guarantee: strong failures: i/o errors More...
xAOD::TEvent *	xaodEvent () const override
	description: the xAOD event and store guarantee: strong failures: out of memory I failures: TEventSvc not configured postcondition: result != 0 More...
xAOD::TStore *	xaodStore () const override
EL::Algorithm *	getAlg (const std::string &name) const override
	effects: returns the algorithms with the given name or NULL if there is none guarantee: strong failures: out of memory II More...
void	skipEvent () override
	effects: skip the current event, i.e. More...
virtual bool	filterPassed () const noexcept final override
	whether the current algorithm passed its filter criterion for the current event More...
virtual void	setFilterPassed (bool val_filterPassed) noexcept final override
	set the value of filterPassed More...
	Worker ()
	standard constructor More...
::StatusCode	directExecute (const SH::SamplePtr &sample, const Job &job, const std::string &location, const SH::MetaObject &options)
	run the job More...
::StatusCode	batchExecute (unsigned job_id, const char *confFile)
	effects: do what is needed to execute the given job segment guarantee: basic failures: job specific More...
::StatusCode	gridExecute (const std::string &sampleName, Long64_t SkipEvents, Long64_t nEventsPerJob)

Protected Member Functions
void	setMetaData (const SH::MetaObject *val_metaData)
	set the metaData More...
void	setOutputHist (const std::string &val_outputTarget)
	set the histogram output list More...
void	setSegmentName (const std::string &val_segmentName)
	set the segment name More...
void	setJobConfig (JobConfig &&jobConfig)
	set the JobConfig More...
void	addModule (std::unique_ptr< Detail::Module > module)
	add the given module to this worker More...
::StatusCode	initialize ()
	initialize the worker More...
::StatusCode	processInputs ()
	process all the inputs More...
::StatusCode	finalize ()
	finalize the worker More...
::StatusCode	processEvents (EventRange &eventRange) override
	process the given event range More...
::StatusCode	openInputFile (const std::string &inputFileUrl) override
	open the given input file without processing it More...
::StatusCode	addOutputStream (const std::string &label, Detail::OutputStreamData output)
	effects: add another output file guarantee: strong failures: low level errors II failures: label already used More...
Long64_t	inputFileNumEntries () const override
	the number of events in the input file More...
uint64_t	eventsProcessed () const noexcept
	the number of events that have been processed More...

Private Types
enum	GridErrorCodes { EC_FAIL = 220, EC_ABORT = 221, EC_NOTFINISHED = 222, EC_BADINPUT = 223 }
typedef std::map< std::string, TH1 * >::const_iterator	OutputHistMapIter
	the output map More...
typedef std::map< std::pair< std::string, std::string >, TTree * >::const_iterator	OutputTreeMapIter
	description: the list of output trees More...

Private Member Functions
void	gridCreateJobSummary (uint64_t eventsProcessed)
void	addOutput (std::unique_ptr< TObject > output)
	add the given output object to the histogram output stream More...

Static Private Member Functions
static bool	fileOpenErrorFilter (int level, bool, const char *, const char *)
	Error handler for file opening. More...

Private Attributes
std::map< std::string, TH1 * >	m_outputHistMap
std::map< std::pair< std::string, std::string >, TTree * >	m_outputTreeMap
std::vector< std::unique_ptr< Detail::Module > >	m_modules
	the list of modules we hold More...
bool	m_newInputFile {false}
	whether this is a new input file (i.e. More...
std::string	m_outputTarget
	the target file to which we will write the histogram output More...
std::string	m_segmentName
	the name of the segment we are processing More...
bool	m_algorithmsInitialized {false}
	whether the algorithms are initialized More...
bool	m_firstEvent {true}
	whether we are still to process the first event More...
std::vector< Detail::AlgorithmData >	m_algs
	the list of algorithms More...
std::string	m_inputFileUrl
	the input file url of the currently opened file More...
std::unique_ptr< TFile >	m_inputFile
	the input file pointer of the currently opened filed More...
TTree *	m_inputTree {nullptr}
	the (main) tree in the input file More...
uint64_t	m_inputTreeEntry {0}
	the entry in the input tree we are currently looking at More...
bool	m_skipEvent = false
	whether we are skipping the current event More...
const SH::MetaObject *	m_metaData {nullptr}
	the meta-data we use More...
uint64_t	m_eventsProcessed {0}
	the number of events that have been processed More...
OutputStreamData *	m_histOutput {nullptr}
	the histogram output stream More...
std::unique_ptr< TTree >	m_jobStats
	Tree saving per-job statistics information. More...
xAOD::TEvent *	m_tevent {nullptr}
	the TEvent structure, if we use one More...
xAOD::TStore *	m_tstore {nullptr}
	the TStore structure, if we use one More...
asg::SgTEvent *	m_evtStore {nullptr}
	the SgTEvent structure, if we use one More...
Worker *	m_worker {nullptr}
	the worker (to pass on to the algorithms) More...
std::map< std::string, std::shared_ptr< Detail::OutputStreamData > >	m_outputs
	the list of output files More...

Detailed Description

Definition at line 24 of file Worker.h.

Member Typedef Documentation

OutputHistMapIter

typedef std::map<std::string,TH1*>::const_iterator EL::Worker::OutputHistMapIter

private

the output map

Definition at line 410 of file Worker.h.

OutputTreeMapIter

typedef std::map<std::pair<std::string,std::string>,TTree*>::const_iterator EL::Worker::OutputTreeMapIter

private

description: the list of output trees

Definition at line 417 of file Worker.h.

Member Enumeration Documentation

GridErrorCodes

enum EL::Worker::GridErrorCodes

private

Enumerator
EC_FAIL
EC_ABORT
EC_NOTFINISHED
EC_BADINPUT

Definition at line 249 of file Worker.h.

                        {
      EC_FAIL = 220,
      EC_ABORT = 221,
      EC_NOTFINISHED = 222,
      EC_BADINPUT = 223
    };

Constructor & Destructor Documentation

~Worker()

EL::Worker::~Worker ( )

virtual

effects: standard destructor guarantee: no-fail

Definition at line 79 of file Worker.cxx.

  {
    RCU_DESTROY_INVARIANT (this);
  }

Worker()

EL::Worker::Worker

(

)

standard constructor

Guarantee

strong

Failures

out of memory I

Definition at line 323 of file Worker.cxx.

  {
    m_worker = this;
 
    RCU_NEW_INVARIANT (this);
  }

Member Function Documentation

addModule()

void EL::Worker::addModule ( std::unique_ptr< Detail::Module >  module )

protected

add the given module to this worker

Guarantee

strong

Failures

out of memory I

Definition at line 760 of file Worker.cxx.

  {
    RCU_CHANGE_INVARIANT (this);
    RCU_REQUIRE (module != nullptr);
    m_modules.push_back (std::move (module));
  }

addOutput() [1/2]

void EL::Detail::ModuleData::addOutput ( std::unique_ptr< TObject >  output )

inherited

add the given output object to the histogram output stream

Guarantee

basic

Failures

out of memory II

Definition at line 33 of file ModuleData.cxx.

    {
      RCU_ASSERT (m_histOutput != nullptr);
      m_histOutput->addOutput (std::move (output));
    }

addOutput() [2/2]

void EL::Worker::addOutput ( TObject *  output_swallow )

finaloverridevirtual

effects: add an object to the output.

the worker takes over ownership of the object, but the caller may maintain a reference to it guarantee: basic, argument is always swallowed failures: out of memory I requires: output_swallow != 0 warning: so far I placed no requirements on the output objects. I may do that at a later stage though, possibly breaking existing code.

Implements EL::IHistogramWorker.

Definition at line 87 of file Worker.cxx.

  {
    std::unique_ptr<TObject> output (output_swallow);
 
    RCU_CHANGE_INVARIANT (this);
    RCU_REQUIRE_SOFT (output_swallow != 0);
 
    RCU::SetDirectory (output_swallow, 0);
    ModuleData::addOutput (std::move (output));
  }

addOutputList()

void EL::Worker::addOutputList ( const std::string &  name, TObject *  output_swallow  )

overridevirtual

effects: add a given object to the output.

instead of trying to merge the outputs from multiple jobs the output file will contain a TList of the given name with the objects from all the output. guarantee: basic, argument is always swallowed failures: out of memory II requires: output_swallow != 0 rationale: This was primarily build to allow the GRL output to be collected from EventLoop jobs, but it can be used for any kind of output that can not or should not be merged.

Implements EL::IWorker.

Definition at line 101 of file Worker.cxx.

  {
    std::unique_ptr<TObject> output (output_swallow);
 
    RCU_CHANGE_INVARIANT (this);
    RCU_REQUIRE_SOFT (output_swallow != 0);
 
    RCU::SetDirectory (output_swallow, 0);
    std::unique_ptr<TList> list (new TList);
    list->SetName (name.c_str());
    list->Add (output.release());
    addOutput (list.release());
  }

addOutputStream()

StatusCode EL::Worker::addOutputStream ( const std::string &  label, Detail::OutputStreamData  output  )

protected

effects: add another output file guarantee: strong failures: low level errors II failures: label already used

Definition at line 712 of file Worker.cxx.

  {
    using namespace msgEventLoop;
    RCU_CHANGE_INVARIANT (this);
 
    if (m_outputs.find (label) != m_outputs.end())
    {
      ANA_MSG_ERROR ("output file already defined for label: " + label);
      return ::StatusCode::FAILURE;
    }
    if (data.file() == nullptr)
    {
      ANA_MSG_ERROR ("output stream does not have a file attached");
      return ::StatusCode::FAILURE;
    }
    if (data.mainStreamName().empty())
      data.setMainStreamName (label);
    m_outputs.insert (std::make_pair (label, std::make_shared<Detail::OutputStreamData>(std::move (data))));
    return ::StatusCode::SUCCESS;
  }

addTree()

StatusCode EL::Worker::addTree ( const TTree &  tree, const std::string &  stream  )

finaloverridevirtual

effects: adds a tree to an output file specified by the stream/label failures: Incorrect stream/label specified, called at the wrong time note: See getOutputFile for failure types...

Implements EL::ITreeWorker.

Definition at line 154 of file Worker.cxx.

  {
    using namespace msgEventLoop;
    RCU_READ_INVARIANT( this );
 
    auto outputIter = m_outputs.find (stream);
    if (outputIter == m_outputs.end())
    {
      ANA_MSG_ERROR ( "No output file with stream name \"" + stream +
                      "\" found" );
      return ::StatusCode::FAILURE;
    }
 
    outputIter->second->addClone (tree);
 
    // Return gracefully:
    return ::StatusCode::SUCCESS;
  }

batchExecute()

StatusCode EL::Worker::batchExecute	(	unsigned	job_id,
		const char *	confFile
	)

effects: do what is needed to execute the given job segment guarantee: basic failures: job specific

Definition at line 816 of file Worker.cxx.

  {
    using namespace msgEventLoop;
    RCU_CHANGE_INVARIANT (this);
 
    try
    {
      std::unique_ptr<TFile> file (TFile::Open (confFile, "READ"));
      if (file.get() == nullptr || file->IsZombie())
      {
        ANA_MSG_ERROR ("failed to open file: " << confFile);
        return ::StatusCode::FAILURE;
      }
 
      std::unique_ptr<BatchJob> job (dynamic_cast<BatchJob*>(file->Get ("job")));
      if (job.get() == nullptr)
      {
        ANA_MSG_ERROR ("failed to retrieve BatchJob object");
        return ::StatusCode::FAILURE;
      }
 
      if (job_id >= job->segments.size())
      {
        ANA_MSG_ERROR ("invalid job-id " << job_id << ", max is " << job->segments.size());
        return ::StatusCode::FAILURE;
      }
      BatchSegment *segment = &job->segments[job_id];
      RCU_ASSERT (segment->job_id == job_id);
      RCU_ASSERT (segment->sample < job->samples.size());
      BatchSample *sample = &job->samples[segment->sample];
 
      gSystem->Exec ("pwd");
      gSystem->MakeDirectory ("output");
 
      setMetaData (&sample->meta);
      setOutputHist (job->location + "/fetch");
      setSegmentName (segment->fullName);
 
      setJobConfig (JobConfig (job->job.jobConfig()));
 
      for (Job::outputIter out = job->job.outputBegin(),
             end = job->job.outputEnd(); out != end; ++ out)
      {
        Detail::OutputStreamData data {
          out->output()->makeWriter (segment->sampleName, segment->segmentName, ".root")};
        ANA_CHECK (addOutputStream (out->label(), std::move (data)));
      }
 
      {
        auto module = std::make_unique<Detail::BatchInputModule> ("BatchInputModule");
        Long64_t maxEvents = metaData()->castDouble (Job::optMaxEvents, -1);
        if (maxEvents != -1)
          module->maxEvents = maxEvents;
        module->sample = sample;
        module->segment = segment;
        addModule (std::move (module));
      }
 
      ANA_CHECK (initialize ());
      ANA_CHECK (processInputs ());
      ANA_CHECK (finalize ());
 
      std::ostringstream job_name;
      job_name << job_id;
      std::ofstream completed ((job->location + "/status/completed-" + job_name.str()).c_str());
      return ::StatusCode::SUCCESS;
    } catch (...)
    {
      Detail::report_exception (std::current_exception());
      return ::StatusCode::FAILURE;
    }
  }

directExecute()

StatusCode EL::Worker::directExecute	(	const SH::SamplePtr &	sample,
		const Job &	job,
		const std::string &	location,
		const SH::MetaObject &	options
	)

run the job

Guarantee

basic

Definition at line 770 of file Worker.cxx.

  {
    using namespace msgEventLoop;
    RCU_CHANGE_INVARIANT (this);
 
    SH::MetaObject meta (*sample->meta());
    meta.fetchDefaults (options);
 
    setMetaData (&meta);
    setOutputHist (location);
    setSegmentName (sample->name());
 
    ANA_MSG_INFO ("Running sample: " << sample->name());
 
    setJobConfig (JobConfig (job.jobConfig()));
 
    for (Job::outputIter out = job.outputBegin(),
           end = job.outputEnd(); out != end; ++ out)
    {
      Detail::OutputStreamData data {
        out->output()->makeWriter (sample->name(), "", ".root")};
      ANA_CHECK (addOutputStream (out->label(), std::move (data)));
    }
 
    {
      auto module = std::make_unique<Detail::DirectInputModule> ("DirectInputModule");
      module->fileList = sample->makeFileList();
      Long64_t maxEvents = metaData()->castDouble (Job::optMaxEvents, -1);
      if (maxEvents != -1)
        module->maxEvents = maxEvents;
      Long64_t skipEvents = metaData()->castDouble (Job::optSkipEvents, 0);
      if (skipEvents != 0)
        module->skipEvents = skipEvents;
      addModule (std::move (module));
    }
 
    ANA_CHECK (initialize ());
    ANA_CHECK (processInputs ());
    ANA_CHECK (finalize ());
    return ::StatusCode::SUCCESS;
  }

eventsProcessed()

uint64_t EL::Worker::eventsProcessed ( ) const

protectednoexcept

the number of events that have been processed

Guarantee

no-fail

Definition at line 751 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    return m_eventsProcessed;
  }

fileOpenErrorFilter()

bool EL::Worker::fileOpenErrorFilter ( int  level, bool  , const char *  s1, const char *  s2  )

staticprivate

Error handler for file opening.

Definition at line 611 of file Worker.cxx.

  {
    // Don't fail on missing dictionary messages.
    if (strstr (s2, "no streamer or dictionary") != nullptr) {
      return true;
    }
 
    // For messages above warning level (SysError, Error, Fatal)
    if( level > kWarning ) {
      // We won't output further; ROOT should have already put something in the log file
      std::string msg = "ROOT error detected in Worker.cxx: ";
      msg += s1;
      msg += " ";
      msg += s2;
      throw std::runtime_error(msg);
 
      // No need for further error handling
      return false;
    }
 
    // Pass to the default error handlers
    return true;
  }

filterPassed()

bool EL::Worker::filterPassed ( ) const

finaloverridevirtualnoexcept

whether the current algorithm passed its filter criterion for the current event

Guarantee

no-fail

Implements EL::IFilterWorker.

Definition at line 305 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    return !m_skipEvent;
  }

finalize()

StatusCode EL::Worker::finalize ( )

protected

finalize the worker

This method ought to be called after all events have been processed. It is meant to ensure that the job is ended properly and all outputs are written out and files are closed.

Guarantee

basic

Failures

finalization failures

Definition at line 480 of file Worker.cxx.

  {
    using namespace msgEventLoop;
 
    RCU_CHANGE_INVARIANT (this);
 
    if (m_algorithmsInitialized == false)
    {
      ANA_MSG_ERROR ("algorithms never got initialized");
      return StatusCode::FAILURE;
    }
 
    ANA_CHECK (openInputFile (""));
    for (auto& module : m_modules)
      ANA_CHECK (module->onFinalize (*this));
    for (auto& output : m_outputs)
    {
      if (output.first != Job::histogramStreamName && output.second->mainStreamName() == output.first)
      {
        output.second->saveOutput ();
        output.second->close ();
        std::string path = output.second->finalFileName ();
        if (!path.empty())
          addOutputList ("EventLoop_OutputStream_" + output.first, new TObjString (path.c_str()));
      }
    }
    for (auto& module : m_modules)
      ANA_CHECK (module->postFinalize (*this));
    if (m_jobStats->GetListOfBranches()->GetEntries() > 0)
    {
      if (m_jobStats->Fill() <= 0)
      {
        ANA_MSG_ERROR ("failed to fill the job statistics tree");
        return ::StatusCode::FAILURE;
      }
      ModuleData::addOutput (std::move (m_jobStats));
    }
    m_histOutput->saveOutput ();
    for (auto& module : m_modules)
      ANA_CHECK (module->onWorkerEnd (*this));
    m_histOutput->saveOutput ();
    m_histOutput->close ();
 
    for (auto& module : m_modules){
      ANA_CHECK (module->postFileClose(*this));
    }
    ANA_MSG_INFO ("worker finished successfully");
    return ::StatusCode::SUCCESS;
  }

getAlg()

Algorithm * EL::Worker::getAlg ( const std::string &  name ) const

overridevirtual

effects: returns the algorithms with the given name or NULL if there is none guarantee: strong failures: out of memory II

Implements EL::IWorker.

Definition at line 282 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    for (auto& alg : m_algs)
    {
      if (alg->hasName (name))
    return alg.m_algorithm->getLegacyAlg();
    }
    return 0;
  }

getOutputFile()

TFile * EL::Worker::getOutputFile ( const std::string &  label ) const

overridevirtual

effects: get the output file that goes into the dataset with the given label.

this dataset has to be registered before the job is submitted. typically that happens in the doSetupJob method. guarantee: strong failures: no dataset with the given label postcondition: result != 0 note: the default value for the argument corresponds to the default label value in the OutputInfo class.

Implements EL::IWorker.

Definition at line 130 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    TFile *result = getOutputFileNull (label);
    if (result == 0)
      RCU_THROW_MSG ("no output dataset defined with label: " + label);
    return result;
  }

getOutputFileNull()

TFile * EL::Worker::getOutputFileNull ( const std::string &  label ) const

overridevirtual

effects: get the output file that goes into the dataset with the given label.

this dataset has to be registered before the job is submitted. typically that happens in the doSetupJob method. guarantee: strong failures: internal errors note: the default value for the argument corresponds to the default label value in the OutputInfo class. rationale: this method was added to support optional output files, i.e.: the algorithm leaves it to the user whether or not the output file gets configured. then the algorithm just checks whether the output file is there. if so it fills it, otherwise it ignores it.

Implements EL::IWorker.

Definition at line 142 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    auto iter = m_outputs.find (label);
    if (iter == m_outputs.end())
      return 0;
    return iter->second->file();
  }

getOutputHist()

TObject * EL::Worker::getOutputHist ( const std::string &  name ) const

finaloverridevirtual

get the output histogram with the given name

This is mostly meant, so that I can emulate the Athena histogram mechanism.

Guarantee

strong

Failures

object not found

Postcondition

result != 0

Implements EL::IHistogramWorker.

Definition at line 118 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
 
    TObject *result = m_histOutput->getOutputHist (name);
    if (result == nullptr) RCU_THROW_MSG ("unknown output histogram: " + name);
    return result;
  }

getOutputTree()

TTree * EL::Worker::getOutputTree ( const std::string &  name, const std::string &  stream  ) const

finaloverridevirtual

effects: get the tree that was added to an output file earlier failures: Tree doesn't exist

Implements EL::ITreeWorker.

Definition at line 176 of file Worker.cxx.

  {
    using namespace msgEventLoop;
    RCU_READ_INVARIANT( this );
 
    auto outputIter = m_outputs.find (stream);
    if (outputIter == m_outputs.end())
    {
      RCU_THROW_MSG ( "No output file with stream name \"" + stream
                      + "\" found" );
    }
 
    TTree *result = outputIter->second->getOutputTree( name );
    if( result == nullptr ) {
      RCU_THROW_MSG ( "No tree with name \"" + name + "\" in stream \"" +
                      stream + "\"" );
    }
    return result;
  }

gridCreateJobSummary()

void EL::Worker::gridCreateJobSummary ( uint64_t  eventsProcessed )

private

gridExecute()

StatusCode EL::Worker::gridExecute	(	const std::string &	sampleName,
		Long64_t	SkipEvents,
		Long64_t	nEventsPerJob
	)

Definition at line 892 of file Worker.cxx.

  {
    using namespace msgEventLoop;
    RCU_CHANGE_INVARIANT (this);
 
    ANA_MSG_INFO ("Running with ROOT version " << gROOT->GetVersion()
                  << " (" << gROOT->GetVersionDate() << ")");
 
    ANA_MSG_INFO ("Loading EventLoop grid job");
 
 
    TList bigOutputs;
    std::unique_ptr<JobConfig> jobConfig;
    SH::MetaObject *mo = 0;
 
    std::unique_ptr<TFile> f (TFile::Open("jobdef.root"));
    if (f == nullptr || f->IsZombie()) {
      ANA_MSG_ERROR ("Could not read jobdef");
      return ::StatusCode::FAILURE;
    }
 
    mo = dynamic_cast<SH::MetaObject*>(f->Get(sampleName.c_str()));
    if (!mo)
      mo = dynamic_cast<SH::MetaObject*>(f->Get("defaultMetaObject"));
    if (!mo) {
      ANA_MSG_ERROR ("Could not read in sample meta object");
      return ::StatusCode::FAILURE;
    }
 
    jobConfig.reset (dynamic_cast<JobConfig*>(f->Get("jobConfig")));
    if (jobConfig == nullptr)
    {
      ANA_MSG_ERROR ("failed to read jobConfig object");
      return ::StatusCode::FAILURE;
    }
 
    {
      std::unique_ptr<TList> outs ((TList*)f->Get("outputs"));
      if (outs == nullptr)
      {
        ANA_MSG_ERROR ("Could not read list of outputs");
        return ::StatusCode::FAILURE;
      }
 
      TIter itr(outs.get());
      TObject *obj = 0;
      while ((obj = itr())) {
        EL::OutputStream * out = dynamic_cast<EL::OutputStream*>(obj);
        if (out) {
          bigOutputs.Add(out);
        }
        else {
          ANA_MSG_ERROR ("Encountered unexpected entry in list of outputs");
          return ::StatusCode::FAILURE;
        }
      }
    }
 
    f->Close();
    f.reset ();
 
    const std::string location = ".";
 
    mo->setBool (Job::optGridReporting, true);
    setMetaData (mo);
    setOutputHist (location);
    setSegmentName ("output");
 
    ANA_MSG_INFO ("Starting EventLoop Grid worker");
 
    {//Create and register the "big" output files with base class
      TIter itr(&bigOutputs);
      TObject *obj = 0;
      while ((obj = itr())) {
        EL::OutputStream *os = dynamic_cast<EL::OutputStream*>(obj);
        if (os == nullptr)
        {
          ANA_MSG_ERROR ("Bad input");
          return ::StatusCode::FAILURE;
        }
        {
          Detail::OutputStreamData data {
            location + "/" + os->label() + ".root", "RECREATE"};
          ANA_CHECK (addOutputStream (os->label(), std::move (data)));
        }
      }
    }
 
    setJobConfig (std::move (*jobConfig));
 
    {
      auto module = std::make_unique<Detail::DirectInputModule> ("DirectInputModule");
      std::ifstream infile("input.txt");
      while (infile) {
        std::string sLine;
        if (!getline(infile, sLine)) break;
        std::istringstream ssLine(sLine);
        while (ssLine) {
          std::string sFile;
          if (!getline(ssLine, sFile, ',')) break;
          module->fileList.push_back(sFile);
        }
      }
      if (module->fileList.size() == 0) {
        ANA_MSG_ERROR ("no input files provided");
        //User was expecting input after all.
        gSystem->Exit(EC_BADINPUT);
      }
 
      if (nEventsPerJob != -1)
        module->maxEvents = nEventsPerJob;
      if (SkipEvents != 0)
        module->skipEvents = SkipEvents;
      addModule (std::move (module));
    }
 
    ANA_CHECK (initialize());
    ANA_CHECK (processInputs ());
    ANA_CHECK (finalize ());
 
    int nEvents = eventsProcessed();
    ANA_MSG_INFO ("Loop finished.");
    ANA_MSG_INFO ("Read/processed " << nEvents << " events.");
 
    ANA_MSG_INFO ("EventLoop Grid worker finished");
    ANA_MSG_INFO ("Saving output");
    return ::StatusCode::SUCCESS;
  }

initialize()

StatusCode EL::Worker::initialize ( )

protected

initialize the worker

This method ought to be called after the options on the worker are set and before any events are processed. It is meant to make sure everything is ready and set up for event processing.

Guarantee

basic

Failures

initialization failures

Definition at line 376 of file Worker.cxx.

  {
    using namespace msgEventLoop;
    RCU_CHANGE_INVARIANT (this);
 
    const bool xAODInput = m_metaData->castBool (Job::optXAODInput, false);
 
    ANA_MSG_INFO ("xAODInput = " << xAODInput);
    if (metaData()->castBool (Job::optAlgorithmMemoryMonitor, false))
      m_modules.push_back (std::make_unique<Detail::MemoryMonitorModule> ("EarlyMemoryMonitorModule"));
    if (xAODInput)
      m_modules.push_back (std::make_unique<Detail::TEventModule> ("TEventModule"));
    auto factoryPreload = metaData()->castString (Job::optFactoryPreload, "");
    if (!factoryPreload.empty())
    {
      auto module = std::make_unique<Detail::FactoryPreloadModule> ("FactoryPreloadModule");
      module->preloader = factoryPreload;
      m_modules.push_back (std::move (module));
    }
    m_modules.push_back (std::make_unique<Detail::LeakCheckModule> ("LeakCheckModule"));
    m_modules.push_back (std::make_unique<Detail::StopwatchModule> ("StopwatchModule"));
    if (metaData()->castBool (Job::optGridReporting, false))
      m_modules.push_back (std::make_unique<Detail::GridReportingModule>("GridReportingModule"));
    if (metaData()->castBool (Job::optAlgorithmTimer, false))
      m_modules.push_back (std::make_unique<Detail::AlgorithmTimerModule> ("AlgorithmTimerModule"));
    if (metaData()->castBool (Job::optAlgorithmMemoryMonitor, false))
      m_modules.push_back (std::make_unique<Detail::AlgorithmMemoryModule> ("AlgorithmMemoryModule"));
    m_modules.push_back (std::make_unique<Detail::FileExecutedModule> ("FileExecutedModule"));
    m_modules.push_back (std::make_unique<Detail::EventCountModule> ("EventCountModule"));
    m_modules.push_back (std::make_unique<Detail::WorkerConfigModule> ("WorkerConfigModule"));
    m_modules.push_back (std::make_unique<Detail::AlgorithmStateModule> ("AlgorithmStateModule"));
    m_modules.push_back (std::make_unique<Detail::PostClosedOutputsModule> ("PostClosedOutputsModule"));
    if (metaData()->castBool (Job::optAlgorithmMemoryMonitor, false))
      m_modules.push_back (std::make_unique<Detail::MemoryMonitorModule> ("LateMemoryMonitorModule"));
 
    if (m_outputs.find (Job::histogramStreamName) == m_outputs.end())
    {
      Detail::OutputStreamData data {
        m_outputTarget + "/hist-" + m_segmentName + ".root", "RECREATE"};
      ANA_CHECK (addOutputStream (Job::histogramStreamName, std::move (data)));
    }
    RCU_ASSERT (m_outputs.find (Job::histogramStreamName) != m_outputs.end());
    m_histOutput = m_outputs.at(Job::histogramStreamName).get();
    if (auto aliases = metaData()->castString (Job::optStreamAliases, ""); !aliases.empty())
    {
      // the format of aliases is "alias1=realname1,alias2=realname2"
      std::istringstream iss (aliases);
      std::string alias;
      while (std::getline (iss, alias, ','))
      {
        auto pos = alias.find ('=');
        if (pos == std::string::npos)
        {
          ANA_MSG_ERROR ("Invalid alias format: " << alias);
          return ::StatusCode::FAILURE;
        }
        auto aliasName = alias.substr (0, pos);
        auto realName = alias.substr (pos + 1);
        auto realOutput = m_outputs.find (realName);
        if (realOutput == m_outputs.end())
        {
          ANA_MSG_ERROR ("output stream " << realName << " not found for alias " << aliasName);
          return ::StatusCode::FAILURE;
        }
        auto [aliasOutput, success] = m_outputs.emplace (aliasName, realOutput->second);
        if (!success)
        {
          ANA_MSG_ERROR ("output stream " << aliasName << " already exists, can't make alias");
          return ::StatusCode::FAILURE;
        }
      }
    }
 
    m_jobStats = std::make_unique<TTree>
      ("EventLoop_JobStats", "EventLoop job statistics");
    m_jobStats->SetDirectory (nullptr);
 
    ANA_MSG_INFO ("calling firstInitialize on all modules");
    for (auto& module : m_modules)
      ANA_CHECK (module->firstInitialize (*this));
    ANA_MSG_INFO ("calling preFileInitialize on all modules");
    for (auto& module : m_modules)
      ANA_CHECK (module->preFileInitialize (*this));
    
    return ::StatusCode::SUCCESS;
  }

inputFile()

TFile * EL::Worker::inputFile ( ) const

overridevirtual

description: the file we are reading the current tree from guarantee: no-fail

Implements EL::IWorker.

Definition at line 226 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    return m_inputFile.get();
  }

inputFileName()

std::string EL::Worker::inputFileName ( ) const

overridevirtual

the name of the file we are reading the current tree from, without the path component

Guarantee

no-fail

Implements EL::IWorker.

Definition at line 235 of file Worker.cxx.

  {
    // no invariant used
    std::string path = inputFile()->GetName();
    auto split = path.rfind ('/');
    if (split != std::string::npos)
      return path.substr (split + 1);
    else
      return path;
  }

inputFileNumEntries()

Long64_t EL::Worker::inputFileNumEntries ( ) const

overrideprotectedvirtual

the number of events in the input file

Guarantee

no-fail

Precondition

inputFile() != 0

Implements EL::Detail::IInputModuleActions.

Definition at line 737 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    RCU_REQUIRE (inputFile() != 0);
 
    if (m_inputTree != 0)
      return m_inputTree->GetEntries();
    else
      return 0;
  }

metaData()

const SH::MetaObject * EL::Worker::metaData ( ) const

overridevirtual

description: the sample meta-data we are working on guarantee: no-fail invariant: metaData != 0 rationale: this can be used for accessing sample meta-data

Implements EL::IWorker.

Definition at line 199 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    return m_metaData;
  }

openInputFile()

StatusCode EL::Worker::openInputFile ( const std::string &  inputFileUrl )

overrideprotectedvirtual

open the given input file without processing it

This is mostly to allow the driver to query the number of events in the input file without processing it, usually to determine the range of events to process.

Guarantee

basic

Failures

file can't be opened

Implements EL::Detail::IInputModuleActions.

Definition at line 636 of file Worker.cxx.

  {
    using namespace msgEventLoop;
 
    // Enable custom error handling in a nice way
    RootUtils::WithRootErrorHandler my_handler( fileOpenErrorFilter );
 
    RCU_CHANGE_INVARIANT (this);
 
    if (m_inputFileUrl == inputFileUrl)
      return ::StatusCode::SUCCESS;
 
    if (!m_inputFileUrl.empty())
    {
      if (m_newInputFile == false)
      {
        for (auto& module : m_modules)
          ANA_CHECK (module->onCloseInputFile (*this));
        for (auto& module : m_modules)
          ANA_CHECK (module->postCloseInputFile (*this));
      }
      m_newInputFile = false;
      m_inputTree = nullptr;
      m_inputFile.reset ();
      m_inputFileUrl.clear ();
    }
 
    if (inputFileUrl.empty())
      return ::StatusCode::SUCCESS;
 
    ANA_MSG_INFO ("Opening file " << inputFileUrl);
    std::unique_ptr<TFile> inputFile;
    try
    {
      inputFile = SH::openFile (inputFileUrl, *metaData());
    } catch (...)
    {
      Detail::report_exception (std::current_exception());
    }
    if (inputFile.get() == 0)
    {
      ANA_MSG_ERROR ("failed to open file " << inputFileUrl);
      for (auto& module : m_modules)
        module->reportInputFailure (*this);
      return ::StatusCode::FAILURE;
    }
    if (inputFile->IsZombie())
    {
      ANA_MSG_ERROR ("input file is a zombie: " << inputFileUrl);
      for (auto& module : m_modules)
        module->reportInputFailure (*this);
      return ::StatusCode::FAILURE;
    }
 
    TTree *tree = 0;
    const std::string treeName
      = m_metaData->castString (SH::MetaFields::treeName, SH::MetaFields::treeName_default);
    tree = dynamic_cast<TTree*>(inputFile->Get (treeName.c_str()));
    if (tree == nullptr)
    {
      ANA_MSG_INFO ("tree " << treeName << " not found in input file: " << inputFileUrl);
      ANA_MSG_INFO ("treating this like a tree with no events");
    }
 
    m_newInputFile = true;
    m_inputTree = tree;
    m_inputTreeEntry = 0;
    m_inputFile = std::move (inputFile);
    m_inputFileUrl = std::move (inputFileUrl);
 
    return ::StatusCode::SUCCESS;
  }

processEvents()

StatusCode EL::Worker::processEvents ( EventRange &  eventRange )

overrideprotectedvirtual

process the given event range

This will update eventRange if the end is set to eof

Guarantee

basic

Failures

file can't be opened
event range exceeds length of file
processing failures

Implements EL::Detail::IInputModuleActions.

Definition at line 533 of file Worker.cxx.

  {
    using namespace msgEventLoop;
 
    RCU_CHANGE_INVARIANT (this);
    RCU_REQUIRE (!eventRange.m_url.empty());
    RCU_REQUIRE (eventRange.m_beginEvent >= 0);
    RCU_REQUIRE (eventRange.m_endEvent == EventRange::eof || eventRange.m_endEvent >= eventRange.m_beginEvent);
 
    ANA_CHECK (openInputFile (eventRange.m_url));
 
    if (eventRange.m_beginEvent > inputFileNumEntries())
    {
      ANA_MSG_ERROR ("first event (" << eventRange.m_beginEvent << ") points beyond last event in file (" << inputFileNumEntries() << ")");
      return ::StatusCode::FAILURE;
    }
    if (eventRange.m_endEvent == EventRange::eof)
    {
      eventRange.m_endEvent = inputFileNumEntries();
    } else if (eventRange.m_endEvent > inputFileNumEntries())
    {
      ANA_MSG_ERROR ("end event (" << eventRange.m_endEvent << ") points beyond last event in file (" << inputFileNumEntries() << ")");
      return ::StatusCode::FAILURE;
    }
 
    m_inputTreeEntry = eventRange.m_beginEvent;
 
    if (m_algorithmsInitialized == false)
    {
      for (auto& module : m_modules)
        ANA_CHECK (module->onInitialize (*this));
      m_algorithmsInitialized = true;
    }
 
    if (m_newInputFile)
    {
      m_newInputFile = false;
      for (auto& module : m_modules)
        ANA_CHECK (module->onNewInputFile (*this));
    }
 
    if (eventRange.m_beginEvent == 0)
    {
      for (auto& module : m_modules)
        ANA_CHECK (module->onFileExecute (*this));
    }
 
    ANA_MSG_INFO ("Processing events " << eventRange.m_beginEvent << "-" << eventRange.m_endEvent << " in file " << eventRange.m_url);
 
    for (uint64_t event = eventRange.m_beginEvent;
         event != uint64_t (eventRange.m_endEvent);
         ++ event)
    {
      m_inputTreeEntry = event;
      for (auto& module : m_modules)
      {
        if (module->onExecute (*this).isFailure())
        {
          ANA_MSG_ERROR ("processing event " << treeEntry() << " on file " << inputFileName());
          return ::StatusCode::FAILURE;
        }
      }
      if (m_firstEvent)
      {
        m_firstEvent = false;
        for (auto& module : m_modules)
          ANA_CHECK (module->postFirstEvent (*this));
      }
      m_eventsProcessed += 1;
      if (m_eventsProcessed % 10000 == 0)
        ANA_MSG_INFO ("Processed " << m_eventsProcessed << " events");
    }
    return ::StatusCode::SUCCESS;
  }

processInputs()

StatusCode EL::Worker::processInputs ( )

protected

process all the inputs

This method ought to be called after initialize and before finalize. It will rely on the defined modules to steer it to the files and events it ought to process.

Definition at line 466 of file Worker.cxx.

  {
    using namespace msgEventLoop;
 
    RCU_CHANGE_INVARIANT (this);
 
    for (auto& module : m_modules)
      ANA_CHECK (module->processInputs (*this, *this));
    return ::StatusCode::SUCCESS;
  }

setFilterPassed()

void EL::Worker::setFilterPassed ( bool  val_filterPassed )

finaloverridevirtualnoexcept

set the value of filterPassed

Guarantee

no-fail

Implements EL::IFilterWorker.

Definition at line 314 of file Worker.cxx.

  {
    RCU_CHANGE_INVARIANT (this);
    m_skipEvent = !val_filterPassed;
  }

setJobConfig()

void EL::Worker::setJobConfig ( JobConfig &&  jobConfig )

protected

set the JobConfig

This takes care of adding the algorithms, etc. (only algorithms for now, 03 Feb 17).

Note the rvalue calling convention here: Algorithms are objects that get modified, so if you use them more than once you need to copy/clone them. However, in practice no driver should need that (though many do for now, 03 Feb 17), as drivers generally stream the JobConfig in for one-time use.

Guarantee

basic

Failures

job configuration/streaming errors

Definition at line 364 of file Worker.cxx.

  {
    RCU_CHANGE_INVARIANT (this);
    for (std::unique_ptr<IAlgorithmWrapper>& alg : jobConfig.extractAlgorithms())
    {
      m_algs.push_back (std::move (alg));
    }
  }

setMetaData()

void EL::Worker::setMetaData ( const SH::MetaObject *  val_metaData )

protected

set the metaData

Guarantee

no-fail

Definition at line 333 of file Worker.cxx.

  {
    RCU_CHANGE_INVARIANT (this);
    RCU_REQUIRE (val_metaData != 0);
 
    m_metaData = val_metaData;
  }

setOutputHist()

void EL::Worker::setOutputHist ( const std::string &  val_outputTarget )

protected

set the histogram output list

Guarantee

no-fail

Definition at line 344 of file Worker.cxx.

  {
    RCU_CHANGE_INVARIANT (this);
 
    m_outputTarget = val_outputTarget;
  }

setSegmentName()

void EL::Worker::setSegmentName ( const std::string &  val_segmentName )

protected

set the segment name

Guarantee

no-fail

Definition at line 354 of file Worker.cxx.

  {
    RCU_CHANGE_INVARIANT (this);
 
    m_segmentName = val_segmentName;
  }

skipEvent()

void EL::Worker::skipEvent ( )

overridevirtual

effects: skip the current event, i.e.

skip the rest of the algorithms for this event guarantee: no-fail rationale: if you have an analysis strategy in which you divide work into multiple algorithms you may want to have dedicated algorithms for event selection that then skip later algorithms that fill histograms

Implements EL::IWorker.

Definition at line 296 of file Worker.cxx.

  {
    RCU_CHANGE_INVARIANT (this);
    m_skipEvent = true;
  }

testInvariant()

void EL::Worker::testInvariant

(

)

const

effects: test the invariant of this object guarantee: no-fail

Definition at line 67 of file Worker.cxx.

  {
    RCU_INVARIANT (this != nullptr);
    for (std::size_t iter = 0, end = m_algs.size(); iter != end; ++ iter)
    {
      RCU_INVARIANT (m_algs[iter].m_algorithm != nullptr);
    }
  }

tree()

TTree * EL::Worker::tree ( ) const

overridevirtual

description: the tree we are running on guarantee: no-fail

Implements EL::IWorker.

Definition at line 208 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    return m_inputTree;
  }

treeEntry()

Long64_t EL::Worker::treeEntry ( ) const

overridevirtual

description: the entry in the tree we are reading guarantee: no-fail

Implements EL::IWorker.

Definition at line 217 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    return m_inputTreeEntry;
  }

triggerConfig()

TTree * EL::Worker::triggerConfig ( ) const

overridevirtual

description: the trigger config tree from the input file, or NULL if we did not find it guarantee: strong failures: i/o errors

Implements EL::IWorker.

Definition at line 249 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
    return dynamic_cast<TTree*>(inputFile()->Get("physicsMeta/TrigConfTree"));
  }

xaodEvent()

xAOD::TEvent * EL::Worker::xaodEvent ( ) const

overridevirtual

description: the xAOD event and store guarantee: strong failures: out of memory I failures: TEventSvc not configured postcondition: result != 0

Implements EL::IWorker.

Definition at line 258 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
 
    if (m_tevent == nullptr)
      RCU_THROW_MSG ("Job not configured for xAOD support");
    return m_tevent;
  }

xaodStore()

xAOD::TStore * EL::Worker::xaodStore ( ) const

overridevirtual

Implements EL::IWorker.

Definition at line 270 of file Worker.cxx.

  {
    RCU_READ_INVARIANT (this);
 
    if (m_tstore == nullptr)
      RCU_THROW_MSG ("Job not configured for xAOD support");
    return m_tstore;
  }

Member Data Documentation

m_algorithmsInitialized

bool EL::Worker::m_algorithmsInitialized {false}

private

whether the algorithms are initialized

Definition at line 444 of file Worker.h.

m_algs

std::vector<Detail::AlgorithmData> EL::Detail::ModuleData::m_algs

inherited

the list of algorithms

Definition at line 66 of file ModuleData.h.

m_eventsProcessed

uint64_t EL::Detail::ModuleData::m_eventsProcessed {0}

inherited

the number of events that have been processed

Definition at line 88 of file ModuleData.h.

m_evtStore

asg::SgTEvent* EL::Detail::ModuleData::m_evtStore {nullptr}

inherited

the SgTEvent structure, if we use one

Definition at line 103 of file ModuleData.h.

m_firstEvent

bool EL::Worker::m_firstEvent {true}

private

whether we are still to process the first event

Definition at line 449 of file Worker.h.

m_histOutput

OutputStreamData* EL::Detail::ModuleData::m_histOutput {nullptr}

inherited

the histogram output stream

Definition at line 91 of file ModuleData.h.

m_inputFile

std::unique_ptr<TFile> EL::Detail::ModuleData::m_inputFile

inherited

the input file pointer of the currently opened filed

Definition at line 72 of file ModuleData.h.

m_inputFileUrl

std::string EL::Detail::ModuleData::m_inputFileUrl

inherited

the input file url of the currently opened file

Definition at line 69 of file ModuleData.h.

m_inputTree

TTree* EL::Detail::ModuleData::m_inputTree {nullptr}

inherited

the (main) tree in the input file

Definition at line 75 of file ModuleData.h.

m_inputTreeEntry

uint64_t EL::Detail::ModuleData::m_inputTreeEntry {0}

inherited

the entry in the input tree we are currently looking at

Definition at line 79 of file ModuleData.h.

m_jobStats

std::unique_ptr<TTree> EL::Detail::ModuleData::m_jobStats

inherited

Tree saving per-job statistics information.

Definition at line 94 of file ModuleData.h.

m_metaData

const SH::MetaObject* EL::Detail::ModuleData::m_metaData {nullptr}

inherited

the meta-data we use

Definition at line 85 of file ModuleData.h.

m_modules

std::vector<std::unique_ptr<Detail::Module> > EL::Worker::m_modules

private

the list of modules we hold

Definition at line 423 of file Worker.h.

m_newInputFile

bool EL::Worker::m_newInputFile {false}

private

whether this is a new input file (i.e.

one that has not yet been connected to the algorithms)

Definition at line 429 of file Worker.h.

m_outputHistMap

std::map<std::string,TH1*> EL::Worker::m_outputHistMap

private

Definition at line 411 of file Worker.h.

m_outputs

std::map<std::string,std::shared_ptr<Detail::OutputStreamData> > EL::Detail::ModuleData::m_outputs

inherited

the list of output files

Definition at line 109 of file ModuleData.h.

m_outputTarget

std::string EL::Worker::m_outputTarget

private

the target file to which we will write the histogram output

Definition at line 434 of file Worker.h.

m_outputTreeMap

std::map<std::pair<std::string,std::string>,TTree*> EL::Worker::m_outputTreeMap

private

Definition at line 418 of file Worker.h.

m_segmentName

std::string EL::Worker::m_segmentName

private

the name of the segment we are processing

Definition at line 439 of file Worker.h.

m_skipEvent

bool EL::Detail::ModuleData::m_skipEvent = false

inherited

whether we are skipping the current event

Definition at line 82 of file ModuleData.h.

m_tevent

xAOD::TEvent* EL::Detail::ModuleData::m_tevent {nullptr}

inherited

the TEvent structure, if we use one

Definition at line 97 of file ModuleData.h.

m_tstore

xAOD::TStore* EL::Detail::ModuleData::m_tstore {nullptr}

inherited

the TStore structure, if we use one

Definition at line 100 of file ModuleData.h.

m_worker

Worker* EL::Detail::ModuleData::m_worker {nullptr}

inherited

the worker (to pass on to the algorithms)

Definition at line 106 of file ModuleData.h.

---

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

• Worker.h
• Worker.cxx