art::EventProcessor Class Reference

#include "EventProcessor.h"

Classes
class	EndPathRunnerTask
class	EndPathTask

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

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

Private Types
template<typename T >
using	tsan = hep::concurrency::thread_sanitize< T >
template<typename T >
using	tsan_unique_ptr = hep::concurrency::thread_sanitize_unique_ptr< T >

Private Member Functions
void	processAllEventsAsync (ScheduleID sid)
void	readAndProcessAsync (ScheduleID sid)
void	processEventAsync (ScheduleID sid)
void	finishEventAsync (ScheduleID sid)
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 ()
Level	advanceItemType ()
void	beginJob ()
void	endJob ()
void	endJobAllSchedules ()
void	openInputFile ()
bool	outputsToOpen ()
void	openSomeOutputFiles ()
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 ()
void	setOutputFileStatus (OutputFileStatus)
void	invokePostBeginJobWorkers_ ()
void	terminateAbnormally_ ()
Schedule &	schedule (ScheduleID const id)
Schedule &	main_schedule ()
actions::ActionCodes	error_action (cet::exception &e) const
template<>
void	process ()

Private Attributes
std::atomic< Level >	nextLevel_ {Level::ReadyToAdvance}
tsan< detail::ExceptionCollector >	ec_ {}
tsan< cet::cpu_timer >	timer_ {}
std::atomic< bool >	beginRunCalled_ {false}
std::atomic< bool >	beginSubRunCalled_ {false}
std::atomic< bool >	finalizeRunEnabled_ {false}
std::atomic< bool >	finalizeSubRunEnabled_ {false}
ActivityRegistry	actReg_ {}
tsan< MFStatusUpdater >	mfStatusUpdater_ {actReg_}
tsan< UpdateOutputCallbacks >	outputCallbacks_ {}
tsan< ProductDescriptions >	producedProductDescriptions_ {}
tsan< ProductTables >	producedProductLookupTables_ {ProductTables::invalid()}
tsan< ProducingServiceSignals >	psSignals_ {}
tsan< Scheduler >	scheduler_
std::unique_ptr< GlobalTaskGroup >	taskGroup_ {nullptr}
detail::SharedResources	sharedResources_ {}
ScheduleIteration	scheduleIteration_
tsan_unique_ptr< ServicesManager >	servicesManager_
tsan< PathManager >	pathManager_
tsan_unique_ptr< InputSource >	input_ {nullptr}
tsan< std::map< ScheduleID, Schedule > >	schedules_ {}
tsan_unique_ptr< FileBlock >	fb_ {nullptr}
tsan_unique_ptr< RunPrincipal >	runPrincipal_ {nullptr}
tsan_unique_ptr< SubRunPrincipal >	subRunPrincipal_ {nullptr}
PerScheduleContainer< std::unique_ptr< EventPrincipal > >	eventPrincipals_ {}
bool const	handleEmptyRuns_
bool const	handleEmptySubRuns_
SharedException	sharedException_
std::atomic< bool >	firstEvent_ {true}
std::atomic< bool >	fileSwitchInProgress_ {false}

Detailed Description

Definition at line 43 of file EventProcessor.h.

Member Typedef Documentation

template<typename T >

using art::EventProcessor::tsan = hep::concurrency::thread_sanitize<T>

private

Definition at line 140 of file EventProcessor.h.

template<typename T >

using art::EventProcessor::tsan_unique_ptr = hep::concurrency::thread_sanitize_unique_ptr<T>

private

Definition at line 143 of file EventProcessor.h.

Member Enumeration Documentation

enum art::EventProcessor::StatusCode

Enumerator
epSuccess
epSignal

Definition at line 49 of file EventProcessor.h.

{ epSuccess = 0, epSignal = 3 };

Constructor & Destructor Documentation

art::EventProcessor::EventProcessor ( fhicl::ParameterSet  pset, detail::EnabledModules  enabled_modules  )

explicit

Definition at line 87 of file EventProcessor.cc.

References actReg_, art::errors::Configuration, e, util::end(), FDEBUG, art::detail::SharedResources::freeze(), fhicl::ParameterSet::get(), art::getReleaseVersion(), handleEmptyRuns_, handleEmptySubRuns_, art::ProcessConfiguration::id(), fhicl::ParameterSet::id(), input_, art::Globals::instance(), art::thread_safe_registry_via_id< K, M >::instance(), art::ConsumesInfo::instance(), art::legacy, art::InputSourceFactory::make(), art::Globals::nschedules(), outputCallbacks_, pathManager_, art::Globals::processName(), producedProductDescriptions_, producedProductLookupTables_, psSignals_, fhicl::ParameterSet::put(), scheduleIteration_, scheduler_, schedules_, servicesManager_, art::ConsumesInfo::setRequireConsumes(), art::setupSignals(), sharedResources_, art::ActivityRegistry::sPostSourceConstruction, taskGroup_, TDEBUG_FUNC, art::Globals::triggerPSet(), fhicl::detail::validationException::what(), and x.

    : scheduler_{pset.get<ParameterSet>("services.scheduler")}
    , scheduleIteration_{scheduler_->num_schedules()}
    , servicesManager_{create_services_manager(
        pset.get<ParameterSet>("services"),
        actReg_,
        sharedResources_)}
    , pathManager_{pset,
                   outputCallbacks_,
                   producedProductDescriptions_,
                   scheduler_->actionTable(),
                   actReg_,
                   std::move(enabled_modules)}
    , handleEmptyRuns_{scheduler_->handleEmptyRuns()}
    , handleEmptySubRuns_{scheduler_->handleEmptySubRuns()}
  {
    auto services_pset = pset.get<ParameterSet>("services");
    auto const scheduler_pset = services_pset.get<ParameterSet>("scheduler");
    {
      // FIXME: Signals and threads require more effort than this!  A
      //        signal is delivered to only one thread, and which
      //        thread is left up to the implementation to decide. To
      //        get control we must block all signals in the main
      //        thread, create a new thread which will handle the
      //        signals we want to handle, unblock the signals in that
      //        thread only, and have it use sigwaitinfo() to suspend
      //        itselt and wait for those signals.
      setupSignals(scheduler_pset.get<bool>("enableSigInt", true));
    }
    ParentageRegistry::instance();
    ProcessConfigurationRegistry::instance();
    ProcessHistoryRegistry::instance();
    // We do this late because the floating point control word, signal
    // masks, etc., are per-thread and inherited from the master
    // thread, so we want to allow system services, user services, and
    // modules to configure these things in their constructors before
    // we let tbb create any threads. This means they cannot use tbb
    // in their constructors, instead they must use the beginJob
    // callout.
    taskGroup_ = scheduler_->global_task_group();
    // Whenever we are ready to enable ROOT's implicit MT, which is
    // equivalent to its use of TBB, the call should be made after our
    // own TBB task manager has been initialized.
    //    ROOT::EnableImplicitMT();
    TDEBUG_FUNC(5) << "nschedules: " << scheduler_->num_schedules()
                   << " nthreads: " << scheduler_->num_threads();

    auto const errorOnMissingConsumes = scheduler_->errorOnMissingConsumes();
    ConsumesInfo::instance()->setRequireConsumes(errorOnMissingConsumes);

    auto const& processName = Globals::instance()->processName();

    // Trigger-names
    servicesManager_->addSystemService<TriggerNamesService>(
      Globals::instance()->triggerPSet(),
      pset.get<ParameterSet>("physics", {}));

    // We have delayed creating the service instances, now actually
    // create them.
    servicesManager_->forceCreation();
    ServiceHandle<FileCatalogMetadata>()->addMetadataString("art.process_name",
                                                            processName);

    // Now that the service module instances have been created we can
    // set the callbacks, set the module description, and register the
    // products for each service module instance.
    ProcessConfiguration const pc{processName, pset.id(), getReleaseVersion()};
    auto const producing_services = servicesManager_->registerProducts(
      producedProductDescriptions_, psSignals_, pc);
    pathManager_->createModulesAndWorkers(
      *taskGroup_, sharedResources_, producing_services);

    ServiceHandle<TriggerNamesService> trigger_names [[maybe_unused]];
    auto const end = Globals::instance()->nschedules();
    for (ScheduleID::size_type i = 0; i != end; ++i) {
      ScheduleID const sid{i};

      // The ordering of the path results in the TriggerPathsInfo (which is used
      // for the TriggerResults object), must be the same as that provided by
      // the TriggerNamesService.
      auto& trigger_paths_info [[maybe_unused]] =
        pathManager_->triggerPathsInfo(sid);
      assert(trigger_names->getTrigPaths() == trigger_paths_info.pathNames());

      schedules_->emplace(std::piecewise_construct,
                          std::forward_as_tuple(sid),
                          std::forward_as_tuple(sid,
                                                pathManager_,
                                                scheduler_->actionTable(),
                                                actReg_,
                                                outputCallbacks_,
                                                *taskGroup_));
    }
    sharedResources_.freeze(taskGroup_->native_group());

    FDEBUG(2) << pset.to_string() << endl;
    // The input source must be created after the end path executor
    // because the end path executor registers a callback that must
    // be invoked after the first input file is opened.
    {
      ParameterSet main_input;
      main_input.put("module_type", "EmptyEvent");
      main_input.put("module_label", "source");
      main_input.put("maxEvents", -1);
      if (!pset.get_if_present("source", main_input)) {
        mf::LogInfo("EventProcessorSourceConfig")
          << "Could not find a source configuration: using default.";
      }
      ModuleDescription const md{
        main_input.id(),
        main_input.get<string>("module_type"),
        main_input.get<string>("module_label"),
        ModuleThreadingType::legacy,
        ProcessConfiguration{processName, pset.id(), getReleaseVersion()}};
      InputSourceDescription isd{md, outputCallbacks_, actReg_};
      try {
        input_.reset(InputSourceFactory::make(main_input, isd).release());
      }
      catch (fhicl::detail::validationException const& e) {
        throw Exception(errors::Configuration)
          << "\n\nModule label: " << cet::bold_fontify(md.moduleLabel())
          << "\nmodule_type : " << cet::bold_fontify(md.moduleName()) << "\n\n"
          << e.what();
      }
      catch (Exception const& x) {
        if (x.categoryCode() == errors::Configuration) {
          throw Exception(errors::Configuration, "FailedInputSource")
            << "Configuration of main input source has failed\n"
            << x;
        }
        throw;
      }
      catch (cet::exception const& x) {
        throw Exception(errors::Configuration, "FailedInputSource")
          << "Configuration of main input source has failed\n"
          << x;
      }
      catch (...) {
        throw;
      }
    }
    actReg_.sPostSourceConstruction.invoke(input_->moduleDescription());
    // Create product tables used for product retrieval within modules.
    producedProductLookupTables_ = ProductTables{producedProductDescriptions_};
    outputCallbacks_->invoke(producedProductLookupTables_);
  }

art::EventProcessor::~EventProcessor ( )

default

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

delete

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

delete

Member Function Documentation

Level art::EventProcessor::advanceItemType ( )

private

Definition at line 1516 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(), readAndProcessAsync(), and recordOutputModuleClosureRequests().

  {
    auto const itemType = input_->nextItemType();
    FDEBUG(1) << string(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 Exception{errors::LogicError}
        << "Invalid next item type presented to the event processor.\n"
        << "Please contact artists@fnal.gov.";
    }
    throw Exception{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 433 of file EventProcessor.cc.

References actReg_, breakpoints::beginJob(), art::Schedule::beginJob(), e, FDEBUG, art::ScheduleIteration::for_each_schedule(), input_, invokePostBeginJobWorkers_(), schedule(), scheduleIteration_, sharedResources_, and art::ActivityRegistry::sPostBeginJob.

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

  {
    FDEBUG(1) << string(8, ' ') << "beginJob\n";
    breakpoints::beginJob();
    // 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'";
      e << "A cet::exception happened while processing"
           " the beginJob of the 'source'\n";
      throw;
    }
    catch (exception const&) {
      mf::LogError("BeginJob") << "A exception happened while processing"
                                  " the beginJob of the 'source'";
      throw;
    }
    catch (...) {
      mf::LogError("BeginJob") << "An unknown exception happened while"
                                  " processing the beginJob of the 'source'";
      throw;
    }
    scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
      schedule(sid).beginJob(sharedResources_);
    });
    actReg_.sPostBeginJob.invoke();
    invokePostBeginJobWorkers_();
  }

void art::EventProcessor::beginRun ( )

private

Definition at line 667 of file EventProcessor.cc.

References actReg_, art::BeginRun, beginRunCalled_, art::errors::EventProcessorFailure, FDEBUG, finalizeRunEnabled_, art::ScheduleIteration::for_each_schedule(), art::Schedule::process(), r, runPrincipal_, schedule(), scheduleIteration_, art::ActivityRegistry::sPostBeginRun, and art::ActivityRegistry::sPreBeginRun.

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

  {
    assert(runPrincipal_);
    RunID const r{runPrincipal_->runID()};
    if (r.isFlush()) {
      return;
    }
    finalizeRunEnabled_ = true;
    try {
      {
        auto const run =
          std::as_const(*runPrincipal_).makeRun(invalid_module_context);
        actReg_.sPreBeginRun.invoke(run);
      }
      scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
        schedule(sid).process(Transition::BeginRun, *runPrincipal_);
      });
      {
        auto const run =
          std::as_const(*runPrincipal_).makeRun(invalid_module_context);
        actReg_.sPostBeginRun.invoke(run);
      }
    }
    catch (cet::exception& ex) {
      throw Exception{
        errors::EventProcessorFailure,
        "EventProcessor: an exception occurred during current event processing",
        ex};
    }
    catch (...) {
      mf::LogError("PassingThrough")
        << "an exception occurred during current event processing";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "beginRun....................(" << r
              << ")\n";
    beginRunCalled_ = true;
  }

void art::EventProcessor::beginRunIfNotDoneAlready ( )

private

Definition at line 707 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 839 of file EventProcessor.cc.

References actReg_, art::BeginSubRun, beginSubRunCalled_, art::errors::EventProcessorFailure, FDEBUG, finalizeSubRunEnabled_, art::ScheduleIteration::for_each_schedule(), art::Schedule::process(), schedule(), scheduleIteration_, art::ActivityRegistry::sPostBeginSubRun, art::ActivityRegistry::sPreBeginSubRun, and subRunPrincipal_.

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

  {
    assert(subRunPrincipal_);
    SubRunID const sr{subRunPrincipal_->subRunID()};
    if (sr.isFlush()) {
      return;
    }
    finalizeSubRunEnabled_ = true;
    try {
      {
        auto const srun =
          std::as_const(*subRunPrincipal_).makeSubRun(invalid_module_context);
        actReg_.sPreBeginSubRun.invoke(srun);
      }
      scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
        schedule(sid).process(Transition::BeginSubRun, *subRunPrincipal_);
      });
      {
        auto const srun =
          std::as_const(*subRunPrincipal_).makeSubRun(invalid_module_context);
        actReg_.sPostBeginSubRun.invoke(srun);
      }
    }
    catch (cet::exception& ex) {
      throw Exception{
        errors::EventProcessorFailure,
        "EventProcessor: an exception occurred during current event processing",
        ex};
    }
    catch (...) {
      mf::LogError("PassingThrough")
        << "an exception occurred during current event processing";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "beginSubRun.................(" << sr
              << ")\n";
    beginSubRunCalled_ = true;
  }

void art::EventProcessor::beginSubRunIfNotDoneAlready ( )

private

Definition at line 879 of file EventProcessor.cc.

References beginSubRun(), and beginSubRunCalled_.

Referenced by process(), and recordOutputModuleClosureRequests().

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

void art::EventProcessor::closeAllFiles ( )

private

Definition at line 510 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 537 of file EventProcessor.cc.

References art::Schedule::closeAllOutputFiles(), FDEBUG, main_schedule(), and respondToCloseOutputFiles().

Referenced by closeAllFiles(), and recordOutputModuleClosureRequests().

  {
    if (!main_schedule().someOutputsOpen()) {
      return;
    }
    respondToCloseOutputFiles();
    main_schedule().closeAllOutputFiles();
    FDEBUG(1) << string(8, ' ') << "closeAllOutputFiles\n";
  }

void art::EventProcessor::closeInputFile ( )

private

Definition at line 517 of file EventProcessor.cc.

References actReg_, closeSomeOutputFiles(), FDEBUG, art::Schedule::incrementInputFileNumber(), input_, art::Granularity::InputFile, main_schedule(), art::Schedule::recordOutputClosureRequests(), respondToCloseInputFile(), art::ActivityRegistry::sPostCloseFile, and art::ActivityRegistry::sPreCloseFile.

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

  {
    main_schedule().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<>.
    main_schedule().recordOutputClosureRequests(Granularity::InputFile);
    if (main_schedule().outputsToClose()) {
      closeSomeOutputFiles();
    }
    respondToCloseInputFile();
    actReg_.sPreCloseFile.invoke();
    input_->closeFile();
    actReg_.sPostCloseFile.invoke();
    FDEBUG(1) << string(8, ' ') << "closeInputFile\n";
  }

void art::EventProcessor::closeSomeOutputFiles ( )

private

Definition at line 585 of file EventProcessor.cc.

References art::Schedule::closeSomeOutputFiles(), FDEBUG, main_schedule(), 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(main_schedule().outputsToClose());
    respondToCloseOutputFiles();
    main_schedule().closeSomeOutputFiles();
    FDEBUG(1) << string(8, ' ') << "closeSomeOutputFiles\n";
  }

void art::EventProcessor::endJob ( )

private

Definition at line 471 of file EventProcessor.cc.

References actReg_, ec_, endJobAllSchedules(), FDEBUG, input_, art::ConsumesInfo::instance(), mf::LogStatistics(), pathManager_, scheduler_, art::ConsumesInfo::showMissingConsumes(), art::ActivityRegistry::sPostEndJob, timer_, and art::detail::writeSummary().

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

  {
    FDEBUG(1) << string(8, ' ') << "endJob\n";
    ec_->call([this] { endJobAllSchedules(); });
    ec_->call([] { ConsumesInfo::instance()->showMissingConsumes(); });
    ec_->call([this] { input_->doEndJob(); });
    ec_->call([this] { actReg_.sPostEndJob.invoke(); });
    ec_->call([] { mf::LogStatistics(); });
    ec_->call([this] {
      detail::writeSummary(pathManager_, scheduler_->wantSummary(), timer_);
    });
  }

void art::EventProcessor::endJobAllSchedules ( )

private

Definition at line 485 of file EventProcessor.cc.

References art::Schedule::endJob(), art::ScheduleIteration::for_each_schedule(), schedule(), and scheduleIteration_.

Referenced by endJob(), and recordOutputModuleClosureRequests().

  {
    scheduleIteration_.for_each_schedule(
      [this](ScheduleID const sid) { schedule(sid).endJob(); });
  }

void art::EventProcessor::endRun ( )

private

Definition at line 758 of file EventProcessor.cc.

References actReg_, beginRunCalled_, art::EndRun, art::errors::EventProcessorFailure, FDEBUG, art::ScheduleIteration::for_each_schedule(), art::Schedule::process(), r, runPrincipal_, schedule(), scheduleIteration_, art::ActivityRegistry::sPostEndRun, and art::ActivityRegistry::sPreEndRun.

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_->runID()};
    assert(!run.isFlush());
    try {
      actReg_.sPreEndRun.invoke(runPrincipal_->runID(),
                                runPrincipal_->endTime());
      scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
        schedule(sid).process(Transition::EndRun, *runPrincipal_);
      });
      auto const r =
        std::as_const(*runPrincipal_).makeRun(invalid_module_context);
      actReg_.sPostEndRun.invoke(r);
    }
    catch (cet::exception& ex) {
      throw Exception{
        errors::EventProcessorFailure,
        "EventProcessor: an exception occurred during current event processing",
        ex};
    }
    catch (...) {
      mf::LogError("PassingThrough")
        << "an exception occurred during current event processing";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "endRun......................(" << run
              << ")\n";
    beginRunCalled_ = false;
  }

void art::EventProcessor::endSubRun ( )

private

Definition at line 987 of file EventProcessor.cc.

References actReg_, beginSubRunCalled_, art::EndSubRun, art::errors::EventProcessorFailure, FDEBUG, art::ScheduleIteration::for_each_schedule(), art::Schedule::process(), schedule(), scheduleIteration_, art::ActivityRegistry::sPostEndSubRun, art::ActivityRegistry::sPreEndSubRun, 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_->subRunID()};
    assert(!sr.isFlush());
    try {
      actReg_.sPreEndSubRun.invoke(subRunPrincipal_->subRunID(),
                                   subRunPrincipal_->endTime());
      scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
        schedule(sid).process(Transition::EndSubRun, *subRunPrincipal_);
      });
      auto const srun =
        std::as_const(*subRunPrincipal_).makeSubRun(invalid_module_context);
      actReg_.sPostEndSubRun.invoke(srun);
    }
    catch (cet::exception& ex) {
      throw Exception{
        errors::EventProcessorFailure,
        "EventProcessor: an exception occurred during current event processing",
        ex};
    }
    catch (...) {
      mf::LogError("PassingThrough")
        << "an exception occurred during current event processing";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "endSubRun...................(" << sr
              << ")\n";
    beginSubRunCalled_ = false;
  }

actions::ActionCodes art::EventProcessor::error_action ( cet::exception &  e ) const

inlineprivate

Definition at line 158 of file EventProcessor.h.

References scheduler_.

Referenced by finishEventAsync(), and processEventAsync().

    {
      return scheduler_->actionTable().find(e.root_cause());
    }

template<Level L>

void art::EventProcessor::finalize ( )

private

template<Level L>

void art::EventProcessor::finalizeContainingLevels ( )

inlineprivate

Definition at line 96 of file EventProcessor.h.

    {}

void art::EventProcessor::finishEventAsync ( ScheduleID  sid )

private

Definition at line 1412 of file EventProcessor.cc.

References actReg_, e, error_action(), art::Granularity::Event, art::Schedule::event_principal(), art::errors::EventProcessorFailure, FDEBUG, art::actions::IgnoreCompletely, openSomeOutputFiles(), processAllEventsAsync(), art::Schedule::recordOutputClosureRequests(), schedule(), sharedException_, art::ActivityRegistry::sPostProcessEvent, art::SharedException::store(), art::SharedException::store_current(), TDEBUG_BEGIN_FUNC_SI, TDEBUG_END_FUNC_SI, TDEBUG_FUNC_SI, and art::Schedule::writeEvent().

  {
    auto& ep = schedule(sid).event_principal();
    actReg_.sPostProcessEvent.invoke(
      std::as_const(ep).makeEvent(invalid_module_context),
      ScheduleContext{sid});

    // Note: We are part of the endPathTask.
    TDEBUG_BEGIN_FUNC_SI(4, sid);
    FDEBUG(1) << string(8, ' ') << "processEvent................("
              << ep.eventID() << ")\n";
    try {
      // Ask the output workers if they have reached their limits, and
      // if so setup to end the job the next time around the event
      // loop.
      FDEBUG(1) << string(8, ' ') << "shouldWeStop\n";
      static std::mutex m;
      std::lock_guard sentry{m};
      // Now we can write the results of processing to the outputs,
      // and delete the event principal.
      if (!ep.eventID().isFlush()) {
        // Possibly open new output files.  This is safe to do because
        // EndPathExecutor functions are called in a serialized
        // context.
        TDEBUG_FUNC_SI(5, sid) << "Calling openSomeOutputFiles()";
        openSomeOutputFiles();
        TDEBUG_FUNC_SI(5, sid) << "Calling schedule(sid).writeEvent()";

        auto const id = ep.eventID();
        schedule(sid).writeEvent();
        FDEBUG(1) << string(8, ' ') << "writeEvent..................(" << id
                  << ")\n";
      }
      TDEBUG_FUNC_SI(5, sid)
        << "Calling schedules_->"
           "recordOutputClosureRequests(Granularity::Event)";
      schedule(sid).recordOutputClosureRequests(Granularity::Event);
    }
    catch (cet::exception& e) {
      if (error_action(e) != actions::IgnoreCompletely) {
        sharedException_.store<Exception>(
          errors::EventProcessorFailure,
          "EventProcessor: an exception occurred "
          "during current event processing",
          e);
        // And then end this task, terminating event processing.
        TDEBUG_END_FUNC_SI(4, sid) << "EXCEPTION";
        return;
      }
      mf::LogWarning(e.category())
        << "exception being ignored for current event:\n"
        << cet::trim_right_copy(e.what(), " \n");
      // WARNING: We continue processing after the catch blocks!!!
    }
    catch (...) {
      mf::LogError("PassingThrough")
        << "an exception occurred during current event processing";
      sharedException_.store_current();
      // And then end this task, terminating event processing.
      TDEBUG_END_FUNC_SI(4, sid) << "EXCEPTION";
      return;
    }

    // The next event processing task is a continuation of this task.
    processAllEventsAsync(sid);
    TDEBUG_END_FUNC_SI(4, sid);
  }

void art::EventProcessor::invokePostBeginJobWorkers_ ( )

private

Definition at line 236 of file EventProcessor.cc.

References actReg_, art::ScheduleID::first(), input_, pathManager_, and art::ActivityRegistry::sPostBeginJobWorkers.

Referenced by beginJob(), and recordOutputModuleClosureRequests().

  {
    using cet::transform_all;
    // Need to convert multiple lists of workers into a long list that
    // the postBeginJobWorkers callbacks can understand.
    vector<Worker*> allWorkers;
    transform_all(pathManager_->triggerPathsInfo(ScheduleID::first()).workers(),
                  back_inserter(allWorkers),
                  [](auto const& pr) { return pr.second.get(); });
    transform_all(pathManager_->endPathInfo(ScheduleID::first()).workers(),
                  back_inserter(allWorkers),
                  [](auto const& pr) { return pr.second.get(); });
    actReg_.sPostBeginJobWorkers.invoke(input_, allWorkers);
  }

template<Level L>

bool art::EventProcessor::levelsToProcess ( )

private

Definition at line 256 of file EventProcessor.cc.

References advanceItemType(), closeSomeOutputFiles(), art::highest_level(), art::errors::LogicError, main_schedule(), nextLevel_, art::Schedule::outputsToClose(), art::ReadyToAdvance, setOutputFileStatus(), and art::Switching.

  {
    if (nextLevel_.load() == Level::ReadyToAdvance) {
      nextLevel_ = advanceItemType();
      // Consider reading right here?
    }
    if (nextLevel_.load() == L) {
      nextLevel_ = Level::ReadyToAdvance;
      if (main_schedule().outputsToClose()) {
        setOutputFileStatus(OutputFileStatus::Switching);
        finalizeContainingLevels<L>();
        closeSomeOutputFiles();
      }
      return true;
    }
    if (nextLevel_.load() < L) {
      return false;
    }
    if (nextLevel_.load() == highest_level()) {
      return false;
    }
    throw Exception{errors::LogicError} << "Incorrect level hierarchy.\n"
                                        << "  Current level: " << L
                                        << "  Next level: " << nextLevel_;
  }

Schedule& art::EventProcessor::main_schedule ( )

inlineprivate

Definition at line 152 of file EventProcessor.h.

References art::ScheduleID::first(), and schedule().

Referenced by closeAllOutputFiles(), closeInputFile(), closeSomeOutputFiles(), art::EventProcessor::recordOutputModuleClosureRequests< Level::Event >(), art::EventProcessor::recordOutputModuleClosureRequests< Level::Run >(), art::EventProcessor::recordOutputModuleClosureRequests< Level::SubRun >(), levelsToProcess(), process(), setOutputFileStatus(), setRunAuxiliaryRangeSetID(), setSubRunAuxiliaryRangeSetID(), writeRun(), and writeSubRun().

    {
      return schedule(ScheduleID::first());
    }

void art::EventProcessor::openInputFile ( )

private

Definition at line 495 of file EventProcessor.cc.

References actReg_, fb_, FDEBUG, input_, art::errors::LogicError, respondToOpenInputFile(), art::ActivityRegistry::sPostOpenFile, and art::ActivityRegistry::sPreOpenFile.

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

  {
    actReg_.sPreOpenFile.invoke();
    FDEBUG(1) << string(8, ' ') << "openInputFile\n";
    fb_.reset(input_->readFile().release());
    if (fb_ == nullptr) {
      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 562 of file EventProcessor.cc.

References fb_, FDEBUG, art::ScheduleIteration::for_each_schedule(), art::Schedule::openSomeOutputFiles(), outputsToOpen(), respondToOpenOutputFiles(), schedule(), and scheduleIteration_.

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

  {
    if (!outputsToOpen()) {
      return;
    }

    auto open_some_outputs = [this](ScheduleID const sid) {
      schedule(sid).openSomeOutputFiles(*fb_);
    };
    scheduleIteration_.for_each_schedule(open_some_outputs);

    FDEBUG(1) << string(8, ' ') << "openSomeOutputFiles\n";
    respondToOpenOutputFiles();
  }

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

delete

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

delete

bool art::EventProcessor::outputsToOpen ( )

private

Definition at line 548 of file EventProcessor.cc.

References art::ScheduleIteration::for_each_schedule(), art::Schedule::outputsToOpen(), schedule(), and scheduleIteration_.

Referenced by openSomeOutputFiles(), and recordOutputModuleClosureRequests().

  {
    bool outputs_to_open{false};
    auto check_outputs_to_open = [this,
                                  &outputs_to_open](ScheduleID const sid) {
      if (schedule(sid).outputsToOpen()) {
        outputs_to_open = true;
      }
    };
    scheduleIteration_.for_each_schedule(check_outputs_to_open);
    return outputs_to_open;
  }

template<Level L>

void art::EventProcessor::process ( )

private

Definition at line 1482 of file EventProcessor.cc.

References ec_, 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)>(); });
    }
    ec_->call([this] {
      finalize<L>();
      recordOutputModuleClosureRequests<L>();
    });
  }

template<>

void art::EventProcessor::process ( )

private

Definition at line 1039 of file EventProcessor.cc.

References beginRunIfNotDoneAlready(), beginSubRunIfNotDoneAlready(), art::Schedule::closeSomeOutputFiles(), ec_, FDEBUG, fileSwitchInProgress_, firstEvent_, main_schedule(), processAllEventsAsync(), respondToCloseOutputFiles(), scheduler_, setOutputFileStatus(), sharedException_, art::shutdown_flag, art::Switching, taskGroup_, and art::SharedException::throw_if_stored_exception().

  {
    if ((shutdown_flag > 0) || !ec_->empty()) {
      return;
    }
    // Note: This loop is to allow output file switching to happen in
    // the main thread.
    firstEvent_ = true;
    bool done = false;
    while (!done) {
      beginRunIfNotDoneAlready();
      beginSubRunIfNotDoneAlready();

      auto const last_schedule_index = scheduler_->num_schedules() - 1;
      for (ScheduleID::size_type i = 0; i != last_schedule_index; ++i) {
        taskGroup_->run([this, i] { processAllEventsAsync(ScheduleID(i)); });
      }
      taskGroup_->native_group().run_and_wait([this, last_schedule_index] {
        processAllEventsAsync(ScheduleID(last_schedule_index));
      });

      // If anything bad happened during event processing, let the
      // user know.
      sharedException_.throw_if_stored_exception();
      if (!fileSwitchInProgress_.load()) {
        done = true;
        continue;
      }
      setOutputFileStatus(OutputFileStatus::Switching);
      finalizeContainingLevels<most_deeply_nested_level()>();
      respondToCloseOutputFiles();
      main_schedule().closeSomeOutputFiles();
      FDEBUG(1) << string(8, ' ') << "closeSomeOutputFiles\n";
      // We started the switch after advancing to the next item type;
      // we must make sure that we read that event before advancing
      // the item type again.
      firstEvent_ = true;
      fileSwitchInProgress_ = false;
    }
  }

void art::EventProcessor::processAllEventsAsync ( ScheduleID  sid )

private

Definition at line 1085 of file EventProcessor.cc.

References readAndProcessAsync(), sharedException_, art::SharedException::store_current(), TDEBUG_BEGIN_FUNC_SI, and TDEBUG_END_FUNC_SI.

Referenced by finishEventAsync(), process(), and readAndProcessAsync().

  {
    // Note: We are part of the processAllEventsTask (schedule head
    // task), and our parent is the eventLoopTask.
    TDEBUG_BEGIN_FUNC_SI(4, sid);
    try {
      readAndProcessAsync(sid);
    }
    catch (...) {
      sharedException_.store_current();
      TDEBUG_END_FUNC_SI(4, sid) << "terminate event loop because of EXCEPTION";
      return;
    }
    // If no exception, then end this task, which does not terminate
    // event processing because readAndProcessAsync creates a
    // continuation task.
    TDEBUG_END_FUNC_SI(4, sid);
  }

void art::EventProcessor::processEvent ( )

private

Referenced by recordOutputModuleClosureRequests().

void art::EventProcessor::processEventAsync ( ScheduleID  sid )

private

Definition at line 1371 of file EventProcessor.cc.

References e, error_action(), art::errors::EventProcessorFailure, art::actions::FailModule, art::actions::FailPath, art::actions::IgnoreCompletely, art::Schedule::process_event_modifiers(), schedule(), sharedException_, art::SharedException::store(), art::SharedException::store_current(), TDEBUG_BEGIN_FUNC_SI, and TDEBUG_END_FUNC_SI.

Referenced by readAndProcessAsync().

      {
    // Note: We are part of the readAndProcessEventTask (schedule head
    // task), and our parent task is the eventLoopTask.
    TDEBUG_BEGIN_FUNC_SI(4, sid);
    assert(!schedule(sid).event_principal().eventID().isFlush());
    // Continue processing via the creation of a continuation.
    auto endPathTask = make_waiting_task<EndPathTask>(this, sid);
    // Start the trigger paths running.  When they finish they will
    // spawn the endPathTask which will run the end path, write the
    // event, and start the next event processing task.
    schedule(sid).process_event_modifiers(endPathTask);
    TDEBUG_END_FUNC_SI(4, sid);
  }
  catch (cet::exception& e) {
    // Upon exiting this scope, end this task, terminating event
    // processing.
    auto const action = error_action(e);
    if (action != actions::IgnoreCompletely) {
      assert(action != actions::FailModule);
      assert(action != actions::FailPath);
      sharedException_.store<Exception>(
        errors::EventProcessorFailure,
        "EventProcessor: an exception occurred during current event processing",
        e);
      TDEBUG_END_FUNC_SI(4, sid) << "terminate event loop because of EXCEPTION";
      return;
    }
    mf::LogWarning(e.category())
      << "exception being ignored for current event:\n"
      << cet::trim_right_copy(e.what(), " \n");
    TDEBUG_END_FUNC_SI(4, sid) << "Ignoring exception.";
  }
  catch (...) {
    mf::LogError("PassingThrough")
      << "an exception occurred during current event processing";
    sharedException_.store_current();
    TDEBUG_END_FUNC_SI(4, sid) << "terminate event loop because of EXCEPTION";
  }

void art::EventProcessor::readAndProcessAsync ( ScheduleID  sid )

private

Definition at line 1110 of file EventProcessor.cc.

References art::Schedule::accept_principal(), actReg_, advanceItemType(), FDEBUG, fileSwitchInProgress_, firstEvent_, art::highest_level(), input_, art::errors::LogicError, art::most_deeply_nested_level(), nextLevel_, processAllEventsAsync(), processEventAsync(), producedProductLookupTables_, psSignals_, art::ReadyToAdvance, sc, schedule(), art::shutdown_flag, art::ActivityRegistry::sPostSourceEvent, art::ActivityRegistry::sPreSourceEvent, subRunPrincipal_, TDEBUG_BEGIN_FUNC_SI, TDEBUG_END_FUNC_SI, and TDEBUG_FUNC_SI.

Referenced by processAllEventsAsync().

  {
    // Note: We are part of the readAndProcessEventTask (schedule head
    // task), and our parent task is the eventLoopTask.
    TDEBUG_BEGIN_FUNC_SI(4, sid);
    // Note: shutdown_flag is a extern global atomic int in
    // art/art/Utilities/UnixSignalHandlers.cc
    if (shutdown_flag) {
      // User called for a clean shutdown using a signal or ctrl-c,
      // end event processing and this task.
      TDEBUG_END_FUNC_SI(4, sid) << "CLEAN SHUTDOWN";
      return;
    }

    // The item type advance and the event read must be done with the
    // input source lock held; however event-processing must not
    // serialized.
    {
      InputSourceMutexSentry lock_input;
      if (fileSwitchInProgress_.load()) {
        // We must avoid advancing the iterator after a schedule has
        // noticed it is time to switch files.  After the switch, we
        // will need to set firstEvent_ true so that the first
        // schedule that resumes after the switch actually reads the
        // event that the first schedule which noticed we needed a
        // switch had advanced the iterator to.

        // Note: We still have the problem that because the schedules
        // do not read events at the same time the file switch point
        // can be up to nschedules-1 ahead of where it would have been
        // if there was only one schedule.  If we are switching output
        // files every event in an attempt to create single event
        // files, this really does not work out too well.
        TDEBUG_END_FUNC_SI(4, sid) << "FILE SWITCH";
        return;
      }
      // Check the next item type and exit this task if it is not an
      // event, or if the user has asynchronously requested a
      // shutdown.
      auto expected = true;
      if (firstEvent_.compare_exchange_strong(expected, false)) {
        // Do not advance the item type on the first event.
      } else {
        // Do the advance item type.
        if (nextLevel_.load() == Level::ReadyToAdvance) {
          // See what the next item is.
          TDEBUG_FUNC_SI(5, sid) << "Calling advanceItemType()";
          nextLevel_ = advanceItemType();
        }
        if ((nextLevel_.load() < most_deeply_nested_level()) ||
            (nextLevel_.load() == highest_level())) {
          // We are popping up, end event processing and this task.
          TDEBUG_END_FUNC_SI(4, sid) << "END OF SUBRUN";
          return;
        }
        if (nextLevel_.load() != most_deeply_nested_level()) {
          // Error: incorrect level hierarchy
          TDEBUG_END_FUNC_SI(4, sid) << "BAD HIERARCHY";
          throw Exception{errors::LogicError} << "Incorrect level hierarchy.";
        }
        nextLevel_ = Level::ReadyToAdvance;
        // At this point we have determined that we are going to read
        // an event and we must do that before dropping the lock on
        // the input source which is what is protecting us against a
        // double-advance caused by a different schedule.
        if (schedule(sid).outputsToClose()) {
          fileSwitchInProgress_ = true;
          TDEBUG_END_FUNC_SI(4, sid) << "FILE SWITCH INITIATED";
          return;
        }
      }

      // Now we can read the event from the source.
      ScheduleContext const sc{sid};
      assert(subRunPrincipal_);
      assert(subRunPrincipal_->subRunID().isValid());
      actReg_.sPreSourceEvent.invoke(sc);
      TDEBUG_FUNC_SI(5, sid) << "Calling input_->readEvent(subRunPrincipal_)";
      auto ep = input_->readEvent(subRunPrincipal_.get());
      assert(ep);
      // The intended behavior here is that the producing services
      // which are called during the sPostReadEvent cannot see each
      // others put products.  We enforce this by creating the groups
      // for the produced products, but do not allow the lookups to
      // find them until after the callbacks have run.
      ep->createGroupsForProducedProducts(producedProductLookupTables_);
      psSignals_->sPostReadEvent.invoke(*ep);
      ep->enableLookupOfProducedProducts();
      actReg_.sPostSourceEvent.invoke(
        std::as_const(*ep).makeEvent(invalid_module_context), sc);
      FDEBUG(1) << string(8, ' ') << "readEvent...................("
                << ep->eventID() << ")\n";
      schedule(sid).accept_principal(std::move(ep));
      // Now we drop the input source lock by exiting the guarded
      // scope.
    }
    if (schedule(sid).event_principal().eventID().isFlush()) {
      // No processing to do, start next event handling task.
      processAllEventsAsync(sid);
      TDEBUG_END_FUNC_SI(4, sid) << "FLUSH EVENT";
      return;
    }

    // Now process the event.
    processEventAsync(sid);
    TDEBUG_END_FUNC_SI(4, sid);
  }

void art::EventProcessor::readEvent ( )

private

Referenced by recordOutputModuleClosureRequests().

void art::EventProcessor::readRun ( )

private

Definition at line 637 of file EventProcessor.cc.

References actReg_, FDEBUG, art::ScheduleIteration::for_each_schedule(), input_, producedProductLookupTables_, psSignals_, r, runPrincipal_, schedule(), scheduleIteration_, art::Schedule::seedRunRangeSet(), art::ActivityRegistry::sPostSourceRun, and art::ActivityRegistry::sPreSourceRun.

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

  {
    actReg_.sPreSourceRun.invoke();
    runPrincipal_.reset(input_->readRun().release());
    assert(runPrincipal_);
    auto rsh = input_->runRangeSetHandler();
    assert(rsh);
    auto seed_range_set = [this, &rsh](ScheduleID const sid) {
      schedule(sid).seedRunRangeSet(*rsh);
    };
    scheduleIteration_.for_each_schedule(seed_range_set);
    // The intended behavior here is that the producing services which
    // are called during the sPostReadRun cannot see each others put
    // products. We enforce this by creating the groups for the
    // produced products, but do not allow the lookups to find them
    // until after the callbacks have run.
    runPrincipal_->createGroupsForProducedProducts(
      producedProductLookupTables_);
    psSignals_->sPostReadRun.invoke(*runPrincipal_);
    runPrincipal_->enableLookupOfProducedProducts();
    {
      auto const r =
        std::as_const(*runPrincipal_).makeRun(invalid_module_context);
      actReg_.sPostSourceRun.invoke(r);
    }
    FDEBUG(1) << string(8, ' ') << "readRun.....................("
              << runPrincipal_->runID() << ")\n";
  }

void art::EventProcessor::readSubRun ( )

private

Definition at line 809 of file EventProcessor.cc.

References actReg_, FDEBUG, art::ScheduleIteration::for_each_schedule(), input_, producedProductLookupTables_, psSignals_, runPrincipal_, schedule(), scheduleIteration_, art::Schedule::seedSubRunRangeSet(), art::ActivityRegistry::sPostSourceSubRun, art::ActivityRegistry::sPreSourceSubRun, and subRunPrincipal_.

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

  {
    actReg_.sPreSourceSubRun.invoke();
    subRunPrincipal_.reset(input_->readSubRun(runPrincipal_.get()).release());
    assert(subRunPrincipal_);
    auto rsh = input_->subRunRangeSetHandler();
    assert(rsh);
    auto seed_range_set = [this, &rsh](ScheduleID const sid) {
      schedule(sid).seedSubRunRangeSet(*rsh);
    };
    scheduleIteration_.for_each_schedule(seed_range_set);
    // The intended behavior here is that the producing services which
    // are called during the sPostReadSubRun cannot see each others
    // put products. We enforce this by creating the groups for the
    // produced products, but do not allow the lookups to find them
    // until after the callbacks have run.
    subRunPrincipal_->createGroupsForProducedProducts(
      producedProductLookupTables_);
    psSignals_->sPostReadSubRun.invoke(*subRunPrincipal_);
    subRunPrincipal_->enableLookupOfProducedProducts();
    {
      auto const sr =
        std::as_const(*subRunPrincipal_).makeSubRun(invalid_module_context);
      actReg_.sPostSourceSubRun.invoke(sr);
    }
    FDEBUG(1) << string(8, ' ') << "readSubRun..................("
              << subRunPrincipal_->subRunID() << ")\n";
  }

template<Level L>

void art::EventProcessor::recordOutputModuleClosureRequests ( )

inlineprivate

Definition at line 100 of file EventProcessor.h.

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

    {}

void art::EventProcessor::respondToCloseInputFile ( )

private

Definition at line 607 of file EventProcessor.cc.

References fb_, FDEBUG, art::ScheduleIteration::for_each_schedule(), art::Schedule::respondToCloseInputFile(), schedule(), and scheduleIteration_.

Referenced by closeInputFile(), and recordOutputModuleClosureRequests().

  {
    scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
      schedule(sid).respondToCloseInputFile(*fb_);
    });
    FDEBUG(1) << string(8, ' ') << "respondToCloseInputFile\n";
  }

void art::EventProcessor::respondToCloseOutputFiles ( )

private

Definition at line 625 of file EventProcessor.cc.

References fb_, FDEBUG, art::ScheduleIteration::for_each_schedule(), art::Schedule::respondToCloseOutputFiles(), schedule(), and scheduleIteration_.

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

  {
    scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
      schedule(sid).respondToCloseOutputFiles(*fb_);
    });
    FDEBUG(1) << string(8, ' ') << "respondToCloseOutputFiles\n";
  }

void art::EventProcessor::respondToOpenInputFile ( )

private

Definition at line 598 of file EventProcessor.cc.

References fb_, FDEBUG, art::ScheduleIteration::for_each_schedule(), art::Schedule::respondToOpenInputFile(), schedule(), and scheduleIteration_.

Referenced by openInputFile(), and recordOutputModuleClosureRequests().

  {
    scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
      schedule(sid).respondToOpenInputFile(*fb_);
    });
    FDEBUG(1) << string(8, ' ') << "respondToOpenInputFile\n";
  }

void art::EventProcessor::respondToOpenOutputFiles ( )

private

Definition at line 616 of file EventProcessor.cc.

References fb_, FDEBUG, art::ScheduleIteration::for_each_schedule(), art::Schedule::respondToOpenOutputFiles(), schedule(), and scheduleIteration_.

Referenced by openSomeOutputFiles(), and recordOutputModuleClosureRequests().

  {
    scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
      schedule(sid).respondToOpenOutputFiles(*fb_);
    });
    FDEBUG(1) << string(8, ' ') << "respondToOpenOutputFiles\n";
  }

EventProcessor::StatusCode art::EventProcessor::runToCompletion

(

)

Definition at line 1499 of file EventProcessor.cc.

References ec_, epSignal, epSuccess, art::shutdown_flag, and terminateAbnormally_().

  {
    StatusCode returnCode{epSuccess};
    ec_->call([this, &returnCode] {
      process<highest_level()>();
      if (shutdown_flag > 0) {
        returnCode = epSignal;
      }
    });
    if (!ec_->empty()) {
      terminateAbnormally_();
      ec_->rethrow();
    }
    return returnCode;
  }

Schedule& art::EventProcessor::schedule ( ScheduleID const  id )

inlineprivate

Definition at line 146 of file EventProcessor.h.

References schedules_.

Referenced by beginJob(), beginRun(), beginSubRun(), endJobAllSchedules(), endRun(), endSubRun(), finishEventAsync(), main_schedule(), openSomeOutputFiles(), outputsToOpen(), processEventAsync(), readAndProcessAsync(), readRun(), readSubRun(), respondToCloseInputFile(), respondToCloseOutputFiles(), respondToOpenInputFile(), respondToOpenOutputFiles(), setRunAuxiliaryRangeSetID(), and setSubRunAuxiliaryRangeSetID().

    {
      return schedules_->at(id);
    }

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

private

Definition at line 578 of file EventProcessor.cc.

References FDEBUG, main_schedule(), and art::Schedule::setOutputFileStatus().

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

  {
    main_schedule().setOutputFileStatus(ofs);
    FDEBUG(1) << string(8, ' ') << "setOutputFileStatus\n";
  }

void art::EventProcessor::setRunAuxiliaryRangeSetID ( )

private

Definition at line 715 of file EventProcessor.cc.

References art::RangeSetHandler::clone(), FDEBUG, art::RangeSetHandler::flushRanges(), art::ScheduleIteration::for_each_schedule(), art::RangeSet::invalid(), main_schedule(), art::RangeSetHandler::Open, runPrincipal_, art::Schedule::runRangeSetHandler(), schedule(), scheduleIteration_, art::RangeSetHandler::seenRanges(), art::Schedule::setRunAuxiliaryRangeSetID(), art::Switching, and tmp.

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

  {
    assert(runPrincipal_);
    FDEBUG(1) << string(8, ' ') << "setRunAuxiliaryRangeSetID...("
              << runPrincipal_->runID() << ")\n";
    if (main_schedule().runRangeSetHandler().type() ==
        RangeSetHandler::HandlerType::Open) {
      // We are using EmptyEvent source, need to merge what the
      // schedules have seen.
      auto mergedRS = RangeSet::invalid();
      auto merge_range_sets = [this, &mergedRS](ScheduleID const sid) {
        auto const& rs = schedule(sid).runRangeSetHandler().seenRanges();
        // The following constructor ensures that the range is sorted
        // before 'merge' is called.
        RangeSet const tmp{rs.run(), rs.ranges()};
        mergedRS.merge(tmp);
      };
      scheduleIteration_.for_each_schedule(merge_range_sets);
      runPrincipal_->updateSeenRanges(mergedRS);
      auto update_executors = [this, &mergedRS](ScheduleID const sid) {
        schedule(sid).setRunAuxiliaryRangeSetID(mergedRS);
      };
      scheduleIteration_.for_each_schedule(update_executors);
      return;
    }

    // Since we are using already existing ranges, all the range set
    // handlers have the same ranges, use the first one.  handler with
    // the largest event number, that will be the one which we will
    // use as the file switch boundary.  Note that is may not match
    // the exactly the schedule that triggered the switch.  Do we need
    // to fix this?
    unique_ptr<RangeSetHandler> rshAtSwitch{
      main_schedule().runRangeSetHandler().clone()};
    if (main_schedule().fileStatus() != OutputFileStatus::Switching) {
      // We are at the end of the job.
      rshAtSwitch->flushRanges();
    }
    runPrincipal_->updateSeenRanges(rshAtSwitch->seenRanges());
    main_schedule().setRunAuxiliaryRangeSetID(rshAtSwitch->seenRanges());
  }

void art::EventProcessor::setSubRunAuxiliaryRangeSetID ( )

private

Definition at line 887 of file EventProcessor.cc.

References art::RangeSetHandler::clone(), art::EventID::event(), art::ClosedRangeSetHandler::eventInfo(), FDEBUG, art::ScheduleID::first(), art::ScheduleIteration::for_each_schedule(), art::ClosedRangeSetHandler::EventInfo::id(), art::RangeSet::invalid(), main_schedule(), art::RangeSetHandler::maybeSplitRange(), art::RangeSetHandler::Open, art::Schedule::runRangeSetHandler(), schedule(), scheduleIteration_, art::Schedule::seedRunRangeSet(), art::Schedule::seedSubRunRangeSet(), art::RangeSetHandler::seenRanges(), art::Schedule::setSubRunAuxiliaryRangeSetID(), subRunPrincipal_, art::Schedule::subRunRangeSetHandler(), art::Switching, and tmp.

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

  {
    assert(subRunPrincipal_);
    FDEBUG(1) << string(8, ' ') << "setSubRunAuxiliaryRangeSetID("
              << subRunPrincipal_->subRunID() << ")\n";
    if (main_schedule().subRunRangeSetHandler().type() ==
        RangeSetHandler::HandlerType::Open) {
      // We are using EmptyEvent source, need to merge what the
      // schedules have seen.
      auto mergedRS = RangeSet::invalid();
      auto merge_range_sets = [this, &mergedRS](ScheduleID const sid) {
        auto const& rs = schedule(sid).subRunRangeSetHandler().seenRanges();
        // The following constructor ensures that the range is sorted
        // before 'merge' is called.
        RangeSet const tmp{rs.run(), rs.ranges()};
        mergedRS.merge(tmp);
      };
      scheduleIteration_.for_each_schedule(merge_range_sets);
      subRunPrincipal_->updateSeenRanges(mergedRS);
      auto update_executors = [this, &mergedRS](ScheduleID const sid) {
        schedule(sid).setSubRunAuxiliaryRangeSetID(mergedRS);
      };
      scheduleIteration_.for_each_schedule(update_executors);
      return;
    }
    // Ranges are split/flushed only for a RangeSetHandler whose
    // dynamic type is 'ClosedRangeSetHandler'.
    //
    // Consider the following range-sets
    //
    //  SubRun RangeSet:
    //
    //    { Run 1 : SubRun 1 : Events [1,7) }  <-- Current
    //
    //  Run RangeSet:
    //
    //    { Run 1 : SubRun 0 : Events [5,11)
    //              SubRun 1 : Events [1,7)    <-- Current
    //              SubRun 1 : Events [9,15) }
    //
    // For a range split just before SubRun 1, Event 6, the
    // range sets should become:
    //
    //  SubRun RangeSet:
    //
    //    { Run 1 : SubRun 1 : Events [1,6)
    //              SubRun 1 : Events [6,7) } <-- Updated
    //
    //  Run RangeSet:
    //
    //    { Run 1 : SubRun 0 : Events [5,11)
    //              SubRun 1 : Events [1,6)
    //              SubRun 1 : Events [6,7)   <-- Updated
    //              SubRun 1 : Events [9,15) }
    //
    // Since we are using already existing ranges, all the range set
    // handlers have the same ranges.  Find the closed range set
    // handler with the largest event number, that will be the one
    // which we will use as the file switch boundary.  Note that is
    // may not match the exactly the schedule that triggered the
    // switch.  Do we need to fix this?
    //
    // If we do not find any handlers with valid event info then we
    // use the first one, which is just fine.  This happens for
    // example when we are dropping all events.
    unsigned largestEvent = 1U;
    ScheduleID idxOfMax{ScheduleID::first()};
    ScheduleID idx{ScheduleID::first()};
    auto& val = main_schedule().subRunRangeSetHandler();
    auto& rsh = dynamic_cast<ClosedRangeSetHandler const&>(val);
    // Make sure the event number is a valid event number before using
    // it. It can be invalid in the handler if we have not yet read an
    // event, which happens with empty subruns and when we are
    // dropping all events.
    if (rsh.eventInfo().id().isValid() && !rsh.eventInfo().id().isFlush()) {
      if (rsh.eventInfo().id().event() > largestEvent) {
        largestEvent = rsh.eventInfo().id().event();
        idxOfMax = idx;
      }
    }
    idx = idx.next();

    unique_ptr<RangeSetHandler> rshAtSwitch{
      main_schedule().subRunRangeSetHandler().clone()};
    if (main_schedule().fileStatus() == OutputFileStatus::Switching) {
      rshAtSwitch->maybeSplitRange();
      unique_ptr<RangeSetHandler> runRSHAtSwitch{
        schedule(idxOfMax).runRangeSetHandler().clone()};
      runRSHAtSwitch->maybeSplitRange();
      main_schedule().seedRunRangeSet(*runRSHAtSwitch);
    } else {
      // We are at the end of the job.
      rshAtSwitch->flushRanges();
    }
    main_schedule().seedSubRunRangeSet(*rshAtSwitch);
    subRunPrincipal_->updateSeenRanges(rshAtSwitch->seenRanges());
    main_schedule().setSubRunAuxiliaryRangeSetID(rshAtSwitch->seenRanges());
  }

void art::EventProcessor::terminateAbnormally_ ( )

private

Definition at line 1544 of file EventProcessor.cc.

Referenced by recordOutputModuleClosureRequests(), and runToCompletion().

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

void art::EventProcessor::writeEvent ( )

private

Referenced by recordOutputModuleClosureRequests().

void art::EventProcessor::writeRun ( )

private

Definition at line 794 of file EventProcessor.cc.

References FDEBUG, main_schedule(), r, runPrincipal_, and art::Schedule::writeRun().

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_->runID()};
    assert(!r.isFlush());
    main_schedule().writeRun(*runPrincipal_);
    FDEBUG(1) << string(8, ' ') << "writeRun....................(" << r
              << ")\n";
  }

void art::EventProcessor::writeSubRun ( )

private

Definition at line 1023 of file EventProcessor.cc.

References FDEBUG, main_schedule(), subRunPrincipal_, and art::Schedule::writeSubRun().

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_->subRunID()};
    assert(!sr.isFlush());
    main_schedule().writeSubRun(*subRunPrincipal_);
    FDEBUG(1) << string(8, ' ') << "writeSubRun.................(" << sr
              << ")\n";
  }

Member Data Documentation

ActivityRegistry art::EventProcessor::actReg_ {}

private

Definition at line 186 of file EventProcessor.h.

Referenced by beginJob(), beginRun(), beginSubRun(), closeInputFile(), endJob(), endRun(), endSubRun(), EventProcessor(), finishEventAsync(), invokePostBeginJobWorkers_(), openInputFile(), readAndProcessAsync(), readRun(), and readSubRun().

std::atomic<bool> art::EventProcessor::beginRunCalled_ {false}

private

Definition at line 173 of file EventProcessor.h.

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

std::atomic<bool> art::EventProcessor::beginSubRunCalled_ {false}

private

Definition at line 176 of file EventProcessor.h.

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

tsan<detail::ExceptionCollector> art::EventProcessor::ec_ {}

private

Definition at line 167 of file EventProcessor.h.

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

PerScheduleContainer<std::unique_ptr<EventPrincipal> > art::EventProcessor::eventPrincipals_ {}

private

Definition at line 246 of file EventProcessor.h.

tsan_unique_ptr<FileBlock> art::EventProcessor::fb_ {nullptr}

private

Definition at line 237 of file EventProcessor.h.

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

std::atomic<bool> art::EventProcessor::fileSwitchInProgress_ {false}

private

Definition at line 266 of file EventProcessor.h.

Referenced by process(), and readAndProcessAsync().

std::atomic<bool> art::EventProcessor::finalizeRunEnabled_ {false}

private

Definition at line 179 of file EventProcessor.h.

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

std::atomic<bool> art::EventProcessor::finalizeSubRunEnabled_ {false}

private

Definition at line 182 of file EventProcessor.h.

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

std::atomic<bool> art::EventProcessor::firstEvent_ {true}

private

Definition at line 263 of file EventProcessor.h.

Referenced by process(), and readAndProcessAsync().

bool const art::EventProcessor::handleEmptyRuns_

private

Definition at line 249 of file EventProcessor.h.

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

bool const art::EventProcessor::handleEmptySubRuns_

private

Definition at line 252 of file EventProcessor.h.

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

tsan_unique_ptr<InputSource> art::EventProcessor::input_ {nullptr}

private

Definition at line 231 of file EventProcessor.h.

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

tsan<MFStatusUpdater> art::EventProcessor::mfStatusUpdater_ {actReg_}

private

Definition at line 189 of file EventProcessor.h.

std::atomic<Level> art::EventProcessor::nextLevel_ {Level::ReadyToAdvance}

private

Definition at line 164 of file EventProcessor.h.

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

tsan<UpdateOutputCallbacks> art::EventProcessor::outputCallbacks_ {}

private

Definition at line 194 of file EventProcessor.h.

Referenced by EventProcessor().

tsan<PathManager> art::EventProcessor::pathManager_

private

Definition at line 228 of file EventProcessor.h.

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

tsan<ProductDescriptions> art::EventProcessor::producedProductDescriptions_ {}

private

Definition at line 200 of file EventProcessor.h.

Referenced by EventProcessor().

tsan<ProductTables> art::EventProcessor::producedProductLookupTables_ {ProductTables::invalid()}

private

Definition at line 208 of file EventProcessor.h.

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

tsan<ProducingServiceSignals> art::EventProcessor::psSignals_ {}

private

Definition at line 210 of file EventProcessor.h.

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

tsan_unique_ptr<RunPrincipal> art::EventProcessor::runPrincipal_ {nullptr}

private

Definition at line 240 of file EventProcessor.h.

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

ScheduleIteration art::EventProcessor::scheduleIteration_

private

Definition at line 221 of file EventProcessor.h.

Referenced by beginJob(), beginRun(), beginSubRun(), endJobAllSchedules(), endRun(), endSubRun(), EventProcessor(), openSomeOutputFiles(), outputsToOpen(), readRun(), readSubRun(), respondToCloseInputFile(), respondToCloseOutputFiles(), respondToOpenInputFile(), respondToOpenOutputFiles(), setRunAuxiliaryRangeSetID(), and setSubRunAuxiliaryRangeSetID().

tsan<Scheduler> art::EventProcessor::scheduler_

private

Definition at line 215 of file EventProcessor.h.

Referenced by endJob(), error_action(), EventProcessor(), and process().

tsan<std::map<ScheduleID, Schedule> > art::EventProcessor::schedules_ {}

private

Definition at line 234 of file EventProcessor.h.

Referenced by EventProcessor(), and schedule().

tsan_unique_ptr<ServicesManager> art::EventProcessor::servicesManager_

private

Definition at line 224 of file EventProcessor.h.

Referenced by EventProcessor().

SharedException art::EventProcessor::sharedException_

private

Definition at line 256 of file EventProcessor.h.

Referenced by finishEventAsync(), process(), processAllEventsAsync(), and processEventAsync().

detail::SharedResources art::EventProcessor::sharedResources_ {}

private

Definition at line 219 of file EventProcessor.h.

Referenced by beginJob(), and EventProcessor().

tsan_unique_ptr<SubRunPrincipal> art::EventProcessor::subRunPrincipal_ {nullptr}

private

Definition at line 243 of file EventProcessor.h.

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

std::unique_ptr<GlobalTaskGroup> art::EventProcessor::taskGroup_ {nullptr}

private

Definition at line 217 of file EventProcessor.h.

Referenced by EventProcessor(), and process().

tsan<cet::cpu_timer> art::EventProcessor::timer_ {}

private

Definition at line 170 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