#include <ProductResolvers.h>

Inheritance diagram for edm::InputProductResolver:

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

void	setupUnscheduled (UnscheduledConfigurator const &) final

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 ()

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 (WaitingTask 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	putOrMergeProduct (std::unique_ptr< WrapperBase > edp, MergeableRunProductMetadata const *mergeableRunProductMetadata=nullptr) const

void	putProduct (std::unique_ptr< WrapperBase > edp) 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_ (WaitingTask waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer sra, ModuleCallingContext const *mcc) const override

void	putProduct_ (std::unique_ptr< WrapperBase > edp) 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::DataManagingProductResolver
ProductStatus	defaultStatus () const

ProductData const &	getProductData () const final

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

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

ProductStatus	status () const

Detailed Description

Definition at line 104 of file ProductResolvers.h.

Constructor & Destructor Documentation

◆ InputProductResolver()

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

inlineexplicit

Definition at line 106 of file ProductResolvers.h.

107 : DataManagingProductResolver(bd, ProductStatus::ResolveNotRun), m_prefetchRequested{false}, aux_{nullptr} {}

Member Function Documentation

◆ isFromCurrentProcess()

bool edm::InputProductResolver::isFromCurrentProcess ( ) const

finalprivatevirtual

Implements edm::DataManagingProductResolver.

Definition at line 313 of file ProductResolvers.cc.

313 { return false; }

◆ prefetchAsync_()

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

overrideprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 246 of file ProductResolvers.cc.

                                                                                    {
     //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) {
       auto workToDo = [this, mcc, &principal, token]() {
         //need to make sure Service system is activated on the reading thread
         ServiceRegistry::Operate operate(token);
         // Caught exception is propagated via WaitingTaskList
         CMS_SA_ALLOW try {
           resolveProductImpl<true>([this, &principal, mcc]() {
             if (principal.branchType() != InEvent) {
               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()) {
                 //another thread could have finished this while we were waiting
                 putProduct(reader->getProduct(branchDescription().branchID(), &principal, mcc));
               }
             }
           });
         } 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(workToDo);
       } else {
         //Have to create a new task
         auto t = make_functor_task(tbb::task::allocate_root(), workToDo);
         tbb::task::spawn(*t);
       }
     }
   }

References edm::WaitingTaskList::add(), edm::ProductResolverBase::branchDescription(), CMS_SA_ALLOW, edm::WaitingTaskList::doneWaiting(), edm::InEvent, m_prefetchRequested, m_waitingTasks, edm::make_functor_task(), edm::ProductResolverBase::productResolved(), edm::ProductResolverBase::putProduct(), createBeamHaloJobs::queue, DQM::reader, submitPVValidationJobs::t, and unpackBuffers-CaloStage2::token.

◆ putProduct_()

void edm::InputProductResolver::putProduct_ ( std::unique_ptr< WrapperBase > edp ) const

overrideprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 482 of file ProductResolvers.cc.

                                                                              {
     if (not productResolved()) {
       //Another thread could have set this
       setProduct(std::move(edp));
     }
   }

References eostools::move(), edm::ProductResolverBase::productResolved(), and edm::DataManagingProductResolver::setProduct().

◆ resetProductData_()

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

overrideprivatevirtual

Reimplemented from edm::DataManagingProductResolver.

Definition at line 301 of file ProductResolvers.cc.

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

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

◆ resolveProduct_()

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

overrideprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 162 of file ProductResolvers.cc.

                                                                                                                {
     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 || branchType == InProcess) and aux_) {
         aux_->preModuleDelayedGetSignal_.emit(*(mcc->getStreamContext()), *mcc);
       }
  
       auto sentry(make_sentry(mcc, [this, branchType](ModuleCallingContext const* iContext) {
         if ((branchType == InEvent || branchType == InProcess) 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()) {
           //another thread could have beaten us here
           putProduct(reader->getProduct(branchDescription().branchID(), &principal, mcc));
         }
       }
     });
   }

References aux_, edm::ProductResolverBase::branchDescription(), edm::Principal::branchType(), edm::signalslot::Signal< T >::emit(), edm::ModuleCallingContext::getStreamContext(), edm::InEvent, edm::InLumi, edm::InProcess, edm::InRun, edm::make_sentry(), edm::UnscheduledAuxiliary::postModuleDelayedGetSignal_, edm::UnscheduledAuxiliary::preModuleDelayedGetSignal_, edm::ProductResolverBase::productResolved(), edm::ProductResolverBase::putProduct(), DQM::reader, and edm::Principal::reader().

◆ retrieveAndMerge_()

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

overrideprivatevirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 196 of file ProductResolvers.cc.

                                                                                                                      {
     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
       std::unique_ptr<WrapperBase> 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())) {
           putOrMergeProduct(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
     }
   }

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, edm::ProductResolverBase::putOrMergeProduct(), DQM::reader, edm::Principal::reader(), edm::DataManagingProductResolver::ResolveFailed, edm::DataManagingProductResolver::setFailedStatus(), and edm::DataManagingProductResolver::status().

◆ setMergeableRunProductMetadata_()

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

overrideprivatevirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 242 of file ProductResolvers.cc.

                                                                                                     {
     setMergeableRunProductMetadataInProductData(mrpm);
   }

References edm::DataManagingProductResolver::setMergeableRunProductMetadataInProductData().

◆ setupUnscheduled()

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

finalvirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 309 of file ProductResolvers.cc.

                                                                                        {
     aux_ = iConfigure.auxiliary();
   }

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

◆ unscheduledWasNotRun_()

bool edm::InputProductResolver::unscheduledWasNotRun_ ( ) const

inlinefinalprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 131 of file ProductResolvers.h.

131 { return false; }

Member Data Documentation

◆ aux_

UnscheduledAuxiliary const* edm::InputProductResolver::aux_

private

Definition at line 137 of file ProductResolvers.h.

Referenced by resolveProduct_(), and setupUnscheduled().

◆ m_prefetchRequested

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

mutableprivate

Definition at line 135 of file ProductResolvers.h.

Referenced by prefetchAsync_(), and resetProductData_().

◆ m_waitingTasks

WaitingTaskList edm::InputProductResolver::m_waitingTasks

mutableprivate

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

◆ InputProductResolver()

Member Function Documentation

◆ isFromCurrentProcess()

◆ prefetchAsync_()

◆ putProduct_()

◆ resetProductData_()

◆ resolveProduct_()

◆ retrieveAndMerge_()

◆ setMergeableRunProductMetadata_()

◆ setupUnscheduled()

◆ unscheduledWasNotRun_()

Member Data Documentation

◆ aux_

◆ m_prefetchRequested

◆ m_waitingTasks