#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 (edm::IOVSyncValue const &iSyncValue, std::shared_ptr< void > const &iRunResource, edm::WaitingTaskHolder iHolder)

void	beginProcessBlock (bool &beginProcessBlockSucceeded)

void	beginRun (ProcessHistoryID const &phid, RunNumber_t run, bool &globalBeginSucceeded, bool &eventSetupForInstanceSucceeded)

void	clearCounters ()
	Clears counters used by trigger report. More...

void	closeInputFile (bool cleaningUpAfterException)

void	closeOutputFiles ()

void	continueLumiAsync (edm::WaitingTaskHolder iHolder)

void	deleteLumiFromCache (LuminosityBlockProcessingStatus &)

void	deleteRunFromCache (ProcessHistoryID const &phid, RunNumber_t run)

void	doErrorStuff ()

void	endJob ()

bool	endOfLoop ()

void	endProcessBlock (bool cleaningUpAfterException, bool beginProcessBlockSucceeded)

void	endRun (ProcessHistoryID const &phid, RunNumber_t run, bool globalBeginSucceeded, bool cleaningUpAfterException)

void	endUnfinishedLumi ()

void	endUnfinishedRun (ProcessHistoryID const &phid, RunNumber_t run, bool globalBeginSucceeded, bool cleaningUpAfterException, bool eventSetupForInstanceSucceeded)

	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 (edm::WaitingTaskHolder iTask, std::shared_ptr< LuminosityBlockProcessingStatus > iLumiStatus)

void	handleEndLumiExceptions (std::exception_ptr const *iPtr, WaitingTaskHolder &holder)

void	inputProcessBlocks ()

InputSource::ItemType	lastTransitionType () const

edm::LuminosityBlockNumber_t	nextLuminosityBlockID ()

std::pair< edm::ProcessHistoryID, edm::RunNumber_t >	nextRunID ()

InputSource::ItemType	nextTransitionType ()

void	openOutputFiles ()

EventProcessor &	operator= (EventProcessor const &)=delete

void	prepareForNextLoop ()

ProcessConfiguration const &	processConfiguration () const

InputSource::ItemType	processLumis (std::shared_ptr< void > const &iRunResource)

int	readAndMergeLumi (LuminosityBlockProcessingStatus &)

std::pair< ProcessHistoryID, RunNumber_t >	readAndMergeRun ()

void	readFile ()

void	readLuminosityBlock (LuminosityBlockProcessingStatus &)

void	readProcessBlock (ProcessBlockPrincipal &)

std::pair< ProcessHistoryID, RunNumber_t >	readRun ()

void	respondToCloseInputFile ()

void	respondToOpenInputFile ()

void	rewindInput ()

StatusCode	run ()

StatusCode	runToCompletion ()

bool	setDeferredException (std::exception_ptr)

void	setExceptionMessageFiles (std::string &message)

void	setExceptionMessageLumis ()

void	setExceptionMessageRuns (std::string &message)

bool	shouldWeCloseOutput () const

bool	shouldWeStop () const

void	startingNewLoop ()

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

void	taskCleanup ()

int	totalEvents () const

int	totalEventsFailed () const

int	totalEventsPassed () const

void	writeLumiAsync (WaitingTaskHolder, LuminosityBlockPrincipal &lumiPrincipal)

void	writeProcessBlockAsync (WaitingTaskHolder, ProcessBlockType)

void	writeRunAsync (WaitingTaskHolder, ProcessHistoryID const &phid, RunNumber_t run, 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 iTask, unsigned int iStreamIndex)

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

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

std::shared_ptr< EDLooperBase > &	looper ()

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	readEvent (unsigned int iStreamIndex)

bool	readNextEventForStream (unsigned int iStreamIndex, LuminosityBlockProcessingStatus &iLumiStatus)

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

std::shared_ptr< ThinnedAssociationsHelper > &	thinnedAssociationsHelper ()

void	throwAboutModulesRequiringLuminosityBlockSynchronization () const

void	warnAboutLegacyModules () const

Private Attributes
std::unique_ptr< ExceptionToActionTable const >	act_table_

std::shared_ptr< ActivityRegistry >	actReg_

bool	asyncStopRequestedWhileProcessingEvents_

StatusCode	asyncStopStatusCodeFromProcessingEvents_

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::string	exceptionMessageRuns_

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::ItemType	lastSourceTransition_

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

bool	looperBeginJobRun_

std::unique_ptr< edm::LimitedTaskQueue >	lumiQueue_

MergeableRunProductProcesses	mergeableRunProductProcesses_

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_

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_

std::vector< SubProcess >	subProcesses_

oneapi::tbb::task_group	taskGroup_

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

Detailed Description

Definition at line 66 of file EventProcessor.h.

Member Typedef Documentation

◆ ExcludedData

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

private

Definition at line 365 of file EventProcessor.h.

◆ ExcludedDataMap

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

private

Definition at line 366 of file EventProcessor.h.

◆ ProcessBlockType

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

Definition at line 243 of file EventProcessor.h.

Member Enumeration Documentation

◆ StatusCode

enum edm::EventProcessor::StatusCode

Enumerator
epSuccess
epException
epOther
epSignal
epInputComplete
epTimedOut
epCountComplete

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

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

       : actReg_(),
         preg_(),
         branchIDListHelper_(),
         serviceToken_(),
         input_(),
         espController_(new eventsetup::EventSetupsController),
         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_(),
         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 271 of file EventProcessor.cc.

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

       : actReg_(),
         preg_(),
         branchIDListHelper_(),
         serviceToken_(),
         input_(),
         espController_(new eventsetup::EventSetupsController),
         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_(),
         exceptionMessageLumis_(false),
         forceLooperToEnd_(false),
         looperBeginJobRun_(false),
         forceESCacheClearOnNewRun_(false),
         asyncStopRequestedWhileProcessingEvents_(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 308 of file EventProcessor.cc.

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

       : actReg_(),
         preg_(),
         branchIDListHelper_(),
         serviceToken_(),
         input_(),
         espController_(new eventsetup::EventSetupsController),
         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_(),
         exceptionMessageLumis_(false),
         forceLooperToEnd_(false),
         looperBeginJobRun_(false),
         forceESCacheClearOnNewRun_(false),
         asyncStopRequestedWhileProcessingEvents_(false),
         eventSetupDataToExcludeFromPrefetching_() {
     init(processDesc, token, legacy);
   }

◆ ~EventProcessor()

edm::EventProcessor::~EventProcessor ( )

Definition at line 579 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_, 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 604 of file EventProcessor.cc.

References actReg_, cms::Exception::addContext(), bk::beginJob(), beginJobCalled_, branchesToDeleteEarly_, c, edm::checkForModuleDependencyCorrectness(), deleteNonConsumedUnscheduledModules_, edmLumisInFiles::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_, MatrixUtil::merge(), eostools::move(), edm::nonConsumedUnscheduledModules(), edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfRuns(), edm::PreallocationConfiguration::numberOfStreams(), edm::PreallocationConfiguration::numberOfThreads(), pathsAndConsumesOfModules_, preallocations_, preg(), preg_, printDependencies_, processBlockHelper_, processConfiguration_, processContext_, edm::PathsAndConsumesOfModules::removeModules(), edm::waiting_task::chain::runLast(), schedule_, serviceToken_, subProcesses_, edm::swap(), std::swap(), throwAboutModulesRequiringLuminosityBlockSynchronization(), createJobs::tmp, warnAboutLegacyModules(), 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()) {
       schedule_->initializeEarlyDelete(branchesToDeleteEarly_, *preg_);
       decltype(branchesToDeleteEarly_)().swap(branchesToDeleteEarly_);
     }
 
     actReg_->preBeginJobSignal_(pathsAndConsumesOfModules_, processContext_);
 
     if (preallocations_.numberOfLuminosityBlocks() > 1) {
       throwAboutModulesRequiringLuminosityBlockSynchronization();
     }
     warnAboutLegacyModules();
 
     //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);
     //}
     try {
       convertException::wrap([&]() { input_->doBeginJob(); });
     } catch (cms::Exception& ex) {
       ex.addContext("Calling beginJob for the source");
       throw;
     }
     espController_->finishConfiguration();
     schedule_->beginJob(*preg_, esp_->recordsToProxyIndices(), *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_->recordsToProxyIndices());
     }
     // toerror.succeeded(); // should we add this?
     for_all(subProcesses_, [](auto& subProcess) { subProcess.doBeginJob(); });
     actReg_->postBeginJobSignal_();
 
     FinalWaitingTask last;
     oneapi::tbb::task_group 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));
     group.wait();
     if (last.exceptionPtr()) {
       std::rethrow_exception(*last.exceptionPtr());
     }
   }

◆ beginLumiAsync()

void edm::EventProcessor::beginLumiAsync	(	edm::IOVSyncValue const &	iSyncValue,
		std::shared_ptr< void > const &	iRunResource,
		edm::WaitingTaskHolder	iHolder
	)

Definition at line 1358 of file EventProcessor.cc.

References actReg_, edm::BeginLuminosityBlock, CMS_SA_ALLOW, edm::signalslot::Signal< T >::emit(), esp_, espController_, edm::PrincipalCache::eventPrincipal(), first, watchdog::group, handleNextEventForStreamAsync(), mps_fire::i, edm::waiting_task::chain::ifThen(), input_, edm::Service< T >::isAvailable(), looper_, lumiQueue_, eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, edm::ActivityRegistry::preESSyncIOVSignal_, principalCache_, processContext_, edm::SerialTaskQueueChain::push(), edm::SerialTaskQueue::push(), createBeamHaloJobs::queue, queueWhichWaitsForIOVsToFinish_, readLuminosityBlock(), edm::waiting_task::chain::runLast(), schedule_, edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, sourceResourcesAcquirer_, mps_update::status, streamEndLumiAsync(), streamLumiActive_, streamLumiStatus_, streamQueues_, subProcesses_, TrackValidation_cff::task, edm::WaitingTaskHolder::taskHasFailed(), edm::waiting_task::chain::then(), and createJobs::tmp.

Referenced by handleNextEventForStreamAsync(), and processLumis().

                                                                     {
     if (iHolder.taskHasFailed()) {
       return;
     }
 
     actReg_->esSyncIOVQueuingSignal_.emit(iSync);
     // We must be careful with the status object here and in code this function calls. IF we want
     // endRun to be called, then we must call resetResources before the things waiting on
     // iHolder are allowed to proceed. Otherwise, there will be race condition (possibly causing
     // endRun to be called much later than it should be, because status is holding iRunResource).
     // Note that this must be done explicitly. Relying on the destructor does not work well
     // because the LimitedTaskQueue for the lumiWork holds the shared_ptr in each of its internal
     // queues, plus it is difficult to guarantee the destructor is called  before iHolder gets
     // destroyed inside this function and lumiWork.
     auto status =
         std::make_shared<LuminosityBlockProcessingStatus>(this, preallocations_.numberOfStreams(), iRunResource);
     chain::first([&](auto nextTask) {
       auto asyncEventSetup = [](ActivityRegistry* actReg,
                                 auto* espController,
                                 auto& queue,
                                 WaitingTaskHolder task,
                                 auto& status,
                                 IOVSyncValue const& iSync) {
         queue.pause();
         CMS_SA_ALLOW try {
           SendSourceTerminationSignalIfException sentry(actReg);
           // Pass in iSync to let the EventSetup system know which run and lumi
           // need to be processed and prepare IOVs for it.
           // Pass in the endIOVWaitingTasks so the lumi can notify them when the
           // lumi is done and no longer needs its EventSetup IOVs.
           actReg->preESSyncIOVSignal_.emit(iSync);
           espController->eventSetupForInstanceAsync(
               iSync, task, status->endIOVWaitingTasks(), status->eventSetupImpls());
           sentry.completedSuccessfully();
         } catch (...) {
           task.doneWaiting(std::current_exception());
         }
       };
       if (espController_->doWeNeedToWaitForIOVsToFinish(iSync)) {
         // We only get here inside this block if there is an EventSetup
         // module not able to handle concurrent IOVs (usually an ESSource)
         // and the new sync value is outside the current IOV of that module.
         auto group = nextTask.group();
         queueWhichWaitsForIOVsToFinish_.push(
             *group, [this, task = std::move(nextTask), iSync, status, asyncEventSetup]() mutable {
               asyncEventSetup(
                   actReg_.get(), espController_.get(), queueWhichWaitsForIOVsToFinish_, std::move(task), status, iSync);
             });
       } else {
         asyncEventSetup(
             actReg_.get(), espController_.get(), queueWhichWaitsForIOVsToFinish_, std::move(nextTask), status, iSync);
       }
     }) | chain::then([this, status, iSync](std::exception_ptr const* iPtr, auto nextTask) {
       actReg_->postESSyncIOVSignal_.emit(iSync);
       //the call to doneWaiting will cause the count to decrement
       auto copyTask = nextTask;
       if (iPtr) {
         nextTask.doneWaiting(*iPtr);
       }
       auto group = copyTask.group();
       lumiQueue_->pushAndPause(
           *group, [this, task = std::move(copyTask), status](edm::LimitedTaskQueue::Resumer iResumer) mutable {
             if (task.taskHasFailed()) {
               status->resetResources();
               return;
             }
 
             status->setResumer(std::move(iResumer));
 
             auto group = task.group();
             sourceResourcesAcquirer_.serialQueueChain().push(
                 *group, [this, postQueueTask = std::move(task), status = std::move(status)]() mutable {
                   //make the services available
                   ServiceRegistry::Operate operate(serviceToken_);
                   // Caught exception is propagated via WaitingTaskHolder
                   CMS_SA_ALLOW try {
                     readLuminosityBlock(*status);
 
                     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, &lumiPrincipal](auto nextTask) {
                       EventSetupImpl const& es = status->eventSetupImpl(esp_->subProcessIndex());
                       {
                         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();
                       //make the services available
                       ServiceRegistry::Operate operateLooper(serviceToken_);
                       looper_->doBeginLuminosityBlock(*(status->lumiPrincipal()), es, &processContext_);
                     }) | then([this, status](std::exception_ptr const* iPtr, auto holder) mutable {
                       if (iPtr) {
                         status->resetResources();
                         holder.doneWaiting(*iPtr);
                       } else {
                         if (not looper_) {
                           status->globalBeginDidSucceed();
                         }
                         EventSetupImpl const& es = status->eventSetupImpl(esp_->subProcessIndex());
                         using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin>;
 
                         for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
                           streamQueues_[i].push(*holder.group(), [this, i, status, holder, &es]() mutable {
                             streamQueues_[i].pause();
 
                             auto& event = principalCache_.eventPrincipal(i);
                             //We need to be sure that 'status' and its internal shared_ptr<LuminosityBlockPrincipal> are only
                             // held by the container as this lambda may not finish executing before all the tasks it
                             // spawns have already started to run.
                             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 tmp(nextTask);
                                 tmp.doneWaiting(*exceptionFromBeginStreamLumi);
                                 streamEndLumiAsync(nextTask, i);
                               } else {
                                 handleNextEventForStreamAsync(std::move(nextTask), i);
                               }
                             }) | runLast(holder);
                           });
                         }
                       }
                     }) | runLast(postQueueTask);
 
                   } catch (...) {
                     status->resetResources();
                     postQueueTask.doneWaiting(std::current_exception());
                   }
                 });  // task in sourceResourcesAcquirer
           });
     }) | chain::runLast(std::move(iHolder));
   }

◆ beginProcessBlock()

void edm::EventProcessor::beginProcessBlock ( bool & beginProcessBlockSucceeded )

Definition at line 1048 of file EventProcessor.cc.

References edm::FinalWaitingTask::done(), edm::WaitingTask::exceptionPtr(), 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;
 
     ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
     beginGlobalTransitionAsync<Traits>(
         WaitingTaskHolder(taskGroup_, &globalWaitTask), *schedule_, transitionInfo, serviceToken_, subProcesses_);
 
     do {
       taskGroup_.wait();
     } while (not globalWaitTask.done());
 
     if (globalWaitTask.exceptionPtr() != nullptr) {
       std::rethrow_exception(*(globalWaitTask.exceptionPtr()));
     }
     beginProcessBlockSucceeded = true;
   }

◆ beginRun()

void edm::EventProcessor::beginRun	(	ProcessHistoryID const &	phid,
		RunNumber_t	run,
		bool &	globalBeginSucceeded,
		bool &	eventSetupForInstanceSucceeded
	)

Definition at line 1142 of file EventProcessor.cc.

References actReg_, edm::BeginRun, edm::RunPrincipal::beginTime(), edm::FinalWaitingTask::done(), esp_, espController_, edm::WaitingTask::exceptionPtr(), FDEBUG, first, forceESCacheClearOnNewRun_, edm::waiting_task::chain::ifThen(), input_, looper_, looperBeginJobRun_, eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, edm::RunPrincipal::run(), run(), edm::waiting_task::chain::runLast(), edm::PrincipalCache::runPrincipal(), schedule_, serviceToken_, subProcesses_, edm::eventsetup::synchronousEventSetupForInstance(), taskGroup_, and edm::waiting_task::chain::then().

                                                                       {
     globalBeginSucceeded = false;
     RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, run);
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
 
       input_->doBeginRun(runPrincipal, &processContext_);
       sentry.completedSuccessfully();
     }
 
     IOVSyncValue ts(EventID(runPrincipal.run(), 0, 0), runPrincipal.beginTime());
     if (forceESCacheClearOnNewRun_) {
       espController_->forceCacheClear();
     }
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
       actReg_->preESSyncIOVSignal_.emit(ts);
       synchronousEventSetupForInstance(ts, taskGroup_, *espController_);
       actReg_->postESSyncIOVSignal_.emit(ts);
       eventSetupForInstanceSucceeded = true;
       sentry.completedSuccessfully();
     }
     auto const& es = esp_->eventSetupImpl();
     if (looper_ && looperBeginJobRun_ == false) {
       looper_->copyInfo(ScheduleInfo(schedule_.get()));
 
       FinalWaitingTask waitTask;
       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(taskGroup_, &waitTask));
 
       do {
         taskGroup_.wait();
       } while (not waitTask.done());
       if (waitTask.exceptionPtr() != nullptr) {
         std::rethrow_exception(*(waitTask.exceptionPtr()));
       }
     }
     {
       using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin>;
       FinalWaitingTask globalWaitTask;
 
       using namespace edm::waiting_task::chain;
       chain::first([&runPrincipal, &es, this](auto waitTask) {
         RunTransitionInfo transitionInfo(runPrincipal, es);
         beginGlobalTransitionAsync<Traits>(
             std::move(waitTask), *schedule_, transitionInfo, serviceToken_, subProcesses_);
       }) | then([&globalBeginSucceeded, run](auto waitTask) mutable {
         globalBeginSucceeded = true;
         FDEBUG(1) << "\tbeginRun " << run << "\n";
       }) | ifThen(looper_, [this, &runPrincipal, &es](auto waitTask) {
         looper_->prefetchAsync(waitTask, serviceToken_, Transition::BeginRun, runPrincipal, es);
       }) | ifThen(looper_, [this, &runPrincipal, &es](auto waitTask) {
         looper_->doBeginRun(runPrincipal, es, &processContext_);
       }) | runLast(WaitingTaskHolder(taskGroup_, &globalWaitTask));
 
       do {
         taskGroup_.wait();
       } while (not globalWaitTask.done());
       if (globalWaitTask.exceptionPtr() != nullptr) {
         std::rethrow_exception(*(globalWaitTask.exceptionPtr()));
       }
     }
     {
       //To wait, the ref count has to be 1+#streams
       FinalWaitingTask streamLoopWaitTask;
 
       using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamBegin>;
 
       RunTransitionInfo transitionInfo(runPrincipal, es);
       beginStreamsTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &streamLoopWaitTask),
                                           *schedule_,
                                           preallocations_.numberOfStreams(),
                                           transitionInfo,
                                           serviceToken_,
                                           subProcesses_);
       do {
         taskGroup_.wait();
       } while (not streamLoopWaitTask.done());
       if (streamLoopWaitTask.exceptionPtr() != nullptr) {
         std::rethrow_exception(*(streamLoopWaitTask.exceptionPtr()));
       }
     }
     FDEBUG(1) << "\tstreamBeginRun " << run << "\n";
     if (looper_) {
       //looper_->doStreamBeginRun(schedule_->streamID(),runPrincipal, es);
     }
   }

◆ branchIDListHelper() [1/2]

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

inlineprivate

Definition at line 287 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 290 of file EventProcessor.h.

References branchIDListHelper_, and edm::get_underlying_safe().

290 { return get_underlying_safe(branchIDListHelper_); }

edm::get_underlying_safe

constexpr std::shared_ptr< T > & get_underlying_safe(propagate_const< std::shared_ptr< T >> &iP)

Definition: get_underlying_safe.h:41

edm::EventProcessor::branchIDListHelper_

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

Definition: EventProcessor.h:314

◆ checkForAsyncStopRequest()

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

private

Definition at line 818 of file EventProcessor.cc.

References epSignal, 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;
       returnCode = epSignal;
     }
     return returnValue;
   }

◆ clearCounters()

void edm::EventProcessor::clearCounters ( )

Clears counters used by trigger report.

Definition at line 812 of file EventProcessor.cc.

References schedule_.

812 { schedule_->clearCounters(); }

edm::EventProcessor::schedule_

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

Definition: EventProcessor.h:328

◆ closeInputFile()

void edm::EventProcessor::closeInputFile ( bool cleaningUpAfterException )

Definition at line 948 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 965 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 1523 of file EventProcessor.cc.

References edm::WaitingTaskHolder::group(), h, handleNextEventForStreamAsync(), eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, mps_update::status, and streamLumiStatus_.

Referenced by processLumis().

                                                                      {
     {
       //all streams are sharing the same status at the moment
       auto status = streamLumiStatus_[0];  //read from streamLumiActive_ happened in calling routine
       status->needToContinueLumi();
       status->startProcessingEvents();
     }
 
     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 = iHolder]() { handleNextEventForStreamAsync(h, streamIndex); });
     }
     iHolder.group()->run(
         [this, streamIndex, h = std::move(iHolder)]() { handleNextEventForStreamAsync(h, streamIndex); });
   }

◆ deleteLumiFromCache()

void edm::EventProcessor::deleteLumiFromCache ( LuminosityBlockProcessingStatus & iStatus )

Definition at line 1821 of file EventProcessor.cc.

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

Referenced by globalEndLumiAsync().

                                                                                    {
     for (auto& s : subProcesses_) {
       s.deleteLumiFromCache(*iStatus.lumiPrincipal());
     }
     iStatus.lumiPrincipal()->setShouldWriteLumi(LuminosityBlockPrincipal::kUninitialized);
     iStatus.lumiPrincipal()->clearPrincipal();
     //FDEBUG(1) << "\tdeleteLumiFromCache " << run << "/" << lumi << "\n";
   }

◆ deleteRunFromCache()

void edm::EventProcessor::deleteRunFromCache	(	ProcessHistoryID const &	phid,
		RunNumber_t	run
	)

Definition at line 1798 of file EventProcessor.cc.

References edm::PrincipalCache::deleteRun(), FDEBUG, edm::for_all(), principalCache_, run(), and subProcesses_.

Referenced by endUnfinishedRun().

                                                                                        {
     principalCache_.deleteRun(phid, run);
     for_all(subProcesses_, [run, phid](auto& subProcess) { subProcess.deleteRunFromCache(phid, run); });
     FDEBUG(1) << "\tdeleteRunFromCache " << run << "\n";
   }

◆ doErrorStuff()

void edm::EventProcessor::doErrorStuff ( )

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

References actReg_, 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;
 
     edm::FinalWaitingTask waitTask;
     oneapi::tbb::task_group 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);
       }
     }
     group.wait();
 
     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 1000 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_->recordsToProxyIndices());
       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 1105 of file EventProcessor.cc.

References edm::Principal::clearPrincipal(), edm::FinalWaitingTask::done(), edm::WaitingTask::exceptionPtr(), 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;
 
     ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
     endGlobalTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &globalWaitTask),
                                      *schedule_,
                                      transitionInfo,
                                      serviceToken_,
                                      subProcesses_,
                                      cleaningUpAfterException);
     do {
       taskGroup_.wait();
     } while (not globalWaitTask.done());
     if (globalWaitTask.exceptionPtr() != nullptr) {
       std::rethrow_exception(*(globalWaitTask.exceptionPtr()));
     }
 
     if (beginProcessBlockSucceeded) {
       FinalWaitingTask writeWaitTask;
       writeProcessBlockAsync(edm::WaitingTaskHolder{taskGroup_, &writeWaitTask}, ProcessBlockType::New);
       do {
         taskGroup_.wait();
       } while (not writeWaitTask.done());
       if (writeWaitTask.exceptionPtr()) {
         std::rethrow_exception(*writeWaitTask.exceptionPtr());
       }
     }
 
     processBlockPrincipal.clearPrincipal();
     for (auto& s : subProcesses_) {
       s.clearProcessBlockPrincipal(ProcessBlockType::New);
     }
   }

◆ endRun()

void edm::EventProcessor::endRun	(	ProcessHistoryID const &	phid,
		RunNumber_t	run,
		bool	globalBeginSucceeded,
		bool	cleaningUpAfterException
	)

Definition at line 1268 of file EventProcessor.cc.

References actReg_, edm::FinalWaitingTask::done(), edm::EndRun, edm::RunPrincipal::endTime(), esp_, espController_, edm::WaitingTask::exceptionPtr(), FDEBUG, first, edm::waiting_task::chain::ifThen(), input_, looper_, edm::EventID::maxEventNumber(), edm::LuminosityBlockID::maxLuminosityBlockNumber(), eostools::move(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, principalCache_, processContext_, edm::RunPrincipal::run(), run(), edm::waiting_task::chain::runLast(), edm::PrincipalCache::runPrincipal(), schedule_, serviceToken_, edm::RunPrincipal::setEndTime(), subProcesses_, edm::eventsetup::synchronousEventSetupForInstance(), and taskGroup_.

Referenced by endUnfinishedRun().

                                                              {
     RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, run);
     runPrincipal.setEndTime(input_->timestamp());
 
     IOVSyncValue ts(
         EventID(runPrincipal.run(), LuminosityBlockID::maxLuminosityBlockNumber(), EventID::maxEventNumber()),
         runPrincipal.endTime());
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
       actReg_->preESSyncIOVSignal_.emit(ts);
       synchronousEventSetupForInstance(ts, taskGroup_, *espController_);
       actReg_->postESSyncIOVSignal_.emit(ts);
       sentry.completedSuccessfully();
     }
     auto const& es = esp_->eventSetupImpl();
     if (globalBeginSucceeded) {
       //To wait, the ref count has to be 1+#streams
       FinalWaitingTask streamLoopWaitTask;
 
       using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamEnd>;
 
       RunTransitionInfo transitionInfo(runPrincipal, es);
       endStreamsTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &streamLoopWaitTask),
                                         *schedule_,
                                         preallocations_.numberOfStreams(),
                                         transitionInfo,
                                         serviceToken_,
                                         subProcesses_,
                                         cleaningUpAfterException);
       do {
         taskGroup_.wait();
       } while (not streamLoopWaitTask.done());
       if (streamLoopWaitTask.exceptionPtr() != nullptr) {
         std::rethrow_exception(*(streamLoopWaitTask.exceptionPtr()));
       }
     }
     FDEBUG(1) << "\tstreamEndRun " << run << "\n";
     if (looper_) {
       //looper_->doStreamEndRun(schedule_->streamID(),runPrincipal, es);
     }
     {
       FinalWaitingTask globalWaitTask;
 
       using namespace edm::waiting_task::chain;
       chain::first([this, &runPrincipal, &es, cleaningUpAfterException](auto nextTask) {
         RunTransitionInfo transitionInfo(runPrincipal, es);
         using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd>;
         endGlobalTransitionAsync<Traits>(
             std::move(nextTask), *schedule_, transitionInfo, serviceToken_, subProcesses_, cleaningUpAfterException);
       }) | ifThen(looper_, [this, &runPrincipal, &es](auto nextTask) {
         looper_->prefetchAsync(std::move(nextTask), serviceToken_, Transition::EndRun, runPrincipal, es);
       }) | ifThen(looper_, [this, &runPrincipal, &es](auto nextTask) {
         looper_->doEndRun(runPrincipal, es, &processContext_);
       }) | runLast(WaitingTaskHolder(taskGroup_, &globalWaitTask));
 
       do {
         taskGroup_.wait();
       } while (not globalWaitTask.done());
       if (globalWaitTask.exceptionPtr() != nullptr) {
         std::rethrow_exception(*(globalWaitTask.exceptionPtr()));
       }
     }
     FDEBUG(1) << "\tendRun " << run << "\n";
   }

◆ endUnfinishedLumi()

void edm::EventProcessor::endUnfinishedLumi ( )

Definition at line 1667 of file EventProcessor.cc.

References edm::FinalWaitingTask::done(), edm::WaitingTask::exceptionPtr(), mps_fire::i, edm::PreallocationConfiguration::numberOfStreams(), preallocations_, streamEndLumiAsync(), streamLumiActive_, streamLumiStatus_, and taskGroup_.

                                          {
     if (streamLumiActive_.load() > 0) {
       FinalWaitingTask globalWaitTask;
       {
         WaitingTaskHolder globalTaskHolder{taskGroup_, &globalWaitTask};
         for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
           if (streamLumiStatus_[i]) {
             streamEndLumiAsync(globalTaskHolder, i);
           }
         }
       }
       do {
         taskGroup_.wait();
       } while (not globalWaitTask.done());
       if (globalWaitTask.exceptionPtr() != nullptr) {
         std::rethrow_exception(*(globalWaitTask.exceptionPtr()));
       }
     }
   }

◆ endUnfinishedRun()

void edm::EventProcessor::endUnfinishedRun	(	ProcessHistoryID const &	phid,
		RunNumber_t	run,
		bool	globalBeginSucceeded,
		bool	cleaningUpAfterException,
		bool	eventSetupForInstanceSucceeded
	)

Definition at line 1239 of file EventProcessor.cc.

References deleteRunFromCache(), endRun(), edm::RunPrincipal::kNo, edm::RunPrincipal::mergeableRunProductMetadata(), edm::MergeableRunProductMetadata::postWriteRun(), edm::MergeableRunProductMetadata::preWriteRun(), principalCache_, run(), edm::PrincipalCache::runPrincipal(), edm::RunPrincipal::shouldWriteRun(), submitPVValidationJobs::t, taskGroup_, and writeRunAsync().

                                                                              {
     if (eventSetupForInstanceSucceeded) {
       //If we skip empty runs, this would be called conditionally
       endRun(phid, run, globalBeginSucceeded, cleaningUpAfterException);
 
       if (globalBeginSucceeded) {
         RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, run);
         if (runPrincipal.shouldWriteRun() != RunPrincipal::kNo) {
           FinalWaitingTask t;
           MergeableRunProductMetadata* mergeableRunProductMetadata = runPrincipal.mergeableRunProductMetadata();
           mergeableRunProductMetadata->preWriteRun();
           writeRunAsync(edm::WaitingTaskHolder{taskGroup_, &t}, phid, run, mergeableRunProductMetadata);
           do {
             taskGroup_.wait();
           } while (not t.done());
           mergeableRunProductMetadata->postWriteRun();
           if (t.exceptionPtr()) {
             std::rethrow_exception(*t.exceptionPtr());
           }
         }
       }
     }
     deleteRunFromCache(phid, run);
   }

◆ fileBlockValid()

bool edm::EventProcessor::fileBlockValid ( )

inline

Definition at line 196 of file EventProcessor.h.

References fb_.

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

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

edm::EventProcessor::fb_

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

Definition: EventProcessor.h:339

◆ 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 802 of file EventProcessor.cc.

References schedule_.

                                                                                      {
     return schedule_->getAllModuleDescriptions();
   }

◆ getToken()

ServiceToken edm::EventProcessor::getToken ( )

Definition at line 800 of file EventProcessor.cc.

References serviceToken_.

Referenced by ~EventProcessor().

800 { return serviceToken_; }

edm::EventProcessor::serviceToken_

ServiceToken serviceToken_

Definition: EventProcessor.h:317

◆ globalEndLumiAsync()

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

Definition at line 1549 of file EventProcessor.cc.

References CMS_SA_ALLOW, deleteLumiFromCache(), edm::EndLuminosityBlock, esp_, 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::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();
     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* iPtr, auto nextTask) mutable {
       std::exception_ptr ptr;
       if (iPtr) {
         ptr = *iPtr;
       }
       ServiceRegistry::Operate operate(serviceToken_);
 
       // 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 { deleteLumiFromCache(*status); } catch (...) {
         if (not ptr) {
           ptr = std::current_exception();
         }
       }
       // Caught exception is passed to handleEndLumiExceptions()
       CMS_SA_ALLOW try {
         status->resumeGlobalLumiQueue();
         queueWhichWaitsForIOVsToFinish_.resume();
       } catch (...) {
         if (not ptr) {
           ptr = std::current_exception();
         }
       }
       // Caught exception is passed to handleEndLumiExceptions()
       CMS_SA_ALLOW try {
         // This call to status.resetResources() must occur before iTask is destroyed.
         // Otherwise there will be a data race which could result in endRun
         // being delayed until it is too late to successfully call it.
         status->resetResources();
         status.reset();
       } catch (...) {
         if (not ptr) {
           ptr = std::current_exception();
         }
       }
 
       if (ptr) {
         handleEndLumiExceptions(&ptr, nextTask);
       }
     }) | runLast(std::move(iTask));
   }

◆ handleEndLumiExceptions()

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

Definition at line 1540 of file EventProcessor.cc.

References setDeferredException(), setExceptionMessageLumis(), and createJobs::tmp.

Referenced by globalEndLumiAsync(), and streamEndLumiAsync().

                                                                                                       {
     if (setDeferredException(*iPtr)) {
       WaitingTaskHolder tmp(holder);
       tmp.doneWaiting(*iPtr);
     } else {
       setExceptionMessageLumis();
     }
   }

◆ handleNextEventForStreamAsync()

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

private

Definition at line 1892 of file EventProcessor.cc.

References beginLumiAsync(), CMS_SA_ALLOW, deferredExceptionPtr_, deferredExceptionPtrIsSet_, edm::WaitingTaskHolder::doneWaiting(), MillePedeFileConverter_cfg::e, edm::WaitingTaskHolder::group(), edm::InputSource::IsLumi, lastTransitionType(), edm::make_waiting_task(), eostools::move(), processEventAsync(), edm::SerialTaskQueueChain::push(), readNextEventForStream(), edm::SharedResourcesAcquirer::serialQueueChain(), serviceToken_, sourceResourcesAcquirer_, mps_update::status, streamEndLumiAsync(), and streamLumiStatus_.

Referenced by beginLumiAsync(), and continueLumiAsync().

                                                                                                        {
     sourceResourcesAcquirer_.serialQueueChain().push(*iTask.group(), [this, iTask, iStreamIndex]() mutable {
       ServiceRegistry::Operate operate(serviceToken_);
       //we do not want to extend the lifetime of the shared_ptr to the end of this function
       // as steramEndLumiAsync may clear the value from streamLumiStatus_[iStreamIndex]
       auto status = streamLumiStatus_[iStreamIndex].get();
       // Caught exception is propagated to EventProcessor::runToCompletion() via deferredExceptionPtr_
       CMS_SA_ALLOW try {
         if (readNextEventForStream(iStreamIndex, *status)) {
           auto recursionTask = make_waiting_task([this, iTask, iStreamIndex](std::exception_ptr const* iPtr) mutable {
             if (iPtr) {
               // Try to end the stream properly even if an exception was
               // thrown on an event.
               bool expected = false;
               if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
                 // This is the case where the exception in iPtr is the primary
                 // exception and we want to see its message.
                 deferredExceptionPtr_ = *iPtr;
                 WaitingTaskHolder tempHolder(iTask);
                 tempHolder.doneWaiting(*iPtr);
               }
               streamEndLumiAsync(std::move(iTask), iStreamIndex);
               //the stream will stop now
               return;
             }
             handleNextEventForStreamAsync(std::move(iTask), iStreamIndex);
           });
 
           processEventAsync(WaitingTaskHolder(*iTask.group(), recursionTask), iStreamIndex);
         } else {
           //the stream will stop now
           if (status->isLumiEnding()) {
             if (lastTransitionType() == InputSource::IsLumi and not status->haveStartedNextLumi()) {
               status->startNextLumi();
               beginLumiAsync(status->nextSyncValue(), status->runResource(), iTask);
             }
             streamEndLumiAsync(std::move(iTask), iStreamIndex);
           } else {
             iTask.doneWaiting(std::exception_ptr{});
           }
         }
       } catch (...) {
         // It is unlikely we will ever get in here ...
         // But if we do try to clean up and propagate the exception
         if (streamLumiStatus_[iStreamIndex]) {
           streamEndLumiAsync(iTask, iStreamIndex);
         }
         bool expected = false;
         if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
           auto e = std::current_exception();
           deferredExceptionPtr_ = e;
           iTask.doneWaiting(e);
         }
       }
     });
   }

◆ init()

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

private

Definition at line 343 of file EventProcessor.cc.

References act_table_, actReg_, cms::cuda::assert(), branchesToDeleteEarly_, branchIDListHelper(), branchIDListHelper_, CMS_SA_ALLOW, trackingPlots::common, edm::errors::Configuration, deleteNonConsumedUnscheduledModules_, edm::dumpOptionsToLogFile(), edm::ensureAvailableAccelerators(), SiStripBadComponentsDQMServiceTemplate_cfg::ep, esp_, espController_, Exception, FDEBUG, processOptions_cff::fileMode, fileModeNoMerge_, edm::fillLooper(), forceESCacheClearOnNewRun_, edm::get_underlying_safe(), edm::ParameterSet::getParameter(), edm::ParameterSet::getUntrackedParameterSet(), historyAppender_, input_, edm::PrincipalCache::insert(), edm::PrincipalCache::insertForInput(), edm::ParentageRegistry::insertMapped(), edm::ParentageRegistry::instance(), mps_monitormerge::items, edm::serviceregistry::kConfigurationOverrides, looper_, lumiQueue_, edm::makeInput(), mergeableRunProductProcesses_, eostools::move(), edm::PreallocationConfiguration::numberOfLuminosityBlocks(), edm::PreallocationConfiguration::numberOfStreams(), or, edm::parameterSet(), edm::popSubProcessVParameterSet(), preallocations_, preg(), preg_, principalCache_, printDependencies_, processBlockHelper_, processConfiguration_, processContext_, edm::ParameterSet::registerIt(), schedule_, serviceToken_, edm::PrincipalCache::setNumberOfConcurrentPrincipals(), edm::ProcessContext::setProcessConfiguration(), edm::MergeableRunProductProcesses::setProcessesWithMergeableRunProducts(), edm::PrincipalCache::setProcessHistoryRegistry(), edm::IllegalParameters::setThrowAnException(), streamLumiStatus_, streamQueues_, 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 nConcurrentRuns = optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentRuns");
     if (nConcurrentRuns != 1) {
       throw Exception(errors::Configuration, "Illegal value nConcurrentRuns : ")
           << "Although the plan is to change this in the future, currently nConcurrentRuns must always be 1.\n";
     }
     unsigned int nConcurrentLumis =
         optionsPset.getUntrackedParameter<unsigned int>("numberOfConcurrentLuminosityBlocks");
     if (nConcurrentLumis == 0) {
       nConcurrentLumis = 2;
     }
     if (nConcurrentLumis > nStreams) {
       nConcurrentLumis = nStreams;
     }
     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.
     unsigned int maxConcurrentIOVs = nConcurrentLumis;
 
     //Check that relationships between threading parameters makes sense
     /*
       if(nThreads<nStreams) {
       //bad
       }
       if(nConcurrentRuns>nStreams) {
       //bad
       }
       if(nConcurrentRuns>nConcurrentLumis) {
       //bad
       }
     */
     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");
     }
 
     // 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 (nStreams > 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};
 
       lumiQueue_ = std::make_unique<LimitedTaskQueue>(nConcurrentLumis);
       streamQueues_.resize(nStreams);
       streamLumiStatus_.resize(nStreams);
 
       processBlockHelper_ = std::make_shared<ProcessBlockHelper>();
 
       // initialize the input source
       input_ = makeInput(*parameterSet,
                          *common,
                          items.preg(),
                          items.branchIDListHelper(),
                          get_underlying_safe(processBlockHelper_),
                          items.thinnedAssociationsHelper(),
                          items.actReg_,
                          items.processConfiguration(),
                          preallocations_);
 
       // initialize the Schedule
       schedule_ =
           items.initSchedule(*parameterSet, 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());
       principalCache_.setProcessHistoryRegistry(input_->processHistoryRegistry());
 
       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_.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_);
       }
     } 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 1069 of file EventProcessor.cc.

References edm::Principal::clearPrincipal(), edm::FinalWaitingTask::done(), edm::WaitingTask::exceptionPtr(), 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;
 
       ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
       beginGlobalTransitionAsync<Traits>(
           WaitingTaskHolder(taskGroup_, &globalWaitTask), *schedule_, transitionInfo, serviceToken_, subProcesses_);
 
       do {
         taskGroup_.wait();
       } while (not globalWaitTask.done());
       if (globalWaitTask.exceptionPtr() != nullptr) {
         std::rethrow_exception(*(globalWaitTask.exceptionPtr()));
       }
 
       FinalWaitingTask writeWaitTask;
       writeProcessBlockAsync(edm::WaitingTaskHolder{taskGroup_, &writeWaitTask}, ProcessBlockType::Input);
       do {
         taskGroup_.wait();
       } while (not writeWaitTask.done());
       if (writeWaitTask.exceptionPtr()) {
         std::rethrow_exception(*writeWaitTask.exceptionPtr());
       }
 
       processBlockPrincipal.clearPrincipal();
       for (auto& s : subProcesses_) {
         s.clearProcessBlockPrincipal(ProcessBlockType::Input);
       }
     }
   }

◆ lastTransitionType()

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

inline

Definition at line 186 of file EventProcessor.h.

References deferredExceptionPtrIsSet_, edm::InputSource::IsStop, and lastSourceTransition_.

Referenced by handleNextEventForStreamAsync(), and processLumis().

                                                    {
       if (deferredExceptionPtrIsSet_) {
         return InputSource::IsStop;
       }
       return lastSourceTransition_;
     }

◆ looper() [1/2]

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

inlineprivate

Definition at line 297 of file EventProcessor.h.

References edm::get_underlying_safe(), and looper_.

Referenced by endJob().

297 { return get_underlying_safe(looper_); }

edm::get_underlying_safe

constexpr std::shared_ptr< T > & get_underlying_safe(propagate_const< std::shared_ptr< T >> &iP)

Definition: get_underlying_safe.h:41

edm::EventProcessor::looper_

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

Definition: EventProcessor.h:340

◆ looper() [2/2]

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

inlineprivate

Definition at line 298 of file EventProcessor.h.

References edm::get_underlying_safe(), and looper_.

298 { return get_underlying_safe(looper_); }

edm::get_underlying_safe

constexpr std::shared_ptr< T > & get_underlying_safe(propagate_const< std::shared_ptr< T >> &iP)

Definition: get_underlying_safe.h:41

edm::EventProcessor::looper_

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

Definition: EventProcessor.h:340

◆ nextLuminosityBlockID()

edm::LuminosityBlockNumber_t edm::EventProcessor::nextLuminosityBlockID ( )

Definition at line 859 of file EventProcessor.cc.

References input_.

Referenced by readNextEventForStream().

859 { return input_->luminosityBlock(); }

edm::EventProcessor::input_

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

Definition: EventProcessor.h:318

◆ nextRunID()

std::pair< edm::ProcessHistoryID, edm::RunNumber_t > edm::EventProcessor::nextRunID ( )

Definition at line 855 of file EventProcessor.cc.

References input_.

                                                                          {
     return std::make_pair(input_->reducedProcessHistoryID(), input_->run());
   }

◆ nextTransitionType()

InputSource::ItemType edm::EventProcessor::nextTransitionType ( )

Definition at line 829 of file EventProcessor.cc.

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

Referenced by readNextEventForStream().

                                                          {
     if (deferredExceptionPtrIsSet_.load()) {
       lastSourceTransition_ = InputSource::IsStop;
       return InputSource::IsStop;
     }
 
     SendSourceTerminationSignalIfException sentry(actReg_.get());
     InputSource::ItemType itemType;
     //For now, do nothing with InputSource::IsSynchronize
     do {
       itemType = input_->nextItemType();
     } while (itemType == InputSource::IsSynchronize);
 
     lastSourceTransition_ = itemType;
     sentry.completedSuccessfully();
 
     StatusCode returnCode = epSuccess;
 
     if (checkForAsyncStopRequest(returnCode)) {
       actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
       lastSourceTransition_ = InputSource::IsStop;
     }
 
     return lastSourceTransition_;
   }

◆ openOutputFiles()

void edm::EventProcessor::openOutputFiles ( )

Definition at line 957 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 285 of file EventProcessor.h.

References edm::get_underlying_safe(), and preg_.

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

285 { return get_underlying_safe(preg_); }

edm::EventProcessor::preg_

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

Definition: EventProcessor.h:313

edm::get_underlying_safe

constexpr std::shared_ptr< T > & get_underlying_safe(propagate_const< std::shared_ptr< T >> &iP)

Definition: get_underlying_safe.h:41

◆ preg() [2/2]

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

inlineprivate

Definition at line 286 of file EventProcessor.h.

References edm::get_underlying_safe(), and preg_.

286 { return get_underlying_safe(preg_); }

edm::EventProcessor::preg_

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

Definition: EventProcessor.h:313

edm::get_underlying_safe

constexpr std::shared_ptr< T > & get_underlying_safe(propagate_const< std::shared_ptr< T >> &iP)

Definition: get_underlying_safe.h:41

◆ prepareForNextLoop()

void edm::EventProcessor::prepareForNextLoop ( )

Definition at line 1021 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 140 of file EventProcessor.h.

References processConfiguration_.

140 { return *processConfiguration_; }

edm::EventProcessor::processConfiguration_

std::shared_ptr< ProcessConfiguration const > processConfiguration_

Definition: EventProcessor.h:324

◆ processEventAsync()

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

private

Definition at line 1963 of file EventProcessor.cc.

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

Referenced by handleNextEventForStreamAsync().

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

◆ processEventAsyncImpl()

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

private

Definition at line 1967 of file EventProcessor.cc.

References esp_, makeMEIFBenchmarkPlots::ev, edm::PrincipalCache::eventPrincipal(), FDEBUG, first, edm::waiting_task::chain::ifThen(), info(), edm::Service< T >::isAvailable(), looper_, eostools::move(), principalCache_, 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([pep](auto nextTask) {
       FDEBUG(1) << "\tprocessEvent\n";
       pep->clearEventPrincipal();
     }) | runLast(iHolder);
   }

◆ processEventWithLooper()

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

private

Definition at line 1996 of file EventProcessor.cc.

References esp_, input_, edm::InputSource::IsStop, edm::EDLooperBase::kContinue, edm::ProcessingController::kToPreviousEvent, edm::ProcessingController::kToSpecifiedEvent, edm::ProcessingController::lastOperationSucceeded(), lastSourceTransition_, 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;
       lastSourceTransition_ = InputSource::IsStop;
     }
   }

◆ processLumis()

InputSource::ItemType edm::EventProcessor::processLumis ( std::shared_ptr< void > const & iRunResource )

Definition at line 1336 of file EventProcessor.cc.

References cms::cuda::assert(), beginLumiAsync(), continueLumiAsync(), edm::FinalWaitingTask::done(), edm::WaitingTask::exceptionPtr(), input_, lastTransitionType(), edm::PreallocationConfiguration::numberOfStreams(), preallocations_, streamLumiActive_, and taskGroup_.

                                                                                           {
     FinalWaitingTask waitTask;
     if (streamLumiActive_ > 0) {
       assert(streamLumiActive_ == preallocations_.numberOfStreams());
       // Continue after opening a new input file
       continueLumiAsync(WaitingTaskHolder{taskGroup_, &waitTask});
     } else {
       beginLumiAsync(IOVSyncValue(EventID(input_->run(), input_->luminosityBlock(), 0),
                                   input_->luminosityBlockAuxiliary()->beginTime()),
                      iRunResource,
                      WaitingTaskHolder{taskGroup_, &waitTask});
     }
     do {
       taskGroup_.wait();
     } while (not waitTask.done());
 
     if (waitTask.exceptionPtr() != nullptr) {
       std::rethrow_exception(*(waitTask.exceptionPtr()));
     }
     return lastTransitionType();
   }

◆ readAndMergeLumi()

int edm::EventProcessor::readAndMergeLumi ( LuminosityBlockProcessingStatus & iStatus )

Definition at line 1747 of file EventProcessor.cc.

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

Referenced by readNextEventForStream().

                                                                                {
     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();
     }
     return input_->luminosityBlock();
   }

◆ readAndMergeRun()

std::pair< ProcessHistoryID, RunNumber_t > edm::EventProcessor::readAndMergeRun ( )

Definition at line 1716 of file EventProcessor.cc.

References actReg_, cms::cuda::assert(), input_, edm::PrincipalCache::merge(), preg(), principalCache_, and edm::PrincipalCache::runPrincipalPtr().

                                                                          {
     principalCache_.merge(input_->runAuxiliary(), preg());
     auto runPrincipal = principalCache_.runPrincipalPtr();
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
       input_->readAndMergeRun(*runPrincipal);
       sentry.completedSuccessfully();
     }
     assert(input_->reducedProcessHistoryID() == runPrincipal->reducedProcessHistoryID());
     return std::make_pair(runPrincipal->reducedProcessHistoryID(), input_->run());
   }

◆ readEvent()

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

private

Definition at line 1949 of file EventProcessor.cc.

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

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);
 
     streamLumiStatus_[iStreamIndex]->updateLastTimestamp(input_->timestamp());
     sentry.completedSuccessfully();
 
     FDEBUG(1) << "\treadEvent\n";
   }

◆ readFile()

void edm::EventProcessor::readFile ( )

Definition at line 930 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_, edm::PrincipalCache::preReadFile(), principalCache_, and findQualityFiles::size.

                                 {
     FDEBUG(1) << " \treadFile\n";
     size_t size = preg_->size();
     SendSourceTerminationSignalIfException sentry(actReg_.get());
 
     principalCache_.preReadFile();
 
     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()

void edm::EventProcessor::readLuminosityBlock ( LuminosityBlockProcessingStatus & iStatus )

Definition at line 1728 of file EventProcessor.cc.

References actReg_, cms::cuda::assert(), Exception, edm::PrincipalCache::getAvailableLumiPrincipalPtr(), edm::PrincipalCache::hasRunPrincipal(), historyAppender_, input_, edm::errors::LogicError, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), eostools::move(), principalCache_, and edm::PrincipalCache::runPrincipalPtr().

Referenced by beginLumiAsync().

                                                                                    {
     if (!principalCache_.hasRunPrincipal()) {
       throw edm::Exception(edm::errors::LogicError) << "EventProcessor::readLuminosityBlock\n"
                                                     << "Illegal attempt to insert lumi into cache\n"
                                                     << "Run is invalid\n"
                                                     << "Contact a Framework Developer\n";
     }
     auto lbp = principalCache_.getAvailableLumiPrincipalPtr();
     assert(lbp);
     lbp->setAux(*input_->luminosityBlockAuxiliary());
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
       input_->readLuminosityBlock(*lbp, *historyAppender_);
       sentry.completedSuccessfully();
     }
     lbp->setRunPrincipal(principalCache_.runPrincipalPtr());
     iStatus.lumiPrincipal() = std::move(lbp);
   }

◆ readNextEventForStream()

bool edm::EventProcessor::readNextEventForStream	(	unsigned int	iStreamIndex,
		LuminosityBlockProcessingStatus &	iLumiStatus
	)

private

Definition at line 1830 of file EventProcessor.cc.

References CMS_SA_ALLOW, edm::LuminosityBlockProcessingStatus::continuingLumi(), deferredExceptionPtr_, deferredExceptionPtrIsSet_, edm::LuminosityBlockProcessingStatus::endLumi(), edm::LuminosityBlockProcessingStatus::haveContinuedLumi(), input_, edm::InputSource::IsEvent, edm::InputSource::IsLumi, edm::InputSource::IsRun, edm::InputSource::IsStop, lastSourceTransition_, edm::LuminosityBlockProcessingStatus::lumiPrincipal(), nextLuminosityBlockID(), nextTransitionType(), or, readAndMergeLumi(), readEvent(), serviceToken_, edm::LuminosityBlockProcessingStatus::setNextSyncValue(), shouldWeStop(), sourceMutex_, edm::LuminosityBlockProcessingStatus::stopProcessingEvents(), and edm::LuminosityBlockProcessingStatus::wasEventProcessingStopped().

Referenced by handleNextEventForStreamAsync().

                                                                                                                  {
     if (deferredExceptionPtrIsSet_.load(std::memory_order_acquire)) {
       iStatus.endLumi();
       return false;
     }
 
     if (iStatus.wasEventProcessingStopped()) {
       return false;
     }
 
     if (shouldWeStop()) {
       lastSourceTransition_ = InputSource::IsStop;
       iStatus.stopProcessingEvents();
       iStatus.endLumi();
       return false;
     }
 
     ServiceRegistry::Operate operate(serviceToken_);
     // Caught exception is propagated to EventProcessor::runToCompletion() via deferredExceptionPtr_
     CMS_SA_ALLOW try {
       //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()));
 
       auto itemType = iStatus.continuingLumi() ? InputSource::IsLumi : nextTransitionType();
       if (InputSource::IsLumi == itemType) {
         iStatus.haveContinuedLumi();
         while (itemType == InputSource::IsLumi and iStatus.lumiPrincipal()->run() == input_->run() and
                iStatus.lumiPrincipal()->luminosityBlock() == nextLuminosityBlockID()) {
           readAndMergeLumi(iStatus);
           itemType = nextTransitionType();
         }
         if (InputSource::IsLumi == itemType) {
           iStatus.setNextSyncValue(IOVSyncValue(EventID(input_->run(), input_->luminosityBlock(), 0),
                                                 input_->luminosityBlockAuxiliary()->beginTime()));
         }
       }
       if (InputSource::IsEvent != itemType) {
         iStatus.stopProcessingEvents();
 
         //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::IsStop == itemType or InputSource::IsLumi == itemType or
             (InputSource::IsRun == itemType and iStatus.lumiPrincipal()->run() != input_->run())) {
           iStatus.endLumi();
         }
         return false;
       }
       readEvent(iStreamIndex);
     } catch (...) {
       bool expected = false;
       if (deferredExceptionPtrIsSet_.compare_exchange_strong(expected, true)) {
         deferredExceptionPtr_ = std::current_exception();
         iStatus.endLumi();
       }
       return false;
     }
     return true;
   }

◆ readProcessBlock()

void edm::EventProcessor::readProcessBlock ( ProcessBlockPrincipal & processBlockPrincipal )

Definition at line 1687 of file EventProcessor.cc.

References actReg_, and input_.

Referenced by inputProcessBlocks().

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

◆ readRun()

std::pair< ProcessHistoryID, RunNumber_t > edm::EventProcessor::readRun ( )

Definition at line 1693 of file EventProcessor.cc.

References actReg_, cms::cuda::assert(), Exception, edm::PrincipalCache::hasRunPrincipal(), historyAppender_, input_, edm::PrincipalCache::insert(), edm::errors::LogicError, mergeableRunProductProcesses_, preg(), principalCache_, and processConfiguration_.

                                                                  {
     if (principalCache_.hasRunPrincipal()) {
       throw edm::Exception(edm::errors::LogicError) << "EventProcessor::readRun\n"
                                                     << "Illegal attempt to insert run into cache\n"
                                                     << "Contact a Framework Developer\n";
     }
     auto rp = std::make_shared<RunPrincipal>(input_->runAuxiliary(),
                                              preg(),
                                              *processConfiguration_,
                                              historyAppender_.get(),
                                              0,
                                              true,
                                              &mergeableRunProductProcesses_);
     {
       SendSourceTerminationSignalIfException sentry(actReg_.get());
       input_->readRun(*rp, *historyAppender_);
       sentry.completedSuccessfully();
     }
     assert(input_->reducedProcessHistoryID() == rp->reducedProcessHistoryID());
     principalCache_.insert(rp);
     return std::make_pair(rp->reducedProcessHistoryID(), input_->run());
   }

◆ respondToCloseInputFile()

void edm::EventProcessor::respondToCloseInputFile ( )

Definition at line 982 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 972 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 1015 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 375 of file EventProcessor.h.

References runToCompletion().

Referenced by beginRun(), Types.EventID::cppID(), Types.LuminosityBlockID::cppID(), deleteRunFromCache(), endRun(), endUnfinishedRun(), and writeRunAsync().

375 { return runToCompletion(); }

edm::EventProcessor::runToCompletion

StatusCode runToCompletion()

Definition: EventProcessor.cc:861

◆ runToCompletion()

EventProcessor::StatusCode edm::EventProcessor::runToCompletion ( )

Definition at line 861 of file EventProcessor.cc.

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

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

                                                            {
     StatusCode returnCode = epSuccess;
     asyncStopStatusCodeFromProcessingEvents_ = epSuccess;
     {
       beginJob();  //make sure this was called
 
       // make the services available
       ServiceRegistry::Operate operate(serviceToken_);
 
       asyncStopRequestedWhileProcessingEvents_ = false;
       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::IsStop) {
               //problem with the source
               doErrorStuff();
 
               throw cms::Exception("BadTransition") << "Unexpected transition change " << 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_.empty()) {
           e.addAdditionalInfo(exceptionMessageRuns_);
           if (e.alreadyPrinted()) {
             LogAbsolute("Additional Exceptions") << exceptionMessageRuns_;
           }
         }
         if (!exceptionMessageFiles_.empty()) {
           e.addAdditionalInfo(exceptionMessageFiles_);
           if (e.alreadyPrinted()) {
             LogAbsolute("Additional Exceptions") << exceptionMessageFiles_;
           }
         }
         throw;
       }
     }
 
     return returnCode;
   }

◆ setDeferredException()

bool edm::EventProcessor::setDeferredException ( std::exception_ptr iException )

Definition at line 2045 of file EventProcessor.cc.

References deferredExceptionPtr_, and deferredExceptionPtrIsSet_.

Referenced by handleEndLumiExceptions().

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

References exceptionMessageFiles_.

2039 { exceptionMessageFiles_ = message; }

edm::EventProcessor::exceptionMessageFiles_

std::string exceptionMessageFiles_

Definition: EventProcessor.h:352

◆ setExceptionMessageLumis()

void edm::EventProcessor::setExceptionMessageLumis ( )

Definition at line 2043 of file EventProcessor.cc.

References exceptionMessageLumis_.

Referenced by handleEndLumiExceptions().

2043 { exceptionMessageLumis_ = true; }

edm::EventProcessor::exceptionMessageLumis_

std::atomic< bool > exceptionMessageLumis_

Definition: EventProcessor.h:354

◆ setExceptionMessageRuns()

void edm::EventProcessor::setExceptionMessageRuns ( std::string & message )

Definition at line 2041 of file EventProcessor.cc.

References exceptionMessageRuns_.

2041 { exceptionMessageRuns_ = message; }

edm::EventProcessor::exceptionMessageRuns_

std::string exceptionMessageRuns_

Definition: EventProcessor.h:353

◆ shouldWeCloseOutput()

bool edm::EventProcessor::shouldWeCloseOutput ( ) const

Definition at line 1026 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 2024 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 990 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";
   }

◆ streamEndLumiAsync()

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

Definition at line 1622 of file EventProcessor.cc.

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

Referenced by beginLumiAsync(), endUnfinishedLumi(), and handleNextEventForStreamAsync().

                                                                                                {
     auto t = edm::make_waiting_task([this, iStreamIndex, iTask](std::exception_ptr const* iPtr) mutable {
       if (iPtr) {
         handleEndLumiExceptions(iPtr, iTask);
       }
       auto status = streamLumiStatus_[iStreamIndex];
       //reset status before releasing queue else get race condtion
       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());
 
     bool cleaningUpAfterException = lumiStatus->cleaningUpAfterException();
     auto eventSetupImpls = &lumiStatus->eventSetupImpls();
 
     if (lumiStatus->didGlobalBeginSucceed()) {
       auto& lumiPrincipal = *lumiStatus->lumiPrincipal();
       IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), EventID::maxEventNumber()),
                       lumiPrincipal.endTime());
       using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd>;
       LumiTransitionInfo transitionInfo(lumiPrincipal, es, eventSetupImpls);
       endStreamTransitionAsync<Traits>(std::move(lumiDoneTask),
                                        *schedule_,
                                        iStreamIndex,
                                        transitionInfo,
                                        serviceToken_,
                                        subProcesses_,
                                        cleaningUpAfterException);
     }
   }

◆ taskCleanup()

void edm::EventProcessor::taskCleanup ( )

Definition at line 597 of file EventProcessor.cc.

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

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

◆ thinnedAssociationsHelper() [1/2]

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

inlineprivate

Definition at line 291 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 294 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 2054 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 806 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEvents().

806 { return schedule_->totalEvents(); }

edm::EventProcessor::schedule_

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

Definition: EventProcessor.h:328

◆ 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 810 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEventsFailed().

810 { return schedule_->totalEventsFailed(); }

edm::EventProcessor::schedule_

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

Definition: EventProcessor.h:328

◆ 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 808 of file EventProcessor.cc.

References schedule_.

Referenced by PythonEventProcessor::totalEventsPassed().

808 { return schedule_->totalEventsPassed(); }

edm::EventProcessor::schedule_

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

Definition: EventProcessor.h:328

◆ warnAboutLegacyModules()

void edm::EventProcessor::warnAboutLegacyModules ( ) const

private

Definition at line 2073 of file EventProcessor.cc.

References edm::Worker::kLegacy, alignCSCRings::s, and schedule_.

Referenced by beginJob().

                                                     {
     std::unique_ptr<LogSystem> s;
     for (auto worker : schedule_->allWorkers()) {
       if (worker->moduleConcurrencyType() == Worker::kLegacy) {
         if (not s) {
           s = std::make_unique<LogSystem>("LegacyModules");
           (*s) << "The following legacy modules are configured. Support for legacy modules\n"
                   "is going to end soon. These modules need to be converted to have type\n"
                   "edm::global, edm::stream, edm::one, or in rare cases edm::limited.";
         }
         (*s) << "\n  " << worker->description()->moduleName() << " " << worker->description()->moduleLabel();
       }
     }
   }

◆ writeLumiAsync()

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

Definition at line 1804 of file EventProcessor.cc.

References actReg_, first, edm::waiting_task::chain::ifThen(), edm::LuminosityBlockPrincipal::kNo, edm::waiting_task::chain::lastTask(), edm::LuminosityBlockPrincipal::luminosityBlock(), edm::RunPrincipal::mergeableRunProductMetadata(), eostools::move(), 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 1764 of file EventProcessor.cc.

References actReg_, first, edm::waiting_task::chain::ifThen(), eostools::move(), 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,
		ProcessHistoryID const &	phid,
		RunNumber_t	run,
		MergeableRunProductMetadata const *	mergeableRunProductMetadata
	)

Definition at line 1778 of file EventProcessor.cc.

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

Referenced by endUnfinishedRun().

                                                                                                      {
     using namespace edm::waiting_task;
     chain::first([&](auto nextTask) {
       ServiceRegistry::Operate op(serviceToken_);
       schedule_->writeRunAsync(nextTask,
                                principalCache_.runPrincipal(phid, run),
                                &processContext_,
                                actReg_.get(),
                                mergeableRunProductMetadata);
     }) | chain::ifThen(not subProcesses_.empty(), [this, phid, run, mergeableRunProductMetadata](auto nextTask) {
       ServiceRegistry::Operate op(serviceToken_);
       for (auto& s : subProcesses_) {
         s.writeRunAsync(nextTask, phid, run, 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 312 of file EventProcessor.h.

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

◆ asyncStopRequestedWhileProcessingEvents_

bool edm::EventProcessor::asyncStopRequestedWhileProcessingEvents_

private

Definition at line 361 of file EventProcessor.h.

Referenced by runToCompletion().

◆ asyncStopStatusCodeFromProcessingEvents_

StatusCode edm::EventProcessor::asyncStopStatusCodeFromProcessingEvents_

private

Definition at line 362 of file EventProcessor.h.

Referenced by runToCompletion().

◆ beginJobCalled_

bool edm::EventProcessor::beginJobCalled_

private

Definition at line 349 of file EventProcessor.h.

Referenced by beginJob().

◆ branchesToDeleteEarly_

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

private

Definition at line 334 of file EventProcessor.h.

Referenced by beginJob(), and init().

◆ branchIDListHelper_

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

private

Definition at line 314 of file EventProcessor.h.

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

◆ deferredExceptionPtr_

std::exception_ptr edm::EventProcessor::deferredExceptionPtr_

private

Definition at line 344 of file EventProcessor.h.

Referenced by handleNextEventForStreamAsync(), readNextEventForStream(), runToCompletion(), and setDeferredException().

◆ deferredExceptionPtrIsSet_

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

private

Definition at line 343 of file EventProcessor.h.

Referenced by handleNextEventForStreamAsync(), lastTransitionType(), nextTransitionType(), readNextEventForStream(), runToCompletion(), and setDeferredException().

◆ deleteNonConsumedUnscheduledModules_

bool edm::EventProcessor::deleteNonConsumedUnscheduledModules_ = true

private

Definition at line 370 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(), beginRun(), endOfLoop(), endRun(), globalEndLumiAsync(), init(), prepareForNextLoop(), processEventAsyncImpl(), processEventWithLooper(), streamEndLumiAsync(), 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(), beginRun(), endRun(), init(), taskCleanup(), and ~EventProcessor().

◆ eventSetupDataToExcludeFromPrefetching_

ExcludedDataMap edm::EventProcessor::eventSetupDataToExcludeFromPrefetching_

private

Definition at line 367 of file EventProcessor.h.

◆ exceptionMessageFiles_

std::string edm::EventProcessor::exceptionMessageFiles_

private

Definition at line 352 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageFiles().

◆ exceptionMessageLumis_

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

private

Definition at line 354 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageLumis().

◆ exceptionMessageRuns_

std::string edm::EventProcessor::exceptionMessageRuns_

private

Definition at line 353 of file EventProcessor.h.

Referenced by runToCompletion(), and setExceptionMessageRuns().

◆ fb_

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

private

Definition at line 339 of file EventProcessor.h.

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

◆ fileModeNoMerge_

bool edm::EventProcessor::fileModeNoMerge_

private

Definition at line 351 of file EventProcessor.h.

Referenced by init(), and runToCompletion().

◆ firstEventInBlock_

bool edm::EventProcessor::firstEventInBlock_ = true

private

Definition at line 363 of file EventProcessor.h.

◆ forceESCacheClearOnNewRun_

bool edm::EventProcessor::forceESCacheClearOnNewRun_

private

Definition at line 357 of file EventProcessor.h.

Referenced by beginRun(), and init().

◆ forceLooperToEnd_

bool edm::EventProcessor::forceLooperToEnd_

private

Definition at line 355 of file EventProcessor.h.

Referenced by endOfLoop().

◆ historyAppender_

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

private

Definition at line 337 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 318 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), closeInputFile(), endJob(), endRun(), init(), inputProcessBlocks(), nextLuminosityBlockID(), nextRunID(), nextTransitionType(), processEventWithLooper(), processLumis(), readAndMergeLumi(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readNextEventForStream(), readProcessBlock(), readRun(), rewindInput(), and ~EventProcessor().

◆ lastSourceTransition_

InputSource::ItemType edm::EventProcessor::lastSourceTransition_

private

Definition at line 319 of file EventProcessor.h.

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

◆ looper_

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

private

Definition at line 340 of file EventProcessor.h.

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

◆ looperBeginJobRun_

bool edm::EventProcessor::looperBeginJobRun_

private

Definition at line 356 of file EventProcessor.h.

Referenced by beginRun(), and startingNewLoop().

◆ lumiQueue_

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

private

Definition at line 330 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(), and readRun().

◆ 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 359 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), continueLumiAsync(), endJob(), endRun(), endUnfinishedLumi(), init(), processLumis(), and readFile().

◆ preg_

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

private

Definition at line 313 of file EventProcessor.h.

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

◆ principalCache_

PrincipalCache edm::EventProcessor::principalCache_

private

Definition at line 348 of file EventProcessor.h.

Referenced by beginLumiAsync(), beginProcessBlock(), beginRun(), deleteRunFromCache(), endProcessBlock(), endRun(), endUnfinishedRun(), init(), inputProcessBlocks(), processEventAsyncImpl(), readAndMergeRun(), readEvent(), readFile(), readLuminosityBlock(), readRun(), writeProcessBlockAsync(), and writeRunAsync().

◆ printDependencies_

bool edm::EventProcessor::printDependencies_ = false

private

Definition at line 369 of file EventProcessor.h.

Referenced by beginJob(), and init().

◆ processBlockHelper_

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

private

Definition at line 315 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(), processConfiguration(), and readRun().

◆ processContext_

ProcessContext edm::EventProcessor::processContext_

private

Definition at line 325 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginRun(), endRun(), globalEndLumiAsync(), init(), processEventWithLooper(), readEvent(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

◆ queueWhichWaitsForIOVsToFinish_

edm::SerialTaskQueue edm::EventProcessor::queueWhichWaitsForIOVsToFinish_

private

Definition at line 322 of file EventProcessor.h.

Referenced by beginLumiAsync(), and globalEndLumiAsync().

◆ 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(), beginRun(), clearCounters(), closeOutputFiles(), Config.Process::dumpPython(), endJob(), endOfLoop(), endProcessBlock(), endRun(), getAllModuleDescriptions(), globalEndLumiAsync(), init(), inputProcessBlocks(), openOutputFiles(), processEventAsyncImpl(), Config.Process::prune(), Config.Process::resolve(), respondToCloseInputFile(), respondToOpenInputFile(), shouldWeCloseOutput(), shouldWeStop(), Config.Process::splitPython(), streamEndLumiAsync(), throwAboutModulesRequiringLuminosityBlockSynchronization(), totalEvents(), totalEventsFailed(), totalEventsPassed(), warnAboutLegacyModules(), writeLumiAsync(), writeProcessBlockAsync(), writeRunAsync(), and ~EventProcessor().

◆ serviceToken_

ServiceToken edm::EventProcessor::serviceToken_

private

Definition at line 317 of file EventProcessor.h.

Referenced by beginJob(), beginLumiAsync(), beginProcessBlock(), beginRun(), endJob(), endProcessBlock(), endRun(), getToken(), globalEndLumiAsync(), handleNextEventForStreamAsync(), init(), inputProcessBlocks(), processEventAsyncImpl(), readNextEventForStream(), runToCompletion(), streamEndLumiAsync(), writeLumiAsync(), writeProcessBlockAsync(), and writeRunAsync().

◆ shouldWeStop_

bool edm::EventProcessor::shouldWeStop_

private

Definition at line 350 of file EventProcessor.h.

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

◆ sourceMutex_

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

private

Definition at line 347 of file EventProcessor.h.

Referenced by readNextEventForStream().

◆ sourceResourcesAcquirer_

SharedResourcesAcquirer edm::EventProcessor::sourceResourcesAcquirer_

private

Definition at line 346 of file EventProcessor.h.

Referenced by beginLumiAsync(), and handleNextEventForStreamAsync().

◆ streamLumiActive_

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

private

Definition at line 332 of file EventProcessor.h.

Referenced by beginLumiAsync(), endUnfinishedLumi(), processLumis(), and streamEndLumiAsync().

◆ streamLumiStatus_

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

private

Definition at line 331 of file EventProcessor.h.

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

◆ streamQueues_

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

private

Definition at line 329 of file EventProcessor.h.

Referenced by beginLumiAsync(), init(), and streamEndLumiAsync().

◆ subProcesses_

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

private

◆ taskGroup_

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

private

Definition at line 310 of file EventProcessor.h.

Referenced by beginProcessBlock(), beginRun(), endProcessBlock(), endRun(), endUnfinishedLumi(), endUnfinishedRun(), inputProcessBlocks(), processLumis(), and taskCleanup().

◆ thinnedAssociationsHelper_

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

private

Definition at line 316 of file EventProcessor.h.

Referenced by init(), and thinnedAssociationsHelper().

Public Types

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ ExcludedData

◆ ExcludedDataMap

◆ ProcessBlockType

Member Enumeration Documentation

◆ StatusCode

Constructor & Destructor Documentation

◆ EventProcessor() [1/4]

◆ EventProcessor() [2/4]

◆ EventProcessor() [3/4]

◆ ~EventProcessor()

◆ EventProcessor() [4/4]

Member Function Documentation

◆ beginJob()

◆ beginLumiAsync()

◆ beginProcessBlock()

◆ beginRun()

◆ branchIDListHelper() [1/2]

◆ branchIDListHelper() [2/2]

◆ checkForAsyncStopRequest()

◆ clearCounters()

◆ closeInputFile()

◆ closeOutputFiles()

◆ continueLumiAsync()

◆ deleteLumiFromCache()

◆ deleteRunFromCache()

◆ doErrorStuff()

◆ endJob()

◆ endOfLoop()

◆ endProcessBlock()

◆ endRun()

◆ endUnfinishedLumi()

◆ endUnfinishedRun()

◆ fileBlockValid()

◆ getAllModuleDescriptions()

◆ getToken()

◆ globalEndLumiAsync()

◆ handleEndLumiExceptions()

◆ handleNextEventForStreamAsync()

◆ init()

◆ inputProcessBlocks()

◆ lastTransitionType()

◆ looper() [1/2]

◆ looper() [2/2]

◆ nextLuminosityBlockID()

◆ nextRunID()

◆ nextTransitionType()

◆ openOutputFiles()

◆ operator=()

◆ preg() [1/2]

◆ preg() [2/2]

◆ prepareForNextLoop()

◆ processConfiguration()

◆ processEventAsync()

◆ processEventAsyncImpl()

◆ processEventWithLooper()

◆ processLumis()

◆ readAndMergeLumi()

◆ readAndMergeRun()

◆ readEvent()

◆ readFile()

◆ readLuminosityBlock()

◆ readNextEventForStream()

◆ readProcessBlock()

◆ readRun()

◆ respondToCloseInputFile()

◆ respondToOpenInputFile()

◆ rewindInput()

◆ run()

◆ runToCompletion()

◆ setDeferredException()

◆ setExceptionMessageFiles()

◆ setExceptionMessageLumis()

◆ setExceptionMessageRuns()