art::EventProcessor Class Reference

#include "EventProcessor.h"

Public Types
enum	Status { epSuccess = 0, epSignal = 3 }
using	StatusCode = Status

Public Member Functions
	EventProcessor (EventProcessor const &)=delete
EventProcessor &	operator= (EventProcessor const &)=delete
	EventProcessor (fhicl::ParameterSet const &pset)
	~EventProcessor ()
StatusCode	runToCompletion ()

Private Member Functions
template<Level L>
bool	levelsToProcess ()
template<Level L>
std::enable_if_t< is_above_most_deeply_nested_level(L)>	begin ()
template<Level L>
void	process ()
template<Level L>
void	finalize ()
template<Level L>
void	finalizeContainingLevels ()
template<Level L>
void	recordOutputModuleClosureRequests ()
void	markLevelAsProcessed ()
Level	advanceItemType ()
void	beginJob ()
void	endJob ()
void	openInputFile ()
void	openSomeOutputFiles ()
void	openAllOutputFiles ()
void	closeInputFile ()
void	closeSomeOutputFiles ()
void	closeAllOutputFiles ()
void	closeAllFiles ()
void	respondToOpenInputFile ()
void	respondToCloseInputFile ()
void	respondToOpenOutputFiles ()
void	respondToCloseOutputFiles ()
void	readRun ()
void	beginRun ()
void	beginRunIfNotDoneAlready ()
void	setRunAuxiliaryRangeSetID ()
void	endRun ()
void	writeRun ()
void	readSubRun ()
void	beginSubRun ()
void	beginSubRunIfNotDoneAlready ()
void	setSubRunAuxiliaryRangeSetID ()
void	endSubRun ()
void	writeSubRun ()
void	readEvent ()
void	processEvent ()
void	writeEvent ()
bool	shouldWeStop () const
void	setOutputFileStatus (OutputFileStatus)
ServiceDirector	initServices_ (fhicl::ParameterSet const &top_pset, ActivityRegistry &areg, ServiceToken &token)
void	initSchedules_ (fhicl::ParameterSet const &pset)
void	invokePostBeginJobWorkers_ ()
template<typename T >
void	process_ (typename T::MyPrincipal &p)
void	servicesActivate_ (ServiceToken st)
void	servicesDeactivate_ ()
void	terminateAbnormally_ ()
template<>
void	process ()

Private Attributes
Level	nextLevel_ {Level::ReadyToAdvance}
std::vector< Level >	activeLevels_ {highest_level()}
detail::ExceptionCollector	ec_ {}
cet::cpu_timer	timer_ {}
bool	beginRunCalled_ {false}
bool	beginSubRunCalled_
bool	finalizeRunEnabled_ {true}
bool	finalizeSubRunEnabled_ {true}
ActionTable	act_table_
ActivityRegistry	actReg_
MFStatusUpdater	mfStatusUpdater_
MasterProductRegistry	preg_ {}
ProductDescriptions	productsToProduce_ {}
ProducingServiceSignals	psSignals_ {}
ServiceToken	serviceToken_ {ServiceToken::createInvalid()}
tbb::task_scheduler_init	tbbManager_ {tbb::task_scheduler_init::deferred}
std::unique_ptr< ServiceRegistry::Operate >	servicesSentry_ {}
PathManager	pathManager_
ServiceDirector	serviceDirector_
std::unique_ptr< InputSource >	input_ {nullptr}
std::unique_ptr< Schedule >	schedule_ {nullptr}
std::unique_ptr< EndPathExecutor >	endPathExecutor_ {nullptr}
std::unique_ptr< FileBlock >	fb_ {nullptr}
std::unique_ptr< RunPrincipal >	runPrincipal_ {nullptr}
std::unique_ptr< SubRunPrincipal >	subRunPrincipal_ {nullptr}
std::unique_ptr< EventPrincipal >	eventPrincipal_ {nullptr}
ProductTables	producedProducts_ {ProductTables::invalid()}
bool	shouldWeStop_ {false}
bool const	handleEmptyRuns_
bool const	handleEmptySubRuns_

Detailed Description

Definition at line 46 of file EventProcessor.h.

Member Typedef Documentation

using art::EventProcessor::StatusCode = Status

Definition at line 57 of file EventProcessor.h.

Member Enumeration Documentation

enum art::EventProcessor::Status

Enumerator
epSuccess
epSignal

Definition at line 52 of file EventProcessor.h.

{ epSuccess = 0, epSignal = 3 };

Constructor & Destructor Documentation

art::EventProcessor::EventProcessor ( EventProcessor const &  )

delete

art::EventProcessor::EventProcessor ( fhicl::ParameterSet const &  pset )

explicit

Definition at line 121 of file EventProcessor.cc.

References act_table_, actReg_, art::ProductMetaData::create_instance(), endPathExecutor_, FDEBUG, art::MasterProductRegistry::finalizeForProcessing(), art::ServiceToken::forceCreation(), fhicl::ParameterSet::get(), art::getReleaseVersion(), handleEmptyRuns_, handleEmptySubRuns_, art::ProcessConfiguration::id(), initSchedules_(), initServices_(), input_, mfStatusUpdater_, pathManager_, preg_, producedProducts_, productsToProduce_, psSignals_, art::ServiceToken::registerProducts(), serviceDirector_, servicesActivate_(), servicesDeactivate_(), and serviceToken_.

  : act_table_{pset.get<ParameterSet>("services.scheduler")}
  , actReg_()
  , mfStatusUpdater_{actReg_}
  , pathManager_{pset, preg_, productsToProduce_, act_table_, actReg_}
  , serviceDirector_{initServices_(pset, actReg_, serviceToken_)}
  , handleEmptyRuns_{pset.get<bool>("services.scheduler.handleEmptyRuns", true)}
  , handleEmptySubRuns_{
      pset.get<bool>("services.scheduler.handleEmptySubRuns", true)}
{
  servicesActivate_(serviceToken_);
  serviceToken_.forceCreation();

  std::string const& processName{pset.get<std::string>("process_name")};
  ProcessConfiguration const pc{processName, pset.id(), getReleaseVersion()};
  serviceToken_.registerProducts(preg_, productsToProduce_, psSignals_, pc);

  // Services
  // System service FileCatalogMetadata needs to know about the process name.
  ServiceHandle<art::FileCatalogMetadata> {}
  ->addMetadataString("process_name", processName);

  input_ = makeInput(pset, processName, preg_, actReg_);
  actReg_.sPostSourceConstruction.invoke(input_->moduleDescription());

  initSchedules_(pset);
  endPathExecutor_ =
    std::make_unique<EndPathExecutor>(pathManager_, act_table_, actReg_, preg_);
  preg_.finalizeForProcessing();
  ProductMetaData::create_instance(preg_);

  producedProducts_ = ProductTables{productsToProduce_};
  FDEBUG(2) << pset.to_string() << std::endl;
  servicesDeactivate_();
}

art::EventProcessor::~EventProcessor

(

)

Definition at line 157 of file EventProcessor.cc.

References servicesActivate_(), and serviceToken_.

{
  // Services must stay usable until they go out of scope, meaning
  // that modules may (say) use services in their destructors.
  servicesActivate_(serviceToken_);
}

Member Function Documentation

art::Level art::EventProcessor::advanceItemType ( )

private

Definition at line 490 of file EventProcessor.cc.

References art::Event, FDEBUG, art::highest_level(), input_, art::InputFile, art::input::IsEvent, art::input::IsFile, art::input::IsInvalid, art::input::IsRun, art::input::IsStop, art::input::IsSubRun, art::errors::LogicError, art::Run, and art::SubRun.

Referenced by levelsToProcess(), and recordOutputModuleClosureRequests().

{
  auto const itemType = input_->nextItemType();
  FDEBUG(1) << spaces(4) << "*** nextItemType: " << itemType << " ***\n";
  switch (itemType) {
    case input::IsStop:
      return highest_level();
    case input::IsFile:
      return Level::InputFile;
    case input::IsRun:
      return Level::Run;
    case input::IsSubRun:
      return Level::SubRun;
    case input::IsEvent:
      return Level::Event;
    case input::IsInvalid: {
      throw art::Exception{art::errors::LogicError}
        << "Invalid next item type presented to the event processor.\n"
        << "Please contact artists@fnal.gov.";
    }
  }
  throw art::Exception{art::errors::LogicError}
    << "Unrecognized next item type presented to the event processor.\n"
    << "Please contact artists@fnal.gov.";
}

template<Level L>

std::enable_if_t<is_above_most_deeply_nested_level(L)> art::EventProcessor::begin ( )

private

void art::EventProcessor::beginJob ( )

private

Definition at line 520 of file EventProcessor.cc.

References breakpoints::beginJob(), endPathExecutor_, FDEBUG, input_, invokePostBeginJobWorkers_(), schedule_, and serviceToken_.

Referenced by art::EventProcessor::begin< Level::Job >(), and recordOutputModuleClosureRequests().

{
  FDEBUG(1) << spaces(8) << "beginJob\n";
  breakpoints::beginJob();
  // make the services available
  ServiceRegistry::Operate op{serviceToken_};
  // NOTE: This implementation assumes 'Job' means one call the
  // EventProcessor::run. If it really means once per 'application'
  // then this code will have to be changed.  Also have to deal with
  // case where have 'run' then new Module added and do 'run' again.
  // In that case the newly added Module needs its 'beginJob' to be
  // called.
  try {
    input_->doBeginJob();
  }
  catch (cet::exception& e) {
    mf::LogError("BeginJob") << "A cet::exception happened while processing"
                                " the beginJob of the 'source'\n";
    e << "A cet::exception happened while processing"
         " the beginJob of the 'source'\n";
    throw;
  }
  catch (std::exception const&) {
    mf::LogError("BeginJob") << "A std::exception happened while processing"
                                " the beginJob of the 'source'\n";
    throw;
  }
  catch (...) {
    mf::LogError("BeginJob") << "An unknown exception happened while"
                                " processing the beginJob of the 'source'\n";
    throw;
  }
  schedule_->beginJob();
  endPathExecutor_->beginJob();
  actReg_.sPostBeginJob.invoke();

  invokePostBeginJobWorkers_();
}

void art::EventProcessor::beginRun ( )

private

Definition at line 727 of file EventProcessor.cc.

References beginRunCalled_, FDEBUG, finalizeRunEnabled_, and runPrincipal_.

Referenced by beginRunIfNotDoneAlready(), art::EventProcessor::begin< Level::Run >(), and recordOutputModuleClosureRequests().

{
  assert(runPrincipal_);
  RunID const r{runPrincipal_->id()};
  if (r.isFlush())
    return;

  finalizeRunEnabled_ = true;
  process_<Begin<Level::Run>>(*runPrincipal_);
  FDEBUG(1) << spaces(8) << "beginRun....................(" << r << ")\n";
  beginRunCalled_ = true;
}

void art::EventProcessor::beginRunIfNotDoneAlready ( )

private

Definition at line 741 of file EventProcessor.cc.

References beginRun(), and beginRunCalled_.

Referenced by art::EventProcessor::begin< Level::SubRun >(), process(), and recordOutputModuleClosureRequests().

{
  if (!beginRunCalled_) {
    beginRun();
  }
}

void art::EventProcessor::beginSubRun ( )

private

Definition at line 807 of file EventProcessor.cc.

References beginSubRunCalled_, FDEBUG, finalizeSubRunEnabled_, and subRunPrincipal_.

Referenced by beginSubRunIfNotDoneAlready(), art::EventProcessor::begin< Level::SubRun >(), and recordOutputModuleClosureRequests().

{
  assert(subRunPrincipal_);
  SubRunID const sr{subRunPrincipal_->id()};
  if (sr.isFlush())
    return;

  finalizeSubRunEnabled_ = true;
  process_<Begin<Level::SubRun>>(*subRunPrincipal_);
  FDEBUG(1) << spaces(8) << "beginSubRun.................(" << sr << ")\n";
  beginSubRunCalled_ = true;
}

void art::EventProcessor::beginSubRunIfNotDoneAlready ( )

private

Definition at line 821 of file EventProcessor.cc.

References beginSubRun(), and beginSubRunCalled_.

Referenced by process(), and recordOutputModuleClosureRequests().

{
  if (!beginSubRunCalled_) {
    beginSubRun();
  }
}

void art::EventProcessor::closeAllFiles ( )

private

Definition at line 596 of file EventProcessor.cc.

References closeAllOutputFiles(), and closeInputFile().

Referenced by art::EventProcessor::finalize< Level::InputFile >(), and recordOutputModuleClosureRequests().

{
  closeAllOutputFiles();
  closeInputFile();
}

void art::EventProcessor::closeAllOutputFiles ( )

private

Definition at line 630 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, and respondToCloseOutputFiles().

Referenced by closeAllFiles(), and recordOutputModuleClosureRequests().

{
  if (!endPathExecutor_->someOutputsOpen())
    return;

  respondToCloseOutputFiles();
  endPathExecutor_->closeAllOutputFiles();
  FDEBUG(1) << spaces(8) << "closeAllOutputFiles\n";
}

void art::EventProcessor::closeInputFile ( )

private

Definition at line 603 of file EventProcessor.cc.

References closeSomeOutputFiles(), endPathExecutor_, FDEBUG, input_, art::Granularity::InputFile, and respondToCloseInputFile().

Referenced by closeAllFiles(), art::EventProcessor::finalize< Level::InputFile >(), and recordOutputModuleClosureRequests().

{
  endPathExecutor_->incrementInputFileNumber();
  // Output-file closing on input-file boundaries are tricky since
  // input files must outlive the output files, which often have data
  // copied forward from the input files.  That's why the
  // recordOutputClosureRequests call is made here instead of in a
  // specialization of recordOutputModuleClosureRequests<>.
  endPathExecutor_->recordOutputClosureRequests(Granularity::InputFile);
  if (endPathExecutor_->outputsToClose()) {
    closeSomeOutputFiles();
  }
  respondToCloseInputFile();
  actReg_.sPreCloseFile.invoke();
  input_->closeFile();
  actReg_.sPostCloseFile.invoke();
  FDEBUG(1) << spaces(8) << "closeInputFile\n";
}

void art::EventProcessor::closeSomeOutputFiles ( )

private

Definition at line 659 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, and respondToCloseOutputFiles().

Referenced by closeInputFile(), levelsToProcess(), and recordOutputModuleClosureRequests().

{
  // Precondition: there are SOME output files that have been
  //               flagged as needing to close.  Otherwise,
  //               'respondtoCloseOutputFiles' will be needlessly
  //               called.
  assert(endPathExecutor_->outputsToClose());
  respondToCloseOutputFiles();
  endPathExecutor_->closeSomeOutputFiles();
  FDEBUG(1) << spaces(8) << "closeSomeOutputFiles\n";
}

void art::EventProcessor::endJob ( )

private

Definition at line 560 of file EventProcessor.cc.

References art::detail::ExceptionCollector::call(), ec_, endPathExecutor_, FDEBUG, input_, pathManager_, schedule_, serviceToken_, timer_, and art::detail::writeSummary().

Referenced by art::EventProcessor::finalize< Level::Job >(), and recordOutputModuleClosureRequests().

{
  FDEBUG(1) << spaces(8) << "endJob\n";
  // Make the services available
  ServiceRegistry::Operate op{serviceToken_};
  ec_.call([this] { schedule_->endJob(); });
  ec_.call([this] { endPathExecutor_->endJob(); });
  ec_.call([this] { input_->doEndJob(); });
  ec_.call([this] { actReg_.sPostEndJob.invoke(); });
  ec_.call([this] {
    detail::writeSummary(
      pathManager_,
      ServiceHandle<TriggerNamesService const> {}->wantSummary(),
      timer_);
  });
}

void art::EventProcessor::endRun ( )

private

Definition at line 758 of file EventProcessor.cc.

References beginRunCalled_, FDEBUG, and runPrincipal_.

Referenced by art::EventProcessor::finalize< Level::Run >(), and recordOutputModuleClosureRequests().

{
  assert(runPrincipal_);
  // Precondition: The RunID does not correspond to a flush ID. --
  // N.B. The flush flag is not explicitly checked here since endRun
  // is only called from finalizeRun, which is where the check
  // happens.
  RunID const run{runPrincipal_->id()};
  assert(!run.isFlush());
  process_<End<Level::Run>>(*runPrincipal_);
  FDEBUG(1) << spaces(8) << "endRun......................(" << run << ")\n";
  beginRunCalled_ = false;
}

void art::EventProcessor::endSubRun ( )

private

Definition at line 838 of file EventProcessor.cc.

References beginSubRunCalled_, FDEBUG, and subRunPrincipal_.

Referenced by art::EventProcessor::finalize< Level::SubRun >(), and recordOutputModuleClosureRequests().

{
  assert(subRunPrincipal_);
  // Precondition: The SubRunID does not correspond to a flush ID.
  // Note: the flush flag is not explicitly checked here since
  // endSubRun is only called from finalizeSubRun, which is where the
  // check happens.
  SubRunID const sr{subRunPrincipal_->id()};
  assert(!sr.isFlush());
  process_<End<Level::SubRun>>(*subRunPrincipal_);
  FDEBUG(1) << spaces(8) << "endSubRun...................(" << sr << ")\n";
  beginSubRunCalled_ = false;
}

template<Level L>

void art::EventProcessor::finalize ( )

private

template<Level L>

void art::EventProcessor::finalizeContainingLevels ( )

inlineprivate

Definition at line 92 of file EventProcessor.h.

  {}

void art::EventProcessor::initSchedules_ ( fhicl::ParameterSet const &  pset )

private

Definition at line 193 of file EventProcessor.cc.

References act_table_, actReg_, art::ScheduleID::first(), fhicl::ParameterSet::get(), pathManager_, preg_, productsToProduce_, schedule_, and tbbManager_.

Referenced by EventProcessor().

{
  // Initialize TBB with desired number of threads.
  auto const num_threads = pset.get<int>("services.scheduler.num_threads");
  tbbManager_.initialize(num_threads);
  mf::LogInfo("MTdiagnostics")
    << "TBB has been configured to use a maximum of "
    << tbb::this_task_arena::max_concurrency() << " threads.";
  schedule_ =
    std::make_unique<Schedule>(ScheduleID::first(),
                               pathManager_,
                               pset,
                               *ServiceHandle<TriggerNamesService const>{},
                               preg_,
                               productsToProduce_,
                               act_table_,
                               actReg_);
}

art::ServiceDirector art::EventProcessor::initServices_ ( fhicl::ParameterSet const &  top_pset, ActivityRegistry &  areg, ServiceToken &  token  )

private

Definition at line 165 of file EventProcessor.cc.

References fhicl::ParameterSet::erase(), fhicl::ParameterSet::get(), pathManager_, and art::PathManager::triggerPathNames().

Referenced by EventProcessor().

{
  auto services = top_pset.get<ParameterSet>("services", {});

  // Save and non-standard service configs, "floating_point_control"
  // to prevent ServiceDirector trying to make one itself.
  auto const fpc_pset =
    services.get<ParameterSet>("floating_point_control", {});
  services.erase("floating_point_control");

  // Remove non-standard non-service config, "message."
  services.erase("message");

  // Create the service director and all user-configured services.
  ServiceDirector director{std::move(services), areg, token};

  // Services requiring special construction.
  director.addSystemService<CurrentModule>(areg);
  director.addSystemService<TriggerNamesService>(
    top_pset, pathManager_.triggerPathNames());
  director.addSystemService<FloatingPointControl>(fpc_pset, areg);
  director.addSystemService<ScheduleContext>();
  return director;
}

void art::EventProcessor::invokePostBeginJobWorkers_ ( )

private

Definition at line 213 of file EventProcessor.cc.

References art::PathManager::endPathInfo(), art::ScheduleID::first(), input_, pathManager_, art::PathManager::triggerPathsInfo(), and art::PathsInfo::workers().

Referenced by beginJob().

{
  // Need to convert multiple lists of workers into a long list that
  // the postBeginJobWorkers callbacks can understand.
  std::vector<Worker*> allWorkers;
  allWorkers.reserve(
    pathManager_.triggerPathsInfo(ScheduleID::first()).workers().size() +
    pathManager_.endPathInfo().workers().size());
  auto workerStripper = [&allWorkers](WorkerMap::value_type const& val) {
    allWorkers.emplace_back(val.second.get());
  };
  cet::for_all(pathManager_.triggerPathsInfo(ScheduleID::first()).workers(),
               workerStripper);
  cet::for_all(pathManager_.endPathInfo().workers(), workerStripper);
  actReg_.sPostBeginJobWorkers.invoke(input_.get(), allWorkers);
}

template<art::Level L>

bool art::EventProcessor::levelsToProcess ( )

private

Definition at line 235 of file EventProcessor.cc.

References activeLevels_, advanceItemType(), closeSomeOutputFiles(), endPathExecutor_, art::highest_level(), art::errors::LogicError, nextLevel_, art::ReadyToAdvance, setOutputFileStatus(), and art::Switching.

{
  if (nextLevel_ == Level::ReadyToAdvance) {
    nextLevel_ = advanceItemType();
    // Consider reading right here?
  }

  if (nextLevel_ == L) {
    activeLevels_.push_back(nextLevel_);
    nextLevel_ = Level::ReadyToAdvance;
    if (endPathExecutor_->outputsToClose()) {
      setOutputFileStatus(OutputFileStatus::Switching);
      finalizeContainingLevels<L>();
      closeSomeOutputFiles();
    }
    return true;
  } else if (nextLevel_ < L) {
    return false;
  } else if (nextLevel_ == highest_level()) {
    return false;
  }

  throw Exception{errors::LogicError} << "Incorrect level hierarchy.";
}

void art::EventProcessor::markLevelAsProcessed ( )

private

Definition at line 483 of file EventProcessor.cc.

References activeLevels_.

Referenced by process(), and recordOutputModuleClosureRequests().

{
  assert(!activeLevels_.empty());
  activeLevels_.pop_back();
}

void art::EventProcessor::openAllOutputFiles ( )

private

Definition at line 623 of file EventProcessor.cc.

References endPathExecutor_, fb_, and FDEBUG.

Referenced by recordOutputModuleClosureRequests().

{
  endPathExecutor_->openAllOutputFiles(*fb_);
  FDEBUG(1) << spaces(8) << "openAllOutputFiles\n";
}

void art::EventProcessor::openInputFile ( )

private

Definition at line 581 of file EventProcessor.cc.

References fb_, FDEBUG, input_, art::errors::LogicError, and respondToOpenInputFile().

Referenced by art::EventProcessor::begin< Level::InputFile >(), and recordOutputModuleClosureRequests().

{
  actReg_.sPreOpenFile.invoke();
  FDEBUG(1) << spaces(8) << "openInputFile\n";
  fb_ = input_->readFile();
  if (!fb_) {
    throw Exception(errors::LogicError)
      << "Source readFile() did not return a valid FileBlock: FileBlock "
      << "should be valid or readFile() should throw.\n";
  }
  actReg_.sPostOpenFile.invoke(fb_->fileName());
  respondToOpenInputFile();
}

void art::EventProcessor::openSomeOutputFiles ( )

private

Definition at line 641 of file EventProcessor.cc.

References endPathExecutor_, fb_, FDEBUG, and respondToOpenOutputFiles().

Referenced by art::EventProcessor::finalize< Level::Event >(), art::EventProcessor::finalize< Level::Run >(), art::EventProcessor::finalize< Level::SubRun >(), and recordOutputModuleClosureRequests().

{
  if (!endPathExecutor_->outputsToOpen())
    return;

  endPathExecutor_->openSomeOutputFiles(*fb_);
  FDEBUG(1) << spaces(8) << "openSomeOutputFiles\n";
  respondToOpenOutputFiles();
}

EventProcessor& art::EventProcessor::operator= ( EventProcessor const &  )

delete

template<art::Level L>

void art::EventProcessor::process ( )

private

Definition at line 439 of file EventProcessor.cc.

References art::detail::ExceptionCollector::call(), ec_, art::detail::ExceptionCollector::empty(), markLevelAsProcessed(), and art::shutdown_flag.

{
  if (shutdown_flag > 0 || !ec_.empty()) {
    return;
  }

  ec_.call([this] { begin<L>(); });

  while (shutdown_flag == 0 && ec_.empty() &&
         levelsToProcess<level_down(L)>()) {
    ec_.call([this] {
      process<level_down(L)>();
      markLevelAsProcessed();
    });
  }
  ec_.call([this] {
    finalize<L>();
    recordOutputModuleClosureRequests<L>();
  });
}

template<>

void art::EventProcessor::process ( )

private

Definition at line 413 of file EventProcessor.cc.

References beginRunIfNotDoneAlready(), beginSubRunIfNotDoneAlready(), ec_, art::detail::ExceptionCollector::empty(), eventPrincipal_, art::highest_level(), nextLevel_, processEvent(), readEvent(), shouldWeStop(), and art::shutdown_flag.

  {
    if (shutdown_flag > 0 || !ec_.empty()) {
      return;
    }

    beginRunIfNotDoneAlready();
    beginSubRunIfNotDoneAlready();
    readEvent();

    assert(eventPrincipal_);
    if (eventPrincipal_->id().isFlush())
      return;
    processEvent();

    if (shouldWeStop()) {
      nextLevel_ = highest_level(); // FIXME: maybe go somewhere else?
    }
    finalize<most_deeply_nested_level()>();
    recordOutputModuleClosureRequests<most_deeply_nested_level()>();
  }

template<typename T >

void art::EventProcessor::process_ ( typename T::MyPrincipal &  p )

private

Definition at line 203 of file EventProcessor.h.

References act_table_, actReg_, endPathExecutor_, art::Event, art::errors::EventProcessorFailure, art::ActionTable::find(), art::actions::IgnoreCompletely, art::actions::Rethrow, and schedule_.

                                                        {
  T::preScheduleSignal(actReg_, p);
  schedule_->process<T>(p);
  endPathExecutor_->process<T>(p);
  T::postScheduleSignal(actReg_, p);
}
catch (cet::exception const& ex) {
  actions::ActionCodes const action{T::level == Level::Event ?
                                      act_table_.find(ex.root_cause()) :
                                      actions::Rethrow};
  switch (action) {
    case actions::IgnoreCompletely: {
      mf::LogWarning(ex.category())
        << "exception being ignored for current event:\n"
        << cet::trim_right_copy(ex.what(), " \n");
      break;
    }
    default: {
      throw art::Exception{
        errors::EventProcessorFailure,
        "EventProcessor: an exception occurred during current event processing",
        ex};
    }
  }
}
catch (...) {
  mf::LogError("PassingThrough")
    << "an exception occurred during current event processing\n";
  throw;
}

void art::EventProcessor::processEvent ( )

private

Definition at line 886 of file EventProcessor.cc.

References eventPrincipal_, and FDEBUG.

Referenced by process(), and recordOutputModuleClosureRequests().

{
  assert(eventPrincipal_);
  EventID const& id{eventPrincipal_->id()};
  // Precondition: The EventID does not correspond to a flush ID.
  assert(!id.isFlush());
  process_<Do<Level::Event>>(*eventPrincipal_);
  FDEBUG(1) << spaces(8) << "processEvent................(" << id << ")\n";
}

void art::EventProcessor::readEvent ( )

private

Definition at line 867 of file EventProcessor.cc.

References eventPrincipal_, FDEBUG, art::ProductTables::get(), art::InEvent, input_, art::Consumer::non_module_context(), producedProducts_, psSignals_, art::ProducingServiceSignals::sPostReadEvent, and subRunPrincipal_.

Referenced by process(), and recordOutputModuleClosureRequests().

{
  assert(subRunPrincipal_);
  assert(subRunPrincipal_->id().isValid());
  {
    actReg_.sPreSourceEvent.invoke();
    eventPrincipal_ = input_->readEvent(subRunPrincipal_.get());
    assert(eventPrincipal_);
    psSignals_.sPostReadEvent.invoke(*eventPrincipal_);
  }
  eventPrincipal_->setProducedProducts(producedProducts_.get(InEvent));
  Event const e{
    *eventPrincipal_, ModuleDescription{}, Consumer::non_module_context()};
  actReg_.sPostSourceEvent.invoke(e);
  FDEBUG(1) << spaces(8) << "readEvent...................("
            << eventPrincipal_->id() << ")\n";
}

void art::EventProcessor::readRun ( )

private

Definition at line 707 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, art::ProductTables::get(), input_, art::InRun, art::Consumer::non_module_context(), producedProducts_, psSignals_, runPrincipal_, and art::ProducingServiceSignals::sPostReadRun.

Referenced by art::EventProcessor::begin< Level::Run >(), and recordOutputModuleClosureRequests().

{
  {
    actReg_.sPreSourceRun.invoke();
    runPrincipal_ = input_->readRun();
    // Seeding the RangeSet is necessary here in case
    // 'sPostReadRun.invoke()' throws.
    endPathExecutor_->seedRunRangeSet(input_->runRangeSetHandler());
    assert(runPrincipal_);
    psSignals_.sPostReadRun.invoke(*runPrincipal_);
  }
  runPrincipal_->setProducedProducts(producedProducts_.get(InRun));
  Run const r{
    *runPrincipal_, ModuleDescription{}, Consumer::non_module_context()};
  actReg_.sPostSourceRun.invoke(r);
  FDEBUG(1) << spaces(8) << "readRun.....................("
            << runPrincipal_->id() << ")\n";
}

void art::EventProcessor::readSubRun ( )

private

Definition at line 787 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, art::ProductTables::get(), input_, art::InSubRun, art::Consumer::non_module_context(), producedProducts_, psSignals_, runPrincipal_, art::ProducingServiceSignals::sPostReadSubRun, and subRunPrincipal_.

Referenced by art::EventProcessor::begin< Level::SubRun >(), and recordOutputModuleClosureRequests().

{
  {
    actReg_.sPreSourceSubRun.invoke();
    subRunPrincipal_ = input_->readSubRun(runPrincipal_.get());
    // Seeding the RangeSet is necessary here in case
    // 'sPostSubRun.invoke()' throws.
    endPathExecutor_->seedSubRunRangeSet(input_->subRunRangeSetHandler());
    assert(subRunPrincipal_);
    psSignals_.sPostReadSubRun.invoke(*subRunPrincipal_);
  }
  subRunPrincipal_->setProducedProducts(producedProducts_.get(InSubRun));
  SubRun const sr{
    *subRunPrincipal_, ModuleDescription{}, Consumer::non_module_context()};
  actReg_.sPostSourceSubRun.invoke(sr);
  FDEBUG(1) << spaces(8) << "readSubRun..................("
            << subRunPrincipal_->id() << ")\n";
}

template<Level L>

void art::EventProcessor::recordOutputModuleClosureRequests ( )

inlineprivate

Definition at line 96 of file EventProcessor.h.

References advanceItemType(), beginJob(), beginRun(), beginRunIfNotDoneAlready(), beginSubRun(), beginSubRunIfNotDoneAlready(), closeAllFiles(), closeAllOutputFiles(), closeInputFile(), closeSomeOutputFiles(), endJob(), endRun(), endSubRun(), markLevelAsProcessed(), openAllOutputFiles(), openInputFile(), openSomeOutputFiles(), processEvent(), readEvent(), readRun(), readSubRun(), respondToCloseInputFile(), respondToCloseOutputFiles(), respondToOpenInputFile(), respondToOpenOutputFiles(), setOutputFileStatus(), setRunAuxiliaryRangeSetID(), setSubRunAuxiliaryRangeSetID(), shouldWeStop(), writeEvent(), writeRun(), and writeSubRun().

  {}

void art::EventProcessor::respondToCloseInputFile ( )

private

Definition at line 680 of file EventProcessor.cc.

References endPathExecutor_, fb_, FDEBUG, and schedule_.

Referenced by closeInputFile(), and recordOutputModuleClosureRequests().

{
  schedule_->respondToCloseInputFile(*fb_);
  endPathExecutor_->respondToCloseInputFile(*fb_);
  FDEBUG(1) << spaces(8) << "respondToCloseInputFile\n";
}

void art::EventProcessor::respondToCloseOutputFiles ( )

private

Definition at line 696 of file EventProcessor.cc.

References endPathExecutor_, fb_, FDEBUG, and schedule_.

Referenced by closeAllOutputFiles(), closeSomeOutputFiles(), and recordOutputModuleClosureRequests().

{
  schedule_->respondToCloseOutputFiles(*fb_);
  endPathExecutor_->respondToCloseOutputFiles(*fb_);
  FDEBUG(1) << spaces(8) << "respondToCloseOutputFiles\n";
}

void art::EventProcessor::respondToOpenInputFile ( )

private

Definition at line 672 of file EventProcessor.cc.

References endPathExecutor_, fb_, FDEBUG, and schedule_.

Referenced by openInputFile(), and recordOutputModuleClosureRequests().

{
  schedule_->respondToOpenInputFile(*fb_);
  endPathExecutor_->respondToOpenInputFile(*fb_);
  FDEBUG(1) << spaces(8) << "respondToOpenInputFile\n";
}

void art::EventProcessor::respondToOpenOutputFiles ( )

private

Definition at line 688 of file EventProcessor.cc.

References endPathExecutor_, fb_, FDEBUG, and schedule_.

Referenced by openSomeOutputFiles(), and recordOutputModuleClosureRequests().

{
  schedule_->respondToOpenOutputFiles(*fb_);
  endPathExecutor_->respondToOpenOutputFiles(*fb_);
  FDEBUG(1) << spaces(8) << "respondToOpenOutputFiles\n";
}

art::EventProcessor::StatusCode art::EventProcessor::runToCompletion

(

)

Definition at line 461 of file EventProcessor.cc.

References art::detail::ExceptionCollector::call(), ec_, art::detail::ExceptionCollector::empty(), epSignal, epSuccess, art::detail::ExceptionCollector::rethrow(), serviceToken_, art::shutdown_flag, and terminateAbnormally_().

{
  StatusCode returnCode{epSuccess};
  // Make the services available
  ServiceRegistry::Operate op{serviceToken_};

  ec_.call([this, &returnCode] {
    process<highest_level()>();
    if (art::shutdown_flag > 0) {
      returnCode = epSignal;
    }
  });

  if (!ec_.empty()) {
    terminateAbnormally_();
    ec_.rethrow();
  }

  return returnCode;
}

void art::EventProcessor::servicesActivate_ ( ServiceToken  st )

private

Definition at line 919 of file EventProcessor.cc.

References servicesSentry_.

Referenced by EventProcessor(), and ~EventProcessor().

{
  servicesSentry_ = std::make_unique<ServiceRegistry::Operate>(st);
}

void art::EventProcessor::servicesDeactivate_ ( )

private

Definition at line 925 of file EventProcessor.cc.

References servicesSentry_.

Referenced by EventProcessor().

{
  servicesSentry_.reset();
}

void art::EventProcessor::setOutputFileStatus ( OutputFileStatus  ofs )

private

Definition at line 652 of file EventProcessor.cc.

References endPathExecutor_, and FDEBUG.

Referenced by levelsToProcess(), and recordOutputModuleClosureRequests().

{
  endPathExecutor_->setOutputFileStatus(ofs);
  FDEBUG(1) << spaces(8) << "setOutputFileStatus\n";
}

void art::EventProcessor::setRunAuxiliaryRangeSetID ( )

private

Definition at line 749 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, and runPrincipal_.

Referenced by art::EventProcessor::finalize< Level::Run >(), and recordOutputModuleClosureRequests().

{
  assert(runPrincipal_);
  endPathExecutor_->setAuxiliaryRangeSetID(*runPrincipal_);
  FDEBUG(1) << spaces(8) << "setRunAuxiliaryRangeSetID...("
            << runPrincipal_->id() << ")\n";
}

void art::EventProcessor::setSubRunAuxiliaryRangeSetID ( )

private

Definition at line 829 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, and subRunPrincipal_.

Referenced by art::EventProcessor::finalize< Level::SubRun >(), and recordOutputModuleClosureRequests().

{
  assert(subRunPrincipal_);
  endPathExecutor_->setAuxiliaryRangeSetID(*subRunPrincipal_);
  FDEBUG(1) << spaces(8) << "setSubRunAuxiliaryRangeSetID("
            << subRunPrincipal_->id() << ")\n";
}

bool art::EventProcessor::shouldWeStop ( ) const

private

Definition at line 909 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, and shouldWeStop_.

Referenced by process(), and recordOutputModuleClosureRequests().

{
  FDEBUG(1) << spaces(8) << "shouldWeStop\n";
  if (shouldWeStop_) {
    return true;
  }
  return endPathExecutor_->terminate();
}

void art::EventProcessor::terminateAbnormally_ ( )

private

Definition at line 931 of file EventProcessor.cc.

Referenced by runToCompletion().

                                            {
  if (ServiceRegistry::isAvailable<RandomNumberGenerator>()) {
    ServiceHandle<RandomNumberGenerator> {}
    ->saveToFile_();
  }
}
catch (...) {
}

void art::EventProcessor::writeEvent ( )

private

Definition at line 897 of file EventProcessor.cc.

References endPathExecutor_, eventPrincipal_, and FDEBUG.

Referenced by art::EventProcessor::finalize< Level::Event >(), and recordOutputModuleClosureRequests().

{
  assert(eventPrincipal_);
  EventID const& id{eventPrincipal_->id()};
  // Precondition: The EventID does not correspond to a flush ID.
  assert(!id.isFlush());
  endPathExecutor_->writeEvent(*eventPrincipal_);
  FDEBUG(1) << spaces(8) << "writeEvent..................(" << id << ")\n";
  eventPrincipal_.reset();
}

void art::EventProcessor::writeRun ( )

private

Definition at line 773 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, and runPrincipal_.

Referenced by art::EventProcessor::finalize< Level::Run >(), and recordOutputModuleClosureRequests().

{
  assert(runPrincipal_);
  // Precondition: The RunID does not correspond to a flush ID.
  RunID const r{runPrincipal_->id()};
  assert(!r.isFlush());
  endPathExecutor_->writeRun(*runPrincipal_);
  FDEBUG(1) << spaces(8) << "writeRun....................(" << r << ")\n";
}

void art::EventProcessor::writeSubRun ( )

private

Definition at line 853 of file EventProcessor.cc.

References endPathExecutor_, FDEBUG, and subRunPrincipal_.

Referenced by art::EventProcessor::finalize< Level::SubRun >(), and recordOutputModuleClosureRequests().

{
  assert(subRunPrincipal_);
  // Precondition: The SubRunID does not correspond to a flush ID.
  SubRunID const& sr{subRunPrincipal_->id()};
  assert(!sr.isFlush());
  endPathExecutor_->writeSubRun(*subRunPrincipal_);
  FDEBUG(1) << spaces(8) << "writeSubRun.................(" << sr << ")\n";
}

Member Data Documentation

ActionTable art::EventProcessor::act_table_

private

Definition at line 174 of file EventProcessor.h.

Referenced by EventProcessor(), initSchedules_(), and process_().

std::vector<Level> art::EventProcessor::activeLevels_ {highest_level()}

private

Definition at line 143 of file EventProcessor.h.

Referenced by levelsToProcess(), and markLevelAsProcessed().

ActivityRegistry art::EventProcessor::actReg_

private

Definition at line 175 of file EventProcessor.h.

Referenced by EventProcessor(), initSchedules_(), and process_().

bool art::EventProcessor::beginRunCalled_ {false}

private

Definition at line 147 of file EventProcessor.h.

Referenced by beginRun(), beginRunIfNotDoneAlready(), endRun(), and art::EventProcessor::finalize< Level::Run >().

bool art::EventProcessor::beginSubRunCalled_

private

Initial value:

{
    false}

Definition at line 148 of file EventProcessor.h.

Referenced by beginSubRun(), beginSubRunIfNotDoneAlready(), endSubRun(), and art::EventProcessor::finalize< Level::SubRun >().

detail::ExceptionCollector art::EventProcessor::ec_ {}

private

Definition at line 144 of file EventProcessor.h.

Referenced by endJob(), process(), and runToCompletion().

std::unique_ptr<EndPathExecutor> art::EventProcessor::endPathExecutor_ {nullptr}

private

Definition at line 187 of file EventProcessor.h.

Referenced by beginJob(), closeAllOutputFiles(), closeInputFile(), closeSomeOutputFiles(), endJob(), EventProcessor(), art::EventProcessor::recordOutputModuleClosureRequests< Level::Event >(), art::EventProcessor::recordOutputModuleClosureRequests< Level::Run >(), art::EventProcessor::recordOutputModuleClosureRequests< Level::SubRun >(), levelsToProcess(), openAllOutputFiles(), openSomeOutputFiles(), process_(), readRun(), readSubRun(), respondToCloseInputFile(), respondToCloseOutputFiles(), respondToOpenInputFile(), respondToOpenOutputFiles(), setOutputFileStatus(), setRunAuxiliaryRangeSetID(), setSubRunAuxiliaryRangeSetID(), shouldWeStop(), writeEvent(), writeRun(), and writeSubRun().

std::unique_ptr<EventPrincipal> art::EventProcessor::eventPrincipal_ {nullptr}

private

Definition at line 192 of file EventProcessor.h.

Referenced by art::EventProcessor::finalize< Level::Event >(), process(), processEvent(), readEvent(), and writeEvent().

std::unique_ptr<FileBlock> art::EventProcessor::fb_ {nullptr}

private

Definition at line 188 of file EventProcessor.h.

Referenced by openAllOutputFiles(), openInputFile(), openSomeOutputFiles(), respondToCloseInputFile(), respondToCloseOutputFiles(), respondToOpenInputFile(), and respondToOpenOutputFiles().

bool art::EventProcessor::finalizeRunEnabled_ {true}

private

Definition at line 151 of file EventProcessor.h.

Referenced by beginRun(), art::EventProcessor::begin< Level::Run >(), and art::EventProcessor::finalize< Level::Run >().

bool art::EventProcessor::finalizeSubRunEnabled_ {true}

private

Definition at line 152 of file EventProcessor.h.

Referenced by beginSubRun(), art::EventProcessor::begin< Level::SubRun >(), and art::EventProcessor::finalize< Level::SubRun >().

bool const art::EventProcessor::handleEmptyRuns_

private

Definition at line 195 of file EventProcessor.h.

Referenced by EventProcessor(), and art::EventProcessor::begin< Level::Run >().

bool const art::EventProcessor::handleEmptySubRuns_

private

Definition at line 196 of file EventProcessor.h.

Referenced by EventProcessor(), and art::EventProcessor::begin< Level::SubRun >().

std::unique_ptr<InputSource> art::EventProcessor::input_ {nullptr}

private

Definition at line 185 of file EventProcessor.h.

Referenced by advanceItemType(), beginJob(), closeInputFile(), endJob(), EventProcessor(), invokePostBeginJobWorkers_(), openInputFile(), readEvent(), readRun(), and readSubRun().

MFStatusUpdater art::EventProcessor::mfStatusUpdater_

private

Definition at line 176 of file EventProcessor.h.

Referenced by EventProcessor().

Level art::EventProcessor::nextLevel_ {Level::ReadyToAdvance}

private

Definition at line 142 of file EventProcessor.h.

Referenced by art::EventProcessor::finalize< Level::InputFile >(), levelsToProcess(), and process().

PathManager art::EventProcessor::pathManager_

private

Definition at line 183 of file EventProcessor.h.

Referenced by endJob(), EventProcessor(), initSchedules_(), initServices_(), and invokePostBeginJobWorkers_().

MasterProductRegistry art::EventProcessor::preg_ {}

private

Definition at line 177 of file EventProcessor.h.

Referenced by EventProcessor(), and initSchedules_().

ProductTables art::EventProcessor::producedProducts_ {ProductTables::invalid()}

private

Definition at line 193 of file EventProcessor.h.

Referenced by EventProcessor(), readEvent(), readRun(), and readSubRun().

ProductDescriptions art::EventProcessor::productsToProduce_ {}

private

Definition at line 178 of file EventProcessor.h.

Referenced by EventProcessor(), and initSchedules_().

ProducingServiceSignals art::EventProcessor::psSignals_ {}

private

Definition at line 179 of file EventProcessor.h.

Referenced by EventProcessor(), readEvent(), readRun(), and readSubRun().

std::unique_ptr<RunPrincipal> art::EventProcessor::runPrincipal_ {nullptr}

private

Definition at line 190 of file EventProcessor.h.

Referenced by beginRun(), endRun(), art::EventProcessor::begin< Level::SubRun >(), art::EventProcessor::finalize< Level::Run >(), readRun(), readSubRun(), setRunAuxiliaryRangeSetID(), and writeRun().

std::unique_ptr<Schedule> art::EventProcessor::schedule_ {nullptr}

private

Definition at line 186 of file EventProcessor.h.

Referenced by beginJob(), endJob(), initSchedules_(), process_(), respondToCloseInputFile(), respondToCloseOutputFiles(), respondToOpenInputFile(), and respondToOpenOutputFiles().

ServiceDirector art::EventProcessor::serviceDirector_

private

Definition at line 184 of file EventProcessor.h.

Referenced by EventProcessor().

std::unique_ptr<ServiceRegistry::Operate> art::EventProcessor::servicesSentry_ {}

private

Definition at line 182 of file EventProcessor.h.

Referenced by servicesActivate_(), and servicesDeactivate_().

ServiceToken art::EventProcessor::serviceToken_ {ServiceToken::createInvalid()}

private

Definition at line 180 of file EventProcessor.h.

Referenced by beginJob(), endJob(), EventProcessor(), runToCompletion(), and ~EventProcessor().

bool art::EventProcessor::shouldWeStop_ {false}

private

Definition at line 194 of file EventProcessor.h.

Referenced by shouldWeStop().

std::unique_ptr<SubRunPrincipal> art::EventProcessor::subRunPrincipal_ {nullptr}

private

Definition at line 191 of file EventProcessor.h.

Referenced by beginSubRun(), endSubRun(), art::EventProcessor::finalize< Level::SubRun >(), readEvent(), readSubRun(), setSubRunAuxiliaryRangeSetID(), and writeSubRun().

tbb::task_scheduler_init art::EventProcessor::tbbManager_ {tbb::task_scheduler_init::deferred}

private

Definition at line 181 of file EventProcessor.h.

Referenced by initSchedules_().

cet::cpu_timer art::EventProcessor::timer_ {}

private

Definition at line 145 of file EventProcessor.h.

Referenced by endJob(), art::EventProcessor::begin< Level::Job >(), and art::EventProcessor::finalize< Level::Job >().

---

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

• art/Framework/EventProcessor/EventProcessor.h
• art/Framework/EventProcessor/EventProcessor.cc