Framework/src/ProductResolvers.cc

0001 /*----------------------------------------------------------------------
0002 ----------------------------------------------------------------------*/
0003 #include "ProductResolvers.h"
0004 #include "FWCore/Framework/interface/maker/Worker.h"
0005 #include "FWCore/Framework/interface/UnscheduledAuxiliary.h"
0006 #include "UnscheduledConfigurator.h"
0007 #include "FWCore/Framework/interface/EventPrincipal.h"
0008 #include "FWCore/Framework/interface/MergeableRunProductMetadata.h"
0009 #include "FWCore/Framework/src/ProductDeletedException.h"
0010 #include "FWCore/Framework/interface/SharedResourcesAcquirer.h"
0011 #include "FWCore/Framework/interface/DelayedReader.h"
0012 #include "FWCore/Framework/interface/TransitionInfoTypes.h"
0013 #include "FWCore/Framework/interface/ProductProvenanceRetriever.h"
0014 #include "DataFormats/Provenance/interface/BranchKey.h"
0015 #include "DataFormats/Provenance/interface/ParentageRegistry.h"
0016 #include "FWCore/MessageLogger/interface/MessageLogger.h"
0017 #include "FWCore/Concurrency/interface/SerialTaskQueue.h"
0018 #include "FWCore/Concurrency/interface/FunctorTask.h"
0019 #include "FWCore/Utilities/interface/TypeID.h"
0020 #include "FWCore/Utilities/interface/make_sentry.h"
0021 #include "FWCore/Utilities/interface/Transition.h"
0022 #include "FWCore/Utilities/interface/thread_safety_macros.h"
0023
0024 #include <cassert>
0025 #include <utility>
0026
0027 static constexpr unsigned int kUnsetOffset = 0;
0028 static constexpr unsigned int kAmbiguousOffset = 1;
0029 static constexpr unsigned int kMissingOffset = 2;
0030
0031 namespace edm {
0032
0033   void DataManagingProductResolver::throwProductDeletedException() const {
0034     ProductDeletedException exception;
0035     exception << "DataManagingProductResolver::resolveProduct_: The product matching all criteria was already deleted\n"
0036               << "Looking for type: " << branchDescription().unwrappedTypeID() << "\n"
0037               << "Looking for module label: " << moduleLabel() << "\n"
0038               << "Looking for productInstanceName: " << productInstanceName() << "\n"
0039               << (processName().empty() ? "" : "Looking for process: ") << processName() << "\n"
0040               << "This means there is a configuration error.\n"
0041               << "The module which is asking for this data must be configured to state that it will read this data.";
0042     throw exception;
0043   }
0044
0045   //This is a templated function in order to avoid calling another virtual function
0046   template <bool callResolver, typename FUNC>
0047   ProductResolverBase::Resolution DataManagingProductResolver::resolveProductImpl(FUNC resolver) const {
0048     if (productWasDeleted()) {
0049       throwProductDeletedException();
0050     }
0051     auto presentStatus = status();
0052
0053     if (callResolver && presentStatus == ProductStatus::ResolveNotRun) {
0054       //if resolver fails because of exception or not setting product
0055       // make sure the status goes to failed
0056       auto failedStatusSetter = [this](ProductStatus* iPresentStatus) {
0057         if (this->status() == ProductStatus::ResolveNotRun) {
0058           this->setFailedStatus();
0059         }
0060         *iPresentStatus = this->status();
0061       };
0062       std::unique_ptr<ProductStatus, decltype(failedStatusSetter)> failedStatusGuard(&presentStatus,
0063                                                                                      failedStatusSetter);
0064
0065       //If successful, this will call setProduct
0066       resolver();
0067     }
0068
0069     if (presentStatus == ProductStatus::ProductSet) {
0070       auto pd = &getProductData();
0071       if (pd->wrapper()->isPresent()) {
0072         return Resolution(pd);
0073       }
0074     }
0075
0076     return Resolution(nullptr);
0077   }
0078
0079   void MergeableInputProductResolver::mergeProduct(
0080       std::shared_ptr<WrapperBase> iFrom, MergeableRunProductMetadata const* mergeableRunProductMetadata) const {
0081     // if its not mergeable and the previous read failed, go ahead and use this one
0082     if (status() == ProductStatus::ResolveFailed) {
0083       setProduct(std::move(iFrom));
0084       return;
0085     }
0086
0087     assert(status() == ProductStatus::ProductSet);
0088     if (not iFrom) {
0089       return;
0090     }
0091
0092     checkType(*iFrom);
0093
0094     auto original = getProductData().unsafe_wrapper();
0095     if (original->isMergeable()) {
0096       if (original->isPresent() != iFrom->isPresent()) {
0097         throw Exception(errors::MismatchedInputFiles)
0098             << "Merge of Run or Lumi product failed for branch " << branchDescription().branchName() << "\n"
0099             << "Was trying to merge objects where one product had been put in the input file and the other had not "
0100                "been."
0101             << "\n"
0102             << "The solution is to drop the branch on input. Or better do not create inconsistent files\n"
0103             << "that need to be merged in the first place.\n";
0104       }
0105       if (original->isPresent()) {
0106         BranchDescription const& desc = branchDescription();
0107         if (mergeableRunProductMetadata == nullptr || desc.branchType() != InRun) {
0108           original->mergeProduct(iFrom.get());
0109         } else {
0110           MergeableRunProductMetadata::MergeDecision decision =
0111               mergeableRunProductMetadata->getMergeDecision(desc.processName());
0112           if (decision == MergeableRunProductMetadata::MERGE) {
0113             original->mergeProduct(iFrom.get());
0114           } else if (decision == MergeableRunProductMetadata::REPLACE) {
0115             // Note this swaps the content of the product where the
0116             // both products branches are present and the product is
0117             // also present (was put) in the branch. A module might
0118             // have already gotten a pointer to the product so we
0119             // keep those pointers valid. This does not call swap
0120             // on the Wrapper.
0121             original->swapProduct(iFrom.get());
0122           }
0123           // If the decision is IGNORE, do nothing
0124         }
0125       }
0126       // If both have isPresent false, do nothing
0127
0128     } else if (original->hasIsProductEqual()) {
0129       if (original->isPresent() && iFrom->isPresent()) {
0130         if (!original->isProductEqual(iFrom.get())) {
0131           auto const& bd = branchDescription();
0132           edm::LogError("RunLumiMerging")
0133               << "ProductResolver::mergeTheProduct\n"
0134               << "Two run/lumi products for the same run/lumi which should be equal are not\n"
0135               << "Using the first, ignoring the second\n"
0136               << "className = " << bd.className() << "\n"
0137               << "moduleLabel = " << bd.moduleLabel() << "\n"
0138               << "instance = " << bd.productInstanceName() << "\n"
0139               << "process = " << bd.processName() << "\n";
0140         }
0141       } else if (!original->isPresent() && iFrom->isPresent()) {
0142         setProduct(std::move(iFrom));
0143       }
0144       // if not iFrom->isPresent(), do nothing
0145     } else {
0146       auto const& bd = branchDescription();
0147       edm::LogWarning("RunLumiMerging") << "ProductResolver::mergeTheProduct\n"
0148                                         << "Run/lumi product has neither a mergeProduct nor isProductEqual function\n"
0149                                         << "Using the first, ignoring the second in merge\n"
0150                                         << "className = " << bd.className() << "\n"
0151                                         << "moduleLabel = " << bd.moduleLabel() << "\n"
0152                                         << "instance = " << bd.productInstanceName() << "\n"
0153                                         << "process = " << bd.processName() << "\n";
0154       if (!original->isPresent() && iFrom->isPresent()) {
0155         setProduct(std::move(iFrom));
0156       }
0157       // In other cases, do nothing
0158     }
0159   }
0160
0161   namespace {
0162     cms::Exception& extendException(cms::Exception& e, BranchDescription const& bd, ModuleCallingContext const* mcc) {
0163       e.addContext(std::string("While reading from source ") + bd.className() + " " + bd.moduleLabel() + " '" +
0164                    bd.productInstanceName() + "' " + bd.processName());
0165       if (mcc) {
0166         edm::exceptionContext(e, *mcc);
0167       }
0168       return e;
0169     }
0170   }  // namespace
0171   ProductResolverBase::Resolution DelayedReaderInputProductResolver::resolveProduct_(
0172       Principal const& principal, bool, SharedResourcesAcquirer*, ModuleCallingContext const* mcc) const {
0173     return resolveProductImpl<true>([this, &principal, mcc]() {
0174       auto branchType = principal.branchType();
0175       if (branchType == InLumi || branchType == InRun) {
0176         //delayed get has not been allowed with Run or Lumis
0177         // The file may already be closed so the reader is invalid
0178         return;
0179       }
0180       if (mcc and branchType == InEvent and aux_) {
0181         aux_->preModuleDelayedGetSignal_.emit(*(mcc->getStreamContext()), *mcc);
0182       }
0183
0184       auto sentry(make_sentry(mcc, [this, branchType](ModuleCallingContext const* iContext) {
0185         if (branchType == InEvent and aux_) {
0186           aux_->postModuleDelayedGetSignal_.emit(*(iContext->getStreamContext()), *iContext);
0187         }
0188       }));
0189
0190       if (auto reader = principal.reader()) {
0191         std::unique_lock<std::recursive_mutex> guard;
0192         if (auto sr = reader->sharedResources().second) {
0193           guard = std::unique_lock<std::recursive_mutex>(*sr);
0194         }
0195         if (not productResolved()) {
0196           try {
0197             //another thread could have beaten us here
0198             setProduct(reader->getProduct(branchDescription().branchID(), &principal, mcc));
0199           } catch (cms::Exception& e) {
0200             throw extendException(e, branchDescription(), mcc);
0201           } catch (std::exception const& e) {
0202             auto newExcept = edm::Exception(errors::StdException) << e.what();
0203             throw extendException(newExcept, branchDescription(), mcc);
0204           }
0205         }
0206       }
0207     });
0208   }
0209
0210   void DelayedReaderInputProductResolver::retrieveAndMerge_(
0211       Principal const& principal, MergeableRunProductMetadata const* mergeableRunProductMetadata) const {
0212     if (auto reader = principal.reader()) {
0213       std::unique_lock<std::recursive_mutex> guard;
0214       if (auto sr = reader->sharedResources().second) {
0215         guard = std::unique_lock<std::recursive_mutex>(*sr);
0216       }
0217
0218       //Can't use resolveProductImpl since it first checks to see
0219       // if the product was already retrieved and then returns if it is
0220       auto edp(reader->getProduct(branchDescription().branchID(), &principal));
0221
0222       if (edp.get() != nullptr) {
0223         if (edp->isMergeable() && branchDescription().branchType() == InRun && !edp->hasSwap()) {
0224           throw Exception(errors::LogicError)
0225               << "Missing definition of member function swap for branch name " << branchDescription().branchName()
0226               << "\n"
0227               << "Mergeable data types written to a Run must have the swap member function defined"
0228               << "\n";
0229         }
0230         if (status() == defaultStatus() || status() == ProductStatus::ProductSet ||
0231             (status() == ProductStatus::ResolveFailed && !branchDescription().isMergeable())) {
0232           setOrMergeProduct(std::move(edp), mergeableRunProductMetadata);
0233         } else {  // status() == ResolveFailed && branchDescription().isMergeable()
0234           throw Exception(errors::MismatchedInputFiles)
0235               << "Merge of Run or Lumi product failed for branch " << branchDescription().branchName() << "\n"
0236               << "The product branch was dropped in the first run or lumi fragment and present in a later one"
0237               << "\n"
0238               << "The solution is to drop the branch on input. Or better do not create inconsistent files\n"
0239               << "that need to be merged in the first place.\n";
0240         }
0241       } else if (status() == defaultStatus()) {
0242         setFailedStatus();
0243       } else if (status() != ProductStatus::ResolveFailed && branchDescription().isMergeable()) {
0244         throw Exception(errors::MismatchedInputFiles)
0245             << "Merge of Run or Lumi product failed for branch " << branchDescription().branchName() << "\n"
0246             << "The product branch was present in first run or lumi fragment and dropped in a later one"
0247             << "\n"
0248             << "The solution is to drop the branch on input. Or better do not create inconsistent files\n"
0249             << "that need to be merged in the first place.\n";
0250       }
0251       // Do nothing in other case. status is ResolveFailed already or
0252       // it is not mergeable and the status is ProductSet
0253     }
0254   }
0255
0256   void MergeableInputProductResolver::setOrMergeProduct(
0257       std::shared_ptr<WrapperBase> prod, MergeableRunProductMetadata const* mergeableRunProductMetadata) const {
0258     if (status() == defaultStatus()) {
0259       //resolveProduct has not been called or it failed
0260       setProduct(std::move(prod));
0261     } else {
0262       mergeProduct(std::move(prod), mergeableRunProductMetadata);
0263     }
0264   }
0265
0266   void DelayedReaderInputProductResolver::setMergeableRunProductMetadata_(MergeableRunProductMetadata const* mrpm) {
0267     setMergeableRunProductMetadataInProductData(mrpm);
0268   }
0269
0270   void DelayedReaderInputProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
0271                                                          Principal const& principal,
0272                                                          bool skipCurrentProcess,
0273                                                          ServiceToken const& token,
0274                                                          SharedResourcesAcquirer* sra,
0275                                                          ModuleCallingContext const* mcc) const {
0276     //need to try changing m_prefetchRequested before adding to m_waitingTasks
0277     bool expected = false;
0278     bool prefetchRequested = m_prefetchRequested.compare_exchange_strong(expected, true);
0279     m_waitingTasks.add(waitTask);
0280
0281     if (prefetchRequested) {
0282       ServiceWeakToken weakToken = token;
0283       auto workToDo = [this, mcc, &principal, weakToken]() {
0284         //need to make sure Service system is activated on the reading thread
0285         ServiceRegistry::Operate operate(weakToken.lock());
0286         // Caught exception is propagated via WaitingTaskList
0287         CMS_SA_ALLOW try {
0288           resolveProductImpl<true>([this, &principal, mcc]() {
0289             if (principal.branchType() != InEvent && principal.branchType() != InProcess) {
0290               return;
0291             }
0292             if (auto reader = principal.reader()) {
0293               std::unique_lock<std::recursive_mutex> guard;
0294               if (auto sr = reader->sharedResources().second) {
0295                 guard = std::unique_lock<std::recursive_mutex>(*sr);
0296               }
0297               if (not productResolved()) {
0298                 try {
0299                   //another thread could have finished this while we were waiting
0300                   setProduct(reader->getProduct(branchDescription().branchID(), &principal, mcc));
0301                 } catch (cms::Exception& e) {
0302                   throw extendException(e, branchDescription(), mcc);
0303                 } catch (std::exception const& e) {
0304                   auto newExcept = edm::Exception(errors::StdException) << e.what();
0305                   throw extendException(newExcept, branchDescription(), mcc);
0306                 }
0307               }
0308             }
0309           });
0310         } catch (...) {
0311           this->m_waitingTasks.doneWaiting(std::current_exception());
0312           return;
0313         }
0314         this->m_waitingTasks.doneWaiting(nullptr);
0315       };
0316
0317       SerialTaskQueueChain* queue = nullptr;
0318       if (auto reader = principal.reader()) {
0319         if (auto shared_res = reader->sharedResources().first) {
0320           queue = &(shared_res->serialQueueChain());
0321         }
0322       }
0323       if (queue) {
0324         queue->push(*waitTask.group(), workToDo);
0325       } else {
0326         //Have to create a new task
0327         auto t = make_functor_task(workToDo);
0328         waitTask.group()->run([t]() {
0329           TaskSentry s{t};
0330           t->execute();
0331         });
0332       }
0333     }
0334   }
0335
0336   void DelayedReaderInputProductResolver::resetProductData_(bool deleteEarly) {
0337     if (not deleteEarly) {
0338       m_prefetchRequested = false;
0339       m_waitingTasks.reset();
0340     }
0341     DataManagingProductResolver::resetProductData_(deleteEarly);
0342   }
0343
0344   void DelayedReaderInputProductResolver::setupUnscheduled(UnscheduledConfigurator const& iConfigure) {
0345     aux_ = iConfigure.auxiliary();
0346   }
0347
0348   bool DelayedReaderInputProductResolver::isFromCurrentProcess() const { return false; }
0349
0350   void PutOnReadInputProductResolver::putProduct(std::unique_ptr<WrapperBase> edp) const {
0351     if (status() != defaultStatus()) {
0352       throw Exception(errors::InsertFailure)
0353           << "Attempt to insert more than one product on branch " << branchDescription().branchName() << "\n";
0354     }
0355
0356     setProduct(std::move(edp));  // ProductResolver takes ownership
0357   }
0358
0359   bool PutOnReadInputProductResolver::isFromCurrentProcess() const { return false; }
0360
0361   ProductResolverBase::Resolution PutOnReadInputProductResolver::resolveProduct_(Principal const&,
0362                                                                                  bool skipCurrentProcess,
0363                                                                                  SharedResourcesAcquirer*,
0364                                                                                  ModuleCallingContext const*) const {
0365     return resolveProductImpl<false>([]() { return; });
0366   }
0367
0368   void PutOnReadInputProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
0369                                                      Principal const& principal,
0370                                                      bool skipCurrentProcess,
0371                                                      ServiceToken const& token,
0372                                                      SharedResourcesAcquirer* sra,
0373                                                      ModuleCallingContext const* mcc) const {}
0374
0375   void PutOnReadInputProductResolver::putOrMergeProduct(std::unique_ptr<WrapperBase> edp) const {
0376     setOrMergeProduct(std::move(edp), nullptr);
0377   }
0378
0379   ProductResolverBase::Resolution PuttableProductResolver::resolveProduct_(Principal const&,
0380                                                                            bool skipCurrentProcess,
0381                                                                            SharedResourcesAcquirer*,
0382                                                                            ModuleCallingContext const*) const {
0383     if (!skipCurrentProcess) {
0384       //'false' means never call the lambda function
0385       return resolveProductImpl<false>([]() { return; });
0386     }
0387     return Resolution(nullptr);
0388   }
0389
0390   void PuttableProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
0391                                                Principal const& principal,
0392                                                bool skipCurrentProcess,
0393                                                ServiceToken const& token,
0394                                                SharedResourcesAcquirer* sra,
0395                                                ModuleCallingContext const* mcc) const {
0396     if (not skipCurrentProcess) {
0397       if (branchDescription().branchType() == InProcess &&
0398           mcc->parent().globalContext()->transition() == GlobalContext::Transition::kAccessInputProcessBlock) {
0399         // This is an accessInputProcessBlock transition
0400         // We cannot access produced products in those transitions
0401         // except for in SubProcesses where they should have already run.
0402         return;
0403       }
0404       if (branchDescription().availableOnlyAtEndTransition() and mcc) {
0405         if (not mcc->parent().isAtEndTransition()) {
0406           return;
0407         }
0408       }
0409
0410       //Need to try modifying prefetchRequested_ before adding to m_waitingTasks
0411       bool expected = false;
0412       bool prefetchRequested = prefetchRequested_.compare_exchange_strong(expected, true);
0413       m_waitingTasks.add(waitTask);
0414
0415       if (worker_ and prefetchRequested) {
0416         //using a waiting task to do a callback guarantees that
0417         // the m_waitingTasks list will be released from waiting even
0418         // if the module does not put this data product or the
0419         // module has an exception while running
0420
0421         auto waiting = make_waiting_task([this](std::exception_ptr const* iException) {
0422           if (nullptr != iException) {
0423             m_waitingTasks.doneWaiting(*iException);
0424           } else {
0425             m_waitingTasks.doneWaiting(std::exception_ptr());
0426           }
0427         });
0428         worker_->callWhenDoneAsync(WaitingTaskHolder(*waitTask.group(), waiting));
0429       }
0430     }
0431   }
0432
0433   void PuttableProductResolver::putProduct(std::unique_ptr<WrapperBase> edp) const {
0434     ProducedProductResolver::putProduct(std::move(edp));
0435     bool expected = false;
0436     if (prefetchRequested_.compare_exchange_strong(expected, true)) {
0437       m_waitingTasks.doneWaiting(std::exception_ptr());
0438     }
0439   }
0440
0441   void PuttableProductResolver::resetProductData_(bool deleteEarly) {
0442     if (not deleteEarly) {
0443       prefetchRequested_ = false;
0444       m_waitingTasks.reset();
0445     }
0446     DataManagingProductResolver::resetProductData_(deleteEarly);
0447   }
0448
0449   void PuttableProductResolver::setupUnscheduled(UnscheduledConfigurator const& iConfigure) {
0450     worker_ = iConfigure.findWorker(branchDescription().moduleLabel());
0451   }
0452
0453   void UnscheduledProductResolver::setupUnscheduled(UnscheduledConfigurator const& iConfigure) {
0454     aux_ = iConfigure.auxiliary();
0455     worker_ = iConfigure.findWorker(branchDescription().moduleLabel());
0456   }
0457
0458   ProductResolverBase::Resolution UnscheduledProductResolver::resolveProduct_(Principal const&,
0459                                                                               bool skipCurrentProcess,
0460                                                                               SharedResourcesAcquirer*,
0461                                                                               ModuleCallingContext const*) const {
0462     if (!skipCurrentProcess and worker_) {
0463       return resolveProductImpl<false>([] {});
0464     }
0465     return Resolution(nullptr);
0466   }
0467
0468   void UnscheduledProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
0469                                                   Principal const& principal,
0470                                                   bool skipCurrentProcess,
0471                                                   ServiceToken const& token,
0472                                                   SharedResourcesAcquirer* sra,
0473                                                   ModuleCallingContext const* mcc) const {
0474     if (skipCurrentProcess) {
0475       return;
0476     }
0477     if (worker_ == nullptr) {
0478       throw cms::Exception("LogicError") << "UnscheduledProductResolver::prefetchAsync_()  called with null worker_. "
0479                                             "This should not happen, please contact framework developers.";
0480     }
0481     //need to try changing prefetchRequested_ before adding to waitingTasks_
0482     bool expected = false;
0483     bool prefetchRequested = prefetchRequested_.compare_exchange_strong(expected, true);
0484     waitingTasks_.add(waitTask);
0485     if (prefetchRequested) {
0486       //Have to create a new task which will make sure the state for UnscheduledProductResolver
0487       // is properly set after the module has run
0488       auto t = make_waiting_task([this](std::exception_ptr const* iPtr) {
0489         //The exception is being rethrown because resolveProductImpl sets the ProductResolver to a failed
0490         // state for the case where an exception occurs during the call to the function.
0491         // Caught exception is propagated via WaitingTaskList
0492         CMS_SA_ALLOW try {
0493           resolveProductImpl<true>([iPtr]() {
0494             if (iPtr) {
0495               std::rethrow_exception(*iPtr);
0496             }
0497           });
0498         } catch (...) {
0499           waitingTasks_.doneWaiting(std::current_exception());
0500           return;
0501         }
0502         waitingTasks_.doneWaiting(nullptr);
0503       });
0504
0505       ParentContext parentContext(mcc);
0506       EventTransitionInfo const& info = aux_->eventTransitionInfo();
0507       worker_->doWorkAsync<OccurrenceTraits<EventPrincipal, BranchActionStreamBegin> >(
0508           WaitingTaskHolder(*waitTask.group(), t),
0509           info,
0510           token,
0511           info.principal().streamID(),
0512           parentContext,
0513           mcc->getStreamContext());
0514     }
0515   }
0516
0517   void UnscheduledProductResolver::resetProductData_(bool deleteEarly) {
0518     if (not deleteEarly) {
0519       prefetchRequested_ = false;
0520       waitingTasks_.reset();
0521     }
0522     DataManagingProductResolver::resetProductData_(deleteEarly);
0523   }
0524
0525   void ProducedProductResolver::putProduct(std::unique_ptr<WrapperBase> edp) const {
0526     if (status() != defaultStatus()) {
0527       throw Exception(errors::InsertFailure)
0528           << "Attempt to insert more than one product on branch " << branchDescription().branchName() << "\n";
0529     }
0530
0531     setProduct(std::move(edp));  // ProductResolver takes ownership
0532   }
0533
0534   bool ProducedProductResolver::isFromCurrentProcess() const { return true; }
0535
0536   void DataManagingProductResolver::connectTo(ProductResolverBase const& iOther, Principal const*) { assert(false); }
0537
0538   void DataManagingProductResolver::checkType(WrapperBase const& prod) const {
0539     // Check if the types match.
0540     TypeID typeID(prod.dynamicTypeInfo());
0541     if (typeID != TypeID{branchDescription().unwrappedType().unvalidatedTypeInfo()}) {
0542       // Types do not match.
0543       throw Exception(errors::EventCorruption)
0544           << "Product on branch " << branchDescription().branchName() << " is of wrong type.\n"
0545           << "It is supposed to be of type " << branchDescription().className() << ".\n"
0546           << "It is actually of type " << typeID.className() << ".\n";
0547     }
0548   }
0549
0550   void DataManagingProductResolver::setProduct(std::unique_ptr<WrapperBase> edp) const {
0551     if (edp) {
0552       checkType(*edp);
0553       productData_.unsafe_setWrapper(std::move(edp));
0554       theStatus_ = ProductStatus::ProductSet;
0555     } else {
0556       setFailedStatus();
0557     }
0558   }
0559   void DataManagingProductResolver::setProduct(std::shared_ptr<WrapperBase> edp) const {
0560     if (edp) {
0561       checkType(*edp);
0562       productData_.unsafe_setWrapper(std::move(edp));
0563       theStatus_ = ProductStatus::ProductSet;
0564     } else {
0565       setFailedStatus();
0566     }
0567   }
0568
0569   // This routine returns true if it is known that currently there is no real product.
0570   // If there is a real product, it returns false.
0571   // If it is not known if there is a real product, it returns false.
0572   bool DataManagingProductResolver::productUnavailable_() const {
0573     auto presentStatus = status();
0574     if (presentStatus == ProductStatus::ProductSet) {
0575       return !(getProductData().wrapper()->isPresent());
0576     }
0577     return presentStatus != ProductStatus::ResolveNotRun;
0578   }
0579
0580   bool DataManagingProductResolver::productResolved_() const {
0581     auto s = status();
0582     return (s != defaultStatus()) or (s == ProductStatus::ProductDeleted);
0583   }
0584
0585   // This routine returns true if the product was deleted early in order to save memory
0586   bool DataManagingProductResolver::productWasDeleted_() const { return status() == ProductStatus::ProductDeleted; }
0587
0588   bool DataManagingProductResolver::productWasFetchedAndIsValid_(bool iSkipCurrentProcess) const {
0589     if (iSkipCurrentProcess and isFromCurrentProcess()) {
0590       return false;
0591     }
0592     if (status() == ProductStatus::ProductSet) {
0593       if (getProductData().wrapper()->isPresent()) {
0594         return true;
0595       }
0596     }
0597     return false;
0598   }
0599
0600   void DataManagingProductResolver::setProductProvenanceRetriever_(ProductProvenanceRetriever const* provRetriever) {
0601     productData_.setProvenance(provRetriever);
0602   }
0603
0604   void DataManagingProductResolver::setProductID_(ProductID const& pid) { productData_.setProductID(pid); }
0605
0606   void DataManagingProductResolver::setMergeableRunProductMetadataInProductData(
0607       MergeableRunProductMetadata const* mrpm) {
0608     productData_.setMergeableRunProductMetadata(mrpm);
0609   }
0610
0611   ProductProvenance const* DataManagingProductResolver::productProvenancePtr_() const {
0612     return provenance()->productProvenance();
0613   }
0614
0615   void DataManagingProductResolver::resetProductData_(bool deleteEarly) {
0616     if (theStatus_ == ProductStatus::ProductSet) {
0617       productData_.resetProductData();
0618     }
0619     if (deleteEarly) {
0620       theStatus_ = ProductStatus::ProductDeleted;
0621     } else {
0622       resetStatus();
0623     }
0624   }
0625
0626   bool DataManagingProductResolver::singleProduct_() const { return true; }
0627
0628   void AliasProductResolver::setProductProvenanceRetriever_(ProductProvenanceRetriever const* provRetriever) {
0629     realProduct_.setProductProvenanceRetriever(provRetriever);
0630   }
0631
0632   void AliasProductResolver::setProductID_(ProductID const& pid) { realProduct_.setProductID(pid); }
0633
0634   ProductProvenance const* AliasProductResolver::productProvenancePtr_() const {
0635     return provenance()->productProvenance();
0636   }
0637
0638   void AliasProductResolver::resetProductData_(bool deleteEarly) { realProduct_.resetProductData_(deleteEarly); }
0639
0640   bool AliasProductResolver::singleProduct_() const { return true; }
0641
0642   SwitchBaseProductResolver::SwitchBaseProductResolver(std::shared_ptr<BranchDescription const> bd,
0643                                                        DataManagingOrAliasProductResolver& realProduct)
0644       : realProduct_(realProduct), productData_(std::move(bd)), prefetchRequested_(false) {
0645     // Parentage of this branch is always the same by construction, so we can compute the ID just "once" here.
0646     Parentage p;
0647     p.setParents(std::vector<BranchID>{realProduct.branchDescription().originalBranchID()});
0648     parentageID_ = p.id();
0649     ParentageRegistry::instance()->insertMapped(p);
0650   }
0651
0652   void SwitchBaseProductResolver::connectTo(ProductResolverBase const& iOther, Principal const* iParentPrincipal) {
0653     throw Exception(errors::LogicError)
0654         << "SwitchBaseProductResolver::connectTo() not implemented and should never be called.\n"
0655         << "Contact a Framework developer\n";
0656   }
0657
0658   void SwitchBaseProductResolver::setupUnscheduled(UnscheduledConfigurator const& iConfigure) {
0659     worker_ = iConfigure.findWorker(branchDescription().moduleLabel());
0660   }
0661
0662   ProductResolverBase::Resolution SwitchBaseProductResolver::resolveProductImpl(Resolution res) const {
0663     if (res.data() == nullptr)
0664       return res;
0665     return Resolution(&productData_);
0666   }
0667
0668   bool SwitchBaseProductResolver::productResolved_() const {
0669     // SwitchProducer will never put anything in the event, and
0670     // "false" will make Event::commit_() to call putProduct() with
0671     // null unique_ptr<WrapperBase> to signal that the produce() was
0672     // run.
0673     return false;
0674   }
0675
0676   void SwitchBaseProductResolver::setProductProvenanceRetriever_(ProductProvenanceRetriever const* provRetriever) {
0677     productData_.setProvenance(provRetriever);
0678   }
0679
0680   void SwitchBaseProductResolver::setProductID_(ProductID const& pid) {
0681     // insertIntoSet is const, so let's exploit that to fake the getting of the "input" product
0682     productData_.setProductID(pid);
0683   }
0684
0685   void SwitchBaseProductResolver::resetProductData_(bool deleteEarly) {
0686     productData_.resetProductData();
0687     realProduct_.resetProductData_(deleteEarly);
0688     if (not deleteEarly) {
0689       prefetchRequested_ = false;
0690       waitingTasks_.reset();
0691     }
0692   }
0693
0694   void SwitchBaseProductResolver::unsafe_setWrapperAndProvenance() const {
0695     // update provenance
0696     productData_.provenance().store()->insertIntoSet(ProductProvenance(branchDescription().branchID(), parentageID_));
0697     // Use the Wrapper of the pointed-to resolver, but the provenance of this resolver
0698     productData_.unsafe_setWrapper(realProduct().getProductData().sharedConstWrapper());
0699   }
0700
0701   SwitchProducerProductResolver::SwitchProducerProductResolver(std::shared_ptr<BranchDescription const> bd,
0702                                                                DataManagingOrAliasProductResolver& realProduct)
0703       : SwitchBaseProductResolver(std::move(bd), realProduct), status_(defaultStatus_) {}
0704
0705   ProductResolverBase::Resolution SwitchProducerProductResolver::resolveProduct_(Principal const& principal,
0706                                                                                  bool skipCurrentProcess,
0707                                                                                  SharedResourcesAcquirer* sra,
0708                                                                                  ModuleCallingContext const* mcc) const {
0709     if (status_ == ProductStatus::ResolveFailed) {
0710       return resolveProductImpl(realProduct().resolveProduct(principal, skipCurrentProcess, sra, mcc));
0711     }
0712     return Resolution(nullptr);
0713   }
0714
0715   void SwitchProducerProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
0716                                                      Principal const& principal,
0717                                                      bool skipCurrentProcess,
0718                                                      ServiceToken const& token,
0719                                                      SharedResourcesAcquirer* sra,
0720                                                      ModuleCallingContext const* mcc) const {
0721     if (skipCurrentProcess) {
0722       return;
0723     }
0724     if (branchDescription().availableOnlyAtEndTransition() and mcc and not mcc->parent().isAtEndTransition()) {
0725       return;
0726     }
0727
0728     //need to try changing prefetchRequested before adding to waitingTasks
0729     bool expected = false;
0730     bool doPrefetchRequested = prefetchRequested().compare_exchange_strong(expected, true);
0731     waitingTasks().add(waitTask);
0732
0733     if (doPrefetchRequested) {
0734       //using a waiting task to do a callback guarantees that
0735       // the waitingTasks() list will be released from waiting even
0736       // if the module does not put this data product or the
0737       // module has an exception while running
0738       auto waiting = make_waiting_task([this](std::exception_ptr const* iException) {
0739         if (nullptr != iException) {
0740           waitingTasks().doneWaiting(*iException);
0741         } else {
0742           unsafe_setWrapperAndProvenance();
0743           waitingTasks().doneWaiting(std::exception_ptr());
0744         }
0745       });
0746       worker()->callWhenDoneAsync(WaitingTaskHolder(*waitTask.group(), waiting));
0747     }
0748   }
0749
0750   void SwitchProducerProductResolver::putProduct(std::unique_ptr<WrapperBase> edp) const {
0751     if (status_ != defaultStatus_) {
0752       throw Exception(errors::InsertFailure)
0753           << "Attempt to insert more than one product for a branch " << branchDescription().branchName()
0754           << "This makes no sense for SwitchProducerProductResolver.\nContact a Framework developer";
0755     }
0756     // Let's use ResolveFailed to signal that produce() was called, as
0757     // there is no real product in this resolver
0758     status_ = ProductStatus::ResolveFailed;
0759     bool expected = false;
0760     if (prefetchRequested().compare_exchange_strong(expected, true)) {
0761       unsafe_setWrapperAndProvenance();
0762       waitingTasks().doneWaiting(std::exception_ptr());
0763     }
0764   }
0765
0766   bool SwitchProducerProductResolver::productUnavailable_() const {
0767     // if produce() was run (ResolveFailed), ask from the real resolver
0768     if (status_ == ProductStatus::ResolveFailed) {
0769       return realProduct().productUnavailable();
0770     }
0771     return true;
0772   }
0773
0774   void SwitchProducerProductResolver::resetProductData_(bool deleteEarly) {
0775     SwitchBaseProductResolver::resetProductData_(deleteEarly);
0776     if (not deleteEarly) {
0777       status_ = defaultStatus_;
0778     }
0779   }
0780
0781   ProductResolverBase::Resolution SwitchAliasProductResolver::resolveProduct_(Principal const& principal,
0782                                                                               bool skipCurrentProcess,
0783                                                                               SharedResourcesAcquirer* sra,
0784                                                                               ModuleCallingContext const* mcc) const {
0785     return resolveProductImpl(realProduct().resolveProduct(principal, skipCurrentProcess, sra, mcc));
0786   }
0787
0788   void SwitchAliasProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
0789                                                   Principal const& principal,
0790                                                   bool skipCurrentProcess,
0791                                                   ServiceToken const& token,
0792                                                   SharedResourcesAcquirer* sra,
0793                                                   ModuleCallingContext const* mcc) const {
0794     if (skipCurrentProcess) {
0795       return;
0796     }
0797
0798     //need to try changing prefetchRequested_ before adding to waitingTasks_
0799     bool expected = false;
0800     bool doPrefetchRequested = prefetchRequested().compare_exchange_strong(expected, true);
0801     waitingTasks().add(waitTask);
0802
0803     if (doPrefetchRequested) {
0804       //using a waiting task to do a callback guarantees that
0805       // the waitingTasks() list will be released from waiting even
0806       // if the module does not put this data product or the
0807       // module has an exception while running
0808       auto waiting = make_waiting_task([this](std::exception_ptr const* iException) {
0809         if (nullptr != iException) {
0810           waitingTasks().doneWaiting(*iException);
0811         } else {
0812           unsafe_setWrapperAndProvenance();
0813           waitingTasks().doneWaiting(std::exception_ptr());
0814         }
0815       });
0816       realProduct().prefetchAsync(
0817           WaitingTaskHolder(*waitTask.group(), waiting), principal, skipCurrentProcess, token, sra, mcc);
0818     }
0819   }
0820
0821   void ParentProcessProductResolver::setProductProvenanceRetriever_(ProductProvenanceRetriever const* provRetriever) {
0822     provRetriever_ = provRetriever;
0823   }
0824
0825   void ParentProcessProductResolver::setProductID_(ProductID const&) {}
0826
0827   ProductProvenance const* ParentProcessProductResolver::productProvenancePtr_() const {
0828     return provRetriever_ ? provRetriever_->branchIDToProvenance(bd_->originalBranchID()) : nullptr;
0829   }
0830
0831   void ParentProcessProductResolver::resetProductData_(bool deleteEarly) {}
0832
0833   bool ParentProcessProductResolver::singleProduct_() const { return true; }
0834
0835   void ParentProcessProductResolver::throwNullRealProduct() const {
0836     // In principle, this ought to be fixed. I noticed one hits this error
0837     // when in a SubProcess and calling the Event::getProvenance function
0838     // with a BranchID to a branch from an earlier SubProcess or the top
0839     // level process and this branch is not kept in this SubProcess. It might
0840     // be possible to hit this in other contexts. I say it ought to be
0841     // fixed because one does not encounter this issue if the SubProcesses
0842     // are split into genuinely different processes (in principle that
0843     // ought to give identical behavior and results). No user has ever
0844     // reported this issue which has been around for some time and it was only
0845     // noticed when testing some rare corner cases after modifying Core code.
0846     // After discussing this with Chris we decided that at least for the moment
0847     // there are higher priorities than fixing this ... I converted it so it
0848     // causes an exception instead of a seg fault. The issue that may need to
0849     // be addressed someday is how ProductResolvers for non-kept branches are
0850     // connected to earlier SubProcesses.
0851     throw Exception(errors::LogicError)
0852         << "ParentProcessProductResolver::throwNullRealProduct RealProduct pointer not set in this context.\n"
0853         << "Contact a Framework developer\n";
0854   }
0855
0856   NoProcessProductResolver::NoProcessProductResolver(std::vector<ProductResolverIndex> const& matchingHolders,
0857                                                      std::vector<bool> const& ambiguous,
0858                                                      bool madeAtEnd)
0859       : matchingHolders_(matchingHolders),
0860         ambiguous_(ambiguous),
0861         lastCheckIndex_(ambiguous_.size() + kUnsetOffset),
0862         lastSkipCurrentCheckIndex_(lastCheckIndex_.load()),
0863         prefetchRequested_(false),
0864         skippingPrefetchRequested_(false),
0865         madeAtEnd_{madeAtEnd} {
0866     assert(ambiguous_.size() == matchingHolders_.size());
0867   }
0868
0869   ProductResolverBase::Resolution NoProcessProductResolver::tryResolver(unsigned int index,
0870                                                                         Principal const& principal,
0871                                                                         bool skipCurrentProcess,
0872                                                                         SharedResourcesAcquirer* sra,
0873                                                                         ModuleCallingContext const* mcc) const {
0874     ProductResolverBase const* productResolver = principal.getProductResolverByIndex(matchingHolders_[index]);
0875     return productResolver->resolveProduct(principal, skipCurrentProcess, sra, mcc);
0876   }
0877
0878   ProductResolverBase::Resolution NoProcessProductResolver::resolveProduct_(Principal const& principal,
0879                                                                             bool skipCurrentProcess,
0880                                                                             SharedResourcesAcquirer* sra,
0881                                                                             ModuleCallingContext const* mcc) const {
0882     //See if we've already cached which Resolver we should call or if
0883     // we know it is ambiguous
0884     const unsigned int choiceSize = ambiguous_.size();
0885
0886     //madeAtEnd_==true and not at end transition is the same as skipping the current process
0887     if ((not skipCurrentProcess) and (madeAtEnd_ and mcc)) {
0888       skipCurrentProcess = not mcc->parent().isAtEndTransition();
0889     }
0890
0891     unsigned int checkCacheIndex = skipCurrentProcess ? lastSkipCurrentCheckIndex_.load() : lastCheckIndex_.load();
0892     if (checkCacheIndex != choiceSize + kUnsetOffset) {
0893       if (checkCacheIndex == choiceSize + kAmbiguousOffset) {
0894         return ProductResolverBase::Resolution::makeAmbiguous();
0895       } else if (checkCacheIndex == choiceSize + kMissingOffset) {
0896         return Resolution(nullptr);
0897       }
0898       return tryResolver(checkCacheIndex, principal, skipCurrentProcess, sra, mcc);
0899     }
0900
0901     std::atomic<unsigned int>& updateCacheIndex = skipCurrentProcess ? lastSkipCurrentCheckIndex_ : lastCheckIndex_;
0902
0903     std::vector<unsigned int> const& lookupProcessOrder = principal.lookupProcessOrder();
0904     for (unsigned int k : lookupProcessOrder) {
0905       assert(k < ambiguous_.size());
0906       if (k == 0)
0907         break;  // Done
0908       if (ambiguous_[k]) {
0909         updateCacheIndex = choiceSize + kAmbiguousOffset;
0910         return ProductResolverBase::Resolution::makeAmbiguous();
0911       }
0912       if (matchingHolders_[k] != ProductResolverIndexInvalid) {
0913         auto resolution = tryResolver(k, principal, skipCurrentProcess, sra, mcc);
0914         if (resolution.data() != nullptr) {
0915           updateCacheIndex = k;
0916           return resolution;
0917         }
0918       }
0919     }
0920
0921     updateCacheIndex = choiceSize + kMissingOffset;
0922     return Resolution(nullptr);
0923   }
0924
0925   void NoProcessProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
0926                                                 Principal const& principal,
0927                                                 bool skipCurrentProcess,
0928                                                 ServiceToken const& token,
0929                                                 SharedResourcesAcquirer* sra,
0930                                                 ModuleCallingContext const* mcc) const {
0931     bool timeToMakeAtEnd = true;
0932     if (madeAtEnd_ and mcc) {
0933       timeToMakeAtEnd = mcc->parent().isAtEndTransition();
0934     }
0935
0936     //If timeToMakeAtEnd is false, then it is equivalent to skipping the current process
0937     if (not skipCurrentProcess and timeToMakeAtEnd) {
0938       //need to try changing prefetchRequested_ before adding to waitingTasks_
0939       bool expected = false;
0940       bool prefetchRequested = prefetchRequested_.compare_exchange_strong(expected, true);
0941       waitingTasks_.add(waitTask);
0942
0943       if (prefetchRequested) {
0944         //we are the first thread to request
0945         tryPrefetchResolverAsync(0, principal, false, sra, mcc, token, waitTask.group());
0946       }
0947     } else {
0948       skippingWaitingTasks_.add(waitTask);
0949       bool expected = false;
0950       if (skippingPrefetchRequested_.compare_exchange_strong(expected, true)) {
0951         //we are the first thread to request
0952         tryPrefetchResolverAsync(0, principal, true, sra, mcc, token, waitTask.group());
0953       }
0954     }
0955   }
0956
0957   void NoProcessProductResolver::setCache(bool iSkipCurrentProcess,
0958                                           ProductResolverIndex iIndex,
0959                                           std::exception_ptr iExceptPtr) const {
0960     if (not iSkipCurrentProcess) {
0961       lastCheckIndex_ = iIndex;
0962       waitingTasks_.doneWaiting(iExceptPtr);
0963     } else {
0964       lastSkipCurrentCheckIndex_ = iIndex;
0965       skippingWaitingTasks_.doneWaiting(iExceptPtr);
0966     }
0967   }
0968
0969   namespace {
0970     class TryNextResolverWaitingTask : public edm::WaitingTask {
0971     public:
0972       TryNextResolverWaitingTask(NoProcessProductResolver const* iResolver,
0973                                  unsigned int iResolverIndex,
0974                                  Principal const* iPrincipal,
0975                                  SharedResourcesAcquirer* iSRA,
0976                                  ModuleCallingContext const* iMCC,
0977                                  bool iSkipCurrentProcess,
0978                                  ServiceToken iToken,
0979                                  oneapi::tbb::task_group* iGroup)
0980           : resolver_(iResolver),
0981             principal_(iPrincipal),
0982             sra_(iSRA),
0983             mcc_(iMCC),
0984             group_(iGroup),
0985             serviceToken_(iToken),
0986             index_(iResolverIndex),
0987             skipCurrentProcess_(iSkipCurrentProcess) {}
0988
0989       void execute() final {
0990         auto exceptPtr = exceptionPtr();
0991         if (exceptPtr) {
0992           resolver_->prefetchFailed(index_, *principal_, skipCurrentProcess_, *exceptPtr);
0993         } else {
0994           if (not resolver_->dataValidFromResolver(index_, *principal_, skipCurrentProcess_)) {
0995             resolver_->tryPrefetchResolverAsync(
0996                 index_ + 1, *principal_, skipCurrentProcess_, sra_, mcc_, serviceToken_.lock(), group_);
0997           }
0998         }
0999       }
1000
1001     private:
1002       NoProcessProductResolver const* resolver_;
1003       Principal const* principal_;
1004       SharedResourcesAcquirer* sra_;
1005       ModuleCallingContext const* mcc_;
1006       oneapi::tbb::task_group* group_;
1007       ServiceWeakToken serviceToken_;
1008       unsigned int index_;
1009       bool skipCurrentProcess_;
1010     };
1011   }  // namespace
1012
1013   void NoProcessProductResolver::prefetchFailed(unsigned int iProcessingIndex,
1014                                                 Principal const& principal,
1015                                                 bool iSkipCurrentProcess,
1016                                                 std::exception_ptr iExceptPtr) const {
1017     std::vector<unsigned int> const& lookupProcessOrder = principal.lookupProcessOrder();
1018     auto k = lookupProcessOrder[iProcessingIndex];
1019
1020     setCache(iSkipCurrentProcess, k, iExceptPtr);
1021   }
1022
1023   bool NoProcessProductResolver::dataValidFromResolver(unsigned int iProcessingIndex,
1024                                                        Principal const& principal,
1025                                                        bool iSkipCurrentProcess) const {
1026     std::vector<unsigned int> const& lookupProcessOrder = principal.lookupProcessOrder();
1027     auto k = lookupProcessOrder[iProcessingIndex];
1028     ProductResolverBase const* productResolver = principal.getProductResolverByIndex(matchingHolders_[k]);
1029
1030     if (productResolver->productWasFetchedAndIsValid(iSkipCurrentProcess)) {
1031       setCache(iSkipCurrentProcess, k, nullptr);
1032       return true;
1033     }
1034     return false;
1035   }
1036
1037   void NoProcessProductResolver::tryPrefetchResolverAsync(unsigned int iProcessingIndex,
1038                                                           Principal const& principal,
1039                                                           bool skipCurrentProcess,
1040                                                           SharedResourcesAcquirer* sra,
1041                                                           ModuleCallingContext const* mcc,
1042                                                           ServiceToken token,
1043                                                           oneapi::tbb::task_group* group) const {
1044     std::vector<unsigned int> const& lookupProcessOrder = principal.lookupProcessOrder();
1045     auto index = iProcessingIndex;
1046
1047     const unsigned int choiceSize = ambiguous_.size();
1048     unsigned int newCacheIndex = choiceSize + kMissingOffset;
1049     while (index < lookupProcessOrder.size()) {
1050       auto k = lookupProcessOrder[index];
1051       if (k == 0) {
1052         break;
1053       }
1054       assert(k < ambiguous_.size());
1055       if (ambiguous_[k]) {
1056         newCacheIndex = choiceSize + kAmbiguousOffset;
1057         break;
1058       }
1059       if (matchingHolders_[k] != ProductResolverIndexInvalid) {
1060         //make new task
1061
1062         auto task = new TryNextResolverWaitingTask(this, index, &principal, sra, mcc, skipCurrentProcess, token, group);
1063         WaitingTaskHolder hTask(*group, task);
1064         ProductResolverBase const* productResolver = principal.getProductResolverByIndex(matchingHolders_[k]);
1065
1066         //Make sure the Services are available on this thread
1067         ServiceRegistry::Operate guard(token);
1068
1069         productResolver->prefetchAsync(hTask, principal, skipCurrentProcess, token, sra, mcc);
1070         return;
1071       }
1072       ++index;
1073     }
1074     //data product unavailable
1075     setCache(skipCurrentProcess, newCacheIndex, nullptr);
1076   }
1077
1078   void NoProcessProductResolver::setProductProvenanceRetriever_(ProductProvenanceRetriever const*) {}
1079
1080   void NoProcessProductResolver::setProductID_(ProductID const&) {}
1081
1082   ProductProvenance const* NoProcessProductResolver::productProvenancePtr_() const { return nullptr; }
1083
1084   inline unsigned int NoProcessProductResolver::unsetIndexValue() const { return ambiguous_.size() + kUnsetOffset; }
1085
1086   void NoProcessProductResolver::resetProductData_(bool) {
1087     // This function should never receive 'true'. On the other hand,
1088     // nothing should break if a 'true' is passed, because
1089     // NoProcessProductResolver just forwards the resolve
1090     const auto resetValue = unsetIndexValue();
1091     lastCheckIndex_ = resetValue;
1092     lastSkipCurrentCheckIndex_ = resetValue;
1093     prefetchRequested_ = false;
1094     skippingPrefetchRequested_ = false;
1095     waitingTasks_.reset();
1096     skippingWaitingTasks_.reset();
1097   }
1098
1099   bool NoProcessProductResolver::singleProduct_() const { return false; }
1100
1101   bool NoProcessProductResolver::unscheduledWasNotRun_() const {
1102     throw Exception(errors::LogicError)
1103         << "NoProcessProductResolver::unscheduledWasNotRun_() not implemented and should never be called.\n"
1104         << "Contact a Framework developer\n";
1105   }
1106
1107   bool NoProcessProductResolver::productUnavailable_() const {
1108     throw Exception(errors::LogicError)
1109         << "NoProcessProductResolver::productUnavailable_() not implemented and should never be called.\n"
1110         << "Contact a Framework developer\n";
1111   }
1112
1113   bool NoProcessProductResolver::productResolved_() const {
1114     throw Exception(errors::LogicError)
1115         << "NoProcessProductResolver::productResolved_() not implemented and should never be called.\n"
1116         << "Contact a Framework developer\n";
1117   }
1118
1119   bool NoProcessProductResolver::productWasDeleted_() const {
1120     throw Exception(errors::LogicError)
1121         << "NoProcessProductResolver::productWasDeleted_() not implemented and should never be called.\n"
1122         << "Contact a Framework developer\n";
1123   }
1124
1125   bool NoProcessProductResolver::productWasFetchedAndIsValid_(bool /*iSkipCurrentProcess*/) const {
1126     throw Exception(errors::LogicError)
1127         << "NoProcessProductResolver::productWasFetchedAndIsValid_() not implemented and should never be called.\n"
1128         << "Contact a Framework developer\n";
1129   }
1130
1131   BranchDescription const& NoProcessProductResolver::branchDescription_() const {
1132     throw Exception(errors::LogicError)
1133         << "NoProcessProductResolver::branchDescription_() not implemented and should never be called.\n"
1134         << "Contact a Framework developer\n";
1135   }
1136
1137   void NoProcessProductResolver::resetBranchDescription_(std::shared_ptr<BranchDescription const>) {
1138     throw Exception(errors::LogicError)
1139         << "NoProcessProductResolver::resetBranchDescription_() not implemented and should never be called.\n"
1140         << "Contact a Framework developer\n";
1141   }
1142
1143   Provenance const* NoProcessProductResolver::provenance_() const {
1144     throw Exception(errors::LogicError)
1145         << "NoProcessProductResolver::provenance_() not implemented and should never be called.\n"
1146         << "Contact a Framework developer\n";
1147   }
1148
1149   void NoProcessProductResolver::connectTo(ProductResolverBase const&, Principal const*) {
1150     throw Exception(errors::LogicError)
1151         << "NoProcessProductResolver::connectTo() not implemented and should never be called.\n"
1152         << "Contact a Framework developer\n";
1153   }
1154
1155   //---- SingleChoiceNoProcessProductResolver ----------------
1156   ProductResolverBase::Resolution SingleChoiceNoProcessProductResolver::resolveProduct_(
1157       Principal const& principal,
1158       bool skipCurrentProcess,
1159       SharedResourcesAcquirer* sra,
1160       ModuleCallingContext const* mcc) const {
1161     //NOTE: Have to lookup the other ProductResolver each time rather than cache
1162     // it's pointer since it appears the pointer can change at some later stage
1163     return principal.getProductResolverByIndex(realResolverIndex_)
1164         ->resolveProduct(principal, skipCurrentProcess, sra, mcc);
1165   }
1166
1167   void SingleChoiceNoProcessProductResolver::prefetchAsync_(WaitingTaskHolder waitTask,
1168                                                             Principal const& principal,
1169                                                             bool skipCurrentProcess,
1170                                                             ServiceToken const& token,
1171                                                             SharedResourcesAcquirer* sra,
1172                                                             ModuleCallingContext const* mcc) const {
1173     principal.getProductResolverByIndex(realResolverIndex_)
1174         ->prefetchAsync(waitTask, principal, skipCurrentProcess, token, sra, mcc);
1175   }
1176
1177   void SingleChoiceNoProcessProductResolver::setProductProvenanceRetriever_(ProductProvenanceRetriever const*) {}
1178
1179   void SingleChoiceNoProcessProductResolver::setProductID_(ProductID const&) {}
1180
1181   ProductProvenance const* SingleChoiceNoProcessProductResolver::productProvenancePtr_() const { return nullptr; }
1182
1183   void SingleChoiceNoProcessProductResolver::resetProductData_(bool) {}
1184
1185   bool SingleChoiceNoProcessProductResolver::singleProduct_() const { return false; }
1186
1187   bool SingleChoiceNoProcessProductResolver::unscheduledWasNotRun_() const {
1188     throw Exception(errors::LogicError)
1189         << "SingleChoiceNoProcessProductResolver::unscheduledWasNotRun_() not implemented and should never be called.\n"
1190         << "Contact a Framework developer\n";
1191   }
1192
1193   bool SingleChoiceNoProcessProductResolver::productUnavailable_() const {
1194     throw Exception(errors::LogicError)
1195         << "SingleChoiceNoProcessProductResolver::productUnavailable_() not implemented and should never be called.\n"
1196         << "Contact a Framework developer\n";
1197   }
1198
1199   bool SingleChoiceNoProcessProductResolver::productResolved_() const {
1200     throw Exception(errors::LogicError)
1201         << "SingleChoiceNoProcessProductResolver::productResolved_() not implemented and should never be called.\n"
1202         << "Contact a Framework developer\n";
1203   }
1204
1205   bool SingleChoiceNoProcessProductResolver::productWasDeleted_() const {
1206     throw Exception(errors::LogicError)
1207         << "SingleChoiceNoProcessProductResolver::productWasDeleted_() not implemented and should never be called.\n"
1208         << "Contact a Framework developer\n";
1209   }
1210
1211   bool SingleChoiceNoProcessProductResolver::productWasFetchedAndIsValid_(bool /*iSkipCurrentProcess*/) const {
1212     throw Exception(errors::LogicError) << "SingleChoiceNoProcessProductResolver::productWasFetchedAndIsValid_() not "
1213                                            "implemented and should never be called.\n"
1214                                         << "Contact a Framework developer\n";
1215   }
1216
1217   BranchDescription const& SingleChoiceNoProcessProductResolver::branchDescription_() const {
1218     throw Exception(errors::LogicError)
1219         << "SingleChoiceNoProcessProductResolver::branchDescription_() not implemented and should never be called.\n"
1220         << "Contact a Framework developer\n";
1221   }
1222
1223   void SingleChoiceNoProcessProductResolver::resetBranchDescription_(std::shared_ptr<BranchDescription const>) {
1224     throw Exception(errors::LogicError) << "SingleChoiceNoProcessProductResolver::resetBranchDescription_() not "
1225                                            "implemented and should never be called.\n"
1226                                         << "Contact a Framework developer\n";
1227   }
1228
1229   Provenance const* SingleChoiceNoProcessProductResolver::provenance_() const {
1230     throw Exception(errors::LogicError)
1231         << "SingleChoiceNoProcessProductResolver::provenance_() not implemented and should never be called.\n"
1232         << "Contact a Framework developer\n";
1233   }
1234
1235   void SingleChoiceNoProcessProductResolver::connectTo(ProductResolverBase const&, Principal const*) {
1236     throw Exception(errors::LogicError)
1237         << "SingleChoiceNoProcessProductResolver::connectTo() not implemented and should never be called.\n"
1238         << "Contact a Framework developer\n";
1239   }
1240
1241 }  // namespace edm