art::OutputWorker Class Reference

#include "OutputWorker.h"

Inheritance diagram for art::OutputWorker:

Public Types
enum	State {   Ready, Pass, Fail, Working,   ExceptionThrown }

Public Member Functions
virtual	~OutputWorker ()
	OutputWorker (OutputModule *mod, WorkerParams const &)
std::string const &	lastClosedFileName () const
void	closeFile ()
bool	fileIsOpen () const
void	incrementInputFileNumber ()
bool	requestsToCloseFile () const
void	openFile (FileBlock const &fb)
void	writeRun (RunPrincipal &rp)
void	writeSubRun (SubRunPrincipal &srp)
void	writeEvent (EventPrincipal &ep, PathContext const &pc)
void	setRunAuxiliaryRangeSetID (RangeSet const &)
void	setSubRunAuxiliaryRangeSetID (RangeSet const &)
void	setFileStatus (OutputFileStatus)
Granularity	fileGranularity () const
void	selectProducts (ProductTables const &)
void	beginJob (detail::SharedResources const &)
void	endJob ()
void	respondToOpenInputFile (FileBlock const &fb)
void	respondToCloseInputFile (FileBlock const &fb)
void	respondToOpenOutputFiles (FileBlock const &fb)
void	respondToCloseOutputFiles (FileBlock const &fb)
void	doWork (Transition, Principal &, ModuleContext const &)
void	doWork_event (hep::concurrency::WaitingTaskPtr workerInPathDoneTask, EventPrincipal &, ModuleContext const &)
void	doWork_event (EventPrincipal &, ModuleContext const &)
ScheduleID	scheduleID () const
bool	returnCode () const
ModuleDescription const &	description () const
hep::concurrency::SerialTaskQueueChain *	serialTaskQueueChain () const
void	reset ()
std::size_t	timesVisited () const
std::size_t	timesRun () const
std::size_t	timesPassed () const
std::size_t	timesFailed () const
std::size_t	timesExcept () const
void	runWorker (EventPrincipal &, ModuleContext const &)
bool	isUnique () const

Protected Member Functions
std::string const &	label () const

Protected Attributes
std::atomic< std::size_t >	counts_visited_ {}
std::atomic< std::size_t >	counts_run_ {}
std::atomic< std::size_t >	counts_passed_ {}
std::atomic< std::size_t >	counts_failed_ {}
std::atomic< std::size_t >	counts_thrown_ {}

Private Member Functions
hep::concurrency::SerialTaskQueueChain *	doSerialTaskQueueChain () const override
void	doBeginJob (detail::SharedResources const &) override
void	doEndJob () override
void	doRespondToOpenInputFile (FileBlock const &) override
void	doRespondToCloseInputFile (FileBlock const &) override
void	doRespondToOpenOutputFiles (FileBlock const &) override
void	doRespondToCloseOutputFiles (FileBlock const &) override
void	doBegin (RunPrincipal &, ModuleContext const &) override
void	doEnd (RunPrincipal &, ModuleContext const &) override
void	doBegin (SubRunPrincipal &, ModuleContext const &) override
void	doEnd (SubRunPrincipal &, ModuleContext const &) override
bool	doProcess (EventPrincipal &, ModuleContext const &) override

Private Attributes
cet::exempt_ptr< OutputModule >	module_
ServiceHandle< CatalogInterface >	ci_ {}
ActivityRegistry const &	actReg_
Granularity	fileGranularity_ {Granularity::Unset}

Detailed Description

Definition at line 25 of file OutputWorker.h.

Member Enumeration Documentation

enum art::Worker::State

inherited

Enumerator
Ready
Pass
Fail
Working
ExceptionThrown

Definition at line 49 of file Worker.h.

{ Ready, Pass, Fail, Working, ExceptionThrown };

Constructor & Destructor Documentation

art::OutputWorker::~OutputWorker ( )

virtualdefault

art::OutputWorker::OutputWorker	(	OutputModule *	mod,
		WorkerParams const &	wp
	)

Definition at line 18 of file OutputWorker.cc.

References actReg_, ci_, art::Worker::description(), art::ScheduleID::first(), fhicl::ParameterSetRegistry::get(), art::Worker::label(), art::module, module_, and art::ModuleBase::moduleDescription().

    : Worker{module->moduleDescription(), wp}
    , module_{module}
    , actReg_{wp.actReg_}
  {
    if (wp.scheduleID_ == ScheduleID::first()) {
      // We only want to register the products (and any shared
      // resources) once, not once for every schedule)
      module_->registerProducts(wp.producedProducts_);
      wp.resources_.registerSharedResources(module_->sharedResources());
    }
    ci_->outputModuleInitiated(
      label(),
      fhicl::ParameterSetRegistry::get(description().parameterSetID()));
  }

Member Function Documentation

void art::Worker::beginJob ( detail::SharedResources const &  resources )

inherited

Definition at line 196 of file Worker.cc.

References art::Worker::actReg_, art::Worker::doBeginJob(), art::Worker::md_, art::ActivityRegistry::sPostModuleBeginJob, and art::ActivityRegistry::sPreModuleBeginJob.

      {
    actReg_.sPreModuleBeginJob.invoke(md_);
    doBeginJob(resources);
    actReg_.sPostModuleBeginJob.invoke(md_);
  }
  catch (...) {
    rethrow_with_context(std::current_exception(), md_, "beginJob");
  }

void art::OutputWorker::closeFile

(

)

Definition at line 116 of file OutputWorker.cc.

References actReg_, ci_, art::Worker::label(), lastClosedFileName(), module_, art::ActivityRegistry::sPostCloseOutputFile, and art::ActivityRegistry::sPreCloseOutputFile.

  {
    actReg_.sPreCloseOutputFile.invoke(label());
    if (module_->doCloseFile()) {
      ci_->outputFileClosed(label(), lastClosedFileName());
    }
    actReg_.sPostCloseOutputFile.invoke(
      OutputFileInfo{label(), lastClosedFileName()});
  }

ModuleDescription const & art::Worker::description ( ) const

inherited

Definition at line 121 of file Worker.cc.

References art::Worker::md_.

Referenced by OutputWorker(), and writeEvent().

  {
    return md_;
  }

void art::OutputWorker::doBegin ( RunPrincipal &  rp, ModuleContext const &  mc  )

overrideprivatevirtual

Implements art::Worker.

Definition at line 77 of file OutputWorker.cc.

References module_.

  {
    module_->doBeginRun(rp, mc);
  }

void art::OutputWorker::doBegin ( SubRunPrincipal &  srp, ModuleContext const &  mc  )

overrideprivatevirtual

Implements art::Worker.

Definition at line 89 of file OutputWorker.cc.

References module_.

  {
    module_->doBeginSubRun(srp, mc);
  }

void art::OutputWorker::doBeginJob ( detail::SharedResources const &  resources )

overrideprivatevirtual

Implements art::Worker.

Definition at line 41 of file OutputWorker.cc.

References module_.

  {
    module_->doBeginJob(resources);
  }

void art::OutputWorker::doEnd ( RunPrincipal &  rp, ModuleContext const &  mc  )

overrideprivatevirtual

Implements art::Worker.

Definition at line 83 of file OutputWorker.cc.

References module_.

  {
    module_->doEndRun(rp, mc);
  }

void art::OutputWorker::doEnd ( SubRunPrincipal &  srp, ModuleContext const &  mc  )

overrideprivatevirtual

Implements art::Worker.

Definition at line 95 of file OutputWorker.cc.

References module_.

  {
    module_->doEndSubRun(srp, mc);
  }

void art::OutputWorker::doEndJob ( )

overrideprivatevirtual

Implements art::Worker.

Definition at line 47 of file OutputWorker.cc.

References module_.

  {
    module_->doEndJob();
  }

bool art::OutputWorker::doProcess ( EventPrincipal &  ep, ModuleContext const &  mc  )

overrideprivatevirtual

Implements art::Worker.

Definition at line 101 of file OutputWorker.cc.

References art::Worker::counts_failed_, art::Worker::counts_passed_, art::Worker::counts_run_, and module_.

  {
    // Note, only filters ever return false, and when they do it means
    // they have rejected.
    return module_->doEvent(
      ep, mc, counts_run_, counts_passed_, counts_failed_);
  }

void art::OutputWorker::doRespondToCloseInputFile ( FileBlock const &  fb )

overrideprivatevirtual

Implements art::Worker.

Definition at line 59 of file OutputWorker.cc.

References module_.

  {
    module_->doRespondToCloseInputFile(fb);
  }

void art::OutputWorker::doRespondToCloseOutputFiles ( FileBlock const &  fb )

overrideprivatevirtual

Implements art::Worker.

Definition at line 71 of file OutputWorker.cc.

References module_.

  {
    module_->doRespondToCloseOutputFiles(fb);
  }

void art::OutputWorker::doRespondToOpenInputFile ( FileBlock const &  fb )

overrideprivatevirtual

Implements art::Worker.

Definition at line 53 of file OutputWorker.cc.

References module_.

  {
    module_->doRespondToOpenInputFile(fb);
  }

void art::OutputWorker::doRespondToOpenOutputFiles ( FileBlock const &  fb )

overrideprivatevirtual

Implements art::Worker.

Definition at line 65 of file OutputWorker.cc.

References module_.

  {
    module_->doRespondToOpenOutputFiles(fb);
  }

hep::concurrency::SerialTaskQueueChain * art::OutputWorker::doSerialTaskQueueChain ( ) const

overrideprivatevirtual

Implements art::Worker.

Definition at line 35 of file OutputWorker.cc.

References module_.

  {
    return module_->serialTaskQueueChain();
  }

void art::Worker::doWork ( Transition  trans, Principal &  principal, ModuleContext const &  mc  )

inherited

Definition at line 250 of file Worker.cc.

References art::Worker::actReg_, art::errors::BadAlloc, art::errors::BadExceptionType, art::BeginRun, art::BeginSubRun, art::Worker::cached_exception_, art::Worker::doBegin(), art::Worker::doEnd(), art::EndRun, art::EndSubRun, art::Worker::ExceptionThrown, art::Worker::Fail, art::Worker::md_, art::errors::OtherArt, art::Worker::Pass, art::Worker::Ready, art::errors::ScheduleExecutionFailure, art::ActivityRegistry::sPostModuleBeginRun, art::ActivityRegistry::sPostModuleBeginSubRun, art::ActivityRegistry::sPostModuleEndRun, art::ActivityRegistry::sPostModuleEndSubRun, art::ActivityRegistry::sPreModuleBeginRun, art::ActivityRegistry::sPreModuleBeginSubRun, art::ActivityRegistry::sPreModuleEndRun, art::ActivityRegistry::sPreModuleEndSubRun, art::Worker::state_, art::errors::StdException, art::errors::Unknown, and art::Worker::Working.

  {
    switch (state_.load()) {
    case Ready:
      break;
    case Pass:
    case Fail:
      return;
    case ExceptionThrown: {
      // Rethrow the cached exception again. It seems impossible to
      // get here a second time unless a cet::exception has been
      // thrown previously.
      mf::LogWarning("repeat") << "A module has been invoked a second time "
                                  "even though it caught an exception during "
                                  "the previous invocation.\nThis may be an "
                                  "indication of a configuration problem.";
      rethrow_exception(cached_exception_);
    }
    case Working:
      break; // See below.
    }

    try {
      if (state_.load() == Working) {
        // Not part of the switch statement above because we want the
        // exception to be caught by our handling mechanism.
        throw Exception(errors::ScheduleExecutionFailure)
          << "A Module has been invoked while it is still being executed.\n"
          << "Product dependencies have invoked a module execution cycle.\n";
      }
      state_ = Working;
      if (trans == Transition::BeginRun) {
        actReg_.sPreModuleBeginRun.invoke(mc);
        doBegin(dynamic_cast<RunPrincipal&>(principal), mc);
        actReg_.sPostModuleBeginRun.invoke(mc);
      } else if (trans == Transition::EndRun) {
        actReg_.sPreModuleEndRun.invoke(mc);
        doEnd(dynamic_cast<RunPrincipal&>(principal), mc);
        actReg_.sPostModuleEndRun.invoke(mc);
      } else if (trans == Transition::BeginSubRun) {
        actReg_.sPreModuleBeginSubRun.invoke(mc);
        doBegin(dynamic_cast<SubRunPrincipal&>(principal), mc);
        actReg_.sPostModuleBeginSubRun.invoke(mc);
      } else if (trans == Transition::EndSubRun) {
        actReg_.sPreModuleEndSubRun.invoke(mc);
        doEnd(dynamic_cast<SubRunPrincipal&>(principal), mc);
        actReg_.sPostModuleEndSubRun.invoke(mc);
      }
      state_ = Pass;
    }
    catch (cet::exception& e) {
      state_ = ExceptionThrown;
      e << "The above exception was thrown while processing module "
        << brief_context(md_, principal) << '\n';
      if (auto edmEx = dynamic_cast<Exception*>(&e)) {
        cached_exception_ = std::make_exception_ptr(*edmEx);
      } else {
        auto art_ex = Exception{errors::OtherArt, std::string(), e};
        cached_exception_ = std::make_exception_ptr(art_ex);
      }
      throw;
    }
    catch (std::bad_alloc const& bda) {
      state_ = ExceptionThrown;
      auto art_ex =
        Exception{errors::BadAlloc}
        << "A bad_alloc exception occurred during a call to the module "
        << brief_context(md_, principal) << '\n'
        << "The job has probably exhausted the virtual memory available to the "
           "process.\n";
      cached_exception_ = make_exception_ptr(art_ex);
      rethrow_exception(cached_exception_);
    }
    catch (std::exception const& e) {
      state_ = ExceptionThrown;
      auto art_ex = Exception{errors::StdException}
                    << "An exception occurred during a call to the module "
                    << brief_context(md_, principal) << e.what();
      cached_exception_ = make_exception_ptr(art_ex);
      rethrow_exception(cached_exception_);
    }
    catch (std::string const& s) {
      state_ = ExceptionThrown;
      auto art_ex = Exception{errors::BadExceptionType, "string"}
                    << "A string thrown as an exception occurred during a call "
                       "to the module "
                    << brief_context(md_, principal) << '\n'
                    << s << '\n';
      cached_exception_ = make_exception_ptr(art_ex);
      rethrow_exception(cached_exception_);
    }
    catch (char const* c) {
      state_ = ExceptionThrown;
      auto art_ex = Exception{errors::BadExceptionType, "char const*"}
                    << "A char const* thrown as an exception occurred during a "
                       "call to the module "
                    << brief_context(md_, principal) << '\n'
                    << c << '\n';
      cached_exception_ = make_exception_ptr(art_ex);
      rethrow_exception(cached_exception_);
    }
    catch (...) {
      state_ = ExceptionThrown;
      auto art_ex =
        Exception{errors::Unknown, "repeated"}
        << "An unknown occurred during a previous call to the module "
        << brief_context(md_, principal) << '\n';
      cached_exception_ = make_exception_ptr(art_ex);
      rethrow_exception(cached_exception_);
    }
  }

void art::Worker::doWork_event ( hep::concurrency::WaitingTaskPtr  workerInPathDoneTask, EventPrincipal &  , ModuleContext const &    )

inherited

Referenced by art::Worker::isUnique().

void art::Worker::doWork_event ( EventPrincipal &  p, ModuleContext const &  mc  )

inherited

Definition at line 366 of file Worker.cc.

References art::Worker::actions_, art::Worker::actReg_, art::errors::BadAlloc, art::errors::BadExceptionType, art::Worker::cached_exception_, art::Worker::counts_failed_, art::Worker::counts_passed_, art::Worker::counts_thrown_, art::Worker::counts_visited_, art::Worker::doProcess(), art::Worker::ExceptionThrown, art::Worker::Fail, art::actions::FailModule, art::ActionTable::find(), art::actions::IgnoreCompletely, art::Worker::md_, art::errors::OtherArt, art::Worker::Pass, art::Worker::returnCode_, art::ActivityRegistry::sPostModule, art::ActivityRegistry::sPreModule, art::Worker::state_, art::errors::StdException, art::errors::Unknown, and art::Worker::Working.

      {
    ++counts_visited_;
    returnCode_ = false;
    // Transition from Ready state to Working state.
    state_ = Working;
    actReg_.sPreModule.invoke(mc);
    // Note: Only the TriggerResults inserter--a producer--calls this
    // function.  The return code is thus always true.
    returnCode_ = doProcess(p, mc);
    actReg_.sPostModule.invoke(mc);
    assert(returnCode_.load());
    state_ = Pass;
  }
  catch (cet::exception& e) {
    auto action = actions_.find(e.root_cause());
    assert(not mc.onEndPath());
    if (action == actions::IgnoreCompletely) {
      state_ = Pass;
      returnCode_ = true;
      ++counts_passed_;
      mf::LogWarning("IgnoreCompletely") << "Module ignored an exception\n"
                                         << e.what();
      // WARNING: We will continue execution below!!!
    } else if (action == actions::FailModule) {
      state_ = Fail;
      returnCode_ = true;
      ++counts_failed_;
      mf::LogWarning("FailModule") << "Module failed due to an exception\n"
                                   << e.what();
      // WARNING: We will continue execution below!!!
    } else {
      state_ = ExceptionThrown;
      ++counts_thrown_;
      e << "The above exception was thrown while processing module "
        << brief_context(md_, p) << '\n';
      if (auto edmEx = dynamic_cast<Exception*>(&e)) {
        cached_exception_ = make_exception_ptr(*edmEx);
      } else {
        cached_exception_ =
          make_exception_ptr(Exception{errors::OtherArt, string(), e});
      }
      rethrow_exception(cached_exception_);
    }
  }
  catch (bad_alloc const& bda) {
    state_ = ExceptionThrown;
    ++counts_thrown_;
    auto art_ex =
      Exception{errors::BadAlloc}
      << "A bad_alloc exception occurred during a call to the module "
      << brief_context(md_, p) << '\n'
      << "The job has probably exhausted the virtual memory available to the "
         "process.\n";
    cached_exception_ = make_exception_ptr(art_ex);
    rethrow_exception(cached_exception_);
  }
  catch (exception const& e) {
    state_ = ExceptionThrown;
    ++counts_thrown_;
    auto art_ex = Exception{errors::StdException}
                  << "An exception occurred during a call to the module "
                  << brief_context(md_, p) << '\n'
                  << e.what();
    cached_exception_ = make_exception_ptr(art_ex);
    rethrow_exception(cached_exception_);
  }
  catch (string const& s) {
    state_ = ExceptionThrown;
    ++counts_thrown_;
    auto art_ex = Exception{errors::BadExceptionType, "string"}
                  << "A string thrown as an exception occurred during a call "
                     "to the module "
                  << brief_context(md_, p) << '\n'
                  << s << '\n';
    cached_exception_ = make_exception_ptr(art_ex);
    rethrow_exception(cached_exception_);
  }
  catch (char const* c) {
    state_ = ExceptionThrown;
    ++counts_thrown_;
    auto art_ex = Exception{errors::BadExceptionType, "char const*"}
                  << "A char const* thrown as an exception occurred during a "
                     "call to the module "
                  << brief_context(md_, p) << '\n'
                  << c << '\n';
    cached_exception_ = make_exception_ptr(art_ex);
    rethrow_exception(cached_exception_);
  }
  catch (...) {
    ++counts_thrown_;
    state_ = ExceptionThrown;
    auto art_ex = Exception{errors::Unknown, "repeated"}
                  << "An unknown occurred during a previous call to the module "
                  << brief_context(md_, p) << '\n';
    cached_exception_ = make_exception_ptr(art_ex);
    rethrow_exception(cached_exception_);
  }

void art::Worker::endJob ( )

inherited

Definition at line 207 of file Worker.cc.

References art::Worker::actReg_, art::Worker::doEndJob(), art::Worker::md_, art::ActivityRegistry::sPostModuleEndJob, and art::ActivityRegistry::sPreModuleEndJob.

      {
    actReg_.sPreModuleEndJob.invoke(md_);
    doEndJob();
    actReg_.sPostModuleEndJob.invoke(md_);
  }
  catch (...) {
    rethrow_with_context(std::current_exception(), md_, "endJob");
  }

Granularity art::OutputWorker::fileGranularity

(

)

const

Definition at line 199 of file OutputWorker.cc.

References module_.

  {
    return module_->fileGranularity();
  }

bool art::OutputWorker::fileIsOpen

(

)

const

Definition at line 181 of file OutputWorker.cc.

References module_.

  {
    return module_->fileIsOpen();
  }

void art::OutputWorker::incrementInputFileNumber

(

)

Definition at line 127 of file OutputWorker.cc.

References module_.

  {
    return module_->incrementInputFileNumber();
  }

bool art::Worker::isUnique ( ) const

inherited

Definition at line 592 of file Worker.cc.

References art::Worker::counts_visited_, art::Worker::doWork_event(), art::ScheduleID::first(), art::Worker::md_, art::ModuleDescription::moduleThreadingType(), art::replicated, art::Worker::runWorker(), art::ModuleContext::scheduleID(), art::Worker::scheduleID_, art::Worker::serialTaskQueueChain(), TDEBUG_BEGIN_FUNC_SI, TDEBUG_END_FUNC_SI, TDEBUG_FUNC_SI, art::Worker::waitingTasks_, and art::Worker::workStarted_.

  {
    if (scheduleID_ == ScheduleID::first()) {
      return true;
    }
    return md_.moduleThreadingType() == ModuleThreadingType::replicated;
  }

string const & art::Worker::label ( ) const

protectedinherited

Definition at line 127 of file Worker.cc.

References art::Worker::md_, and art::ModuleDescription::moduleLabel().

Referenced by closeFile(), openFile(), and OutputWorker().

  {
    return md_.moduleLabel();
  }

std::string const & art::OutputWorker::lastClosedFileName

(

)

const

Definition at line 110 of file OutputWorker.cc.

References module_.

Referenced by closeFile().

  {
    return module_->lastClosedFileName();
  }

void art::OutputWorker::openFile

(

FileBlock const &

fb

)

Definition at line 139 of file OutputWorker.cc.

References actReg_, ci_, art::Worker::label(), module_, and art::ActivityRegistry::sPostOpenOutputFile.

  {
    if (module_->doOpenFile(fb)) {
      ci_->outputFileOpened(label());
      actReg_.sPostOpenOutputFile.invoke(label());
    }
  }

bool art::OutputWorker::requestsToCloseFile

(

)

const

Definition at line 133 of file OutputWorker.cc.

References module_.

  {
    return module_->requestsToCloseFile();
  }

void art::Worker::reset ( )

inherited

Definition at line 149 of file Worker.cc.

References art::Worker::cached_exception_, art::Worker::Ready, art::Worker::returnCode_, art::Worker::scheduleID_, art::Worker::state_, TDEBUG_FUNC_SI, art::Worker::waitingTasks_, and art::Worker::workStarted_.

  {
    state_ = Ready;
    cached_exception_ = std::exception_ptr{};
    TDEBUG_FUNC_SI(6, scheduleID_)
      << hex << this << dec << " Resetting waitingTasks_";
    waitingTasks_.reset();
    workStarted_ = false;
    returnCode_ = false;
  }

void art::Worker::respondToCloseInputFile ( FileBlock const &  fb )

inherited

Definition at line 226 of file Worker.cc.

References art::Worker::actReg_, art::Worker::doRespondToCloseInputFile(), art::Worker::md_, art::ActivityRegistry::sPostModuleRespondToCloseInputFile, and art::ActivityRegistry::sPreModuleRespondToCloseInputFile.

  {
    actReg_.sPreModuleRespondToCloseInputFile.invoke(md_);
    doRespondToCloseInputFile(fb);
    actReg_.sPostModuleRespondToCloseInputFile.invoke(md_);
  }

void art::Worker::respondToCloseOutputFiles ( FileBlock const &  fb )

inherited

Definition at line 242 of file Worker.cc.

References art::Worker::actReg_, art::Worker::doRespondToCloseOutputFiles(), art::Worker::md_, art::ActivityRegistry::sPostModuleRespondToCloseOutputFiles, and art::ActivityRegistry::sPreModuleRespondToCloseOutputFiles.

  {
    actReg_.sPreModuleRespondToCloseOutputFiles.invoke(md_);
    doRespondToCloseOutputFiles(fb);
    actReg_.sPostModuleRespondToCloseOutputFiles.invoke(md_);
  }

void art::Worker::respondToOpenInputFile ( FileBlock const &  fb )

inherited

Definition at line 218 of file Worker.cc.

References art::Worker::actReg_, art::Worker::doRespondToOpenInputFile(), art::Worker::md_, art::ActivityRegistry::sPostModuleRespondToOpenInputFile, and art::ActivityRegistry::sPreModuleRespondToOpenInputFile.

  {
    actReg_.sPreModuleRespondToOpenInputFile.invoke(md_);
    doRespondToOpenInputFile(fb);
    actReg_.sPostModuleRespondToOpenInputFile.invoke(md_);
  }

void art::Worker::respondToOpenOutputFiles ( FileBlock const &  fb )

inherited

Definition at line 234 of file Worker.cc.

References art::Worker::actReg_, art::Worker::doRespondToOpenOutputFiles(), art::Worker::md_, art::ActivityRegistry::sPostModuleRespondToOpenOutputFiles, and art::ActivityRegistry::sPreModuleRespondToOpenOutputFiles.

  {
    actReg_.sPreModuleRespondToOpenOutputFiles.invoke(md_);
    doRespondToOpenOutputFiles(fb);
    actReg_.sPostModuleRespondToOpenOutputFiles.invoke(md_);
  }

bool art::Worker::returnCode ( ) const

inherited

Definition at line 134 of file Worker.cc.

References art::Worker::returnCode_.

  {
    return returnCode_.load();
  }

void art::Worker::runWorker ( EventPrincipal &  p, ModuleContext const &  mc  )

inherited

Definition at line 466 of file Worker.cc.

References art::Worker::actions_, art::Worker::actReg_, art::errors::BadAlloc, art::errors::BadExceptionType, art::Worker::cached_exception_, art::Worker::counts_failed_, art::Worker::counts_passed_, art::Worker::counts_thrown_, art::Worker::doProcess(), art::Worker::ExceptionThrown, art::Worker::Fail, art::actions::FailModule, art::actions::FailPath, art::ActionTable::find(), art::actions::IgnoreCompletely, art::Worker::md_, art::ModuleContext::onEndPath(), art::errors::OtherArt, art::Worker::Pass, art::Worker::returnCode_, art::ModuleContext::scheduleID(), art::actions::SkipEvent, art::ActivityRegistry::sPostModule, art::ActivityRegistry::sPreModule, art::Worker::state_, art::errors::StdException, TDEBUG_BEGIN_TASK_SI, TDEBUG_END_TASK_SI, art::errors::Unknown, art::Worker::waitingTasks_, and art::Worker::Working.

Referenced by art::Worker::isUnique().

  {
    auto const sid = mc.scheduleID();
    TDEBUG_BEGIN_TASK_SI(4, sid);
    returnCode_ = false;
    try {
      // Transition from Ready state to Working state.
      state_ = Working;
      actReg_.sPreModule.invoke(mc);
      // Note: Only filters ever return false, and when they do it
      // means they have rejected.
      returnCode_ = doProcess(p, mc);
      actReg_.sPostModule.invoke(mc);
      state_ = Fail;
      if (returnCode_.load()) {
        state_ = Pass;
      }
    }
    catch (cet::exception& e) {
      auto action = actions_.find(e.root_cause());
      // If we are processing an endPath, treat SkipEvent or FailPath
      // as FailModule, so any subsequent OutputModules are still run.
      if (mc.onEndPath()) {
        if ((action == actions::SkipEvent) || (action == actions::FailPath)) {
          action = actions::FailModule;
        }
      }
      if (action == actions::IgnoreCompletely) {
        state_ = Pass;
        returnCode_ = true;
        ++counts_passed_;
        mf::LogWarning("IgnoreCompletely") << "Module ignored an exception\n"
                                           << e.what();
        // WARNING: We will continue execution below!!!
      } else if (action == actions::FailModule) {
        state_ = Fail;
        returnCode_ = true;
        ++counts_failed_;
        mf::LogWarning("FailModule") << "Module failed due to an exception\n"
                                     << e.what();
        // WARNING: We will continue execution below!!!
      } else {
        state_ = ExceptionThrown;
        ++counts_thrown_;
        e << "The above exception was thrown while processing module "
          << brief_context(md_, p);
        if (auto art_ex = dynamic_cast<Exception*>(&e)) {
          cached_exception_ = make_exception_ptr(*art_ex);
        } else {
          cached_exception_ =
            make_exception_ptr(Exception{errors::OtherArt, string(), e});
        }
        waitingTasks_.doneWaiting(cached_exception_);
        TDEBUG_END_TASK_SI(4, sid) << "because of EXCEPTION";
        return;
      }
    }
    catch (bad_alloc const& bda) {
      state_ = ExceptionThrown;
      ++counts_thrown_;
      auto art_ex =
        Exception{errors::BadAlloc}
        << "A bad_alloc exception was thrown while processing module "
        << brief_context(md_, p) << '\n'
        << "The job has probably exhausted the virtual memory available to the "
           "process.\n";
      cached_exception_ = make_exception_ptr(art_ex);
      waitingTasks_.doneWaiting(cached_exception_);
      TDEBUG_END_TASK_SI(4, sid) << "because of EXCEPTION";
      return;
    }
    catch (exception const& e) {
      state_ = ExceptionThrown;
      ++counts_thrown_;
      auto art_ex = Exception{errors::StdException}
                    << "An exception was thrown while processing module "
                    << brief_context(md_, p) << '\n'
                    << e.what();
      cached_exception_ = make_exception_ptr(art_ex);
      waitingTasks_.doneWaiting(cached_exception_);
      TDEBUG_END_TASK_SI(4, sid) << "because of EXCEPTION";
      return;
    }
    catch (string const& s) {
      state_ = ExceptionThrown;
      ++counts_thrown_;
      auto art_ex =
        Exception{errors::BadExceptionType, "string"}
        << "A string was thrown as an exception while processing module "
        << brief_context(md_, p) << '\n'
        << s << '\n';
      cached_exception_ = make_exception_ptr(art_ex);
      waitingTasks_.doneWaiting(cached_exception_);
      TDEBUG_END_TASK_SI(4, sid) << "because of EXCEPTION";
      return;
    }
    catch (char const* c) {
      state_ = ExceptionThrown;
      ++counts_thrown_;
      auto art_ex =
        Exception{errors::BadExceptionType, "char const*"}
        << "A char const* was thrown as an exception while processing module "
        << brief_context(md_, p) << '\n'
        << c << '\n';
      cached_exception_ = make_exception_ptr(art_ex);
      waitingTasks_.doneWaiting(cached_exception_);
      TDEBUG_END_TASK_SI(4, sid) << "because of EXCEPTION";
      return;
    }
    catch (...) {
      ++counts_thrown_;
      state_ = ExceptionThrown;
      auto art_ex =
        Exception{errors::Unknown, "repeated"}
        << "An unknown exception was thrown while processing module "
        << brief_context(md_, p) << '\n';
      cached_exception_ = make_exception_ptr(art_ex);
      waitingTasks_.doneWaiting(cached_exception_);
      TDEBUG_END_TASK_SI(4, sid) << "because of EXCEPTION";
      return;
    }
    waitingTasks_.doneWaiting(exception_ptr{});
    TDEBUG_END_TASK_SI(4, sid);
  }

ScheduleID art::Worker::scheduleID ( ) const

inlineinherited

Definition at line 70 of file Worker.h.

    {
      return scheduleID_;
    }

void art::OutputWorker::selectProducts

(

ProductTables const &

tables

)

Definition at line 193 of file OutputWorker.cc.

References module_.

  {
    module_->selectProducts(tables);
  }

SerialTaskQueueChain * art::Worker::serialTaskQueueChain ( ) const

inherited

Definition at line 140 of file Worker.cc.

References art::Worker::doSerialTaskQueueChain().

Referenced by art::Worker::isUnique().

  {
    return doSerialTaskQueueChain();
  }

void art::OutputWorker::setFileStatus

(

OutputFileStatus

ofs

)

Definition at line 187 of file OutputWorker.cc.

References module_.

  {
    return module_->setFileStatus(ofs);
  }

void art::OutputWorker::setRunAuxiliaryRangeSetID

(

RangeSet const &

rs

)

Definition at line 169 of file OutputWorker.cc.

References module_.

  {
    module_->doSetRunAuxiliaryRangeSetID(rs);
  }

void art::OutputWorker::setSubRunAuxiliaryRangeSetID

(

RangeSet const &

rs

)

Definition at line 175 of file OutputWorker.cc.

References module_.

  {
    module_->doSetSubRunAuxiliaryRangeSetID(rs);
  }

size_t art::Worker::timesExcept ( ) const

inherited

Definition at line 190 of file Worker.cc.

References art::Worker::counts_thrown_.

  {
    return counts_thrown_.load();
  }

size_t art::Worker::timesFailed ( ) const

inherited

Definition at line 183 of file Worker.cc.

References art::Worker::counts_failed_.

  {
    return counts_failed_.load();
  }

size_t art::Worker::timesPassed ( ) const

inherited

Definition at line 176 of file Worker.cc.

References art::Worker::counts_passed_.

  {
    return counts_passed_.load();
  }

size_t art::Worker::timesRun ( ) const

inherited

Definition at line 169 of file Worker.cc.

References art::Worker::counts_run_.

  {
    return counts_run_.load();
  }

size_t art::Worker::timesVisited ( ) const

inherited

Definition at line 162 of file Worker.cc.

References art::Worker::counts_visited_.

  {
    return counts_visited_.load();
  }

void art::OutputWorker::writeEvent	(	EventPrincipal &	ep,
		PathContext const &	pc
	)

Definition at line 160 of file OutputWorker.cc.

References actReg_, art::Worker::description(), module_, art::ActivityRegistry::sPostWriteEvent, and art::ActivityRegistry::sPreWriteEvent.

  {
    ModuleContext const mc{pc, description()};
    actReg_.sPreWriteEvent.invoke(mc);
    module_->doWriteEvent(ep, mc);
    actReg_.sPostWriteEvent.invoke(mc);
  }

void art::OutputWorker::writeRun

(

RunPrincipal &

rp

)

Definition at line 148 of file OutputWorker.cc.

References module_.

  {
    module_->doWriteRun(rp);
  }

void art::OutputWorker::writeSubRun

(

SubRunPrincipal &

srp

)

Definition at line 154 of file OutputWorker.cc.

References module_.

  {
    module_->doWriteSubRun(srp);
  }

Member Data Documentation

ActivityRegistry const& art::OutputWorker::actReg_

private

Definition at line 65 of file OutputWorker.h.

Referenced by closeFile(), openFile(), OutputWorker(), and writeEvent().

ServiceHandle<CatalogInterface> art::OutputWorker::ci_ {}

private

Definition at line 64 of file OutputWorker.h.

Referenced by closeFile(), openFile(), and OutputWorker().

std::atomic<std::size_t> art::Worker::counts_failed_ {}

protectedinherited

Definition at line 101 of file Worker.h.

Referenced by art::WorkerT< T >::doProcess(), doProcess(), art::Worker::doWork_event(), art::Worker::runWorker(), and art::Worker::timesFailed().

std::atomic<std::size_t> art::Worker::counts_passed_ {}

protectedinherited

Definition at line 100 of file Worker.h.

Referenced by art::WorkerT< T >::doProcess(), doProcess(), art::Worker::doWork_event(), art::Worker::runWorker(), and art::Worker::timesPassed().

std::atomic<std::size_t> art::Worker::counts_run_ {}

protectedinherited

Definition at line 99 of file Worker.h.

Referenced by art::WorkerT< T >::doProcess(), doProcess(), and art::Worker::timesRun().

std::atomic<std::size_t> art::Worker::counts_thrown_ {}

protectedinherited

Definition at line 102 of file Worker.h.

Referenced by art::Worker::doWork_event(), art::Worker::runWorker(), and art::Worker::timesExcept().

std::atomic<std::size_t> art::Worker::counts_visited_ {}

protectedinherited

Definition at line 98 of file Worker.h.

Referenced by art::Worker::doWork_event(), art::Worker::isUnique(), and art::Worker::timesVisited().

Granularity art::OutputWorker::fileGranularity_ {Granularity::Unset}

private

Definition at line 66 of file OutputWorker.h.

cet::exempt_ptr<OutputModule> art::OutputWorker::module_

private

Definition at line 63 of file OutputWorker.h.

Referenced by closeFile(), doBegin(), doBeginJob(), doEnd(), doEndJob(), doProcess(), doRespondToCloseInputFile(), doRespondToCloseOutputFiles(), doRespondToOpenInputFile(), doRespondToOpenOutputFiles(), doSerialTaskQueueChain(), fileGranularity(), fileIsOpen(), incrementInputFileNumber(), lastClosedFileName(), openFile(), OutputWorker(), requestsToCloseFile(), selectProducts(), setFileStatus(), setRunAuxiliaryRangeSetID(), setSubRunAuxiliaryRangeSetID(), writeEvent(), writeRun(), and writeSubRun().

---

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

• art/Framework/Core/OutputWorker.h
• art/Framework/Core/OutputWorker.cc