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 const &pset, detail::EnabledModules const &enabled_modules)
	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 139 of file EventProcessor.h.

template<typename T >

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

private

Definition at line 142 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 const &  pset, detail::EnabledModules const &  enabled_modules  )

explicit

Definition at line 124 of file EventProcessor.cc.

    : 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_,
                   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 = pathManager_->triggerPathsInfo(sid);
      assert(trigger_names->getTrigPaths() == trigger_paths_info.pathNames());

      auto results_inserter =
        maybe_trigger_results_inserter(sid, //
                                       processName,
                                       pset,
                                       Globals::instance()->triggerPSet(),
                                       outputCallbacks_, //
                                       producedProductDescriptions_,
                                       scheduler_->actionTable(), //
                                       actReg_,                   //
                                       trigger_paths_info,        //
                                       *taskGroup_,               //
                                       sharedResources_);
      schedules_->emplace(std::piecewise_construct,
                          std::forward_as_tuple(sid),
                          std::forward_as_tuple(sid,
                                                pathManager_,
                                                scheduler_->actionTable(),
                                                actReg_,
                                                outputCallbacks_,
                                                move(results_inserter),
                                                *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 ( EventProcessor const &  )

delete

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

delete

Member Function Documentation

Level art::EventProcessor::advanceItemType ( )

private

Definition at line 1544 of file EventProcessor.cc.

  {
    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 480 of file EventProcessor.cc.

  {
    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'\n";
      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'\n";
      throw;
    }
    catch (...) {
      mf::LogError("BeginJob") << "An unknown exception happened while"
                                  " processing the beginJob of the 'source'\n";
      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 713 of file EventProcessor.cc.

  {
    assert(runPrincipal_);
    RunID const r{runPrincipal_->runID()};
    if (r.isFlush()) {
      return;
    }
    finalizeRunEnabled_ = true;
    try {
      {
        Run const run{*runPrincipal_, invalid_module_context};
        actReg_.sPreBeginRun.invoke(run);
      }
      scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
        schedule(sid).process(Transition::BeginRun, *runPrincipal_);
      });
      {
        Run const run{*runPrincipal_, 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\n";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "beginRun....................(" << r
              << ")\n";
    beginRunCalled_ = true;
  }

void art::EventProcessor::beginRunIfNotDoneAlready ( )

private

Definition at line 751 of file EventProcessor.cc.

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

void art::EventProcessor::beginSubRun ( )

private

Definition at line 882 of file EventProcessor.cc.

  {
    assert(subRunPrincipal_);
    SubRunID const sr{subRunPrincipal_->subRunID()};
    if (sr.isFlush()) {
      return;
    }
    finalizeSubRunEnabled_ = true;
    try {
      {
        SubRun const srun{*subRunPrincipal_, invalid_module_context};
        actReg_.sPreBeginSubRun.invoke(srun);
      }
      scheduleIteration_.for_each_schedule([this](ScheduleID const sid) {
        schedule(sid).process(Transition::BeginSubRun, *subRunPrincipal_);
      });
      {
        SubRun const srun{*subRunPrincipal_, 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\n";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "beginSubRun.................(" << sr
              << ")\n";
    beginSubRunCalled_ = true;
  }

void art::EventProcessor::beginSubRunIfNotDoneAlready ( )

private

Definition at line 920 of file EventProcessor.cc.

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

void art::EventProcessor::closeAllFiles ( )

private

Definition at line 557 of file EventProcessor.cc.

  {
    closeAllOutputFiles();
    closeInputFile();
  }

void art::EventProcessor::closeAllOutputFiles ( )

private

Definition at line 584 of file EventProcessor.cc.

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

void art::EventProcessor::closeInputFile ( )

private

Definition at line 564 of file EventProcessor.cc.

  {
    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 632 of file EventProcessor.cc.

  {
    // 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 518 of file EventProcessor.cc.

  {
    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 532 of file EventProcessor.cc.

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

void art::EventProcessor::endRun ( )

private

Definition at line 802 of file EventProcessor.cc.

  {
    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_);
      });
      Run const r{*runPrincipal_, 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\n";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "endRun......................(" << run
              << ")\n";
    beginRunCalled_ = false;
  }

void art::EventProcessor::endSubRun ( )

private

Definition at line 1028 of file EventProcessor.cc.

  {
    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_);
      });
      SubRun const srun{*subRunPrincipal_, 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\n";
      throw;
    }
    FDEBUG(1) << string(8, ' ') << "endSubRun...................(" << sr
              << ")\n";
    beginSubRunCalled_ = false;
  }

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

inlineprivate

Definition at line 157 of file EventProcessor.h.

    {
      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 95 of file EventProcessor.h.

    {}

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

private

Definition at line 1436 of file EventProcessor.cc.

  {
    auto& ep = schedule(sid).event_principal();
    actReg_.sPostProcessEvent.invoke(Event{ep, 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";
      TDEBUG_FUNC_SI(5, sid) << "Calling schedules_->allAtLimit()";
      static std::mutex m;
      std::lock_guard sentry{m};
      if (schedule(sid).allAtLimit()) {
        // Set to return to the File level.
        nextLevel_ = highest_level();
      }
      // 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\n";
      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 285 of file EventProcessor.cc.

  {
    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 305 of file EventProcessor.cc.

  {
    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;
    } else if (nextLevel_.load() < L) {
      return false;
    } else 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 151 of file EventProcessor.h.

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

void art::EventProcessor::openInputFile ( )

private

Definition at line 542 of file EventProcessor.cc.

  {
    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 609 of file EventProcessor.cc.

  {
    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 595 of file EventProcessor.cc.

  {
    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 1510 of file EventProcessor.cc.

  {
    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 1079 of file EventProcessor.cc.

  {
    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 1125 of file EventProcessor.cc.

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

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

private

Definition at line 1399 of file EventProcessor.cc.

                                                            {
    // 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.
    if (error_action(e) != actions::IgnoreCompletely) {
      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\n";
    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 1150 of file EventProcessor.cc.

  {
    // 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(producedProductLookupTables_);
      actReg_.sPostSourceEvent.invoke(Event{*ep, invalid_module_context}, sc);
      FDEBUG(1) << string(8, ' ') << "readEvent...................("
                << ep->eventID() << ")\n";
      schedule(sid).accept_principal(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

void art::EventProcessor::readRun ( )

private

Definition at line 684 of file EventProcessor.cc.

  {
    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(producedProductLookupTables_);
    {
      Run const r{*runPrincipal_, invalid_module_context};
      actReg_.sPostSourceRun.invoke(r);
    }
    FDEBUG(1) << string(8, ' ') << "readRun.....................("
              << runPrincipal_->runID() << ")\n";
  }

void art::EventProcessor::readSubRun ( )

private

Definition at line 852 of file EventProcessor.cc.

  {
    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(
      producedProductLookupTables_);
    {
      SubRun const sr{*subRunPrincipal_, 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 99 of file EventProcessor.h.

    {}

void art::EventProcessor::respondToCloseInputFile ( )

private

Definition at line 654 of file EventProcessor.cc.

  {
    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 672 of file EventProcessor.cc.

  {
    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 645 of file EventProcessor.cc.

  {
    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 663 of file EventProcessor.cc.

  {
    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 1527 of file EventProcessor.cc.

  {
    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 145 of file EventProcessor.h.

    {
      return schedules_->at(id);
    }

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

private

Definition at line 625 of file EventProcessor.cc.

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

void art::EventProcessor::setRunAuxiliaryRangeSetID ( )

private

Definition at line 759 of file EventProcessor.cc.

  {
    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.
      RangeSet mergedRS;
      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 928 of file EventProcessor.cc.

  {
    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.
      RangeSet mergedRS;
      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 1572 of file EventProcessor.cc.

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

void art::EventProcessor::writeEvent ( )

private

void art::EventProcessor::writeRun ( )

private

Definition at line 837 of file EventProcessor.cc.

  {
    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 1063 of file EventProcessor.cc.

  {
    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 185 of file EventProcessor.h.

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

private

Definition at line 172 of file EventProcessor.h.

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

private

Definition at line 175 of file EventProcessor.h.

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

private

Definition at line 166 of file EventProcessor.h.

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

private

Definition at line 245 of file EventProcessor.h.

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

private

Definition at line 236 of file EventProcessor.h.

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

private

Definition at line 265 of file EventProcessor.h.

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

private

Definition at line 178 of file EventProcessor.h.

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

private

Definition at line 181 of file EventProcessor.h.

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

private

Definition at line 262 of file EventProcessor.h.

bool const art::EventProcessor::handleEmptyRuns_

private

Definition at line 248 of file EventProcessor.h.

bool const art::EventProcessor::handleEmptySubRuns_

private

Definition at line 251 of file EventProcessor.h.

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

private

Definition at line 230 of file EventProcessor.h.

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

private

Definition at line 188 of file EventProcessor.h.

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

private

Definition at line 163 of file EventProcessor.h.

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

private

Definition at line 193 of file EventProcessor.h.

tsan<PathManager> art::EventProcessor::pathManager_

private

Definition at line 227 of file EventProcessor.h.

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

private

Definition at line 199 of file EventProcessor.h.

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

private

Definition at line 207 of file EventProcessor.h.

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

private

Definition at line 209 of file EventProcessor.h.

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

private

Definition at line 239 of file EventProcessor.h.

ScheduleIteration art::EventProcessor::scheduleIteration_

private

Definition at line 220 of file EventProcessor.h.

tsan<Scheduler> art::EventProcessor::scheduler_

private

Definition at line 214 of file EventProcessor.h.

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

private

Definition at line 233 of file EventProcessor.h.

tsan_unique_ptr<ServicesManager> art::EventProcessor::servicesManager_

private

Definition at line 223 of file EventProcessor.h.

SharedException art::EventProcessor::sharedException_

private

Definition at line 255 of file EventProcessor.h.

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

private

Definition at line 218 of file EventProcessor.h.

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

private

Definition at line 242 of file EventProcessor.h.

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

private

Definition at line 216 of file EventProcessor.h.

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

private

Definition at line 169 of file EventProcessor.h.

---

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

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