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

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

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	setProcessHistory (ProcessHistory const &ph)

void	setProvenance (ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid)

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

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 98 of file ProductResolvers.h.

Constructor & Destructor Documentation

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

inlineexplicit

Definition at line 100 of file ProductResolvers.h.

References edm::DataManagingProductResolver::isFromCurrentProcess(), edm::ProductResolverBase::prefetchAsync_(), cmsRelvalreport::principal(), edm::ProductResolverBase::putProduct_(), edm::ProductResolverBase::resolveProduct_(), edm::ProductResolverBase::retrieveAndMerge_(), edm::ProductResolverBase::setMergeableRunProductMetadata_(), and edm::ProductResolverBase::setupUnscheduled().

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

Member Function Documentation

bool edm::InputProductResolver::isFromCurrentProcess ( ) const

finalprivatevirtual

Implements edm::DataManagingProductResolver.

Definition at line 311 of file ProductResolvers.cc.

                                                    {
     return false;
   }

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 245 of file ProductResolvers.cc.

References edm::ProductResolverBase::branchDescription(), edm::InEvent, edm::make_functor_task(), cmsRelvalreport::principal(), edm::ProductResolverBase::productResolved(), edm::SerialTaskQueueChain::push(), edm::ProductResolverBase::putProduct(), createBeamHaloJobs::queue, matplotRender::reader, and lumiQTWidget::t.

                                                                                    {
     m_waitingTasks.add(waitTask);
     
     bool expected = false;
     if( m_prefetchRequested.compare_exchange_strong(expected, true) ) {
       
       auto workToDo = [this, mcc, &principal, token] () {
         //need to make sure Service system is activated on the reading thread
         ServiceRegistry::Operate guard(token);
         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);
       }
     }
   }

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

overrideprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 501 of file ProductResolvers.cc.

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

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

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

overrideprivatevirtual

Reimplemented from edm::DataManagingProductResolver.

Definition at line 298 of file ProductResolvers.cc.

References edm::DataManagingProductResolver::resetProductData_().

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

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

overrideprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 160 of file ProductResolvers.cc.

References edm::ProductResolverBase::branchDescription(), revisionDML::branchType(), edm::ModuleCallingContext::getStreamContext(), edm::InEvent, edm::make_sentry(), cmsRelvalreport::principal(), edm::ProductResolverBase::productResolved(), edm::ProductResolverBase::putProduct(), and matplotRender::reader.

                                                                                {
     return resolveProductImpl<true>([this,&principal,mcc]() {
       auto branchType = principal.branchType();
       if(branchType != InEvent) {
         //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()) {
           //another thread could have beaten us here
           putProduct( reader->getProduct(branchDescription().branchID(), &principal, mcc));
         }
       }
     });
                               
   }

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

overrideprivatevirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 196 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, edm::ProductResolverBase::putOrMergeProduct(), edm::Principal::reader(), matplotRender::reader, edm::DataManagingProductResolver::ResolveFailed, edm::DataManagingProductResolver::setFailedStatus(), 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
       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
     }
   }

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

overrideprivatevirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 241 of file ProductResolvers.cc.

References edm::DataManagingProductResolver::setMergeableRunProductMetadataInProductData().

                                                                                                {
     setMergeableRunProductMetadataInProductData(mrpm);
   }

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

finalvirtual

Reimplemented from edm::ProductResolverBase.

Definition at line 305 of file ProductResolvers.cc.

References edm::UnscheduledConfigurator::auxiliary().

                                                                                   {
     aux_ = iConfigure.auxiliary();
   }

bool edm::InputProductResolver::unscheduledWasNotRun_ ( ) const

inlinefinalprivatevirtual

Implements edm::ProductResolverBase.

Definition at line 127 of file ProductResolvers.h.

References edm::DataManagingProductResolver::resetProductData_().

127 {return false;}

Member Data Documentation

UnscheduledAuxiliary const* edm::InputProductResolver::aux_

private

Definition at line 133 of file ProductResolvers.h.

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

mutableprivate

Definition at line 131 of file ProductResolvers.h.

WaitingTaskList edm::InputProductResolver::m_waitingTasks

mutableprivate

Definition at line 132 of file ProductResolvers.h.

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation