#include <ProductResolvers.h>

Inheritance diagram for edm::DelayedReaderInputProductResolver:

Public Member Functions
	DelayedReaderInputProductResolver (std::shared_ptr< BranchDescription const > bd)

void	setupUnscheduled (UnscheduledConfigurator const &) final

Public Member Functions inherited from edm::MergeableInputProductResolver
	MergeableInputProductResolver (std::shared_ptr< BranchDescription const > bd, ProductStatus iDefaultStatus)

Public Member Functions inherited from edm::DataManagingProductResolver
void	connectTo (ProductResolverBase const &, Principal const *) final

	DataManagingProductResolver (std::shared_ptr< BranchDescription const > bd, ProductStatus iDefaultStatus)

void	resetStatus ()

template<bool callResolver, typename FUNC >
ProductResolverBase::Resolution	resolveProductImpl (FUNC resolver) const

Public Member Functions inherited from edm::DataManagingOrAliasProductResolver
	DataManagingOrAliasProductResolver ()

Public Member Functions inherited from edm::ProductResolverBase
BranchDescription const &	branchDescription () const

std::string const &	moduleLabel () const

ProductResolverBase &	operator= (ProductResolverBase const &)=delete

void	prefetchAsync (WaitingTaskHolder waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer sra, ModuleCallingContext const mcc) const

std::string const &	processName () const

ProductID const &	productID () const

std::string const &	productInstanceName () const

ProductProvenance const *	productProvenancePtr () const

bool	productResolved () const

	ProductResolverBase ()

	ProductResolverBase (ProductResolverBase const &)=delete

TypeID	productType () const

bool	productUnavailable () const

bool	productWasDeleted () const

bool	productWasFetchedAndIsValid (bool iSkipCurrentProcess) const

Provenance const *	provenance () const

bool	provenanceAvailable () const

void	resetBranchDescription (std::shared_ptr< BranchDescription const > bd)

void	resetProductData ()

std::string const &	resolvedModuleLabel () const

Resolution	resolveProduct (Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer sra, ModuleCallingContext const mcc) const

void	retrieveAndMerge (Principal const &principal, MergeableRunProductMetadata const *mergeableRunProductMetadata) const

void	setMergeableRunProductMetadata (MergeableRunProductMetadata const *mrpm)

void	setProductID (ProductID const &pid)

void	setProductProvenanceRetriever (ProductProvenanceRetriever const *provRetriever)

bool	singleProduct () const

StableProvenance const *	stableProvenance () const

void	unsafe_deleteProduct () const

bool	unscheduledWasNotRun () const

void	write (std::ostream &os) const

virtual	~ProductResolverBase ()

Private Member Functions
bool	isFromCurrentProcess () const final

void	prefetchAsync_ (WaitingTaskHolder waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer sra, ModuleCallingContext const mcc) const override

void	resetProductData_ (bool deleteEarly) override

Resolution	resolveProduct_ (Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer sra, ModuleCallingContext const mcc) const override

void	retrieveAndMerge_ (Principal const &principal, MergeableRunProductMetadata const *mergeableRunProductMetadata) const override

void	setMergeableRunProductMetadata_ (MergeableRunProductMetadata const *) override

bool	unscheduledWasNotRun_ () const final

Private Attributes
UnscheduledAuxiliary const *	aux_

std::atomic< bool >	m_prefetchRequested

WaitingTaskList	m_waitingTasks

Additional Inherited Members
Public Types inherited from edm::DataManagingProductResolver
enum	ProductStatus { ProductStatus::ProductSet, ProductStatus::NotPut, ProductStatus::ResolveFailed, ProductStatus::ResolveNotRun, ProductStatus::ProductDeleted }

Protected Member Functions inherited from edm::MergeableInputProductResolver
void	mergeProduct (std::shared_ptr< WrapperBase > edp, MergeableRunProductMetadata const *) const

void	setOrMergeProduct (std::shared_ptr< WrapperBase > prod, MergeableRunProductMetadata const *mergeableRunProductMetadata) const

Protected Member Functions inherited from edm::DataManagingProductResolver
void	checkType (WrapperBase const &prod) const

ProductStatus	defaultStatus () const

ProductData const &	getProductData () const final

template<bool callResolver, typename FUNC >
Resolution	resolveProductImpl (FUNC resolver) const

void	setFailedStatus () const

void	setMergeableRunProductMetadataInProductData (MergeableRunProductMetadata const *)

void	setProduct (std::unique_ptr< WrapperBase > edp) const

void	setProduct (std::shared_ptr< WrapperBase > edp) const

ProductStatus	status () const

Detailed Description

Definition at line 119 of file ProductResolvers.h.

Constructor & Destructor Documentation

◆ DelayedReaderInputProductResolver()

edm::DelayedReaderInputProductResolver::DelayedReaderInputProductResolver ( std::shared_ptr< BranchDescription const > bd )

inlineexplicit

Definition at line 121 of file ProductResolvers.h.

         : MergeableInputProductResolver(bd, ProductStatus::ResolveNotRun), m_prefetchRequested{false}, aux_{nullptr} {
       assert(bd->onDemand());
       assert(not bd->produced());
     }

Member Function Documentation

◆ isFromCurrentProcess()

bool edm::DelayedReaderInputProductResolver::isFromCurrentProcess ( ) const

finalprivatevirtual

Implements edm::DataManagingProductResolver.

Definition at line 351 of file ProductResolvers.cc.

351 { return false; }

◆ prefetchAsync_()

void edm::DelayedReaderInputProductResolver::prefetchAsync_	(	WaitingTaskHolder	waitTask,
		Principal const &	principal,
		bool	skipCurrentProcess,
		ServiceToken const &	token,
		SharedResourcesAcquirer *	sra,
		ModuleCallingContext const *	mcc
	)		const

overrideprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 271 of file ProductResolvers.cc.

References edm::WaitingTaskList::add(), edm::ProductResolverBase::branchDescription(), CMS_SA_ALLOW, edm::WaitingTaskList::doneWaiting(), MillePedeFileConverter_cfg::e, cppFunctionSkipper::exception, Exception, edm::WaitingTaskHolder::group(), edm::InEvent, edm::InProcess, edm::ServiceWeakToken::lock(), m_prefetchRequested, m_waitingTasks, edm::make_functor_task(), edm::ProductResolverBase::productResolved(), createBeamHaloJobs::queue, DQM::reader, alignCSCRings::s, edm::DataManagingProductResolver::setProduct(), edm::errors::StdException, submitPVValidationJobs::t, and unpackBuffers-CaloStage2::token.

                                                                                                 {
     //need to try changing m_prefetchRequested before adding to m_waitingTasks
     bool expected = false;
     bool prefetchRequested = m_prefetchRequested.compare_exchange_strong(expected, true);
     m_waitingTasks.add(waitTask);
 
     if (prefetchRequested) {
       ServiceWeakToken weakToken = token;
       auto workToDo = [this, mcc, &principal, weakToken]() {
         //need to make sure Service system is activated on the reading thread
         ServiceRegistry::Operate operate(weakToken.lock());
         // Caught exception is propagated via WaitingTaskList
         CMS_SA_ALLOW try {
           resolveProductImpl<true>([this, &principal, mcc]() {
             if (principal.branchType() != InEvent && principal.branchType() != InProcess) {
               return;
             }
             if (auto reader = principal.reader()) {
               std::unique_lock<std::recursive_mutex> guard;
               if (auto sr = reader->sharedResources().second) {
                 guard = std::unique_lock<std::recursive_mutex>(*sr);
               }
               if (not productResolved()) {
                 try {
                   //another thread could have finished this while we were waiting
                   setProduct(reader->getProduct(branchDescription().branchID(), &principal, mcc));
                 } catch (cms::Exception& e) {
                   extendException(e, branchDescription(), mcc);
                   throw;
                 } catch (std::exception const& e) {
                   auto newExcept = edm::Exception(errors::StdException) << e.what();
                   extendException(newExcept, branchDescription(), mcc);
                   throw newExcept;
                 }
               }
             }
           });
         } catch (...) {
           this->m_waitingTasks.doneWaiting(std::current_exception());
           return;
         }
         this->m_waitingTasks.doneWaiting(nullptr);
       };
 
       SerialTaskQueueChain* queue = nullptr;
       if (auto reader = principal.reader()) {
         if (auto shared_res = reader->sharedResources().first) {
           queue = &(shared_res->serialQueueChain());
         }
       }
       if (queue) {
         queue->push(*waitTask.group(), workToDo);
       } else {
         //Have to create a new task
         auto t = make_functor_task(workToDo);
         waitTask.group()->run([t]() {
           TaskSentry s{t};
           t->execute();
         });
       }
     }
   }

◆ resetProductData_()

void edm::DelayedReaderInputProductResolver::resetProductData_ ( bool deleteEarly )

overrideprivatevirtual

Reimplemented from edm::DataManagingProductResolver.

Definition at line 339 of file ProductResolvers.cc.

References m_prefetchRequested, m_waitingTasks, edm::WaitingTaskList::reset(), and edm::DataManagingProductResolver::resetProductData_().

                                                                             {
     if (not deleteEarly) {
       m_prefetchRequested = false;
       m_waitingTasks.reset();
     }
     DataManagingProductResolver::resetProductData_(deleteEarly);
   }

◆ resolveProduct_()

ProductResolverBase::Resolution edm::DelayedReaderInputProductResolver::resolveProduct_	(	Principal const &	principal,
		bool	skipCurrentProcess,
		SharedResourcesAcquirer *	sra,
		ModuleCallingContext const *	mcc
	)		const

overrideprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 170 of file ProductResolvers.cc.

References aux_, edm::ProductResolverBase::branchDescription(), MillePedeFileConverter_cfg::e, edm::signalslot::Signal< T >::emit(), cppFunctionSkipper::exception, Exception, edm::ModuleCallingContext::getStreamContext(), edm::InEvent, edm::InLumi, edm::InRun, edm::make_sentry(), edm::UnscheduledAuxiliary::postModuleDelayedGetSignal_, edm::UnscheduledAuxiliary::preModuleDelayedGetSignal_, edm::ProductResolverBase::productResolved(), DQM::reader, edm::DataManagingProductResolver::setProduct(), and edm::errors::StdException.

                                                                                                          {
     return resolveProductImpl<true>([this, &principal, mcc]() {
       auto branchType = principal.branchType();
       if (branchType == InLumi || branchType == InRun) {
         //delayed get has not been allowed with Run or Lumis
         // The file may already be closed so the reader is invalid
         return;
       }
       if (mcc and branchType == InEvent and aux_) {
         aux_->preModuleDelayedGetSignal_.emit(*(mcc->getStreamContext()), *mcc);
       }
 
       auto sentry(make_sentry(mcc, [this, branchType](ModuleCallingContext const* iContext) {
         if (branchType == InEvent and aux_) {
           aux_->postModuleDelayedGetSignal_.emit(*(iContext->getStreamContext()), *iContext);
         }
       }));
 
       if (auto reader = principal.reader()) {
         std::unique_lock<std::recursive_mutex> guard;
         if (auto sr = reader->sharedResources().second) {
           guard = std::unique_lock<std::recursive_mutex>(*sr);
         }
         if (not productResolved()) {
           try {
             //another thread could have beaten us here
             setProduct(reader->getProduct(branchDescription().branchID(), &principal, mcc));
           } catch (cms::Exception& e) {
             extendException(e, branchDescription(), mcc);
             throw;
           } catch (std::exception const& e) {
             auto newExcept = edm::Exception(errors::StdException) << e.what();
             extendException(newExcept, branchDescription(), mcc);
             throw newExcept;
           }
         }
       }
     });
   }

◆ retrieveAndMerge_()

void edm::DelayedReaderInputProductResolver::retrieveAndMerge_	(	Principal const &	principal,
		MergeableRunProductMetadata const *	mergeableRunProductMetadata
	)		const

overrideprivatevirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 211 of file ProductResolvers.cc.

References edm::ProductResolverBase::branchDescription(), edm::BranchDescription::branchName(), edm::BranchDescription::branchType(), edm::DataManagingProductResolver::defaultStatus(), Exception, edm::InRun, edm::errors::LogicError, edm::errors::MismatchedInputFiles, eostools::move(), edm::DataManagingProductResolver::ProductSet, DQM::reader, edm::Principal::reader(), edm::DataManagingProductResolver::ResolveFailed, edm::DataManagingProductResolver::setFailedStatus(), edm::MergeableInputProductResolver::setOrMergeProduct(), and edm::DataManagingProductResolver::status().

                                                                                                         {
     if (auto reader = principal.reader()) {
       std::unique_lock<std::recursive_mutex> guard;
       if (auto sr = reader->sharedResources().second) {
         guard = std::unique_lock<std::recursive_mutex>(*sr);
       }
 
       //Can't use resolveProductImpl since it first checks to see
       // if the product was already retrieved and then returns if it is
       auto edp(reader->getProduct(branchDescription().branchID(), &principal));
 
       if (edp.get() != nullptr) {
         if (edp->isMergeable() && branchDescription().branchType() == InRun && !edp->hasSwap()) {
           throw Exception(errors::LogicError)
               << "Missing definition of member function swap for branch name " << branchDescription().branchName()
               << "\n"
               << "Mergeable data types written to a Run must have the swap member function defined"
               << "\n";
         }
         if (status() == defaultStatus() || status() == ProductStatus::ProductSet ||
             (status() == ProductStatus::ResolveFailed && !branchDescription().isMergeable())) {
           setOrMergeProduct(std::move(edp), mergeableRunProductMetadata);
         } else {  // status() == ResolveFailed && branchDescription().isMergeable()
           throw Exception(errors::MismatchedInputFiles)
               << "Merge of Run or Lumi product failed for branch " << branchDescription().branchName() << "\n"
               << "The product branch was dropped in the first run or lumi fragment and present in a later one"
               << "\n"
               << "The solution is to drop the branch on input. Or better do not create inconsistent files\n"
               << "that need to be merged in the first place.\n";
         }
       } else if (status() == defaultStatus()) {
         setFailedStatus();
       } else if (status() != ProductStatus::ResolveFailed && branchDescription().isMergeable()) {
         throw Exception(errors::MismatchedInputFiles)
             << "Merge of Run or Lumi product failed for branch " << branchDescription().branchName() << "\n"
             << "The product branch was present in first run or lumi fragment and dropped in a later one"
             << "\n"
             << "The solution is to drop the branch on input. Or better do not create inconsistent files\n"
             << "that need to be merged in the first place.\n";
       }
       // Do nothing in other case. status is ResolveFailed already or
       // it is not mergeable and the status is ProductSet
     }
   }

◆ setMergeableRunProductMetadata_()

void edm::DelayedReaderInputProductResolver::setMergeableRunProductMetadata_ ( MergeableRunProductMetadata const * mrpm )

overrideprivatevirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 267 of file ProductResolvers.cc.

References edm::DataManagingProductResolver::setMergeableRunProductMetadataInProductData().

                                                                                                                  {
     setMergeableRunProductMetadataInProductData(mrpm);
   }

◆ setupUnscheduled()

void edm::DelayedReaderInputProductResolver::setupUnscheduled ( UnscheduledConfigurator const & iConfigure )

finalvirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 347 of file ProductResolvers.cc.

References aux_, and edm::UnscheduledConfigurator::auxiliary().

                                                                                                     {
     aux_ = iConfigure.auxiliary();
   }

◆ unscheduledWasNotRun_()

bool edm::DelayedReaderInputProductResolver::unscheduledWasNotRun_ ( ) const

inlinefinalprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 148 of file ProductResolvers.h.

148 { return false; }

Member Data Documentation

◆ aux_

UnscheduledAuxiliary const* edm::DelayedReaderInputProductResolver::aux_

private

Definition at line 154 of file ProductResolvers.h.

Referenced by resolveProduct_(), and setupUnscheduled().

◆ m_prefetchRequested

std::atomic<bool> edm::DelayedReaderInputProductResolver::m_prefetchRequested

mutableprivate

Definition at line 152 of file ProductResolvers.h.

Referenced by prefetchAsync_(), and resetProductData_().

◆ m_waitingTasks

WaitingTaskList edm::DelayedReaderInputProductResolver::m_waitingTasks

mutableprivate

Definition at line 153 of file ProductResolvers.h.

Referenced by prefetchAsync_(), and resetProductData_().

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ DelayedReaderInputProductResolver()

Member Function Documentation

◆ isFromCurrentProcess()

◆ prefetchAsync_()

◆ resetProductData_()

◆ resolveProduct_()

◆ retrieveAndMerge_()

◆ setMergeableRunProductMetadata_()

◆ setupUnscheduled()

◆ unscheduledWasNotRun_()

Member Data Documentation

◆ aux_

◆ m_prefetchRequested

◆ m_waitingTasks