edm::EventProcessor Class Reference

#include <EventProcessor.h>

Public Types
using	ProcessBlockType = PrincipalCache::ProcessBlockType
enum	StatusCode {   epSuccess = 0, epException = 1, epOther = 2, epSignal = 3,   epInputComplete = 4, epTimedOut = 5, epCountComplete = 6 }

Public Member Functions
void	beginJob ()
void	beginLumiAsync (IOVSyncValue const &, std::shared_ptr< RunProcessingStatus >, WaitingTaskHolder)
void	beginProcessBlock (bool &beginProcessBlockSucceeded)
void	beginRunAsync (IOVSyncValue const &, WaitingTaskHolder)
void	clearCounters ()
	Clears counters used by trigger report. More...
void	clearLumiPrincipal (LuminosityBlockProcessingStatus &)
void	clearRunPrincipal (RunProcessingStatus &)
void	closeInputFile (bool cleaningUpAfterException)
void	closeOutputFiles ()
void	continueLumiAsync (WaitingTaskHolder)
void	doErrorStuff ()
void	endJob ()
bool	endOfLoop ()
void	endProcessBlock (bool cleaningUpAfterException, bool beginProcessBlockSucceeded)
void	endRunAsync (std::shared_ptr< RunProcessingStatus >, WaitingTaskHolder)
void	endUnfinishedLumi (bool cleaningUpAfterException)
void	endUnfinishedRun (bool cleaningUpAfterException)
	EventProcessor (std::unique_ptr< ParameterSet > parameterSet, ServiceToken const &token=ServiceToken(), serviceregistry::ServiceLegacy=serviceregistry::kOverlapIsError, std::vector< std::string > const &defaultServices=std::vector< std::string >(), std::vector< std::string > const &forcedServices=std::vector< std::string >())
	EventProcessor (std::unique_ptr< ParameterSet > parameterSet, std::vector< std::string > const &defaultServices, std::vector< std::string > const &forcedServices=std::vector< std::string >())
	EventProcessor (std::shared_ptr< ProcessDesc > processDesc, ServiceToken const &token, serviceregistry::ServiceLegacy legacy)
	EventProcessor (EventProcessor const &)=delete
bool	fileBlockValid ()
std::vector< ModuleDescription const * >	getAllModuleDescriptions () const
ServiceToken	getToken ()
void	globalEndLumiAsync (WaitingTaskHolder, std::shared_ptr< LuminosityBlockProcessingStatus >)
void	globalEndRunAsync (WaitingTaskHolder, std::shared_ptr< RunProcessingStatus >)
void	handleEndLumiExceptions (std::exception_ptr, WaitingTaskHolder const &)
void	handleEndRunExceptions (std::exception_ptr, WaitingTaskHolder const &)
void	inputProcessBlocks ()
InputSource::ItemTypeInfo	lastTransitionType () const
InputSource::ItemTypeInfo	nextTransitionType ()
void	nextTransitionTypeAsync (std::shared_ptr< RunProcessingStatus > iRunStatus, WaitingTaskHolder nextTask)
void	openOutputFiles ()
EventProcessor &	operator= (EventProcessor const &)=delete
void	prepareForNextLoop ()
ProcessConfiguration const &	processConfiguration () const
InputSource::ItemType	processRuns ()
void	readAndMergeLumi (LuminosityBlockProcessingStatus &)
void	readAndMergeRun (RunProcessingStatus &)
void	readFile ()
std::shared_ptr< LuminosityBlockPrincipal >	readLuminosityBlock (std::shared_ptr< RunPrincipal > rp)
void	readProcessBlock (ProcessBlockPrincipal &)
std::shared_ptr< RunPrincipal >	readRun ()
void	releaseBeginRunResources (unsigned int iStream)
void	respondToCloseInputFile ()
void	respondToOpenInputFile ()
void	rewindInput ()
StatusCode	run ()
StatusCode	runToCompletion ()
bool	setDeferredException (std::exception_ptr)
void	setExceptionMessageFiles (std::string &message)
void	setExceptionMessageLumis ()
void	setExceptionMessageRuns ()
bool	shouldWeCloseOutput () const
bool	shouldWeStop () const
void	startingNewLoop ()
void	streamBeginRunAsync (unsigned int iStream, std::shared_ptr< RunProcessingStatus >, bool precedingTasksSucceeded, WaitingTaskHolder)
void	streamEndLumiAsync (WaitingTaskHolder, unsigned int iStreamIndex)
void	streamEndRunAsync (WaitingTaskHolder, unsigned int iStreamIndex)
void	taskCleanup ()
int	totalEvents () const
int	totalEventsFailed () const
int	totalEventsPassed () const
void	writeLumiAsync (WaitingTaskHolder, LuminosityBlockPrincipal &)
void	writeProcessBlockAsync (WaitingTaskHolder, ProcessBlockType)
void	writeRunAsync (WaitingTaskHolder, RunPrincipal const &, MergeableRunProductMetadata const *)
	~EventProcessor ()

Private Types
typedef std::set< std::pair< std::string, std::string > >	ExcludedData
typedef std::map< std::string, ExcludedData >	ExcludedDataMap

Private Member Functions
std::shared_ptr< BranchIDListHelper const >	branchIDListHelper () const
std::shared_ptr< BranchIDListHelper > &	branchIDListHelper ()
bool	checkForAsyncStopRequest (StatusCode &)
void	handleNextEventForStreamAsync (WaitingTaskHolder, unsigned int iStreamIndex)
void	handleNextItemAfterMergingRunEntries (std::shared_ptr< RunProcessingStatus >, WaitingTaskHolder)
void	init (std::shared_ptr< ProcessDesc > &processDesc, ServiceToken const &token, serviceregistry::ServiceLegacy)
std::shared_ptr< EDLooperBase const >	looper () const
std::shared_ptr< EDLooperBase > &	looper ()
bool	needToCallNext () const
std::shared_ptr< ProductRegistry const >	preg () const
std::shared_ptr< ProductRegistry > &	preg ()
void	processEventAsync (WaitingTaskHolder iHolder, unsigned int iStreamIndex)
void	processEventAsyncImpl (WaitingTaskHolder iHolder, unsigned int iStreamIndex)
void	processEventWithLooper (EventPrincipal &, unsigned int iStreamIndex)
void	readAndMergeLumiEntriesAsync (std::shared_ptr< LuminosityBlockProcessingStatus >, WaitingTaskHolder)
void	readAndMergeRunEntriesAsync (std::shared_ptr< RunProcessingStatus >, WaitingTaskHolder)
void	readEvent (unsigned int iStreamIndex)
bool	readNextEventForStream (WaitingTaskHolder const &, unsigned int iStreamIndex, LuminosityBlockProcessingStatus &)
void	setNeedToCallNext (bool val)
std::shared_ptr< ThinnedAssociationsHelper const >	thinnedAssociationsHelper () const
std::shared_ptr< ThinnedAssociationsHelper > &	thinnedAssociationsHelper ()
void	throwAboutModulesRequiringLuminosityBlockSynchronization () const
void	warnAboutModulesRequiringRunSynchronization () const

Private Attributes
std::unique_ptr< ExceptionToActionTable const >	act_table_
std::shared_ptr< ActivityRegistry >	actReg_
bool	beginJobCalled_
std::vector< std::string >	branchesToDeleteEarly_
edm::propagate_const< std::shared_ptr< BranchIDListHelper > >	branchIDListHelper_
std::exception_ptr	deferredExceptionPtr_
std::atomic< bool >	deferredExceptionPtrIsSet_
bool	deleteNonConsumedUnscheduledModules_ = true
edm::propagate_const< std::shared_ptr< eventsetup::EventSetupProvider > >	esp_
edm::propagate_const< std::unique_ptr< eventsetup::EventSetupsController > >	espController_
ExcludedDataMap	eventSetupDataToExcludeFromPrefetching_
std::string	exceptionMessageFiles_
std::atomic< bool >	exceptionMessageLumis_
std::atomic< bool >	exceptionMessageRuns_
std::shared_ptr< RunProcessingStatus >	exceptionRunStatus_
edm::propagate_const< std::shared_ptr< FileBlock > >	fb_
bool	fileModeNoMerge_
bool	firstEventInBlock_ = true
bool	forceESCacheClearOnNewRun_
bool	forceLooperToEnd_
edm::propagate_const< std::unique_ptr< HistoryAppender > >	historyAppender_
edm::propagate_const< std::unique_ptr< InputSource > >	input_
InputSource::ItemTypeInfo	lastSourceTransition_
edm::propagate_const< std::shared_ptr< EDLooperBase > >	looper_
bool	looperBeginJobRun_
std::unique_ptr< edm::LimitedTaskQueue >	lumiQueue_
MergeableRunProductProcesses	mergeableRunProductProcesses_
std::vector< std::string >	modulesToIgnoreForDeleteEarly_
edm::propagate_const< std::unique_ptr< ModuleTypeResolverMaker const > >	moduleTypeResolverMaker_
bool	needToCallNext_ = true
PathsAndConsumesOfModules	pathsAndConsumesOfModules_
PreallocationConfiguration	preallocations_
edm::propagate_const< std::shared_ptr< ProductRegistry > >	preg_
PrincipalCache	principalCache_
bool	printDependencies_ = false
edm::propagate_const< std::shared_ptr< ProcessBlockHelper > >	processBlockHelper_
std::shared_ptr< ProcessConfiguration const >	processConfiguration_
ProcessContext	processContext_
edm::SerialTaskQueue	queueWhichWaitsForIOVsToFinish_
std::multimap< std::string, std::string >	referencesToBranches_
std::unique_ptr< edm::LimitedTaskQueue >	runQueue_
edm::propagate_const< std::unique_ptr< Schedule > >	schedule_
ServiceToken	serviceToken_
bool	shouldWeStop_
std::shared_ptr< std::recursive_mutex >	sourceMutex_
SharedResourcesAcquirer	sourceResourcesAcquirer_
std::atomic< unsigned int >	streamLumiActive_ {0}
std::vector< std::shared_ptr< LuminosityBlockProcessingStatus > >	streamLumiStatus_
std::vector< edm::SerialTaskQueue >	streamQueues_
SerialTaskQueue	streamQueuesInserter_
std::atomic< unsigned int >	streamRunActive_ {0}
std::vector< std::shared_ptr< RunProcessingStatus > >	streamRunStatus_
std::vector< SubProcess >	subProcesses_
oneapi::tbb::task_group	taskGroup_
edm::propagate_const< std::shared_ptr< ThinnedAssociationsHelper > >	thinnedAssociationsHelper_

Detailed Description

Definition at line 69 of file EventProcessor.h.

Member Typedef Documentation

ExcludedData

typedef std::set<std::pair<std::string, std::string> > edm::EventProcessor::ExcludedData

private

Definition at line 370 of file EventProcessor.h.

ExcludedDataMap

typedef std::map<std::string, ExcludedData> edm::EventProcessor::ExcludedDataMap

private

Definition at line 371 of file EventProcessor.h.

ProcessBlockType

using edm::EventProcessor::ProcessBlockType = PrincipalCache::ProcessBlockType

Definition at line 236 of file EventProcessor.h.

Member Enumeration Documentation

StatusCode

enum edm::EventProcessor::StatusCode

Enumerator
epSuccess
epException
epOther
epSignal
epInputComplete
epTimedOut
epCountComplete

Definition at line 79 of file EventProcessor.h.

                    {
      epSuccess = 0,
      epException = 1,
      epOther = 2,
      epSignal = 3,
      epInputComplete = 4,
      epTimedOut = 5,
      epCountComplete = 6
    };

Constructor & Destructor Documentation

EventProcessor() [1/4]

edm::EventProcessor::EventProcessor ( std::unique_ptr< ParameterSet >  parameterSet, ServiceToken const &  token = ServiceToken(), serviceregistry::ServiceLegacy  iLegacy = serviceregistry::kOverlapIsError, std::vector< std::string > const &  defaultServices = std::vector<std::string>(), std::vector< std::string > const &  forcedServices = std::vector<std::string>()  )

explicit

Definition at line 214 of file EventProcessor.cc.

References init(), eostools::move(), and edm::parameterSet().

      : actReg_(),
        preg_(),
        branchIDListHelper_(),
        serviceToken_(),
        input_(),
        moduleTypeResolverMaker_(makeModuleTypeResolverMaker(*parameterSet)),
        espController_(std::make_unique<eventsetup::EventSetupsController>(moduleTypeResolverMaker_.get())),
        esp_(),
        act_table_(),
        processConfiguration_(),
        schedule_(),
        subProcesses_(),
        historyAppender_(new HistoryAppender),
        fb_(),
        looper_(),
        deferredExceptionPtrIsSet_(false),
        sourceResourcesAcquirer_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().first),
        sourceMutex_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().second),
        principalCache_(),
        beginJobCalled_(false),
        shouldWeStop_(false),
        fileModeNoMerge_(false),
        exceptionMessageFiles_(),
        exceptionMessageRuns_(false),
        exceptionMessageLumis_(false),
        forceLooperToEnd_(false),
        looperBeginJobRun_(false),
        forceESCacheClearOnNewRun_(false),
        eventSetupDataToExcludeFromPrefetching_() {
    auto processDesc = std::make_shared<ProcessDesc>(std::move(parameterSet));
    processDesc->addServices(defaultServices, forcedServices);
    init(processDesc, iToken, iLegacy);
  }

EventProcessor() [2/4]

edm::EventProcessor::EventProcessor	(	std::unique_ptr< ParameterSet >	parameterSet,
		std::vector< std::string > const &	defaultServices,
		std::vector< std::string > const &	forcedServices = std::vector<std::string>()
	)

Definition at line 253 of file EventProcessor.cc.

References init(), edm::serviceregistry::kOverlapIsError, eostools::move(), and edm::parameterSet().

      : actReg_(),
        preg_(),
        branchIDListHelper_(),
        serviceToken_(),
        input_(),
        moduleTypeResolverMaker_(makeModuleTypeResolverMaker(*parameterSet)),
        espController_(std::make_unique<eventsetup::EventSetupsController>(moduleTypeResolverMaker_.get())),
        esp_(),
        act_table_(),
        processConfiguration_(),
        schedule_(),
        subProcesses_(),
        historyAppender_(new HistoryAppender),
        fb_(),
        looper_(),
        deferredExceptionPtrIsSet_(false),
        sourceResourcesAcquirer_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().first),
        sourceMutex_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().second),
        principalCache_(),
        beginJobCalled_(false),
        shouldWeStop_(false),
        fileModeNoMerge_(false),
        exceptionMessageFiles_(),
        exceptionMessageRuns_(false),
        exceptionMessageLumis_(false),
        forceLooperToEnd_(false),
        looperBeginJobRun_(false),
        forceESCacheClearOnNewRun_(false),
        eventSetupDataToExcludeFromPrefetching_() {
    auto processDesc = std::make_shared<ProcessDesc>(std::move(parameterSet));
    processDesc->addServices(defaultServices, forcedServices);
    init(processDesc, ServiceToken(), serviceregistry::kOverlapIsError);
  }

EventProcessor() [3/4]

edm::EventProcessor::EventProcessor	(	std::shared_ptr< ProcessDesc >	processDesc,
		ServiceToken const &	token,
		serviceregistry::ServiceLegacy	legacy
	)

Definition at line 290 of file EventProcessor.cc.

References init(), and unpackBuffers-CaloStage2::token.

      : actReg_(),
        preg_(),
        branchIDListHelper_(),
        serviceToken_(),
        input_(),
        moduleTypeResolverMaker_(makeModuleTypeResolverMaker(*processDesc->getProcessPSet())),
        espController_(std::make_unique<eventsetup::EventSetupsController>(moduleTypeResolverMaker_.get())),
        esp_(),
        act_table_(),
        processConfiguration_(),
        schedule_(),
        subProcesses_(),
        historyAppender_(new HistoryAppender),
        fb_(),
        looper_(),
        deferredExceptionPtrIsSet_(false),
        sourceResourcesAcquirer_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().first),
        sourceMutex_(SharedResourcesRegistry::instance()->createAcquirerForSourceDelayedReader().second),
        principalCache_(),
        beginJobCalled_(false),
        shouldWeStop_(false),
        fileModeNoMerge_(false),
        exceptionMessageFiles_(),
        exceptionMessageRuns_(false),
        exceptionMessageLumis_(false),
        forceLooperToEnd_(false),
        looperBeginJobRun_(false),
        forceESCacheClearOnNewRun_(false),
        eventSetupDataToExcludeFromPrefetching_() {
    init(processDesc, token, legacy);
  }

~EventProcessor()

edm::EventProcessor::~EventProcessor

(

)

Definition at line 607 of file EventProcessor.cc.

References actReg_, edm::ParentageRegistry::clear(), edm::pset::Registry::clear(), esp_, espController_, getToken(), input_, edm::ParentageRegistry::instance(), edm::pset::Registry::instance(), looper_, findAndChange::op, schedule_, and unpackBuffers-CaloStage2::token.

                                  {
    // Make the services available while everything is being deleted.
    ServiceToken token = getToken();
    ServiceRegistry::Operate op(token);

    // manually destroy all these thing that may need the services around
    // propagate_const<T> has no reset() function
    espController_ = nullptr;
    esp_ = nullptr;
    schedule_ = nullptr;
    input_ = nullptr;
    looper_ = nullptr;
    actReg_ = nullptr;

    pset::Registry::instance()->clear();
    ParentageRegistry::instance()->clear();
  }

EventProcessor() [4/4]

edm::EventProcessor::EventProcessor ( EventProcessor const &  )

delete

Member Function Documentation

beginJob()

void edm::EventProcessor::beginJob

(

void

)

This should be called before the first call to 'run' If this is not called in time, it will automatically be called the first time 'run' is called

Definition at line 632 of file EventProcessor.cc.

References actReg_, cms::Exception::addContext(), bk::beginJob(), beginJobCalled_, branchesToDeleteEarly_, HltBtagPostValidation_cff::c, edm::checkForModuleDependencyCorrectness(), ALPAKA_ACCELERATOR_NAMESPACE::brokenline::constexpr(), deleteNonConsumedUnscheduledModules_, makeListRunsInFiles::description, esp_, espController_, edm::first(), edm::for_all(), watchdog::group, mps_fire::i, edm::waiting_task::chain::ifThen(), edm::InEvent, edm::PathsAndConsumesOfModules::initialize(), edm::InLumi, edm::InProcess, input_, edm::InRun, MainPageGenerator::l, dqmdumpme::last, edm::waiting_task::chain::lastTask(), looper_, merge(), modulesToIgnoreForDeleteEarly_, eostools::move(), edm::nonConsumedUnscheduledModules(), edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfRuns(), edm::PreallocationConfiguration::numberOfStreams(), edm::PreallocationConfiguration::numberOfThreads(), pathsAndConsumesOfModules_, preallocations_, preg(), preg_, printDependencies_, processBlockHelper_, processConfiguration_, processContext_, referencesToBranches_, edm::PathsAndConsumesOfModules::removeModules(), edm::waiting_task::chain::runLast(), schedule_, serviceToken_, subProcesses_, edm::swap(), throwAboutModulesRequiringLuminosityBlockSynchronization(), createJobs::tmp, warnAboutModulesRequiringRunSynchronization(), and edm::convertException::wrap().

Referenced by runToCompletion().

                                {
    if (beginJobCalled_)
      return;
    beginJobCalled_ = true;
    bk::beginJob();

    // StateSentry toerror(this); // should we add this ?
    //make the services available
    ServiceRegistry::Operate operate(serviceToken_);

    service::SystemBounds bounds(preallocations_.numberOfStreams(),
                                 preallocations_.numberOfLuminosityBlocks(),
                                 preallocations_.numberOfRuns(),
                                 preallocations_.numberOfThreads());
    actReg_->preallocateSignal_(bounds);
    schedule_->convertCurrentProcessAlias(processConfiguration_->processName());
    pathsAndConsumesOfModules_.initialize(schedule_.get(), preg());

    std::vector<ModuleProcessName> consumedBySubProcesses;
    for_all(subProcesses_,
            [&consumedBySubProcesses, deleteModules = deleteNonConsumedUnscheduledModules_](auto& subProcess) {
              auto c = subProcess.keepOnlyConsumedUnscheduledModules(deleteModules);
              if (consumedBySubProcesses.empty()) {
                consumedBySubProcesses = std::move(c);
              } else if (not c.empty()) {
                std::vector<ModuleProcessName> tmp;
                tmp.reserve(consumedBySubProcesses.size() + c.size());
                std::merge(consumedBySubProcesses.begin(),
                           consumedBySubProcesses.end(),
                           c.begin(),
                           c.end(),
                           std::back_inserter(tmp));
                std::swap(consumedBySubProcesses, tmp);
              }
            });

    // Note: all these may throw
    checkForModuleDependencyCorrectness(pathsAndConsumesOfModules_, printDependencies_);
    if (deleteNonConsumedUnscheduledModules_) {
      if (auto const unusedModules = nonConsumedUnscheduledModules(pathsAndConsumesOfModules_, consumedBySubProcesses);
          not unusedModules.empty()) {
        pathsAndConsumesOfModules_.removeModules(unusedModules);

        edm::LogInfo("DeleteModules").log([&unusedModules](auto& l) {
          l << "Following modules are not in any Path or EndPath, nor is their output consumed by any other module, "
               "and "
               "therefore they are deleted before beginJob transition.";
          for (auto const& description : unusedModules) {
            l << "\n " << description->moduleLabel();
          }
        });
        for (auto const& description : unusedModules) {
          schedule_->deleteModule(description->moduleLabel(), actReg_.get());
        }
      }
    }
    // Initialize after the deletion of non-consumed unscheduled
    // modules to avoid non-consumed non-run modules to keep the
    // products unnecessarily alive
    if (not branchesToDeleteEarly_.empty()) {
      auto modulesToSkip = std::move(modulesToIgnoreForDeleteEarly_);
      auto branchesToDeleteEarly = std::move(branchesToDeleteEarly_);
      auto referencesToBranches = std::move(referencesToBranches_);
      schedule_->initializeEarlyDelete(branchesToDeleteEarly, referencesToBranches, modulesToSkip, *preg_);
    }

    if (preallocations_.numberOfLuminosityBlocks() > 1) {
      throwAboutModulesRequiringLuminosityBlockSynchronization();
    }
    if (preallocations_.numberOfRuns() > 1) {
      warnAboutModulesRequiringRunSynchronization();
    }

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

    //NOTE: in future we should have a beginOfJob for looper that takes no arguments
    //  For now we delay calling beginOfJob until first beginOfRun
    //if(looper_) {
    //   looper_->beginOfJob(es);
    //}
    espController_->finishConfiguration();
    actReg_->eventSetupConfigurationSignal_(esp_->recordsToResolverIndices(), processContext_);
    actReg_->preBeginJobSignal_(pathsAndConsumesOfModules_, processContext_);
    try {
      convertException::wrap([&]() { input_->doBeginJob(); });
    } catch (cms::Exception& ex) {
      ex.addContext("Calling beginJob for the source");
      throw;
    }

    schedule_->beginJob(*preg_, esp_->recordsToResolverIndices(), *processBlockHelper_);
    if (looper_) {
      constexpr bool mustPrefetchMayGet = true;
      auto const processBlockLookup = preg_->productLookup(InProcess);
      auto const runLookup = preg_->productLookup(InRun);
      auto const lumiLookup = preg_->productLookup(InLumi);
      auto const eventLookup = preg_->productLookup(InEvent);
      looper_->updateLookup(InProcess, *processBlockLookup, mustPrefetchMayGet);
      looper_->updateLookup(InRun, *runLookup, mustPrefetchMayGet);
      looper_->updateLookup(InLumi, *lumiLookup, mustPrefetchMayGet);
      looper_->updateLookup(InEvent, *eventLookup, mustPrefetchMayGet);
      looper_->updateLookup(esp_->recordsToResolverIndices());
    }
    // toerror.succeeded(); // should we add this?
    for_all(subProcesses_, [](auto& subProcess) { subProcess.doBeginJob(); });
    actReg_->postBeginJobSignal_();

    oneapi::tbb::task_group group;
    FinalWaitingTask last{group};
    using namespace edm::waiting_task::chain;
    first([this](auto nextTask) {
      for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
        first([i, this](auto nextTask) {
          ServiceRegistry::Operate operate(serviceToken_);
          schedule_->beginStream(i);
        }) | ifThen(not subProcesses_.empty(), [this, i](auto nextTask) {
          ServiceRegistry::Operate operate(serviceToken_);
          for_all(subProcesses_, [i](auto& subProcess) { subProcess.doBeginStream(i); });
        }) | lastTask(nextTask);
      }
    }) | runLast(WaitingTaskHolder(group, &last));
    last.wait();
  }

beginLumiAsync()

void edm::EventProcessor::beginLumiAsync	(	IOVSyncValue const &	iSync,
		std::shared_ptr< RunProcessingStatus >	iRunStatus,
		edm::WaitingTaskHolder	iHolder
	)

Definition at line 1666 of file EventProcessor.cc.

References actReg_, edm::BeginLuminosityBlock, CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), endRunAsync(), esp_, espController_, edm::PrincipalCache::eventPrincipal(), dqmdumpme::first, handleNextEventForStreamAsync(), mps_fire::i, edm::waiting_task::chain::ifThen(), input_, edm::Service< T >::isAvailable(), edm::InputSource::LastItemToBeMerged, lastTransitionType(), looper_, lumiQueue_, eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, edm::SerialTaskQueueChain::push(), edm::SerialTaskQueue::push(), queueWhichWaitsForIOVsToFinish_, readAndMergeLumiEntriesAsync(), readLuminosityBlock(), edm::SerialTaskQueue::resume(), edm::waiting_task::chain::runLast(), schedule_, edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, setNeedToCallNext(), sourceResourcesAcquirer_, mps_update::status, streamLumiActive_, streamLumiStatus_, streamQueues_, streamQueuesInserter_, subProcesses_, and edm::waiting_task::chain::then().

Referenced by handleNextEventForStreamAsync(), and handleNextItemAfterMergingRunEntries().

                                                                    {
    actReg_->esSyncIOVQueuingSignal_.emit(iSync);

    auto status = std::make_shared<LuminosityBlockProcessingStatus>();
    chain::first([this, &iSync, &status](auto nextTask) {
      espController_->runOrQueueEventSetupForInstanceAsync(iSync,
                                                           nextTask,
                                                           status->endIOVWaitingTasks(),
                                                           status->eventSetupImpls(),
                                                           queueWhichWaitsForIOVsToFinish_,
                                                           actReg_.get(),
                                                           serviceToken_);
    }) | chain::then([this, status, iRunStatus](std::exception_ptr const* iException, auto nextTask) {
      CMS_SA_ALLOW try {
        //the call to doneWaiting will cause the count to decrement
        if (iException) {
          WaitingTaskHolder copyHolder(nextTask);
          copyHolder.doneWaiting(*iException);
        }

        lumiQueue_->pushAndPause(
            *nextTask.group(),
            [this, postLumiQueueTask = nextTask, status, iRunStatus](edm::LimitedTaskQueue::Resumer iResumer) mutable {
              CMS_SA_ALLOW try {
                if (postLumiQueueTask.taskHasFailed()) {
                  status->resetResources();
                  queueWhichWaitsForIOVsToFinish_.resume();
                  endRunAsync(iRunStatus, postLumiQueueTask);
                  return;
                }

                status->setResumer(std::move(iResumer));

                sourceResourcesAcquirer_.serialQueueChain().push(
                    *postLumiQueueTask.group(),
                    [this, postSourceTask = postLumiQueueTask, status, iRunStatus]() mutable {
                      CMS_SA_ALLOW try {
                        ServiceRegistry::Operate operate(serviceToken_);

                        if (postSourceTask.taskHasFailed()) {
                          status->resetResources();
                          queueWhichWaitsForIOVsToFinish_.resume();
                          endRunAsync(iRunStatus, postSourceTask);
                          return;
                        }

                        status->setLumiPrincipal(readLuminosityBlock(iRunStatus->runPrincipal()));

                        LuminosityBlockPrincipal& lumiPrincipal = *status->lumiPrincipal();
                        {
                          SendSourceTerminationSignalIfException sentry(actReg_.get());
                          input_->doBeginLumi(lumiPrincipal, &processContext_);
                          sentry.completedSuccessfully();
                        }

                        Service<RandomNumberGenerator> rng;
                        if (rng.isAvailable()) {
                          LuminosityBlock lb(lumiPrincipal, ModuleDescription(), nullptr, false);
                          rng->preBeginLumi(lb);
                        }

                        EventSetupImpl const& es = status->eventSetupImpl(esp_->subProcessIndex());

                        using namespace edm::waiting_task::chain;
                        chain::first([this, status](auto nextTask) mutable {
                          if (lastTransitionType().itemPosition() != InputSource::ItemPosition::LastItemToBeMerged) {
                            readAndMergeLumiEntriesAsync(std::move(status), std::move(nextTask));
                          } else {
                            setNeedToCallNext(true);
                          }
                        }) | then([this, status, &es, &lumiPrincipal](auto nextTask) {
                          LumiTransitionInfo transitionInfo(lumiPrincipal, es, &status->eventSetupImpls());
                          using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin>;
                          beginGlobalTransitionAsync<Traits>(
                              nextTask, *schedule_, transitionInfo, serviceToken_, subProcesses_);
                        }) | ifThen(looper_, [this, status, &es](auto nextTask) {
                          looper_->prefetchAsync(
                              nextTask, serviceToken_, Transition::BeginLuminosityBlock, *(status->lumiPrincipal()), es);
                        }) | ifThen(looper_, [this, status, &es](auto nextTask) {
                          status->globalBeginDidSucceed();
                          ServiceRegistry::Operate operateLooper(serviceToken_);
                          looper_->doBeginLuminosityBlock(*(status->lumiPrincipal()), es, &processContext_);
                        }) | then([this, status, iRunStatus](std::exception_ptr const* iException, auto holder) mutable {
                          if (iException) {
                            status->resetResources();
                            queueWhichWaitsForIOVsToFinish_.resume();
                            WaitingTaskHolder copyHolder(holder);
                            copyHolder.doneWaiting(*iException);
                            endRunAsync(iRunStatus, holder);
                          } else {
                            if (not looper_) {
                              status->globalBeginDidSucceed();
                            }

                            status->setGlobalEndRunHolder(iRunStatus->globalEndRunHolder());

                            EventSetupImpl const& es = status->eventSetupImpl(esp_->subProcessIndex());
                            using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin>;

                            streamQueuesInserter_.push(*holder.group(), [this, status, holder, &es]() mutable {
                              for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
                                streamQueues_[i].push(*holder.group(), [this, i, status, holder, &es]() mutable {
                                  if (!status->shouldStreamStartLumi()) {
                                    return;
                                  }
                                  streamQueues_[i].pause();

                                  auto& event = principalCache_.eventPrincipal(i);
                                  auto eventSetupImpls = &status->eventSetupImpls();
                                  auto lp = status->lumiPrincipal().get();
                                  streamLumiStatus_[i] = std::move(status);
                                  ++streamLumiActive_;
                                  event.setLuminosityBlockPrincipal(lp);
                                  LumiTransitionInfo transitionInfo(*lp, es, eventSetupImpls);
                                  using namespace edm::waiting_task::chain;
                                  chain::first([this, i, &transitionInfo](auto nextTask) {
                                    beginStreamTransitionAsync<Traits>(std::move(nextTask),
                                                                       *schedule_,
                                                                       i,
                                                                       transitionInfo,
                                                                       serviceToken_,
                                                                       subProcesses_);
                                  }) |
                                      then([this, i](std::exception_ptr const* exceptionFromBeginStreamLumi,
                                                     auto nextTask) {
                                        if (exceptionFromBeginStreamLumi) {
                                          WaitingTaskHolder copyHolder(nextTask);
                                          copyHolder.doneWaiting(*exceptionFromBeginStreamLumi);
                                        }
                                        handleNextEventForStreamAsync(std::move(nextTask), i);
                                      }) |
                                      runLast(std::move(holder));
                                });
                              }  // end for loop over streams
                            });
                          }
                        }) | runLast(postSourceTask);
                      } catch (...) {
                        status->resetResources();
                        queueWhichWaitsForIOVsToFinish_.resume();
                        WaitingTaskHolder copyHolder(postSourceTask);
                        copyHolder.doneWaiting(std::current_exception());
                        endRunAsync(iRunStatus, postSourceTask);
                      }
                    });  // task in sourceResourcesAcquirer
              } catch (...) {
                status->resetResources();
                queueWhichWaitsForIOVsToFinish_.resume();
                WaitingTaskHolder copyHolder(postLumiQueueTask);
                copyHolder.doneWaiting(std::current_exception());
                endRunAsync(iRunStatus, postLumiQueueTask);
              }
            });  // task in lumiQueue
      } catch (...) {
        status->resetResources();
        queueWhichWaitsForIOVsToFinish_.resume();
        WaitingTaskHolder copyHolder(nextTask);
        copyHolder.doneWaiting(std::current_exception());
        endRunAsync(iRunStatus, nextTask);
      }
    }) | chain::runLast(std::move(iHolder));
  }

beginProcessBlock()

void edm::EventProcessor::beginProcessBlock

(

bool &

beginProcessBlockSucceeded

)

Definition at line 1108 of file EventProcessor.cc.

References edm::ProcessBlockPrincipal::fillProcessBlockPrincipal(), principalCache_, edm::PrincipalCache::processBlockPrincipal(), processConfiguration_, schedule_, serviceToken_, subProcesses_, and taskGroup_.

                                                                         {
    ProcessBlockPrincipal& processBlockPrincipal = principalCache_.processBlockPrincipal();
    processBlockPrincipal.fillProcessBlockPrincipal(processConfiguration_->processName());

    using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionGlobalBegin>;
    FinalWaitingTask globalWaitTask{taskGroup_};

    ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
    beginGlobalTransitionAsync<Traits>(
        WaitingTaskHolder(taskGroup_, &globalWaitTask), *schedule_, transitionInfo, serviceToken_, subProcesses_);

    globalWaitTask.wait();
    beginProcessBlockSucceeded = true;
  }

beginRunAsync()

void edm::EventProcessor::beginRunAsync	(	IOVSyncValue const &	iSync,
		WaitingTaskHolder	iHolder
	)

Definition at line 1211 of file EventProcessor.cc.

References actReg_, edm::BeginRun, CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), esp_, espController_, exceptionRunStatus_, dqmdumpme::first, forceESCacheClearOnNewRun_, globalEndRunAsync(), edm::WaitingTaskHolder::group(), watchdog::group, handleNextItemAfterMergingRunEntries(), mps_fire::i, edm::waiting_task::chain::ifThen(), input_, edm::InputSource::LastItemToBeMerged, lastTransitionType(), looper_, looperBeginJobRun_, edm::make_waiting_task(), eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, processContext_, edm::SerialTaskQueueChain::push(), edm::SerialTaskQueue::push(), queueWhichWaitsForIOVsToFinish_, readAndMergeRunEntriesAsync(), readRun(), edm::SerialTaskQueue::resume(), edm::waiting_task::chain::runLast(), runQueue_, schedule_, edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, setNeedToCallNext(), sourceResourcesAcquirer_, mps_update::status, streamBeginRunAsync(), streamQueues_, streamQueuesInserter_, subProcesses_, edm::WaitingTaskHolder::taskHasFailed(), and edm::waiting_task::chain::then().

Referenced by endRunAsync(), and processRuns().

                                                                                         {
    if (iHolder.taskHasFailed()) {
      return;
    }

    actReg_->esSyncIOVQueuingSignal_.emit(iSync);

    auto status = std::make_shared<RunProcessingStatus>(preallocations_.numberOfStreams(), iHolder);

    chain::first([this, &status, iSync](auto nextTask) {
      espController_->runOrQueueEventSetupForInstanceAsync(iSync,
                                                           nextTask,
                                                           status->endIOVWaitingTasks(),
                                                           status->eventSetupImpls(),
                                                           queueWhichWaitsForIOVsToFinish_,
                                                           actReg_.get(),
                                                           serviceToken_,
                                                           forceESCacheClearOnNewRun_);
    }) | chain::then([this, status](std::exception_ptr const* iException, auto nextTask) {
      CMS_SA_ALLOW try {
        if (iException) {
          WaitingTaskHolder copyHolder(nextTask);
          copyHolder.doneWaiting(*iException);
          // Finish handling the exception in the task pushed to runQueue_
        }
        ServiceRegistry::Operate operate(serviceToken_);

        runQueue_->pushAndPause(
            *nextTask.group(),
            [this, postRunQueueTask = nextTask, status](edm::LimitedTaskQueue::Resumer iResumer) mutable {
              CMS_SA_ALLOW try {
                if (postRunQueueTask.taskHasFailed()) {
                  status->resetBeginResources();
                  queueWhichWaitsForIOVsToFinish_.resume();
                  return;
                }

                status->setResumer(std::move(iResumer));

                sourceResourcesAcquirer_.serialQueueChain().push(
                    *postRunQueueTask.group(), [this, postSourceTask = postRunQueueTask, status]() mutable {
                      CMS_SA_ALLOW try {
                        ServiceRegistry::Operate operate(serviceToken_);

                        if (postSourceTask.taskHasFailed()) {
                          status->resetBeginResources();
                          queueWhichWaitsForIOVsToFinish_.resume();
                          status->resumeGlobalRunQueue();
                          return;
                        }

                        status->setRunPrincipal(readRun());

                        RunPrincipal& runPrincipal = *status->runPrincipal();
                        {
                          SendSourceTerminationSignalIfException sentry(actReg_.get());
                          input_->doBeginRun(runPrincipal, &processContext_);
                          sentry.completedSuccessfully();
                        }

                        EventSetupImpl const& es = status->eventSetupImpl(esp_->subProcessIndex());
                        if (looper_ && looperBeginJobRun_ == false) {
                          looper_->copyInfo(ScheduleInfo(schedule_.get()));

                          oneapi::tbb::task_group group;
                          FinalWaitingTask waitTask{group};
                          using namespace edm::waiting_task::chain;
                          chain::first([this, &es](auto nextTask) {
                            looper_->esPrefetchAsync(nextTask, es, Transition::BeginRun, serviceToken_);
                          }) | then([this, &es](auto nextTask) mutable {
                            looper_->beginOfJob(es);
                            looperBeginJobRun_ = true;
                            looper_->doStartingNewLoop();
                          }) | runLast(WaitingTaskHolder(group, &waitTask));
                          waitTask.wait();
                        }

                        using namespace edm::waiting_task::chain;
                        chain::first([this, status](auto nextTask) mutable {
                          CMS_SA_ALLOW try {
                            if (lastTransitionType().itemPosition() != InputSource::ItemPosition::LastItemToBeMerged) {
                              readAndMergeRunEntriesAsync(status, nextTask);
                            } else {
                              setNeedToCallNext(true);
                            }
                          } catch (...) {
                            status->setStopBeforeProcessingRun(true);
                            nextTask.doneWaiting(std::current_exception());
                          }
                        }) | then([this, status, &es](auto nextTask) {
                          if (status->stopBeforeProcessingRun()) {
                            return;
                          }
                          RunTransitionInfo transitionInfo(*status->runPrincipal(), es, &status->eventSetupImpls());
                          using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin>;
                          beginGlobalTransitionAsync<Traits>(
                              nextTask, *schedule_, transitionInfo, serviceToken_, subProcesses_);
                        }) | then([status](auto nextTask) mutable {
                          if (status->stopBeforeProcessingRun()) {
                            return;
                          }
                          status->globalBeginDidSucceed();
                        }) | ifThen(looper_, [this, status, &es](auto nextTask) {
                          if (status->stopBeforeProcessingRun()) {
                            return;
                          }
                          looper_->prefetchAsync(
                              nextTask, serviceToken_, Transition::BeginRun, *status->runPrincipal(), es);
                        }) | ifThen(looper_, [this, status, &es](auto nextTask) {
                          if (status->stopBeforeProcessingRun()) {
                            return;
                          }
                          ServiceRegistry::Operate operateLooper(serviceToken_);
                          looper_->doBeginRun(*status->runPrincipal(), es, &processContext_);
                        }) | then([this, status](std::exception_ptr const* iException, auto holder) mutable {
                          bool precedingTasksSucceeded = true;
                          if (iException) {
                            precedingTasksSucceeded = false;
                            WaitingTaskHolder copyHolder(holder);
                            copyHolder.doneWaiting(*iException);
                          }

                          if (status->stopBeforeProcessingRun()) {
                            // We just quit now if there was a failure when merging runs
                            status->resetBeginResources();
                            queueWhichWaitsForIOVsToFinish_.resume();
                            status->resumeGlobalRunQueue();
                            return;
                          }
                          CMS_SA_ALLOW try {
                            // Under normal circumstances, this task runs after endRun has completed for all streams
                            // and global endLumi has completed for all lumis contained in this run
                            auto globalEndRunTask =
                                edm::make_waiting_task([this, status](std::exception_ptr const*) mutable {
                                  WaitingTaskHolder taskHolder = status->holderOfTaskInProcessRuns();
                                  status->holderOfTaskInProcessRuns().doneWaiting(std::exception_ptr{});
                                  globalEndRunAsync(std::move(taskHolder), std::move(status));
                                });
                            status->setGlobalEndRunHolder(WaitingTaskHolder{*holder.group(), globalEndRunTask});
                          } catch (...) {
                            status->resetBeginResources();
                            queueWhichWaitsForIOVsToFinish_.resume();
                            status->resumeGlobalRunQueue();
                            holder.doneWaiting(std::current_exception());
                            return;
                          }

                          // After this point we are committed to end the run via endRunAsync

                          ServiceRegistry::Operate operate(serviceToken_);

                          // The only purpose of the pause is to cause stream begin run to execute before
                          // global begin lumi in the single threaded case (maintains consistency with
                          // the order that existed before concurrent runs were implemented).
                          PauseQueueSentry pauseQueueSentry(streamQueuesInserter_);

                          CMS_SA_ALLOW try {
                            streamQueuesInserter_.push(
                                *holder.group(), [this, status, precedingTasksSucceeded, holder]() mutable {
                                  for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
                                    CMS_SA_ALLOW try {
                                      streamQueues_[i].push(
                                          *holder.group(),
                                          [this, i, status, precedingTasksSucceeded, holder]() mutable {
                                            streamBeginRunAsync(
                                                i, std::move(status), precedingTasksSucceeded, std::move(holder));
                                          });
                                    } catch (...) {
                                      if (status->streamFinishedBeginRun()) {
                                        WaitingTaskHolder copyHolder(holder);
                                        copyHolder.doneWaiting(std::current_exception());
                                        status->resetBeginResources();
                                        queueWhichWaitsForIOVsToFinish_.resume();
                                        exceptionRunStatus_ = status;
                                      }
                                    }
                                  }
                                });
                          } catch (...) {
                            WaitingTaskHolder copyHolder(holder);
                            copyHolder.doneWaiting(std::current_exception());
                            status->resetBeginResources();
                            queueWhichWaitsForIOVsToFinish_.resume();
                            exceptionRunStatus_ = status;
                          }
                          handleNextItemAfterMergingRunEntries(status, holder);
                        }) | runLast(postSourceTask);
                      } catch (...) {
                        status->resetBeginResources();
                        queueWhichWaitsForIOVsToFinish_.resume();
                        status->resumeGlobalRunQueue();
                        postSourceTask.doneWaiting(std::current_exception());
                      }
                    });  // task in sourceResourcesAcquirer
              } catch (...) {
                status->resetBeginResources();
                queueWhichWaitsForIOVsToFinish_.resume();
                status->resumeGlobalRunQueue();
                postRunQueueTask.doneWaiting(std::current_exception());
              }
            });  // task in runQueue
      } catch (...) {
        status->resetBeginResources();
        queueWhichWaitsForIOVsToFinish_.resume();
        nextTask.doneWaiting(std::current_exception());
      }
    }) | chain::runLast(std::move(iHolder));
  }

branchIDListHelper() [1/2]

std::shared_ptr<BranchIDListHelper const> edm::EventProcessor::branchIDListHelper ( ) const

inlineprivate

Definition at line 282 of file EventProcessor.h.

References branchIDListHelper_, and edm::get_underlying_safe().

Referenced by init().

                                                                       {
      return get_underlying_safe(branchIDListHelper_);
    }

branchIDListHelper() [2/2]

std::shared_ptr<BranchIDListHelper>& edm::EventProcessor::branchIDListHelper ( )

inlineprivate

Definition at line 285 of file EventProcessor.h.

References branchIDListHelper_, and edm::get_underlying_safe().

{ return get_underlying_safe(branchIDListHelper_); }

checkForAsyncStopRequest()

bool edm::EventProcessor::checkForAsyncStopRequest ( StatusCode &  returnCode )

private

Definition at line 848 of file EventProcessor.cc.

References epSignal, edm::JobReport::reportShutdownSignal(), runEdmFileComparison::returnCode, and edm::shutdown_flag.

Referenced by nextTransitionType().

                                                                      {
    bool returnValue = false;

    // Look for a shutdown signal
    if (shutdown_flag.load(std::memory_order_acquire)) {
      returnValue = true;
      edm::LogSystem("ShutdownSignal") << "an external signal was sent to shutdown the job early.";
      edm::Service<edm::JobReport> jr;
      jr->reportShutdownSignal();
      returnCode = epSignal;
    }
    return returnValue;
  }

clearCounters()

void edm::EventProcessor::clearCounters

(

)

Clears counters used by trigger report.

Definition at line 842 of file EventProcessor.cc.

References schedule_.

{ schedule_->clearCounters(); }

clearLumiPrincipal()

void edm::EventProcessor::clearLumiPrincipal

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 2109 of file EventProcessor.cc.

References edm::LuminosityBlockPrincipal::kUninitialized, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), alignCSCRings::s, and subProcesses_.

Referenced by globalEndLumiAsync().

                                                                                  {
    for (auto& s : subProcesses_) {
      s.clearLumiPrincipal(*iStatus.lumiPrincipal());
    }
    iStatus.lumiPrincipal()->setRunPrincipal(std::shared_ptr<RunPrincipal>());
    iStatus.lumiPrincipal()->setShouldWriteLumi(LuminosityBlockPrincipal::kUninitialized);
    iStatus.lumiPrincipal()->clearPrincipal();
  }

clearRunPrincipal()

void edm::EventProcessor::clearRunPrincipal

(

RunProcessingStatus &

iStatus

)

Definition at line 2084 of file EventProcessor.cc.

References edm::RunPrincipal::kUninitialized, edm::RunProcessingStatus::runPrincipal(), alignCSCRings::s, and subProcesses_.

Referenced by globalEndRunAsync().

                                                                     {
    for (auto& s : subProcesses_) {
      s.clearRunPrincipal(*iStatus.runPrincipal());
    }
    iStatus.runPrincipal()->setShouldWriteRun(RunPrincipal::kUninitialized);
    iStatus.runPrincipal()->clearPrincipal();
  }

closeInputFile()

void edm::EventProcessor::closeInputFile

(

bool

cleaningUpAfterException

)

Definition at line 1008 of file EventProcessor.cc.

References actReg_, fb_, FDEBUG, fileBlockValid(), and input_.

                                                                   {
    if (fileBlockValid()) {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->closeFile(fb_.get(), cleaningUpAfterException);
      sentry.completedSuccessfully();
    }
    FDEBUG(1) << "\tcloseInputFile\n";
  }

closeOutputFiles()

void edm::EventProcessor::closeOutputFiles

(

)

Definition at line 1025 of file EventProcessor.cc.

References FDEBUG, edm::for_all(), processBlockHelper_, schedule_, and subProcesses_.

                                        {
    schedule_->closeOutputFiles();
    for_all(subProcesses_, [](auto& subProcess) { subProcess.closeOutputFiles(); });
    processBlockHelper_->clearAfterOutputFilesClose();
    FDEBUG(1) << "\tcloseOutputFiles\n";
  }

continueLumiAsync()

void edm::EventProcessor::continueLumiAsync

(

edm::WaitingTaskHolder

iHolder

)

Definition at line 1831 of file EventProcessor.cc.

References dqmdumpme::first, h, handleNextEventForStreamAsync(), input_, edm::InputSource::IsLumi, edm::LuminosityBlockProcessingStatus::kProcessing, lastTransitionType(), eostools::move(), nextTransitionType(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, readAndMergeLumi(), edm::waiting_task::chain::runLast(), mps_update::status, streamLumiStatus_, and edm::waiting_task::chain::then().

Referenced by processRuns().

                                                                     {
    chain::first([this](auto nextTask) {
      //all streams are sharing the same status at the moment
      auto status = streamLumiStatus_[0];  //read from streamLumiActive_ happened in calling routine
      status->setEventProcessingState(LuminosityBlockProcessingStatus::EventProcessingState::kProcessing);

      while (lastTransitionType() == InputSource::ItemType::IsLumi and
             status->lumiPrincipal()->luminosityBlock() == input_->luminosityBlock()) {
        readAndMergeLumi(*status);
        nextTransitionType();
      }
    }) | chain::then([this](auto nextTask) mutable {
      unsigned int streamIndex = 0;
      oneapi::tbb::task_arena arena{oneapi::tbb::task_arena::attach()};
      for (; streamIndex < preallocations_.numberOfStreams() - 1; ++streamIndex) {
        arena.enqueue([this, streamIndex, h = nextTask]() { handleNextEventForStreamAsync(h, streamIndex); });
      }
      nextTask.group()->run(
          [this, streamIndex, h = std::move(nextTask)]() { handleNextEventForStreamAsync(h, streamIndex); });
    }) | chain::runLast(std::move(iHolder));
  }

doErrorStuff()

void edm::EventProcessor::doErrorStuff

(

)

Definition at line 1099 of file EventProcessor.cc.

References FDEBUG.

Referenced by runToCompletion().

                                    {
    FDEBUG(1) << "\tdoErrorStuff\n";
    LogError("StateMachine") << "The EventProcessor state machine encountered an unexpected event\n"
                             << "and went to the error state\n"
                             << "Will attempt to terminate processing normally\n"
                             << "(IF using the looper the next loop will be attempted)\n"
                             << "This likely indicates a bug in an input module or corrupted input or both\n";
  }

endJob()

void edm::EventProcessor::endJob

(

void

)

This should be called before the EventProcessor is destroyed throws if any module's endJob throws an exception.

Definition at line 763 of file EventProcessor.cc.

References actReg_, HltBtagPostValidation_cff::c, edm::SubProcess::doEndJob(), edm::InputSource::doEndJob(), edm::EDLooperBase::endOfJob(), SiStripBadComponentsDQMServiceTemplate_cfg::ep, edm::first(), watchdog::group, mps_fire::i, input_, MainPageGenerator::l, edm::waiting_task::chain::lastTask(), looper(), looper_, mutex, edm::PreallocationConfiguration::numberOfStreams(), preallocations_, schedule_, serviceToken_, subProcesses_, and edm::waiting_task::chain::then().

Referenced by PythonEventProcessor::~PythonEventProcessor().

                              {
    // Collects exceptions, so we don't throw before all operations are performed.
    ExceptionCollector c(
        "Multiple exceptions were thrown while executing endJob. An exception message follows for each.\n");

    //make the services available
    ServiceRegistry::Operate operate(serviceToken_);

    using namespace edm::waiting_task::chain;

    oneapi::tbb::task_group group;
    edm::FinalWaitingTask waitTask{group};

    {
      //handle endStream transitions
      edm::WaitingTaskHolder taskHolder(group, &waitTask);
      std::mutex collectorMutex;
      for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
        first([this, i, &c, &collectorMutex](auto nextTask) {
          std::exception_ptr ep;
          try {
            ServiceRegistry::Operate operate(serviceToken_);
            this->schedule_->endStream(i);
          } catch (...) {
            ep = std::current_exception();
          }
          if (ep) {
            std::lock_guard<std::mutex> l(collectorMutex);
            c.call([&ep]() { std::rethrow_exception(ep); });
          }
        }) | then([this, i, &c, &collectorMutex](auto nextTask) {
          for (auto& subProcess : subProcesses_) {
            first([this, i, &c, &collectorMutex, &subProcess](auto nextTask) {
              std::exception_ptr ep;
              try {
                ServiceRegistry::Operate operate(serviceToken_);
                subProcess.doEndStream(i);
              } catch (...) {
                ep = std::current_exception();
              }
              if (ep) {
                std::lock_guard<std::mutex> l(collectorMutex);
                c.call([&ep]() { std::rethrow_exception(ep); });
              }
            }) | lastTask(nextTask);
          }
        }) | lastTask(taskHolder);
      }
    }
    waitTask.waitNoThrow();

    auto actReg = actReg_.get();
    c.call([actReg]() { actReg->preEndJobSignal_(); });
    schedule_->endJob(c);
    for (auto& subProcess : subProcesses_) {
      c.call(std::bind(&SubProcess::doEndJob, &subProcess));
    }
    c.call(std::bind(&InputSource::doEndJob, input_.get()));
    if (looper_) {
      c.call(std::bind(&EDLooperBase::endOfJob, looper()));
    }
    c.call([actReg]() { actReg->postEndJobSignal_(); });
    if (c.hasThrown()) {
      c.rethrow();
    }
  }

endOfLoop()

bool edm::EventProcessor::endOfLoop

(

)

Definition at line 1060 of file EventProcessor.cc.

References esp_, FDEBUG, forceLooperToEnd_, edm::EDLooperBase::kContinue, looper_, preg_, schedule_, and mps_update::status.

Referenced by runToCompletion().

                                 {
    if (looper_) {
      ModuleChanger changer(schedule_.get(), preg_.get(), esp_->recordsToResolverIndices());
      looper_->setModuleChanger(&changer);
      EDLooperBase::Status status = looper_->doEndOfLoop(esp_->eventSetupImpl());
      looper_->setModuleChanger(nullptr);
      if (status != EDLooperBase::kContinue || forceLooperToEnd_)
        return true;
      else
        return false;
    }
    FDEBUG(1) << "\tendOfLoop\n";
    return true;
  }

endProcessBlock()

void edm::EventProcessor::endProcessBlock	(	bool	cleaningUpAfterException,
		bool	beginProcessBlockSucceeded
	)

Definition at line 1149 of file EventProcessor.cc.

References edm::Principal::clearPrincipal(), edm::PrincipalCache::New, principalCache_, edm::PrincipalCache::processBlockPrincipal(), alignCSCRings::s, schedule_, serviceToken_, subProcesses_, taskGroup_, and writeProcessBlockAsync().

                                                                                                     {
    ProcessBlockPrincipal& processBlockPrincipal = principalCache_.processBlockPrincipal();

    using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionGlobalEnd>;
    FinalWaitingTask globalWaitTask{taskGroup_};

    ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
    endGlobalTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &globalWaitTask),
                                     *schedule_,
                                     transitionInfo,
                                     serviceToken_,
                                     subProcesses_,
                                     cleaningUpAfterException);
    globalWaitTask.wait();

    if (beginProcessBlockSucceeded) {
      FinalWaitingTask writeWaitTask{taskGroup_};
      writeProcessBlockAsync(edm::WaitingTaskHolder{taskGroup_, &writeWaitTask}, ProcessBlockType::New);
      writeWaitTask.wait();
    }

    processBlockPrincipal.clearPrincipal();
    for (auto& s : subProcesses_) {
      s.clearProcessBlockPrincipal(ProcessBlockType::New);
    }
  }

endRunAsync()

void edm::EventProcessor::endRunAsync	(	std::shared_ptr< RunProcessingStatus >	iRunStatus,
		WaitingTaskHolder	iHolder
	)

Definition at line 1461 of file EventProcessor.cc.

References actReg_, beginRunAsync(), CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), edm::RunPrincipal::endTime(), espController_, dqmdumpme::first, handleEndRunExceptions(), mps_fire::i, input_, edm::InputSource::IsRun, lastTransitionType(), edm::EventID::maxEventNumber(), edm::LuminosityBlockID::maxLuminosityBlockNumber(), eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, edm::SerialTaskQueue::push(), queueWhichWaitsForIOVsToFinish_, edm::RunPrincipal::run(), edm::waiting_task::chain::runLast(), serviceToken_, edm::RunPrincipal::setEndTime(), streamEndRunAsync(), streamQueues_, streamQueuesInserter_, and edm::waiting_task::chain::then().

Referenced by beginLumiAsync(), endUnfinishedRun(), handleNextEventForStreamAsync(), and handleNextItemAfterMergingRunEntries().

                                                                                                           {
    RunPrincipal& runPrincipal = *iRunStatus->runPrincipal();
    iRunStatus->setEndTime();
    IOVSyncValue ts(
        EventID(runPrincipal.run(), LuminosityBlockID::maxLuminosityBlockNumber(), EventID::maxEventNumber()),
        runPrincipal.endTime());
    CMS_SA_ALLOW try { actReg_->esSyncIOVQueuingSignal_.emit(ts); } catch (...) {
      WaitingTaskHolder copyHolder(iHolder);
      copyHolder.doneWaiting(std::current_exception());
    }

    chain::first([this, &iRunStatus, &ts](auto nextTask) {
      espController_->runOrQueueEventSetupForInstanceAsync(ts,
                                                           nextTask,
                                                           iRunStatus->endIOVWaitingTasksEndRun(),
                                                           iRunStatus->eventSetupImplsEndRun(),
                                                           queueWhichWaitsForIOVsToFinish_,
                                                           actReg_.get(),
                                                           serviceToken_);
    }) | chain::then([this, iRunStatus](std::exception_ptr const* iException, auto nextTask) {
      if (iException) {
        iRunStatus->setEndingEventSetupSucceeded(false);
        handleEndRunExceptions(*iException, nextTask);
      }
      ServiceRegistry::Operate operate(serviceToken_);
      streamQueuesInserter_.push(*nextTask.group(), [this, nextTask]() mutable {
        for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
          CMS_SA_ALLOW try {
            streamQueues_[i].push(*nextTask.group(), [this, i, nextTask]() mutable {
              streamQueues_[i].pause();
              streamEndRunAsync(std::move(nextTask), i);
            });
          } catch (...) {
            WaitingTaskHolder copyHolder(nextTask);
            copyHolder.doneWaiting(std::current_exception());
          }
        }
      });

      if (lastTransitionType() == InputSource::ItemType::IsRun) {
        CMS_SA_ALLOW try {
          beginRunAsync(IOVSyncValue(EventID(input_->run(), 0, 0), input_->runAuxiliary()->beginTime()), nextTask);
        } catch (...) {
          WaitingTaskHolder copyHolder(nextTask);
          copyHolder.doneWaiting(std::current_exception());
        }
      }
    }) | chain::runLast(std::move(iHolder));
  }

endUnfinishedLumi()

void edm::EventProcessor::endUnfinishedLumi

(

bool

cleaningUpAfterException

)

Definition at line 1974 of file EventProcessor.cc.

References cms::cuda::assert(), mps_fire::i, edm::PreallocationConfiguration::numberOfStreams(), preallocations_, streamEndLumiAsync(), streamLumiActive_, streamLumiStatus_, streamRunActive_, and taskGroup_.

                                                                      {
    if (streamRunActive_ == 0) {
      assert(streamLumiActive_ == 0);
    } else {
      assert(streamRunActive_ == preallocations_.numberOfStreams());
      if (streamLumiActive_ > 0) {
        FinalWaitingTask globalWaitTask{taskGroup_};
        assert(streamLumiActive_ == preallocations_.numberOfStreams());
        streamLumiStatus_[0]->noMoreEventsInLumi();
        streamLumiStatus_[0]->setCleaningUpAfterException(cleaningUpAfterException);
        for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
          streamEndLumiAsync(WaitingTaskHolder{taskGroup_, &globalWaitTask}, i);
        }
        globalWaitTask.wait();
      }
    }
  }

endUnfinishedRun()

void edm::EventProcessor::endUnfinishedRun

(

bool

cleaningUpAfterException

)

Definition at line 1652 of file EventProcessor.cc.

References endRunAsync(), edm::InputSource::IsStop, lastSourceTransition_, eostools::move(), streamRunActive_, streamRunStatus_, and taskGroup_.

                                                                     {
    if (streamRunActive_ > 0) {
      FinalWaitingTask waitTask{taskGroup_};

      auto runStatus = streamRunStatus_[0].get();
      runStatus->setCleaningUpAfterException(cleaningUpAfterException);
      WaitingTaskHolder holder{taskGroup_, &waitTask};
      runStatus->setHolderOfTaskInProcessRuns(holder);
      lastSourceTransition_ = InputSource::ItemType::IsStop;
      endRunAsync(streamRunStatus_[0], std::move(holder));
      waitTask.wait();
    }
  }

fileBlockValid()

bool edm::EventProcessor::fileBlockValid ( )

inline

Definition at line 193 of file EventProcessor.h.

References fb_.

Referenced by closeInputFile(), openOutputFiles(), respondToCloseInputFile(), and respondToOpenInputFile().

{ return fb_.get() != nullptr; }

getAllModuleDescriptions()

std::vector< ModuleDescription const * > edm::EventProcessor::getAllModuleDescriptions

(

)

const

Return a vector allowing const access to all the ModuleDescriptions for this EventProccessor. *** N.B. *** Ownership of the ModuleDescriptions is not *** passed to the caller. Do not call delete on these *** pointers!

Definition at line 832 of file EventProcessor.cc.

References schedule_.

                                                                                     {
    return schedule_->getAllModuleDescriptions();
  }

getToken()

ServiceToken edm::EventProcessor::getToken

(

)

Definition at line 830 of file EventProcessor.cc.

References serviceToken_.

Referenced by ~EventProcessor().

{ return serviceToken_; }

globalEndLumiAsync()

void edm::EventProcessor::globalEndLumiAsync	(	edm::WaitingTaskHolder	iTask,
		std::shared_ptr< LuminosityBlockProcessingStatus >	iLumiStatus
	)

Definition at line 1862 of file EventProcessor.cc.

References clearLumiPrincipal(), CMS_SA_ALLOW, edm::EndLuminosityBlock, esp_, dqmdumpme::first, handleEndLumiExceptions(), edm::waiting_task::chain::ifThen(), looper_, edm::EventID::maxEventNumber(), eostools::move(), processContext_, queueWhichWaitsForIOVsToFinish_, edm::SerialTaskQueue::resume(), edm::waiting_task::chain::runLast(), schedule_, serviceToken_, mps_update::status, subProcesses_, edm::WaitingTaskHolder::taskHasFailed(), edm::waiting_task::chain::then(), and writeLumiAsync().

Referenced by streamEndLumiAsync().

                                                                                                      {
    // Get some needed info out of the status object before moving
    // it into finalTaskForThisLumi.
    auto& lp = *(iLumiStatus->lumiPrincipal());
    bool didGlobalBeginSucceed = iLumiStatus->didGlobalBeginSucceed();
    bool cleaningUpAfterException = iLumiStatus->cleaningUpAfterException() || iTask.taskHasFailed();
    EventSetupImpl const& es = iLumiStatus->eventSetupImpl(esp_->subProcessIndex());
    std::vector<std::shared_ptr<const EventSetupImpl>> const* eventSetupImpls = &iLumiStatus->eventSetupImpls();

    using namespace edm::waiting_task::chain;
    chain::first([this, &lp, &es, &eventSetupImpls, cleaningUpAfterException](auto nextTask) {
      IOVSyncValue ts(EventID(lp.run(), lp.luminosityBlock(), EventID::maxEventNumber()), lp.beginTime());

      LumiTransitionInfo transitionInfo(lp, es, eventSetupImpls);
      using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd>;
      endGlobalTransitionAsync<Traits>(
          std::move(nextTask), *schedule_, transitionInfo, serviceToken_, subProcesses_, cleaningUpAfterException);
    }) | then([this, didGlobalBeginSucceed, &lumiPrincipal = lp](auto nextTask) {
      //Only call writeLumi if beginLumi succeeded
      if (didGlobalBeginSucceed) {
        writeLumiAsync(std::move(nextTask), lumiPrincipal);
      }
    }) | ifThen(looper_, [this, &lp, &es](auto nextTask) {
      looper_->prefetchAsync(std::move(nextTask), serviceToken_, Transition::EndLuminosityBlock, lp, es);
    }) | ifThen(looper_, [this, &lp, &es](auto nextTask) {
      //any thrown exception auto propagates to nextTask via the chain
      ServiceRegistry::Operate operate(serviceToken_);
      looper_->doEndLuminosityBlock(lp, es, &processContext_);
    }) | then([status = std::move(iLumiStatus), this](std::exception_ptr const* iException, auto nextTask) mutable {
      if (iException) {
        handleEndLumiExceptions(*iException, nextTask);
      }
      ServiceRegistry::Operate operate(serviceToken_);

      std::exception_ptr ptr;

      // Try hard to clean up resources so the
      // process can terminate in a controlled
      // fashion even after exceptions have occurred.
      // Caught exception is passed to handleEndLumiExceptions()
      CMS_SA_ALLOW try { clearLumiPrincipal(*status); } catch (...) {
        if (not ptr) {
          ptr = std::current_exception();
        }
      }
      // Caught exception is passed to handleEndLumiExceptions()
      CMS_SA_ALLOW try { queueWhichWaitsForIOVsToFinish_.resume(); } catch (...) {
        if (not ptr) {
          ptr = std::current_exception();
        }
      }
      // Caught exception is passed to handleEndLumiExceptions()
      CMS_SA_ALLOW try {
        status->resetResources();
        status->globalEndRunHolderDoneWaiting();
        status.reset();
      } catch (...) {
        if (not ptr) {
          ptr = std::current_exception();
        }
      }

      if (ptr && !iException) {
        handleEndLumiExceptions(ptr, nextTask);
      }
    }) | runLast(std::move(iTask));
  }

globalEndRunAsync()

void edm::EventProcessor::globalEndRunAsync	(	WaitingTaskHolder	iTask,
		std::shared_ptr< RunProcessingStatus >	iRunStatus
	)

Definition at line 1520 of file EventProcessor.cc.

References clearRunPrincipal(), CMS_SA_ALLOW, edm::EndRun, esp_, dqmdumpme::first, handleEndRunExceptions(), edm::waiting_task::chain::ifThen(), looper_, eostools::move(), edm::MergeableRunProductMetadata::preWriteRun(), processContext_, queueWhichWaitsForIOVsToFinish_, edm::SerialTaskQueue::resume(), edm::waiting_task::chain::runLast(), schedule_, serviceToken_, mps_update::status, subProcesses_, edm::WaitingTaskHolder::taskHasFailed(), edm::waiting_task::chain::then(), and writeRunAsync().

Referenced by beginRunAsync().

                                                                                                               {
    auto& runPrincipal = *(iRunStatus->runPrincipal());
    bool didGlobalBeginSucceed = iRunStatus->didGlobalBeginSucceed();
    bool cleaningUpAfterException = iRunStatus->cleaningUpAfterException() || iTask.taskHasFailed();
    EventSetupImpl const& es = iRunStatus->eventSetupImplEndRun(esp_->subProcessIndex());
    std::vector<std::shared_ptr<const EventSetupImpl>> const* eventSetupImpls = &iRunStatus->eventSetupImplsEndRun();
    bool endingEventSetupSucceeded = iRunStatus->endingEventSetupSucceeded();

    MergeableRunProductMetadata* mergeableRunProductMetadata = runPrincipal.mergeableRunProductMetadata();
    using namespace edm::waiting_task::chain;
    chain::first([this, &runPrincipal, &es, &eventSetupImpls, cleaningUpAfterException, endingEventSetupSucceeded](
                     auto nextTask) {
      if (endingEventSetupSucceeded) {
        RunTransitionInfo transitionInfo(runPrincipal, es, eventSetupImpls);
        using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd>;
        endGlobalTransitionAsync<Traits>(
            std::move(nextTask), *schedule_, transitionInfo, serviceToken_, subProcesses_, cleaningUpAfterException);
      }
    }) |
        ifThen(looper_ && endingEventSetupSucceeded,
               [this, &runPrincipal, &es](auto nextTask) {
                 looper_->prefetchAsync(std::move(nextTask), serviceToken_, Transition::EndRun, runPrincipal, es);
               }) |
        ifThen(looper_ && endingEventSetupSucceeded,
               [this, &runPrincipal, &es](auto nextTask) {
                 ServiceRegistry::Operate operate(serviceToken_);
                 looper_->doEndRun(runPrincipal, es, &processContext_);
               }) |
        ifThen(didGlobalBeginSucceed && endingEventSetupSucceeded,
               [this, mergeableRunProductMetadata, &runPrincipal = runPrincipal](auto nextTask) {
                 mergeableRunProductMetadata->preWriteRun();
                 writeRunAsync(nextTask, runPrincipal, mergeableRunProductMetadata);
               }) |
        then([status = std::move(iRunStatus),
              this,
              didGlobalBeginSucceed,
              mergeableRunProductMetadata,
              endingEventSetupSucceeded](std::exception_ptr const* iException, auto nextTask) mutable {
          if (didGlobalBeginSucceed && endingEventSetupSucceeded) {
            mergeableRunProductMetadata->postWriteRun();
          }
          if (iException) {
            handleEndRunExceptions(*iException, nextTask);
          }
          ServiceRegistry::Operate operate(serviceToken_);

          std::exception_ptr ptr;

          // Try hard to clean up resources so the
          // process can terminate in a controlled
          // fashion even after exceptions have occurred.
          CMS_SA_ALLOW try { clearRunPrincipal(*status); } catch (...) {
            if (not ptr) {
              ptr = std::current_exception();
            }
          }
          CMS_SA_ALLOW try {
            status->resumeGlobalRunQueue();
            queueWhichWaitsForIOVsToFinish_.resume();
          } catch (...) {
            if (not ptr) {
              ptr = std::current_exception();
            }
          }
          CMS_SA_ALLOW try {
            status->resetEndResources();
            status.reset();
          } catch (...) {
            if (not ptr) {
              ptr = std::current_exception();
            }
          }

          if (ptr && !iException) {
            handleEndRunExceptions(ptr, nextTask);
          }
        }) |
        runLast(std::move(iTask));
  }

handleEndLumiExceptions()

void edm::EventProcessor::handleEndLumiExceptions	(	std::exception_ptr	iException,
		WaitingTaskHolder const &	holder
	)

Definition at line 1853 of file EventProcessor.cc.

References setExceptionMessageLumis(), edm::WaitingTaskHolder::taskHasFailed(), and createJobs::tmp.

Referenced by globalEndLumiAsync(), and streamEndLumiAsync().

                                                                                                           {
    if (holder.taskHasFailed()) {
      setExceptionMessageLumis();
    } else {
      WaitingTaskHolder tmp(holder);
      tmp.doneWaiting(iException);
    }
  }

handleEndRunExceptions()

void edm::EventProcessor::handleEndRunExceptions	(	std::exception_ptr	iException,
		WaitingTaskHolder const &	holder
	)

Definition at line 1511 of file EventProcessor.cc.

References setExceptionMessageRuns(), edm::WaitingTaskHolder::taskHasFailed(), and createJobs::tmp.

Referenced by endRunAsync(), globalEndRunAsync(), and streamEndRunAsync().

                                                                                                          {
    if (holder.taskHasFailed()) {
      setExceptionMessageRuns();
    } else {
      WaitingTaskHolder tmp(holder);
      tmp.doneWaiting(iException);
    }
  }

handleNextEventForStreamAsync()

void edm::EventProcessor::handleNextEventForStreamAsync ( WaitingTaskHolder  iTask, unsigned int  iStreamIndex  )

private

Definition at line 2280 of file EventProcessor.cc.

References cms::cuda::assert(), beginLumiAsync(), CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), endRunAsync(), edm::WaitingTaskHolder::group(), watchdog::group, input_, edm::InputSource::IsLumi, edm::LuminosityBlockProcessingStatus::kPauseForFileTransition, edm::LuminosityBlockProcessingStatus::kStopLumi, lastTransitionType(), edm::make_waiting_task(), eostools::move(), processEventAsync(), edm::SerialTaskQueueChain::push(), readNextEventForStream(), edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, sourceResourcesAcquirer_, mps_update::status, streamEndLumiAsync(), streamLumiStatus_, streamRunStatus_, and edm::WaitingTaskHolder::taskHasFailed().

Referenced by beginLumiAsync(), and continueLumiAsync().

                                                                                                       {
    if (streamLumiStatus_[iStreamIndex]->haveStartedNextLumiOrEndedRun()) {
      streamEndLumiAsync(iTask, iStreamIndex);
      return;
    }
    auto group = iTask.group();
    sourceResourcesAcquirer_.serialQueueChain().push(*group, [this, iTask = std::move(iTask), iStreamIndex]() mutable {
      CMS_SA_ALLOW try {
        auto status = streamLumiStatus_[iStreamIndex].get();
        ServiceRegistry::Operate operate(serviceToken_);

        if (readNextEventForStream(iTask, iStreamIndex, *status)) {
          auto recursionTask =
              make_waiting_task([this, iTask, iStreamIndex](std::exception_ptr const* iEventException) mutable {
                if (iEventException) {
                  WaitingTaskHolder copyHolder(iTask);
                  copyHolder.doneWaiting(*iEventException);
                  // Intentionally, we don't return here. The recursive call to
                  // handleNextEvent takes care of immediately ending the run properly
                  // using the same code it uses to end the run in other situations.
                }
                handleNextEventForStreamAsync(std::move(iTask), iStreamIndex);
              });

          processEventAsync(WaitingTaskHolder(*iTask.group(), recursionTask), iStreamIndex);
        } else {
          // the stream will stop processing this lumi now
          if (status->eventProcessingState() == LuminosityBlockProcessingStatus::EventProcessingState::kStopLumi) {
            if (not status->haveStartedNextLumiOrEndedRun()) {
              status->noMoreEventsInLumi();
              status->startNextLumiOrEndRun();
              if (lastTransitionType() == InputSource::ItemType::IsLumi && !iTask.taskHasFailed()) {
                CMS_SA_ALLOW try {
                  beginLumiAsync(IOVSyncValue(EventID(input_->run(), input_->luminosityBlock(), 0),
                                              input_->luminosityBlockAuxiliary()->beginTime()),
                                 streamRunStatus_[iStreamIndex],
                                 iTask);
                } catch (...) {
                  WaitingTaskHolder copyHolder(iTask);
                  copyHolder.doneWaiting(std::current_exception());
                  endRunAsync(streamRunStatus_[iStreamIndex], iTask);
                }
              } else {
                // If appropriate, this will also start the next run.
                endRunAsync(streamRunStatus_[iStreamIndex], iTask);
              }
            }
            streamEndLumiAsync(iTask, iStreamIndex);
          } else {
            assert(status->eventProcessingState() ==
                   LuminosityBlockProcessingStatus::EventProcessingState::kPauseForFileTransition);
            auto runStatus = streamRunStatus_[iStreamIndex].get();

            if (runStatus->holderOfTaskInProcessRuns().hasTask()) {
              runStatus->holderOfTaskInProcessRuns().doneWaiting(std::exception_ptr{});
            }
          }
        }
      } catch (...) {
        WaitingTaskHolder copyHolder(iTask);
        copyHolder.doneWaiting(std::current_exception());
        handleNextEventForStreamAsync(std::move(iTask), iStreamIndex);
      }
    });
  }

handleNextItemAfterMergingRunEntries()

void edm::EventProcessor::handleNextItemAfterMergingRunEntries ( std::shared_ptr< RunProcessingStatus >  iRunStatus, WaitingTaskHolder  iHolder  )

private

Definition at line 2180 of file EventProcessor.cc.

References beginLumiAsync(), CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), endRunAsync(), dqmdumpme::first, input_, edm::InputSource::IsFile, edm::InputSource::IsLumi, lastTransitionType(), eostools::move(), needToCallNext(), nextTransitionTypeAsync(), edm::waiting_task::chain::runLast(), serviceToken_, and edm::waiting_task::chain::then().

Referenced by beginRunAsync(), and processRuns().

                                                                                       {
    chain::first([this, iRunStatus](auto nextTask) mutable {
      if (needToCallNext()) {
        nextTransitionTypeAsync(std::move(iRunStatus), std::move(nextTask));
      }
    }) | chain::then([this, iRunStatus](std::exception_ptr const* iException, auto nextTask) {
      ServiceRegistry::Operate operate(serviceToken_);
      if (iException) {
        WaitingTaskHolder copyHolder(nextTask);
        copyHolder.doneWaiting(*iException);
      }
      if (lastTransitionType() == InputSource::ItemType::IsFile) {
        iRunStatus->holderOfTaskInProcessRuns().doneWaiting(std::exception_ptr{});
        return;
      }
      if (lastTransitionType() == InputSource::ItemType::IsLumi && !nextTask.taskHasFailed()) {
        CMS_SA_ALLOW try {
          beginLumiAsync(IOVSyncValue(EventID(input_->run(), input_->luminosityBlock(), 0),
                                      input_->luminosityBlockAuxiliary()->beginTime()),
                         iRunStatus,
                         nextTask);
          return;
        } catch (...) {
          WaitingTaskHolder copyHolder(nextTask);
          copyHolder.doneWaiting(std::current_exception());
        }
      }
      // Note that endRunAsync will call beginRunAsync for the following run
      // if appropriate.
      endRunAsync(iRunStatus, std::move(nextTask));
    }) | chain::runLast(std::move(iHolder));
  }

init()

void edm::EventProcessor::init ( std::shared_ptr< ProcessDesc > &  processDesc, ServiceToken const &  token, serviceregistry::ServiceLegacy  iLegacy  )

private

Definition at line 325 of file EventProcessor.cc.

References act_table_, actReg_, cms::cuda::assert(), branchesToDeleteEarly_, branchIDListHelper(), branchIDListHelper_, CMS_SA_ALLOW, trackingPlots::common, edm::errors::Configuration, deleteNonConsumedUnscheduledModules_, testHGCalDigi_cfg::dumpOptions, edm::dumpOptionsToLogFile(), edm::ensureAvailableAccelerators(), SiStripBadComponentsDQMServiceTemplate_cfg::ep, esp_, espController_, Exception, FDEBUG, DiMuonV_cfg::fileMode, fileModeNoMerge_, edm::fillLooper(), forceESCacheClearOnNewRun_, edm::ServiceRegistry::get(), edm::get_underlying_safe(), edm::ParameterSet::getParameter(), edm::ModuleDescription::getUniqueID(), edm::ParameterSet::getUntrackedParameterSet(), watchdog::group, historyAppender_, input_, edm::PrincipalCache::insert(), edm::PrincipalCache::insertForInput(), edm::ParentageRegistry::insertMapped(), edm::ParentageRegistry::instance(), edm::ServiceRegistry::instance(), mps_monitormerge::items, edm::serviceregistry::kConfigurationOverrides, looper_, lumiQueue_, edm::makeInput(), mergeableRunProductProcesses_, modulesToIgnoreForDeleteEarly_, moduleTypeResolverMaker_, eostools::move(), edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfRuns(), edm::PreallocationConfiguration::numberOfStreams(), or, edm::parameterSet(), edm::popSubProcessVParameterSet(), preallocations_, preg(), preg_, principalCache_, printDependencies_, processBlockHelper_, processConfiguration_, processContext_, muonDTDigis_cfi::pset, referencesToBranches_, edm::ParameterSet::registerIt(), runQueue_, schedule_, serviceToken_, edm::PrincipalCache::setNumberOfConcurrentPrincipals(), edm::ProcessContext::setProcessConfiguration(), edm::MergeableRunProductProcesses::setProcessesWithMergeableRunProducts(), edm::IllegalParameters::setThrowAnException(), streamLumiStatus_, streamQueues_, streamRunStatus_, AlCaHLTBitMon_QueryRunRegistry::string, subProcesses_, thinnedAssociationsHelper(), thinnedAssociationsHelper_, unpackBuffers-CaloStage2::token, and edm::validateTopLevelParameterSets().

Referenced by EventProcessor().

                                                                  {
    //std::cerr << processDesc->dump() << std::endl;

    // register the empty parentage vector , once and for all
    ParentageRegistry::instance()->insertMapped(Parentage());

    // register the empty parameter set, once and for all.
    ParameterSet().registerIt();

    std::shared_ptr<ParameterSet> parameterSet = processDesc->getProcessPSet();

    // If there are subprocesses, pop the subprocess parameter sets out of the process parameter set
    auto subProcessVParameterSet = popSubProcessVParameterSet(*parameterSet);
    bool const hasSubProcesses = !subProcessVParameterSet.empty();

    // Validates the parameters in the 'options', 'maxEvents', 'maxLuminosityBlocks',
    // and 'maxSecondsUntilRampdown' top level parameter sets. Default values are also
    // set in here if the parameters were not explicitly set.
    validateTopLevelParameterSets(parameterSet.get());

    // Now set some parameters specific to the main process.
    ParameterSet const& optionsPset(parameterSet->getUntrackedParameterSet("options"));
    auto const& fileMode = optionsPset.getUntrackedParameter<std::string>("fileMode");
    if (fileMode != "NOMERGE" and fileMode != "FULLMERGE") {
      throw Exception(errors::Configuration, "Illegal fileMode parameter value: ")
          << fileMode << ".\n"
          << "Legal values are 'NOMERGE' and 'FULLMERGE'.\n";
    } else {
      fileModeNoMerge_ = (fileMode == "NOMERGE");
    }
    forceESCacheClearOnNewRun_ = optionsPset.getUntrackedParameter<bool>("forceEventSetupCacheClearOnNewRun");
    ensureAvailableAccelerators(*parameterSet);

    //threading
    unsigned int nThreads = optionsPset.getUntrackedParameter<unsigned int>("numberOfThreads");

    // Even if numberOfThreads was set to zero in the Python configuration, the code
    // in cmsRun.cpp should have reset it to something else.
    assert(nThreads != 0);

    unsigned int nStreams = optionsPset.getUntrackedParameter<unsigned int>("numberOfStreams");
    if (nStreams == 0) {
      nStreams = nThreads;
    }
    unsigned int nConcurrentLumis =
        optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentLuminosityBlocks");
    if (nConcurrentLumis == 0) {
      nConcurrentLumis = 2;
    }
    if (nConcurrentLumis > nStreams) {
      nConcurrentLumis = nStreams;
    }
    unsigned int nConcurrentRuns = optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentRuns");
    if (nConcurrentRuns == 0 || nConcurrentRuns > nConcurrentLumis) {
      nConcurrentRuns = nConcurrentLumis;
    }
    std::vector<std::string> loopers = parameterSet->getParameter<std::vector<std::string>>("@all_loopers");
    if (!loopers.empty()) {
      //For now loopers make us run only 1 transition at a time
      if (nStreams != 1 || nConcurrentLumis != 1 || nConcurrentRuns != 1) {
        edm::LogWarning("ThreadStreamSetup") << "There is a looper, so the number of streams, the number "
                                                "of concurrent runs, and the number of concurrent lumis "
                                                "are all being reset to 1. Loopers cannot currently support "
                                                "values greater than 1.";
        nStreams = 1;
        nConcurrentLumis = 1;
        nConcurrentRuns = 1;
      }
    }
    bool dumpOptions = optionsPset.getUntrackedParameter<bool>("dumpOptions");
    if (dumpOptions) {
      dumpOptionsToLogFile(nThreads, nStreams, nConcurrentLumis, nConcurrentRuns);
    } else {
      if (nThreads > 1 or nStreams > 1) {
        edm::LogInfo("ThreadStreamSetup") << "setting # threads " << nThreads << "\nsetting # streams " << nStreams;
      }
    }

    // The number of concurrent IOVs is configured individually for each record in
    // the class NumberOfConcurrentIOVs to values less than or equal to this.
    // This maximum simplifies to being equal nConcurrentLumis if nConcurrentRuns is 1.
    // Considering endRun, beginRun, and beginLumi we might need 3 concurrent IOVs per
    // concurrent run past the first in use cases where IOVs change within a run.
    unsigned int maxConcurrentIOVs =
        3 * nConcurrentRuns - 2 + ((nConcurrentLumis > nConcurrentRuns) ? (nConcurrentLumis - nConcurrentRuns) : 0);

    IllegalParameters::setThrowAnException(optionsPset.getUntrackedParameter<bool>("throwIfIllegalParameter"));

    printDependencies_ = optionsPset.getUntrackedParameter<bool>("printDependencies");
    deleteNonConsumedUnscheduledModules_ =
        optionsPset.getUntrackedParameter<bool>("deleteNonConsumedUnscheduledModules");
    //for now, if have a subProcess, don't allow early delete
    // In the future we should use the SubProcess's 'keep list' to decide what can be kept
    if (not hasSubProcesses) {
      branchesToDeleteEarly_ = optionsPset.getUntrackedParameter<std::vector<std::string>>("canDeleteEarly");
    }
    if (not branchesToDeleteEarly_.empty()) {
      auto referencePSets =
          optionsPset.getUntrackedParameter<std::vector<edm::ParameterSet>>("holdsReferencesToDeleteEarly");
      for (auto const& pset : referencePSets) {
        auto product = pset.getParameter<std::string>("product");
        auto references = pset.getParameter<std::vector<std::string>>("references");
        for (auto const& ref : references) {
          referencesToBranches_.emplace(product, ref);
        }
      }
      modulesToIgnoreForDeleteEarly_ =
          optionsPset.getUntrackedParameter<std::vector<std::string>>("modulesToIgnoreForDeleteEarly");
    }

    // Now do general initialization
    ScheduleItems items;

    //initialize the services
    auto& serviceSets = processDesc->getServicesPSets();
    ServiceToken token = items.initServices(serviceSets, *parameterSet, iToken, iLegacy, true);
    serviceToken_ = items.addCPRandTNS(*parameterSet, token);

    //make the services available
    ServiceRegistry::Operate operate(serviceToken_);

    CMS_SA_ALLOW try {
      if (nThreads > 1) {
        edm::Service<RootHandlers> handler;
        handler->willBeUsingThreads();
      }

      // intialize miscellaneous items
      std::shared_ptr<CommonParams> common(items.initMisc(*parameterSet));

      // intialize the event setup provider
      ParameterSet const& eventSetupPset(optionsPset.getUntrackedParameterSet("eventSetup"));
      esp_ = espController_->makeProvider(
          *parameterSet, items.actReg_.get(), &eventSetupPset, maxConcurrentIOVs, dumpOptions);

      // initialize the looper, if any
      if (!loopers.empty()) {
        looper_ = fillLooper(*espController_, *esp_, *parameterSet, loopers);
        looper_->setActionTable(items.act_table_.get());
        looper_->attachTo(*items.actReg_);

        // in presence of looper do not delete modules
        deleteNonConsumedUnscheduledModules_ = false;
      }

      preallocations_ = PreallocationConfiguration{nThreads, nStreams, nConcurrentLumis, nConcurrentRuns};

      runQueue_ = std::make_unique<LimitedTaskQueue>(nConcurrentRuns);
      lumiQueue_ = std::make_unique<LimitedTaskQueue>(nConcurrentLumis);
      streamQueues_.resize(nStreams);
      streamRunStatus_.resize(nStreams);
      streamLumiStatus_.resize(nStreams);

      processBlockHelper_ = std::make_shared<ProcessBlockHelper>();

      {
        std::optional<ScheduleItems::MadeModules> madeModules;

        //setup input and modules concurrently
        tbb::task_group group;

        // initialize the input source
        auto tempReg = std::make_shared<ProductRegistry>();
        auto sourceID = ModuleDescription::getUniqueID();

        group.run([&, this]() {
          // initialize the Schedule
          ServiceRegistry::Operate operate(serviceToken_);
          auto const& tns = ServiceRegistry::instance().get<service::TriggerNamesService>();
          madeModules =
              items.initModules(*parameterSet, tns, preallocations_, &processContext_, moduleTypeResolverMaker_.get());
        });

        group.run([&, this, tempReg]() {
          ServiceRegistry::Operate operate(serviceToken_);
          input_ = makeInput(sourceID,
                             *parameterSet,
                             *common,
                             /*items.preg(),*/ tempReg,
                             items.branchIDListHelper(),
                             get_underlying_safe(processBlockHelper_),
                             items.thinnedAssociationsHelper(),
                             items.actReg_,
                             items.processConfiguration(),
                             preallocations_);
        });

        group.wait();
        items.preg()->addFromInput(*tempReg);
        input_->switchTo(items.preg());

        {
          auto const& tns = ServiceRegistry::instance().get<service::TriggerNamesService>();
          schedule_ = items.finishSchedule(std::move(*madeModules),
                                           *parameterSet,
                                           tns,
                                           hasSubProcesses,
                                           preallocations_,
                                           &processContext_,
                                           *processBlockHelper_);
        }
      }

      // set the data members
      act_table_ = std::move(items.act_table_);
      actReg_ = items.actReg_;
      preg_ = items.preg();
      mergeableRunProductProcesses_.setProcessesWithMergeableRunProducts(*preg_);
      branchIDListHelper_ = items.branchIDListHelper();
      thinnedAssociationsHelper_ = items.thinnedAssociationsHelper();
      processConfiguration_ = items.processConfiguration();
      processContext_.setProcessConfiguration(processConfiguration_.get());

      FDEBUG(2) << parameterSet << std::endl;

      principalCache_.setNumberOfConcurrentPrincipals(preallocations_);
      for (unsigned int index = 0; index < preallocations_.numberOfStreams(); ++index) {
        // Reusable event principal
        auto ep = std::make_shared<EventPrincipal>(preg(),
                                                   branchIDListHelper(),
                                                   thinnedAssociationsHelper(),
                                                   *processConfiguration_,
                                                   historyAppender_.get(),
                                                   index,
                                                   true /*primary process*/,
                                                   &*processBlockHelper_);
        principalCache_.insert(std::move(ep));
      }

      for (unsigned int index = 0; index < preallocations_.numberOfRuns(); ++index) {
        auto rp = std::make_unique<RunPrincipal>(
            preg(), *processConfiguration_, historyAppender_.get(), index, true, &mergeableRunProductProcesses_);
        principalCache_.insert(std::move(rp));
      }

      for (unsigned int index = 0; index < preallocations_.numberOfLuminosityBlocks(); ++index) {
        auto lp =
            std::make_unique<LuminosityBlockPrincipal>(preg(), *processConfiguration_, historyAppender_.get(), index);
        principalCache_.insert(std::move(lp));
      }

      {
        auto pb = std::make_unique<ProcessBlockPrincipal>(preg(), *processConfiguration_);
        principalCache_.insert(std::move(pb));

        auto pbForInput = std::make_unique<ProcessBlockPrincipal>(preg(), *processConfiguration_);
        principalCache_.insertForInput(std::move(pbForInput));
      }

      // fill the subprocesses, if there are any
      subProcesses_.reserve(subProcessVParameterSet.size());
      for (auto& subProcessPSet : subProcessVParameterSet) {
        subProcesses_.emplace_back(subProcessPSet,
                                   *parameterSet,
                                   preg(),
                                   branchIDListHelper(),
                                   *processBlockHelper_,
                                   *thinnedAssociationsHelper_,
                                   SubProcessParentageHelper(),
                                   *espController_,
                                   *actReg_,
                                   token,
                                   serviceregistry::kConfigurationOverrides,
                                   preallocations_,
                                   &processContext_,
                                   moduleTypeResolverMaker_);
      }
    } catch (...) {
      //in case of an exception, make sure Services are available
      // during the following destructors
      espController_ = nullptr;
      esp_ = nullptr;
      schedule_ = nullptr;
      input_ = nullptr;
      looper_ = nullptr;
      actReg_ = nullptr;
      throw;
    }
  }

inputProcessBlocks()

void edm::EventProcessor::inputProcessBlocks

(

)

Definition at line 1123 of file EventProcessor.cc.

References edm::Principal::clearPrincipal(), edm::PrincipalCache::Input, input_, edm::PrincipalCache::inputProcessBlockPrincipal(), principalCache_, readProcessBlock(), alignCSCRings::s, schedule_, serviceToken_, subProcesses_, taskGroup_, and writeProcessBlockAsync().

                                          {
    input_->fillProcessBlockHelper();
    ProcessBlockPrincipal& processBlockPrincipal = principalCache_.inputProcessBlockPrincipal();
    while (input_->nextProcessBlock(processBlockPrincipal)) {
      readProcessBlock(processBlockPrincipal);

      using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionProcessBlockInput>;
      FinalWaitingTask globalWaitTask{taskGroup_};

      ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
      beginGlobalTransitionAsync<Traits>(
          WaitingTaskHolder(taskGroup_, &globalWaitTask), *schedule_, transitionInfo, serviceToken_, subProcesses_);

      globalWaitTask.wait();

      FinalWaitingTask writeWaitTask{taskGroup_};
      writeProcessBlockAsync(edm::WaitingTaskHolder{taskGroup_, &writeWaitTask}, ProcessBlockType::Input);
      writeWaitTask.wait();

      processBlockPrincipal.clearPrincipal();
      for (auto& s : subProcesses_) {
        s.clearProcessBlockPrincipal(ProcessBlockType::Input);
      }
    }
  }

lastTransitionType()

InputSource::ItemTypeInfo edm::EventProcessor::lastTransitionType ( ) const

inline

Definition at line 189 of file EventProcessor.h.

References lastSourceTransition_.

Referenced by beginLumiAsync(), beginRunAsync(), continueLumiAsync(), endRunAsync(), handleNextEventForStreamAsync(), handleNextItemAfterMergingRunEntries(), nextTransitionTypeAsync(), processRuns(), readAndMergeLumiEntriesAsync(), readAndMergeRunEntriesAsync(), and readNextEventForStream().

{ return lastSourceTransition_; }

looper() [1/2]

std::shared_ptr<EDLooperBase const> edm::EventProcessor::looper ( ) const

inlineprivate

Definition at line 292 of file EventProcessor.h.

References edm::get_underlying_safe(), and looper_.

Referenced by endJob().

{ return get_underlying_safe(looper_); }

looper() [2/2]

std::shared_ptr<EDLooperBase>& edm::EventProcessor::looper ( )

inlineprivate

Definition at line 293 of file EventProcessor.h.

References edm::get_underlying_safe(), and looper_.

{ return get_underlying_safe(looper_); }

needToCallNext()

bool edm::EventProcessor::needToCallNext ( ) const

inlineprivate

Definition at line 298 of file EventProcessor.h.

References needToCallNext_.

Referenced by handleNextItemAfterMergingRunEntries(), and readNextEventForStream().

{ return needToCallNext_; }

nextTransitionType()

InputSource::ItemTypeInfo edm::EventProcessor::nextTransitionType

(

)

Definition at line 870 of file EventProcessor.cc.

References actReg_, checkForAsyncStopRequest(), epSuccess, edm::ExternalSignal, input_, edm::InputSource::IsStop, edm::InputSource::IsSynchronize, lastSourceTransition_, and runEdmFileComparison::returnCode.

Referenced by continueLumiAsync(), nextTransitionTypeAsync(), processRuns(), readAndMergeLumiEntriesAsync(), readAndMergeRunEntriesAsync(), and readNextEventForStream().

                                                             {
    SendSourceTerminationSignalIfException sentry(actReg_.get());
    InputSource::ItemTypeInfo itemTypeInfo;
    {
      SourceNextGuard guard(*actReg_.get());
      //For now, do nothing with InputSource::IsSynchronize
      do {
        itemTypeInfo = input_->nextItemType();
      } while (itemTypeInfo == InputSource::ItemType::IsSynchronize);
    }
    lastSourceTransition_ = itemTypeInfo;
    sentry.completedSuccessfully();

    StatusCode returnCode = epSuccess;

    if (checkForAsyncStopRequest(returnCode)) {
      actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
      lastSourceTransition_ = InputSource::ItemType::IsStop;
    }

    return lastSourceTransition_;
  }

nextTransitionTypeAsync()

void edm::EventProcessor::nextTransitionTypeAsync	(	std::shared_ptr< RunProcessingStatus >	iRunStatus,
		WaitingTaskHolder	nextTask
	)

Definition at line 893 of file EventProcessor.cc.

References CMS_SA_ALLOW, Exception, edm::WaitingTaskHolder::group(), watchdog::group, input_, edm::InputSource::IsRun, lastTransitionType(), edm::errors::LogicError, eostools::move(), nextTransitionType(), edm::SerialTaskQueueChain::push(), edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, sourceMutex_, and sourceResourcesAcquirer_.

Referenced by handleNextItemAfterMergingRunEntries().

                                                                           {
    auto group = nextTask.group();
    sourceResourcesAcquirer_.serialQueueChain().push(
        *group, [this, runStatus = std::move(iRunStatus), nextHolder = std::move(nextTask)]() mutable {
          CMS_SA_ALLOW try {
            ServiceRegistry::Operate operate(serviceToken_);
            std::lock_guard<std::recursive_mutex> guard(*(sourceMutex_.get()));
            nextTransitionType();
            if (lastTransitionType() == InputSource::ItemType::IsRun &&
                runStatus->runPrincipal()->run() == input_->run() &&
                runStatus->runPrincipal()->reducedProcessHistoryID() == input_->reducedProcessHistoryID()) {
              throw Exception(errors::LogicError)
                  << "InputSource claimed previous Run Entry was last to be merged in this file,\n"
                  << "but the next entry has the same run number and reduced ProcessHistoryID.\n"
                  << "This is probably a bug in the InputSource. Please report to the Core group.\n";
            }
          } catch (...) {
            nextHolder.doneWaiting(std::current_exception());
          }
        });
  }

openOutputFiles()

void edm::EventProcessor::openOutputFiles

(

)

Definition at line 1017 of file EventProcessor.cc.

References fb_, FDEBUG, fileBlockValid(), edm::for_all(), schedule_, and subProcesses_.

                                       {
    if (fileBlockValid()) {
      schedule_->openOutputFiles(*fb_);
      for_all(subProcesses_, [this](auto& subProcess) { subProcess.openOutputFiles(*fb_); });
    }
    FDEBUG(1) << "\topenOutputFiles\n";
  }

operator=()

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

delete

preg() [1/2]

std::shared_ptr<ProductRegistry const> edm::EventProcessor::preg ( ) const

inlineprivate

Definition at line 280 of file EventProcessor.h.

References edm::get_underlying_safe(), and preg_.

Referenced by beginJob(), init(), readAndMergeLumi(), and readFile().

{ return get_underlying_safe(preg_); }

preg() [2/2]

std::shared_ptr<ProductRegistry>& edm::EventProcessor::preg ( )

inlineprivate

Definition at line 281 of file EventProcessor.h.

References edm::get_underlying_safe(), and preg_.

{ return get_underlying_safe(preg_); }

prepareForNextLoop()

void edm::EventProcessor::prepareForNextLoop

(

)

Definition at line 1081 of file EventProcessor.cc.

References esp_, FDEBUG, and looper_.

Referenced by runToCompletion().

                                          {
    looper_->prepareForNextLoop(esp_.get());
    FDEBUG(1) << "\tprepareForNextLoop\n";
  }

processConfiguration()

ProcessConfiguration const& edm::EventProcessor::processConfiguration ( ) const

inline

Definition at line 143 of file EventProcessor.h.

References processConfiguration_.

{ return *processConfiguration_; }

processEventAsync()

void edm::EventProcessor::processEventAsync ( WaitingTaskHolder  iHolder, unsigned int  iStreamIndex  )

private

Definition at line 2361 of file EventProcessor.cc.

References edm::WaitingTaskHolder::group(), and processEventAsyncImpl().

Referenced by handleNextEventForStreamAsync().

                                                                                             {
    iHolder.group()->run([this, iHolder, iStreamIndex]() { processEventAsyncImpl(iHolder, iStreamIndex); });
  }

processEventAsyncImpl()

void edm::EventProcessor::processEventAsyncImpl ( WaitingTaskHolder  iHolder, unsigned int  iStreamIndex  )

private

Definition at line 2377 of file EventProcessor.cc.

References actReg_, esp_, makeMEIFBenchmarkPlots::ev, edm::PrincipalCache::eventPrincipal(), FDEBUG, dqmdumpme::first, edm::waiting_task::chain::ifThen(), info(), edm::Service< T >::isAvailable(), edm::StreamContext::kEvent, looper_, eostools::move(), principalCache_, processContext_, processEventWithLooper(), groupFilesInBlocks::reverse, edm::waiting_task::chain::runLast(), schedule_, serviceToken_, streamLumiStatus_, subProcesses_, and edm::waiting_task::chain::then().

Referenced by processEventAsync().

                                                                                                 {
    auto pep = &(principalCache_.eventPrincipal(iStreamIndex));

    ServiceRegistry::Operate operate(serviceToken_);
    Service<RandomNumberGenerator> rng;
    if (rng.isAvailable()) {
      Event ev(*pep, ModuleDescription(), nullptr);
      rng->postEventRead(ev);
    }

    EventSetupImpl const& es = streamLumiStatus_[iStreamIndex]->eventSetupImpl(esp_->subProcessIndex());
    using namespace edm::waiting_task::chain;
    chain::first([this, &es, pep, iStreamIndex](auto nextTask) {
      EventTransitionInfo info(*pep, es);
      schedule_->processOneEventAsync(std::move(nextTask), iStreamIndex, info, serviceToken_);
    }) | ifThen(not subProcesses_.empty(), [this, pep, iStreamIndex](auto nextTask) {
      for (auto& subProcess : boost::adaptors::reverse(subProcesses_)) {
        subProcess.doEventAsync(nextTask, *pep, &streamLumiStatus_[iStreamIndex]->eventSetupImpls());
      }
    }) | ifThen(looper_, [this, iStreamIndex, pep](auto nextTask) {
      //NOTE: behavior change. previously if an exception happened looper was still called. Now it will not be called
      ServiceRegistry::Operate operateLooper(serviceToken_);
      processEventWithLooper(*pep, iStreamIndex);
    }) | then([this, pep](auto nextTask) {
      FDEBUG(1) << "\tprocessEvent\n";
      StreamContext streamContext(pep->streamID(),
                                  StreamContext::Transition::kEvent,
                                  pep->id(),
                                  pep->runPrincipal().index(),
                                  pep->luminosityBlockPrincipal().index(),
                                  pep->time(),
                                  &processContext_);
      ClearEventGuard guard(*this->actReg_.get(), streamContext);
      pep->clearEventPrincipal();
    }) | runLast(iHolder);
  }

processEventWithLooper()

void edm::EventProcessor::processEventWithLooper ( EventPrincipal &  iPrincipal, unsigned int  iStreamIndex  )

private

Definition at line 2414 of file EventProcessor.cc.

References esp_, input_, edm::EDLooperBase::kContinue, edm::ProcessingController::kToPreviousEvent, edm::ProcessingController::kToSpecifiedEvent, edm::ProcessingController::lastOperationSucceeded(), looper_, processContext_, edm::ProcessingController::requestedTransition(), edm::ProcessingController::setLastOperationSucceeded(), shouldWeStop_, edm::ProcessingController::specifiedEventTransition(), mps_update::status, edm::EventPrincipal::streamID(), streamLumiStatus_, and summarizeEdmComparisonLogfiles::succeeded.

Referenced by processEventAsyncImpl().

                                                                                                   {
    bool randomAccess = input_->randomAccess();
    ProcessingController::ForwardState forwardState = input_->forwardState();
    ProcessingController::ReverseState reverseState = input_->reverseState();
    ProcessingController pc(forwardState, reverseState, randomAccess);

    EDLooperBase::Status status = EDLooperBase::kContinue;
    do {
      StreamContext streamContext(iPrincipal.streamID(), &processContext_);
      EventSetupImpl const& es = streamLumiStatus_[iStreamIndex]->eventSetupImpl(esp_->subProcessIndex());
      status = looper_->doDuringLoop(iPrincipal, es, pc, &streamContext);

      bool succeeded = true;
      if (randomAccess) {
        if (pc.requestedTransition() == ProcessingController::kToPreviousEvent) {
          input_->skipEvents(-2);
        } else if (pc.requestedTransition() == ProcessingController::kToSpecifiedEvent) {
          succeeded = input_->goToEvent(pc.specifiedEventTransition());
        }
      }
      pc.setLastOperationSucceeded(succeeded);
    } while (!pc.lastOperationSucceeded());
    if (status != EDLooperBase::kContinue) {
      shouldWeStop_ = true;
    }
  }

processRuns()

InputSource::ItemType edm::EventProcessor::processRuns

(

)

Definition at line 1176 of file EventProcessor.cc.

References cms::cuda::assert(), beginRunAsync(), continueLumiAsync(), handleNextItemAfterMergingRunEntries(), input_, edm::InputSource::IsRun, lastTransitionType(), eostools::move(), nextTransitionType(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, readAndMergeRun(), setNeedToCallNext(), streamLumiActive_, streamRunActive_, streamRunStatus_, and taskGroup_.

                                                  {
    FinalWaitingTask waitTask{taskGroup_};
    assert(lastTransitionType() == InputSource::ItemType::IsRun);
    if (streamRunActive_ == 0) {
      assert(streamLumiActive_ == 0);

      beginRunAsync(IOVSyncValue(EventID(input_->run(), 0, 0), input_->runAuxiliary()->beginTime()),
                    WaitingTaskHolder{taskGroup_, &waitTask});
    } else {
      assert(streamRunActive_ == preallocations_.numberOfStreams());

      auto runStatus = streamRunStatus_[0];

      while (lastTransitionType() == InputSource::ItemType::IsRun and
             runStatus->runPrincipal()->run() == input_->run() and
             runStatus->runPrincipal()->reducedProcessHistoryID() == input_->reducedProcessHistoryID()) {
        readAndMergeRun(*runStatus);
        nextTransitionType();
      }

      setNeedToCallNext(false);

      WaitingTaskHolder holder{taskGroup_, &waitTask};
      runStatus->setHolderOfTaskInProcessRuns(holder);
      if (streamLumiActive_ > 0) {
        assert(streamLumiActive_ == preallocations_.numberOfStreams());
        continueLumiAsync(std::move(holder));
      } else {
        handleNextItemAfterMergingRunEntries(std::move(runStatus), std::move(holder));
      }
    }
    waitTask.wait();
    return lastTransitionType();
  }

readAndMergeLumi()

void edm::EventProcessor::readAndMergeLumi

(

LuminosityBlockProcessingStatus &

iStatus

)

Definition at line 2037 of file EventProcessor.cc.

References actReg_, cms::cuda::assert(), input_, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), or, and preg().

Referenced by continueLumiAsync(), and readAndMergeLumiEntriesAsync().

                                                                                {
    auto& lumiPrincipal = *iStatus.lumiPrincipal();
    assert(lumiPrincipal.aux().sameIdentity(*input_->luminosityBlockAuxiliary()) or
           input_->processHistoryRegistry().reducedProcessHistoryID(lumiPrincipal.aux().processHistoryID()) ==
               input_->processHistoryRegistry().reducedProcessHistoryID(
                   input_->luminosityBlockAuxiliary()->processHistoryID()));
    bool lumiOK = lumiPrincipal.adjustToNewProductRegistry(*preg());
    assert(lumiOK);
    lumiPrincipal.mergeAuxiliary(*input_->luminosityBlockAuxiliary());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readAndMergeLumi(*iStatus.lumiPrincipal());
      sentry.completedSuccessfully();
    }
  }

readAndMergeLumiEntriesAsync()

void edm::EventProcessor::readAndMergeLumiEntriesAsync ( std::shared_ptr< LuminosityBlockProcessingStatus >  iLumiStatus, WaitingTaskHolder  iHolder  )

private

Definition at line 2152 of file EventProcessor.cc.

References CMS_SA_ALLOW, edm::WaitingTaskHolder::group(), watchdog::group, input_, edm::InputSource::IsLumi, edm::InputSource::LastItemToBeMerged, lastTransitionType(), eostools::move(), nextTransitionType(), edm::SerialTaskQueueChain::push(), readAndMergeLumi(), edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, setNeedToCallNext(), sourceMutex_, and sourceResourcesAcquirer_.

Referenced by beginLumiAsync().

                                                                               {
    auto group = iHolder.group();
    sourceResourcesAcquirer_.serialQueueChain().push(
        *group, [this, iLumiStatus = std::move(iLumiStatus), holder = std::move(iHolder)]() mutable {
          CMS_SA_ALLOW try {
            ServiceRegistry::Operate operate(serviceToken_);

            std::lock_guard<std::recursive_mutex> guard(*(sourceMutex_.get()));

            nextTransitionType();
            setNeedToCallNext(false);

            while (lastTransitionType() == InputSource::ItemType::IsLumi and
                   iLumiStatus->lumiPrincipal()->luminosityBlock() == input_->luminosityBlock()) {
              readAndMergeLumi(*iLumiStatus);
              if (lastTransitionType().itemPosition() == InputSource::ItemPosition::LastItemToBeMerged) {
                setNeedToCallNext(true);
                return;
              }
              nextTransitionType();
            }
          } catch (...) {
            holder.doneWaiting(std::current_exception());
          }
        });
  }

readAndMergeRun()

void edm::EventProcessor::readAndMergeRun

(

RunProcessingStatus &

iStatus

)

Definition at line 2011 of file EventProcessor.cc.

References actReg_, edm::Principal::adjustToNewProductRegistry(), cms::cuda::assert(), input_, edm::RunPrincipal::mergeAuxiliary(), preg_, and edm::RunProcessingStatus::runPrincipal().

Referenced by processRuns(), and readAndMergeRunEntriesAsync().

                                                                   {
    RunPrincipal& runPrincipal = *iStatus.runPrincipal();

    bool runOK = runPrincipal.adjustToNewProductRegistry(*preg_);
    assert(runOK);
    runPrincipal.mergeAuxiliary(*input_->runAuxiliary());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readAndMergeRun(runPrincipal);
      sentry.completedSuccessfully();
    }
  }

readAndMergeRunEntriesAsync()

void edm::EventProcessor::readAndMergeRunEntriesAsync ( std::shared_ptr< RunProcessingStatus >  iRunStatus, WaitingTaskHolder  iHolder  )

private

Definition at line 2118 of file EventProcessor.cc.

References CMS_SA_ALLOW, edm::WaitingTaskHolder::group(), watchdog::group, input_, edm::InputSource::IsRun, edm::InputSource::LastItemToBeMerged, lastTransitionType(), eostools::move(), nextTransitionType(), edm::SerialTaskQueueChain::push(), readAndMergeRun(), edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, setNeedToCallNext(), sourceMutex_, sourceResourcesAcquirer_, and mps_update::status.

Referenced by beginRunAsync().

                                                                              {
    auto group = iHolder.group();
    sourceResourcesAcquirer_.serialQueueChain().push(
        *group, [this, status = std::move(iRunStatus), holder = std::move(iHolder)]() mutable {
          CMS_SA_ALLOW try {
            ServiceRegistry::Operate operate(serviceToken_);

            std::lock_guard<std::recursive_mutex> guard(*(sourceMutex_.get()));

            nextTransitionType();
            setNeedToCallNext(false);

            while (lastTransitionType() == InputSource::ItemType::IsRun and
                   status->runPrincipal()->run() == input_->run() and
                   status->runPrincipal()->reducedProcessHistoryID() == input_->reducedProcessHistoryID()) {
              if (status->holderOfTaskInProcessRuns().taskHasFailed()) {
                status->setStopBeforeProcessingRun(true);
                return;
              }
              readAndMergeRun(*status);
              if (lastTransitionType().itemPosition() == InputSource::ItemPosition::LastItemToBeMerged) {
                setNeedToCallNext(true);
                return;
              }
              nextTransitionType();
            }
          } catch (...) {
            status->setStopBeforeProcessingRun(true);
            holder.doneWaiting(std::current_exception());
          }
        });
  }

readEvent()

void edm::EventProcessor::readEvent ( unsigned int  iStreamIndex )

private

Definition at line 2346 of file EventProcessor.cc.

References actReg_, edm::PrincipalCache::eventPrincipal(), FDEBUG, input_, principalCache_, processContext_, streamLumiStatus_, and streamRunStatus_.

Referenced by readNextEventForStream().

                                                          {
    //TODO this will have to become per stream
    auto& event = principalCache_.eventPrincipal(iStreamIndex);
    StreamContext streamContext(event.streamID(), &processContext_);

    SendSourceTerminationSignalIfException sentry(actReg_.get());
    input_->readEvent(event, streamContext);

    streamRunStatus_[iStreamIndex]->updateLastTimestamp(input_->timestamp());
    streamLumiStatus_[iStreamIndex]->updateLastTimestamp(input_->timestamp());
    sentry.completedSuccessfully();

    FDEBUG(1) << "\treadEvent\n";
  }

readFile()

void edm::EventProcessor::readFile

(

)

Definition at line 983 of file EventProcessor.cc.

References actReg_, edm::PrincipalCache::adjustEventsToNewProductRegistry(), edm::PrincipalCache::adjustIndexesAfterProductRegistryAddition(), fb_, FDEBUG, input_, edm::PreallocationConfiguration::numberOfStreams(), edm::PreallocationConfiguration::numberOfThreads(), edm::FileBlock::ParallelProcesses, preallocations_, preg(), preg_, principalCache_, streamLumiActive_, streamLumiStatus_, streamRunActive_, and streamRunStatus_.

                                {
    FDEBUG(1) << " \treadFile\n";
    size_t size = preg_->size();
    SendSourceTerminationSignalIfException sentry(actReg_.get());

    if (streamRunActive_ > 0) {
      streamRunStatus_[0]->runPrincipal()->preReadFile();
      streamRunStatus_[0]->runPrincipal()->adjustIndexesAfterProductRegistryAddition();
    }

    if (streamLumiActive_ > 0) {
      streamLumiStatus_[0]->lumiPrincipal()->adjustIndexesAfterProductRegistryAddition();
    }

    fb_ = input_->readFile();
    if (size < preg_->size()) {
      principalCache_.adjustIndexesAfterProductRegistryAddition();
    }
    principalCache_.adjustEventsToNewProductRegistry(preg());
    if (preallocations_.numberOfStreams() > 1 and preallocations_.numberOfThreads() > 1) {
      fb_->setNotFastClonable(FileBlock::ParallelProcesses);
    }
    sentry.completedSuccessfully();
  }

readLuminosityBlock()

std::shared_ptr< LuminosityBlockPrincipal > edm::EventProcessor::readLuminosityBlock

(

std::shared_ptr< RunPrincipal >

rp

)

Definition at line 2024 of file EventProcessor.cc.

References actReg_, cms::cuda::assert(), edm::PrincipalCache::getAvailableLumiPrincipalPtr(), historyAppender_, input_, eostools::move(), and principalCache_.

Referenced by beginLumiAsync().

                                                                                                            {
    auto lbp = principalCache_.getAvailableLumiPrincipalPtr();
    assert(lbp);
    lbp->setAux(*input_->luminosityBlockAuxiliary());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readLuminosityBlock(*lbp, *historyAppender_);
      sentry.completedSuccessfully();
    }
    lbp->setRunPrincipal(std::move(rp));
    return lbp;
  }

readNextEventForStream()

bool edm::EventProcessor::readNextEventForStream ( WaitingTaskHolder const &  iTask, unsigned int  iStreamIndex, LuminosityBlockProcessingStatus &  iStatus  )

private

Definition at line 2214 of file EventProcessor.cc.

References edm::LuminosityBlockProcessingStatus::eventProcessingState(), Exception, input_, edm::InputSource::IsEvent, edm::InputSource::IsLumi, edm::InputSource::IsRun, edm::InputSource::IsStop, edm::LuminosityBlockProcessingStatus::kPauseForFileTransition, edm::LuminosityBlockProcessingStatus::kProcessing, edm::LuminosityBlockProcessingStatus::kStopLumi, lastSourceTransition_, lastTransitionType(), edm::errors::LogicError, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), needToCallNext(), nextTransitionType(), or, readEvent(), serviceToken_, edm::LuminosityBlockProcessingStatus::setEventProcessingState(), setNeedToCallNext(), shouldWeStop(), sourceMutex_, and edm::WaitingTaskHolder::taskHasFailed().

Referenced by handleNextEventForStreamAsync().

                                                                                        {
    // This function returns true if it successfully reads an event for the stream and that
    // requires both that an event is next and there are no problems or requests to stop.

    if (iTask.taskHasFailed()) {
      // We want all streams to stop or all streams to pause. If we are already in the
      // middle of pausing streams, then finish pausing all of them and the lumi will be
      // ended later. Otherwise, just end it now.
      if (iStatus.eventProcessingState() == LuminosityBlockProcessingStatus::EventProcessingState::kProcessing) {
        iStatus.setEventProcessingState(LuminosityBlockProcessingStatus::EventProcessingState::kStopLumi);
      }
      return false;
    }

    // Did another stream already stop or pause this lumi?
    if (iStatus.eventProcessingState() != LuminosityBlockProcessingStatus::EventProcessingState::kProcessing) {
      return false;
    }

    // Are output modules or the looper requesting we stop?
    if (shouldWeStop()) {
      lastSourceTransition_ = InputSource::ItemType::IsStop;
      iStatus.setEventProcessingState(LuminosityBlockProcessingStatus::EventProcessingState::kStopLumi);
      return false;
    }

    ServiceRegistry::Operate operate(serviceToken_);

    // need to use lock in addition to the serial task queue because
    // of delayed provenance reading and reading data in response to
    // edm::Refs etc
    std::lock_guard<std::recursive_mutex> guard(*(sourceMutex_.get()));

    // If we didn't already call nextTransitionType while merging lumis, call it here.
    // This asks the input source what is next and also checks for signals.

    InputSource::ItemType itemType = needToCallNext() ? nextTransitionType() : lastTransitionType();
    setNeedToCallNext(true);

    if (InputSource::ItemType::IsEvent != itemType) {
      // IsFile may continue processing the lumi and
      // looper_ can cause the input source to declare a new IsRun which is actually
      // just a continuation of the previous run
      if (InputSource::ItemType::IsStop == itemType or InputSource::ItemType::IsLumi == itemType or
          (InputSource::ItemType::IsRun == itemType and
           (iStatus.lumiPrincipal()->run() != input_->run() or
            iStatus.lumiPrincipal()->runPrincipal().reducedProcessHistoryID() != input_->reducedProcessHistoryID()))) {
        if (itemType == InputSource::ItemType::IsLumi &&
            iStatus.lumiPrincipal()->luminosityBlock() == input_->luminosityBlock()) {
          throw Exception(errors::LogicError)
              << "InputSource claimed previous Lumi Entry was last to be merged in this file,\n"
              << "but the next lumi entry has the same lumi number.\n"
              << "This is probably a bug in the InputSource. Please report to the Core group.\n";
        }
        iStatus.setEventProcessingState(LuminosityBlockProcessingStatus::EventProcessingState::kStopLumi);
      } else {
        iStatus.setEventProcessingState(LuminosityBlockProcessingStatus::EventProcessingState::kPauseForFileTransition);
      }
      return false;
    }
    readEvent(iStreamIndex);
    return true;
  }

readProcessBlock()

void edm::EventProcessor::readProcessBlock

(

ProcessBlockPrincipal &

processBlockPrincipal

)

Definition at line 1992 of file EventProcessor.cc.

References actReg_, and input_.

Referenced by inputProcessBlocks().

                                                                                    {
    SendSourceTerminationSignalIfException sentry(actReg_.get());
    input_->readProcessBlock(processBlockPrincipal);
    sentry.completedSuccessfully();
  }

readRun()

std::shared_ptr< RunPrincipal > edm::EventProcessor::readRun

(

)

Definition at line 1998 of file EventProcessor.cc.

References actReg_, cms::cuda::assert(), edm::PrincipalCache::getAvailableRunPrincipalPtr(), historyAppender_, input_, and principalCache_.

Referenced by beginRunAsync().

                                                      {
    auto rp = principalCache_.getAvailableRunPrincipalPtr();
    assert(rp);
    rp->setAux(*input_->runAuxiliary());
    {
      SendSourceTerminationSignalIfException sentry(actReg_.get());
      input_->readRun(*rp, *historyAppender_);
      sentry.completedSuccessfully();
    }
    assert(input_->reducedProcessHistoryID() == rp->reducedProcessHistoryID());
    return rp;
  }

releaseBeginRunResources()

void edm::EventProcessor::releaseBeginRunResources

(

unsigned int

iStream

)

Definition at line 1452 of file EventProcessor.cc.

References queueWhichWaitsForIOVsToFinish_, edm::SerialTaskQueue::resume(), mps_update::status, streamQueues_, and streamRunStatus_.

Referenced by streamBeginRunAsync().

                                                                    {
    auto& status = streamRunStatus_[iStream];
    if (status->streamFinishedBeginRun()) {
      status->resetBeginResources();
      queueWhichWaitsForIOVsToFinish_.resume();
    }
    streamQueues_[iStream].resume();
  }

respondToCloseInputFile()

void edm::EventProcessor::respondToCloseInputFile

(

)

Definition at line 1042 of file EventProcessor.cc.

References fb_, FDEBUG, fileBlockValid(), edm::for_all(), schedule_, and subProcesses_.

                                               {
    if (fileBlockValid()) {
      schedule_->respondToCloseInputFile(*fb_);
      for_all(subProcesses_, [this](auto& subProcess) { subProcess.respondToCloseInputFile(*fb_); });
    }
    FDEBUG(1) << "\trespondToCloseInputFile\n";
  }

respondToOpenInputFile()

void edm::EventProcessor::respondToOpenInputFile

(

)

Definition at line 1032 of file EventProcessor.cc.

References branchIDListHelper_, fb_, FDEBUG, fileBlockValid(), edm::for_all(), schedule_, and subProcesses_.

                                              {
    if (fileBlockValid()) {
      for_all(subProcesses_,
              [this](auto& subProcess) { subProcess.updateBranchIDListHelper(branchIDListHelper_->branchIDLists()); });
      schedule_->respondToOpenInputFile(*fb_);
      for_all(subProcesses_, [this](auto& subProcess) { subProcess.respondToOpenInputFile(*fb_); });
    }
    FDEBUG(1) << "\trespondToOpenInputFile\n";
  }

rewindInput()

void edm::EventProcessor::rewindInput

(

)

Definition at line 1075 of file EventProcessor.cc.

References FDEBUG, and input_.

Referenced by runToCompletion().

                                   {
    input_->repeat();
    input_->rewind();
    FDEBUG(1) << "\trewind\n";
  }

run()

EventProcessor::StatusCode edm::EventProcessor::run ( )

inline

Definition at line 381 of file EventProcessor.h.

References runToCompletion().

Referenced by Types.EventID::cppID(), and Types.LuminosityBlockID::cppID().

{ return runToCompletion(); }

runToCompletion()

EventProcessor::StatusCode edm::EventProcessor::runToCompletion

(

)

Definition at line 916 of file EventProcessor.cc.

References actReg_, beginJob(), deferredExceptionPtr_, deferredExceptionPtrIsSet_, doErrorStuff(), MillePedeFileConverter_cfg::e, endOfLoop(), epSuccess, Exception, exceptionMessageFiles_, exceptionMessageLumis_, exceptionMessageRuns_, fileModeNoMerge_, personalPlayback::fp, edm::InputSource::IsStop, prepareForNextLoop(), rewindInput(), serviceToken_, startingNewLoop(), AlCaHLTBitMon_QueryRunRegistry::string, and edm::convertException::wrap().

Referenced by PythonEventProcessor::run(), and run().

                                                           {
    beginJob();  //make sure this was called

    // make the services available
    ServiceRegistry::Operate operate(serviceToken_);
    actReg_->beginProcessingSignal_();
    auto endSignal = [](ActivityRegistry* iReg) { iReg->endProcessingSignal_(); };
    std::unique_ptr<ActivityRegistry, decltype(endSignal)> guard(actReg_.get(), endSignal);
    try {
      FilesProcessor fp(fileModeNoMerge_);

      convertException::wrap([&]() {
        bool firstTime = true;
        do {
          if (not firstTime) {
            prepareForNextLoop();
            rewindInput();
          } else {
            firstTime = false;
          }
          startingNewLoop();

          auto trans = fp.processFiles(*this);

          fp.normalEnd();

          if (deferredExceptionPtrIsSet_.load()) {
            std::rethrow_exception(deferredExceptionPtr_);
          }
          if (trans != InputSource::ItemType::IsStop) {
            //problem with the source
            doErrorStuff();

            throw cms::Exception("BadTransition") << "Unexpected transition change " << static_cast<int>(trans);
          }
        } while (not endOfLoop());
      });  // convertException::wrap

    }  // Try block
    catch (cms::Exception& e) {
      if (exceptionMessageLumis_) {
        std::string message(
            "Another exception was caught while trying to clean up lumis after the primary fatal exception.");
        e.addAdditionalInfo(message);
        if (e.alreadyPrinted()) {
          LogAbsolute("Additional Exceptions") << message;
        }
      }
      if (exceptionMessageRuns_) {
        std::string message(
            "Another exception was caught while trying to clean up runs after the primary fatal exception.");
        e.addAdditionalInfo(message);
        if (e.alreadyPrinted()) {
          LogAbsolute("Additional Exceptions") << message;
        }
      }
      if (!exceptionMessageFiles_.empty()) {
        e.addAdditionalInfo(exceptionMessageFiles_);
        if (e.alreadyPrinted()) {
          LogAbsolute("Additional Exceptions") << exceptionMessageFiles_;
        }
      }
      throw;
    }
    return epSuccess;
  }

setDeferredException()

bool edm::EventProcessor::setDeferredException

(

std::exception_ptr

iException

)

Definition at line 2462 of file EventProcessor.cc.

References deferredExceptionPtr_, and deferredExceptionPtrIsSet_.

                                                                       {
    bool expected = false;
    if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
      deferredExceptionPtr_ = iException;
      return true;
    }
    return false;
  }

setExceptionMessageFiles()

void edm::EventProcessor::setExceptionMessageFiles

(

std::string &

message

)

Definition at line 2456 of file EventProcessor.cc.

References exceptionMessageFiles_.

{ exceptionMessageFiles_ = message; }

setExceptionMessageLumis()

void edm::EventProcessor::setExceptionMessageLumis

(

)

Definition at line 2460 of file EventProcessor.cc.

References exceptionMessageLumis_.

Referenced by handleEndLumiExceptions().

{ exceptionMessageLumis_ = true; }

setExceptionMessageRuns()

void edm::EventProcessor::setExceptionMessageRuns

(

)

Definition at line 2458 of file EventProcessor.cc.

References exceptionMessageRuns_.

Referenced by handleEndRunExceptions().

{ exceptionMessageRuns_ = true; }

setNeedToCallNext()

void edm::EventProcessor::setNeedToCallNext ( bool  val )

inlineprivate

Definition at line 299 of file EventProcessor.h.

References needToCallNext_, and heppy_batch::val.

Referenced by beginLumiAsync(), beginRunAsync(), processRuns(), readAndMergeLumiEntriesAsync(), readAndMergeRunEntriesAsync(), and readNextEventForStream().

{ needToCallNext_ = val; }

shouldWeCloseOutput()

bool edm::EventProcessor::shouldWeCloseOutput

(

)

const

Definition at line 1086 of file EventProcessor.cc.

References FDEBUG, schedule_, and subProcesses_.

                                                 {
    FDEBUG(1) << "\tshouldWeCloseOutput\n";
    if (!subProcesses_.empty()) {
      for (auto const& subProcess : subProcesses_) {
        if (subProcess.shouldWeCloseOutput()) {
          return true;
        }
      }
      return false;
    }
    return schedule_->shouldWeCloseOutput();
  }

shouldWeStop()

bool edm::EventProcessor::shouldWeStop

(

)

const

Definition at line 2441 of file EventProcessor.cc.

References FDEBUG, schedule_, shouldWeStop_, and subProcesses_.

Referenced by readNextEventForStream().

                                          {
    FDEBUG(1) << "\tshouldWeStop\n";
    if (shouldWeStop_)
      return true;
    if (!subProcesses_.empty()) {
      for (auto const& subProcess : subProcesses_) {
        if (subProcess.terminate()) {
          return true;
        }
      }
      return false;
    }
    return schedule_->terminate();
  }

startingNewLoop()

void edm::EventProcessor::startingNewLoop

(

)

Definition at line 1050 of file EventProcessor.cc.

References FDEBUG, looper_, looperBeginJobRun_, and shouldWeStop_.

Referenced by runToCompletion().

                                       {
    shouldWeStop_ = false;
    //NOTE: for first loop, need to delay calling 'doStartingNewLoop'
    // until after we've called beginOfJob
    if (looper_ && looperBeginJobRun_) {
      looper_->doStartingNewLoop();
    }
    FDEBUG(1) << "\tstartingNewLoop\n";
  }

streamBeginRunAsync()

void edm::EventProcessor::streamBeginRunAsync	(	unsigned int	iStream,
		std::shared_ptr< RunProcessingStatus >	status,
		bool	precedingTasksSucceeded,
		WaitingTaskHolder	iHolder
	)

Definition at line 1420 of file EventProcessor.cc.

References CMS_SA_ALLOW, edm::WaitingTaskHolder::doneWaiting(), esp_, edm::RunProcessingStatus::eventSetupImpl(), edm::RunProcessingStatus::eventSetupImpls(), dqmdumpme::first, eostools::move(), releaseBeginRunResources(), edm::waiting_task::chain::runLast(), edm::RunProcessingStatus::runPrincipal(), schedule_, serviceToken_, mps_update::status, streamQueues_, streamRunActive_, streamRunStatus_, subProcesses_, and edm::waiting_task::chain::then().

Referenced by beginRunAsync().

                                                                      {
    // These shouldn't throw
    streamQueues_[iStream].pause();
    ++streamRunActive_;
    streamRunStatus_[iStream] = std::move(status);

    CMS_SA_ALLOW try {
      using namespace edm::waiting_task::chain;
      chain::first([this, iStream, precedingTasksSucceeded](auto nextTask) {
        if (precedingTasksSucceeded) {
          RunProcessingStatus& rs = *streamRunStatus_[iStream];
          RunTransitionInfo transitionInfo(
              *rs.runPrincipal(), rs.eventSetupImpl(esp_->subProcessIndex()), &rs.eventSetupImpls());
          using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamBegin>;
          beginStreamTransitionAsync<Traits>(
              std::move(nextTask), *schedule_, iStream, transitionInfo, serviceToken_, subProcesses_);
        }
      }) | then([this, iStream](std::exception_ptr const* exceptionFromBeginStreamRun, auto nextTask) {
        if (exceptionFromBeginStreamRun) {
          nextTask.doneWaiting(*exceptionFromBeginStreamRun);
        }
        releaseBeginRunResources(iStream);
      }) | runLast(iHolder);
    } catch (...) {
      releaseBeginRunResources(iStream);
      iHolder.doneWaiting(std::current_exception());
    }
  }

streamEndLumiAsync()

void edm::EventProcessor::streamEndLumiAsync	(	edm::WaitingTaskHolder	iTask,
		unsigned int	iStreamIndex
	)

Definition at line 1931 of file EventProcessor.cc.

References esp_, globalEndLumiAsync(), edm::WaitingTaskHolder::group(), handleEndLumiExceptions(), edm::make_waiting_task(), eostools::move(), schedule_, serviceToken_, mps_update::status, streamLumiActive_, streamLumiStatus_, streamQueues_, subProcesses_, submitPVValidationJobs::t, and edm::WaitingTaskHolder::taskHasFailed().

Referenced by endUnfinishedLumi(), and handleNextEventForStreamAsync().

                                                                                               {
    auto t = edm::make_waiting_task([this, iStreamIndex, iTask](std::exception_ptr const* iException) mutable {
      auto status = streamLumiStatus_[iStreamIndex];
      if (iException) {
        handleEndLumiExceptions(*iException, iTask);
      }

      // reset status before releasing queue else get race condition
      streamLumiStatus_[iStreamIndex].reset();
      --streamLumiActive_;
      streamQueues_[iStreamIndex].resume();

      //are we the last one?
      if (status->streamFinishedLumi()) {
        globalEndLumiAsync(iTask, std::move(status));
      }
    });

    edm::WaitingTaskHolder lumiDoneTask{*iTask.group(), t};

    // Need to be sure the lumi status is released before lumiDoneTask can every be called.
    // therefore we do not want to hold the shared_ptr
    auto lumiStatus = streamLumiStatus_[iStreamIndex].get();
    lumiStatus->setEndTime();

    EventSetupImpl const& es = lumiStatus->eventSetupImpl(esp_->subProcessIndex());
    auto eventSetupImpls = &lumiStatus->eventSetupImpls();
    bool cleaningUpAfterException = lumiStatus->cleaningUpAfterException() || iTask.taskHasFailed();

    if (lumiStatus->didGlobalBeginSucceed()) {
      auto& lumiPrincipal = *lumiStatus->lumiPrincipal();
      using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd>;
      LumiTransitionInfo transitionInfo(lumiPrincipal, es, eventSetupImpls);
      endStreamTransitionAsync<Traits>(std::move(lumiDoneTask),
                                       *schedule_,
                                       iStreamIndex,
                                       transitionInfo,
                                       serviceToken_,
                                       subProcesses_,
                                       cleaningUpAfterException);
    }
  }

streamEndRunAsync()

void edm::EventProcessor::streamEndRunAsync	(	WaitingTaskHolder	iTask,
		unsigned int	iStreamIndex
	)

Definition at line 1600 of file EventProcessor.cc.

References CMS_SA_ALLOW, esp_, exceptionRunStatus_, edm::WaitingTaskHolder::group(), handleEndRunExceptions(), edm::make_waiting_task(), eostools::move(), schedule_, serviceToken_, streamQueues_, streamRunActive_, streamRunStatus_, subProcesses_, and edm::WaitingTaskHolder::taskHasFailed().

Referenced by endRunAsync().

                                                                                           {
    CMS_SA_ALLOW try {
      if (!streamRunStatus_[iStreamIndex]) {
        if (exceptionRunStatus_->streamFinishedRun()) {
          exceptionRunStatus_->globalEndRunHolder().doneWaiting(std::exception_ptr());
          exceptionRunStatus_.reset();
        }
        return;
      }

      auto runDoneTask =
          edm::make_waiting_task([this, iTask, iStreamIndex](std::exception_ptr const* iException) mutable {
            if (iException) {
              handleEndRunExceptions(*iException, iTask);
            }

            auto runStatus = streamRunStatus_[iStreamIndex];

            //reset status before releasing queue else get race condition
            if (runStatus->streamFinishedRun()) {
              runStatus->globalEndRunHolder().doneWaiting(std::exception_ptr());
            }
            streamRunStatus_[iStreamIndex].reset();
            --streamRunActive_;
            streamQueues_[iStreamIndex].resume();
          });

      WaitingTaskHolder runDoneTaskHolder{*iTask.group(), runDoneTask};

      auto runStatus = streamRunStatus_[iStreamIndex].get();

      if (runStatus->didGlobalBeginSucceed() && runStatus->endingEventSetupSucceeded()) {
        EventSetupImpl const& es = runStatus->eventSetupImplEndRun(esp_->subProcessIndex());
        auto eventSetupImpls = &runStatus->eventSetupImplsEndRun();
        bool cleaningUpAfterException = runStatus->cleaningUpAfterException() || iTask.taskHasFailed();

        auto& runPrincipal = *runStatus->runPrincipal();
        using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamEnd>;
        RunTransitionInfo transitionInfo(runPrincipal, es, eventSetupImpls);
        endStreamTransitionAsync<Traits>(std::move(runDoneTaskHolder),
                                         *schedule_,
                                         iStreamIndex,
                                         transitionInfo,
                                         serviceToken_,
                                         subProcesses_,
                                         cleaningUpAfterException);
      }
    } catch (...) {
      handleEndRunExceptions(std::current_exception(), iTask);
    }
  }

taskCleanup()

void edm::EventProcessor::taskCleanup

(

)

Definition at line 625 of file EventProcessor.cc.

References cms::cuda::assert(), espController_, TrackValidation_cff::task, and taskGroup_.

                                   {
    edm::FinalWaitingTask task{taskGroup_};
    espController_->endIOVsAsync(edm::WaitingTaskHolder{taskGroup_, &task});
    task.waitNoThrow();
    assert(task.done());
  }

thinnedAssociationsHelper() [1/2]

std::shared_ptr<ThinnedAssociationsHelper const> edm::EventProcessor::thinnedAssociationsHelper ( ) const

inlineprivate

Definition at line 286 of file EventProcessor.h.

References edm::get_underlying_safe(), and thinnedAssociationsHelper_.

Referenced by init().

                                                                                     {
      return get_underlying_safe(thinnedAssociationsHelper_);
    }

thinnedAssociationsHelper() [2/2]

std::shared_ptr<ThinnedAssociationsHelper>& edm::EventProcessor::thinnedAssociationsHelper ( )

inlineprivate

Definition at line 289 of file EventProcessor.h.

References edm::get_underlying_safe(), and thinnedAssociationsHelper_.

                                                                          {
      return get_underlying_safe(thinnedAssociationsHelper_);
    }

throwAboutModulesRequiringLuminosityBlockSynchronization()

void edm::EventProcessor::throwAboutModulesRequiringLuminosityBlockSynchronization ( ) const

private

Definition at line 2471 of file EventProcessor.cc.

References newFWLiteAna::found, and schedule_.

Referenced by beginJob().

                                                                                      {
    cms::Exception ex("ModulesSynchingOnLumis");
    ex << "The framework is configured to use at least two streams, but the following modules\n"
       << "require synchronizing on LuminosityBlock boundaries:";
    bool found = false;
    for (auto worker : schedule_->allWorkers()) {
      if (worker->wantsGlobalLuminosityBlocks() and worker->globalLuminosityBlocksQueue()) {
        found = true;
        ex << "\n  " << worker->description()->moduleName() << " " << worker->description()->moduleLabel();
      }
    }
    if (found) {
      ex << "\n\nThe situation can be fixed by either\n"
         << " * modifying the modules to support concurrent LuminosityBlocks (preferred), or\n"
         << " * setting 'process.options.numberOfConcurrentLuminosityBlocks = 1' in the configuration file";
      throw ex;
    }
  }

totalEvents()

int edm::EventProcessor::totalEvents

(

)

const

Return the number of events this EventProcessor has tried to process (inclues both successes and failures, including failures due to exceptions during processing).

Definition at line 836 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEvents().

{ return schedule_->totalEvents(); }

totalEventsFailed()

int edm::EventProcessor::totalEventsFailed

(

)

const

Return the number of events that have not passed any trigger. (N.B. totalEventsFailed() + totalEventsPassed() == totalEvents()

Definition at line 840 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEventsFailed().

{ return schedule_->totalEventsFailed(); }

totalEventsPassed()

int edm::EventProcessor::totalEventsPassed

(

)

const

Return the number of events processed by this EventProcessor which have been passed by one or more trigger paths.

Definition at line 838 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEventsPassed().

{ return schedule_->totalEventsPassed(); }

warnAboutModulesRequiringRunSynchronization()

void edm::EventProcessor::warnAboutModulesRequiringRunSynchronization ( ) const

private

Definition at line 2490 of file EventProcessor.cc.

References alignCSCRings::s, and schedule_.

Referenced by beginJob().

                                                                         {
    std::unique_ptr<LogSystem> s;
    for (auto worker : schedule_->allWorkers()) {
      if (worker->wantsGlobalRuns() and worker->globalRunsQueue()) {
        if (not s) {
          s = std::make_unique<LogSystem>("ModulesSynchingOnRuns");
          (*s) << "The following modules require synchronizing on Run boundaries:";
        }
        (*s) << "\n  " << worker->description()->moduleName() << " " << worker->description()->moduleLabel();
      }
    }
  }

writeLumiAsync()

void edm::EventProcessor::writeLumiAsync	(	WaitingTaskHolder	task,
		LuminosityBlockPrincipal &	lumiPrincipal
	)

Definition at line 2092 of file EventProcessor.cc.

References actReg_, dqmdumpme::first, edm::waiting_task::chain::ifThen(), edm::LuminosityBlockPrincipal::kNo, edm::waiting_task::chain::lastTask(), edm::LuminosityBlockPrincipal::luminosityBlock(), edm::RunPrincipal::mergeableRunProductMetadata(), eostools::move(), findAndChange::op, processContext_, edm::LuminosityBlockPrincipal::runPrincipal(), alignCSCRings::s, schedule_, serviceToken_, edm::LuminosityBlockPrincipal::shouldWriteLumi(), subProcesses_, TrackValidation_cff::task, and edm::MergeableRunProductMetadata::writeLumi().

Referenced by globalEndLumiAsync().

                                                                                                     {
    using namespace edm::waiting_task;
    if (lumiPrincipal.shouldWriteLumi() != LuminosityBlockPrincipal::kNo) {
      chain::first([&](auto nextTask) {
        ServiceRegistry::Operate op(serviceToken_);

        lumiPrincipal.runPrincipal().mergeableRunProductMetadata()->writeLumi(lumiPrincipal.luminosityBlock());
        schedule_->writeLumiAsync(nextTask, lumiPrincipal, &processContext_, actReg_.get());
      }) | chain::ifThen(not subProcesses_.empty(), [this, &lumiPrincipal](auto nextTask) {
        ServiceRegistry::Operate op(serviceToken_);
        for (auto& s : subProcesses_) {
          s.writeLumiAsync(nextTask, lumiPrincipal);
        }
      }) | chain::lastTask(std::move(task));
    }
  }

writeProcessBlockAsync()

void edm::EventProcessor::writeProcessBlockAsync	(	WaitingTaskHolder	task,
		ProcessBlockType	processBlockType
	)

Definition at line 2053 of file EventProcessor.cc.

References actReg_, dqmdumpme::first, edm::waiting_task::chain::ifThen(), eostools::move(), findAndChange::op, principalCache_, edm::PrincipalCache::processBlockPrincipal(), processContext_, edm::waiting_task::chain::runLast(), alignCSCRings::s, schedule_, serviceToken_, subProcesses_, and TrackValidation_cff::task.

Referenced by endProcessBlock(), and inputProcessBlocks().

                                                                                                       {
    using namespace edm::waiting_task;
    chain::first([&](auto nextTask) {
      ServiceRegistry::Operate op(serviceToken_);
      schedule_->writeProcessBlockAsync(
          nextTask, principalCache_.processBlockPrincipal(processBlockType), &processContext_, actReg_.get());
    }) | chain::ifThen(not subProcesses_.empty(), [this, processBlockType](auto nextTask) {
      ServiceRegistry::Operate op(serviceToken_);
      for (auto& s : subProcesses_) {
        s.writeProcessBlockAsync(nextTask, processBlockType);
      }
    }) | chain::runLast(std::move(task));
  }

writeRunAsync()

void edm::EventProcessor::writeRunAsync	(	WaitingTaskHolder	task,
		RunPrincipal const &	runPrincipal,
		MergeableRunProductMetadata const *	mergeableRunProductMetadata
	)

Definition at line 2067 of file EventProcessor.cc.

References actReg_, dqmdumpme::first, edm::waiting_task::chain::ifThen(), edm::RunPrincipal::kNo, eostools::move(), findAndChange::op, processContext_, edm::waiting_task::chain::runLast(), alignCSCRings::s, schedule_, serviceToken_, edm::RunPrincipal::shouldWriteRun(), subProcesses_, and TrackValidation_cff::task.

Referenced by globalEndRunAsync().

                                                                                                     {
    using namespace edm::waiting_task;
    if (runPrincipal.shouldWriteRun() != RunPrincipal::kNo) {
      chain::first([&](auto nextTask) {
        ServiceRegistry::Operate op(serviceToken_);
        schedule_->writeRunAsync(nextTask, runPrincipal, &processContext_, actReg_.get(), mergeableRunProductMetadata);
      }) | chain::ifThen(not subProcesses_.empty(), [this, &runPrincipal, mergeableRunProductMetadata](auto nextTask) {
        ServiceRegistry::Operate op(serviceToken_);
        for (auto& s : subProcesses_) {
          s.writeRunAsync(nextTask, runPrincipal, mergeableRunProductMetadata);
        }
      }) | chain::runLast(std::move(task));
    }
  }

Member Data Documentation

act_table_

std::unique_ptr<ExceptionToActionTable const> edm::EventProcessor::act_table_

private

Definition at line 323 of file EventProcessor.h.

Referenced by init().

actReg_

std::shared_ptr<ActivityRegistry> edm::EventProcessor::actReg_

private

Definition at line 311 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRunAsync(), closeInputFile(), endJob(), endRunAsync(), init(), nextTransitionType(), processEventAsyncImpl(), readAndMergeLumi(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readProcessBlock(), readRun(), runToCompletion(), writeLumiAsync(), writeProcessBlockAsync(), writeRunAsync(), and ~EventProcessor().

beginJobCalled_

bool edm::EventProcessor::beginJobCalled_

private

Definition at line 356 of file EventProcessor.h.

Referenced by beginJob().

branchesToDeleteEarly_

std::vector<std::string> edm::EventProcessor::branchesToDeleteEarly_

private

Definition at line 339 of file EventProcessor.h.

Referenced by beginJob(), and init().

branchIDListHelper_

edm::propagate_const<std::shared_ptr<BranchIDListHelper> > edm::EventProcessor::branchIDListHelper_

private

Definition at line 313 of file EventProcessor.h.

Referenced by branchIDListHelper(), init(), and respondToOpenInputFile().

deferredExceptionPtr_

std::exception_ptr edm::EventProcessor::deferredExceptionPtr_

private

Definition at line 351 of file EventProcessor.h.

Referenced by runToCompletion(), and setDeferredException().

deferredExceptionPtrIsSet_

std::atomic<bool> edm::EventProcessor::deferredExceptionPtrIsSet_

private

Definition at line 350 of file EventProcessor.h.

Referenced by runToCompletion(), and setDeferredException().

deleteNonConsumedUnscheduledModules_

bool edm::EventProcessor::deleteNonConsumedUnscheduledModules_ = true

private

Definition at line 375 of file EventProcessor.h.

Referenced by beginJob(), and init().

esp_

edm::propagate_const<std::shared_ptr<eventsetup::EventSetupProvider> > edm::EventProcessor::esp_

private

Definition at line 321 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRunAsync(), endOfLoop(), globalEndLumiAsync(), globalEndRunAsync(), init(), prepareForNextLoop(), processEventAsyncImpl(), processEventWithLooper(), streamBeginRunAsync(), streamEndLumiAsync(), streamEndRunAsync(), and ~EventProcessor().

espController_

edm::propagate_const<std::unique_ptr<eventsetup::EventSetupsController> > edm::EventProcessor::espController_

private

Definition at line 320 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRunAsync(), endRunAsync(), init(), taskCleanup(), and ~EventProcessor().

eventSetupDataToExcludeFromPrefetching_

ExcludedDataMap edm::EventProcessor::eventSetupDataToExcludeFromPrefetching_

private

Definition at line 372 of file EventProcessor.h.

exceptionMessageFiles_

std::string edm::EventProcessor::exceptionMessageFiles_

private

Definition at line 359 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageFiles().

exceptionMessageLumis_

std::atomic<bool> edm::EventProcessor::exceptionMessageLumis_

private

Definition at line 361 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageLumis().

exceptionMessageRuns_

std::atomic<bool> edm::EventProcessor::exceptionMessageRuns_

private

Definition at line 360 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageRuns().

exceptionRunStatus_

std::shared_ptr<RunProcessingStatus> edm::EventProcessor::exceptionRunStatus_

private

Definition at line 334 of file EventProcessor.h.

Referenced by beginRunAsync(), and streamEndRunAsync().

fb_

edm::propagate_const<std::shared_ptr<FileBlock> > edm::EventProcessor::fb_

private

Definition at line 346 of file EventProcessor.h.

Referenced by closeInputFile(), fileBlockValid(), openOutputFiles(), readFile(), respondToCloseInputFile(), and respondToOpenInputFile().

fileModeNoMerge_

bool edm::EventProcessor::fileModeNoMerge_

private

Definition at line 358 of file EventProcessor.h.

Referenced by init(), and runToCompletion().

firstEventInBlock_

bool edm::EventProcessor::firstEventInBlock_ = true

private

Definition at line 368 of file EventProcessor.h.

forceESCacheClearOnNewRun_

bool edm::EventProcessor::forceESCacheClearOnNewRun_

private

Definition at line 364 of file EventProcessor.h.

Referenced by beginRunAsync(), and init().

forceLooperToEnd_

bool edm::EventProcessor::forceLooperToEnd_

private

Definition at line 362 of file EventProcessor.h.

Referenced by endOfLoop().

historyAppender_

edm::propagate_const<std::unique_ptr<HistoryAppender> > edm::EventProcessor::historyAppender_

private

Definition at line 344 of file EventProcessor.h.

Referenced by init(), readLuminosityBlock(), and readRun().

input_

edm::propagate_const<std::unique_ptr<InputSource> > edm::EventProcessor::input_

private

Definition at line 317 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRunAsync(), closeInputFile(), continueLumiAsync(), endJob(), endRunAsync(), handleNextEventForStreamAsync(), handleNextItemAfterMergingRunEntries(), init(), inputProcessBlocks(), nextTransitionType(), nextTransitionTypeAsync(), processEventWithLooper(), processRuns(), readAndMergeLumi(), readAndMergeLumiEntriesAsync(), readAndMergeRun(), readAndMergeRunEntriesAsync(), readEvent(), readFile(), readLuminosityBlock(), readNextEventForStream(), readProcessBlock(), readRun(), rewindInput(), and ~EventProcessor().

lastSourceTransition_

InputSource::ItemTypeInfo edm::EventProcessor::lastSourceTransition_

private

Definition at line 318 of file EventProcessor.h.

Referenced by endUnfinishedRun(), lastTransitionType(), nextTransitionType(), and readNextEventForStream().

looper_

edm::propagate_const<std::shared_ptr<EDLooperBase> > edm::EventProcessor::looper_

private

Definition at line 347 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRunAsync(), Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), endOfLoop(), globalEndLumiAsync(), globalEndRunAsync(), init(), looper(), prepareForNextLoop(), processEventAsyncImpl(), processEventWithLooper(), Config.Process::splitPython(), startingNewLoop(), and ~EventProcessor().

looperBeginJobRun_

bool edm::EventProcessor::looperBeginJobRun_

private

Definition at line 363 of file EventProcessor.h.

Referenced by beginRunAsync(), and startingNewLoop().

lumiQueue_

std::unique_ptr<edm::LimitedTaskQueue> edm::EventProcessor::lumiQueue_

private

Definition at line 332 of file EventProcessor.h.

Referenced by beginLumiAsync(), and init().

mergeableRunProductProcesses_

MergeableRunProductProcesses edm::EventProcessor::mergeableRunProductProcesses_

private

Definition at line 327 of file EventProcessor.h.

Referenced by init().

modulesToIgnoreForDeleteEarly_

std::vector<std::string> edm::EventProcessor::modulesToIgnoreForDeleteEarly_

private

Definition at line 341 of file EventProcessor.h.

Referenced by beginJob(), and init().

moduleTypeResolverMaker_

edm::propagate_const<std::unique_ptr<ModuleTypeResolverMaker const> > edm::EventProcessor::moduleTypeResolverMaker_

private

Definition at line 319 of file EventProcessor.h.

Referenced by init().

needToCallNext_

bool edm::EventProcessor::needToCallNext_ = true

private

Definition at line 376 of file EventProcessor.h.

Referenced by needToCallNext(), and setNeedToCallNext().

pathsAndConsumesOfModules_

PathsAndConsumesOfModules edm::EventProcessor::pathsAndConsumesOfModules_

private

Definition at line 326 of file EventProcessor.h.

Referenced by beginJob().

preallocations_

PreallocationConfiguration edm::EventProcessor::preallocations_

private

Definition at line 366 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRunAsync(), continueLumiAsync(), endJob(), endRunAsync(), endUnfinishedLumi(), init(), processRuns(), and readFile().

preg_

edm::propagate_const<std::shared_ptr<ProductRegistry> > edm::EventProcessor::preg_

private

Definition at line 312 of file EventProcessor.h.

Referenced by beginJob(), endOfLoop(), init(), preg(), readAndMergeRun(), and readFile().

principalCache_

PrincipalCache edm::EventProcessor::principalCache_

private

Definition at line 355 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginProcessBlock(), endProcessBlock(), init(), inputProcessBlocks(), processEventAsyncImpl(), readEvent(), readFile(), readLuminosityBlock(), readRun(), and writeProcessBlockAsync().

printDependencies_

bool edm::EventProcessor::printDependencies_ = false

private

Definition at line 374 of file EventProcessor.h.

Referenced by beginJob(), and init().

processBlockHelper_

edm::propagate_const<std::shared_ptr<ProcessBlockHelper> > edm::EventProcessor::processBlockHelper_

private

Definition at line 314 of file EventProcessor.h.

Referenced by beginJob(), closeOutputFiles(), and init().

processConfiguration_

std::shared_ptr<ProcessConfiguration const> edm::EventProcessor::processConfiguration_

private

Definition at line 324 of file EventProcessor.h.

Referenced by beginJob(), beginProcessBlock(), init(), and processConfiguration().

processContext_

ProcessContext edm::EventProcessor::processContext_

private

Definition at line 325 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRunAsync(), globalEndLumiAsync(), globalEndRunAsync(), init(), processEventAsyncImpl(), processEventWithLooper(), readEvent(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

queueWhichWaitsForIOVsToFinish_

edm::SerialTaskQueue edm::EventProcessor::queueWhichWaitsForIOVsToFinish_

private

Definition at line 322 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginRunAsync(), endRunAsync(), globalEndLumiAsync(), globalEndRunAsync(), and releaseBeginRunResources().

referencesToBranches_

std::multimap<std::string, std::string> edm::EventProcessor::referencesToBranches_

private

Definition at line 340 of file EventProcessor.h.

Referenced by beginJob(), and init().

runQueue_

std::unique_ptr<edm::LimitedTaskQueue> edm::EventProcessor::runQueue_

private

Definition at line 331 of file EventProcessor.h.

Referenced by beginRunAsync(), and init().

schedule_

edm::propagate_const<std::unique_ptr<Schedule> > edm::EventProcessor::schedule_

private

Definition at line 328 of file EventProcessor.h.

Referenced by Config.Process::_insertPaths(), Config.Process::_replaceInSchedule(), Config.Process::_replaceInScheduleDirectly(), beginJob(), beginLumiAsync(), beginProcessBlock(), beginRunAsync(), clearCounters(), closeOutputFiles(), Config.Process::dumpPython(), endJob(), endOfLoop(), endProcessBlock(), getAllModuleDescriptions(), globalEndLumiAsync(), globalEndRunAsync(), init(), inputProcessBlocks(), openOutputFiles(), processEventAsyncImpl(), Config.Process::prune(), Config.Process::resolve(), respondToCloseInputFile(), respondToOpenInputFile(), shouldWeCloseOutput(), shouldWeStop(), Config.Process::splitPython(), streamBeginRunAsync(), streamEndLumiAsync(), streamEndRunAsync(), throwAboutModulesRequiringLuminosityBlockSynchronization(), totalEvents(), totalEventsFailed(), totalEventsPassed(), warnAboutModulesRequiringRunSynchronization(), writeLumiAsync(), writeProcessBlockAsync(), writeRunAsync(), and ~EventProcessor().

serviceToken_

ServiceToken edm::EventProcessor::serviceToken_

private

Definition at line 316 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginProcessBlock(), beginRunAsync(), endJob(), endProcessBlock(), endRunAsync(), getToken(), globalEndLumiAsync(), globalEndRunAsync(), handleNextEventForStreamAsync(), handleNextItemAfterMergingRunEntries(), init(), inputProcessBlocks(), nextTransitionTypeAsync(), processEventAsyncImpl(), readAndMergeLumiEntriesAsync(), readAndMergeRunEntriesAsync(), readNextEventForStream(), runToCompletion(), streamBeginRunAsync(), streamEndLumiAsync(), streamEndRunAsync(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

shouldWeStop_

bool edm::EventProcessor::shouldWeStop_

private

Definition at line 357 of file EventProcessor.h.

Referenced by processEventWithLooper(), shouldWeStop(), and startingNewLoop().

sourceMutex_

std::shared_ptr<std::recursive_mutex> edm::EventProcessor::sourceMutex_

private

Definition at line 354 of file EventProcessor.h.

Referenced by nextTransitionTypeAsync(), readAndMergeLumiEntriesAsync(), readAndMergeRunEntriesAsync(), and readNextEventForStream().

sourceResourcesAcquirer_

SharedResourcesAcquirer edm::EventProcessor::sourceResourcesAcquirer_

private

Definition at line 353 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginRunAsync(), handleNextEventForStreamAsync(), nextTransitionTypeAsync(), readAndMergeLumiEntriesAsync(), and readAndMergeRunEntriesAsync().

streamLumiActive_

std::atomic<unsigned int> edm::EventProcessor::streamLumiActive_ {0}

private

Definition at line 337 of file EventProcessor.h.

Referenced by beginLumiAsync(), endUnfinishedLumi(), processRuns(), readFile(), and streamEndLumiAsync().

streamLumiStatus_

std::vector<std::shared_ptr<LuminosityBlockProcessingStatus> > edm::EventProcessor::streamLumiStatus_

private

Definition at line 335 of file EventProcessor.h.

Referenced by beginLumiAsync(), continueLumiAsync(), endUnfinishedLumi(), handleNextEventForStreamAsync(), init(), processEventAsyncImpl(), processEventWithLooper(), readEvent(), readFile(), and streamEndLumiAsync().

streamQueues_

std::vector<edm::SerialTaskQueue> edm::EventProcessor::streamQueues_

private

Definition at line 329 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginRunAsync(), endRunAsync(), init(), releaseBeginRunResources(), streamBeginRunAsync(), streamEndLumiAsync(), and streamEndRunAsync().

streamQueuesInserter_

SerialTaskQueue edm::EventProcessor::streamQueuesInserter_

private

Definition at line 330 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginRunAsync(), and endRunAsync().

streamRunActive_

std::atomic<unsigned int> edm::EventProcessor::streamRunActive_ {0}

private

Definition at line 336 of file EventProcessor.h.

Referenced by endUnfinishedLumi(), endUnfinishedRun(), processRuns(), readFile(), streamBeginRunAsync(), and streamEndRunAsync().

streamRunStatus_

std::vector<std::shared_ptr<RunProcessingStatus> > edm::EventProcessor::streamRunStatus_

private

Definition at line 333 of file EventProcessor.h.

Referenced by endUnfinishedRun(), handleNextEventForStreamAsync(), init(), processRuns(), readEvent(), readFile(), releaseBeginRunResources(), streamBeginRunAsync(), and streamEndRunAsync().

subProcesses_

std::vector<SubProcess> edm::EventProcessor::subProcesses_

private

Definition at line 343 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginProcessBlock(), beginRunAsync(), clearLumiPrincipal(), clearRunPrincipal(), closeOutputFiles(), Config.Process::dumpConfig(), Config.Process::dumpPython(), endJob(), endProcessBlock(), globalEndLumiAsync(), globalEndRunAsync(), init(), inputProcessBlocks(), openOutputFiles(), processEventAsyncImpl(), respondToCloseInputFile(), respondToOpenInputFile(), shouldWeCloseOutput(), shouldWeStop(), Config.Process::splitPython(), streamBeginRunAsync(), streamEndLumiAsync(), streamEndRunAsync(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

taskGroup_

oneapi::tbb::task_group edm::EventProcessor::taskGroup_

private

Definition at line 309 of file EventProcessor.h.

Referenced by beginProcessBlock(), endProcessBlock(), endUnfinishedLumi(), endUnfinishedRun(), inputProcessBlocks(), processRuns(), and taskCleanup().

thinnedAssociationsHelper_

edm::propagate_const<std::shared_ptr<ThinnedAssociationsHelper> > edm::EventProcessor::thinnedAssociationsHelper_

private

Definition at line 315 of file EventProcessor.h.

Referenced by init(), and thinnedAssociationsHelper().