33 exception <<
"ProductResolverBase::resolveProduct_: The product matching all criteria was already deleted\n" 35 <<
"Looking for module label: " <<
moduleLabel() <<
"\n" 38 <<
"This means there is a configuration error.\n" 39 <<
"The module which is asking for this data must be configured to state that it will read this data.";
44 template <
bool callResolver,
typename FUNC>
49 auto presentStatus =
status();
54 auto failedStatusSetter = [
this](
ProductStatus* presentStatus) {
58 *presentStatus = this->
status();
60 std::unique_ptr<ProductStatus, decltype(failedStatusSetter)> failedStatusGuard(&presentStatus,
69 if (pd->wrapper()->isPresent()) {
94 if (
original->isPresent() != iFrom->isPresent()) {
97 <<
"Was trying to merge objects where one product had been put in the input file and the other had not " 100 <<
"The solution is to drop the branch on input. Or better do not create inconsistent files\n" 101 <<
"that need to be merged in the first place.\n";
105 if (mergeableRunProductMetadata ==
nullptr || desc.
branchType() !=
InRun) {
106 original->mergeProduct(iFrom.get());
111 original->mergeProduct(iFrom.get());
126 }
else if (
original->hasIsProductEqual()) {
127 if (
original->isPresent() && iFrom->isPresent()) {
128 if (!
original->isProductEqual(iFrom.get())) {
131 <<
"ProductResolver::mergeTheProduct\n" 132 <<
"Two run/lumi products for the same run/lumi which should be equal are not\n" 133 <<
"Using the first, ignoring the second\n" 134 <<
"className = " << bd.className() <<
"\n" 135 <<
"moduleLabel = " << bd.moduleLabel() <<
"\n" 136 <<
"instance = " << bd.productInstanceName() <<
"\n" 137 <<
"process = " << bd.processName() <<
"\n";
139 }
else if (!
original->isPresent() && iFrom->isPresent()) {
145 edm::LogWarning(
"RunLumiMerging") <<
"ProductResolver::mergeTheProduct\n" 146 <<
"Run/lumi product has neither a mergeProduct nor isProductEqual function\n" 147 <<
"Using the first, ignoring the second in merge\n" 148 <<
"className = " << bd.className() <<
"\n" 149 <<
"moduleLabel = " << bd.moduleLabel() <<
"\n" 150 <<
"instance = " << bd.productInstanceName() <<
"\n" 151 <<
"process = " << bd.processName() <<
"\n";
152 if (!
original->isPresent() && iFrom->isPresent()) {
163 return resolveProductImpl<true>([
this, &
principal, mcc]() {
176 aux_->postModuleDelayedGetSignal_.emit(*(iContext->
getStreamContext()), *iContext);
180 if (
auto reader = principal.reader()) {
181 std::unique_lock<std::recursive_mutex> guard;
182 if (
auto sr =
reader->sharedResources().second) {
183 guard = std::unique_lock<std::recursive_mutex>(*sr);
196 std::unique_lock<std::recursive_mutex> guard;
197 if (
auto sr =
reader->sharedResources().second) {
198 guard = std::unique_lock<std::recursive_mutex>(*sr);
205 if (edp.get() !=
nullptr) {
210 <<
"Mergeable data types written to a Run must have the swap member function defined" 219 <<
"The product branch was dropped in the first run or lumi fragment and present in a later one" 221 <<
"The solution is to drop the branch on input. Or better do not create inconsistent files\n" 222 <<
"that need to be merged in the first place.\n";
229 <<
"The product branch was present in first run or lumi fragment and dropped in a later one" 231 <<
"The solution is to drop the branch on input. Or better do not create inconsistent files\n" 232 <<
"that need to be merged in the first place.\n";
245 bool skipCurrentProcess,
249 m_waitingTasks.add(waitTask);
251 bool expected =
false;
252 if (m_prefetchRequested.compare_exchange_strong(expected,
true)) {
253 auto workToDo = [
this, mcc, &
principal, token]() {
257 resolveProductImpl<true>([
this, &
principal, mcc]() {
258 if (principal.branchType() !=
InEvent) {
261 if (
auto reader = principal.reader()) {
262 std::unique_lock<std::recursive_mutex> guard;
263 if (
auto sr =
reader->sharedResources().second) {
264 guard = std::unique_lock<std::recursive_mutex>(*sr);
273 this->m_waitingTasks.doneWaiting(std::current_exception());
276 this->m_waitingTasks.doneWaiting(
nullptr);
280 if (
auto reader = principal.reader()) {
281 if (
auto shared_res =
reader->sharedResources().first) {
282 queue = &(shared_res->serialQueueChain());
286 queue->
push(workToDo);
290 tbb::task::spawn(*
t);
296 m_prefetchRequested =
false;
297 m_waitingTasks.reset();
308 bool skipCurrentProcess,
311 if (!skipCurrentProcess) {
313 return resolveProductImpl<false>([]() {
return; });
320 bool skipCurrentProcess,
324 if (not skipCurrentProcess) {
326 if (not mcc->parent().isAtEndTransition()) {
330 m_waitingTasks.add(waitTask);
332 bool expected =
false;
333 if (worker_ and prefetchRequested_.compare_exchange_strong(expected,
true)) {
339 auto waiting =
make_waiting_task(tbb::task::allocate_root(), [
this](std::exception_ptr
const* iException) {
340 if (
nullptr != iException) {
341 m_waitingTasks.doneWaiting(*iException);
343 m_waitingTasks.doneWaiting(std::exception_ptr());
346 worker_->callWhenDoneAsync(waiting);
353 bool expected =
false;
354 if (prefetchRequested_.compare_exchange_strong(expected,
true)) {
355 m_waitingTasks.doneWaiting(std::exception_ptr());
360 m_waitingTasks.reset();
362 prefetchRequested_ =
false;
372 assert(worker_ !=
nullptr);
376 bool skipCurrentProcess,
379 if (!skipCurrentProcess and worker_) {
380 return resolveProductImpl<true>([&
principal,
this, sra, mcc]() {
386 auto workCall = [
this, &
event, &parentContext, mcc]() {
388 aux_->postModuleDelayedGetSignal_.emit(*(iContext->
getStreamContext()), *iContext);
402 std::ostringstream ost;
403 ost <<
"Calling produce method for unscheduled module " << worker_->description().moduleName() <<
"/'" 404 << worker_->description().moduleLabel() <<
"'";
415 bool skipCurrentProcess,
419 if (skipCurrentProcess) {
422 waitingTasks_.add(waitTask);
423 bool expected =
false;
424 if (prefetchRequested_.compare_exchange_strong(expected,
true)) {
427 auto t =
make_waiting_task(tbb::task::allocate_root(), [
this](std::exception_ptr
const* iPtr) {
431 resolveProductImpl<true>([iPtr]() {
433 std::rethrow_exception(*iPtr);
437 waitingTasks_.doneWaiting(std::current_exception());
440 waitingTasks_.doneWaiting(
nullptr);
451 prefetchRequested_ =
false;
452 waitingTasks_.reset();
497 <<
"It is actually of type " << typeID.className() <<
".\n";
514 auto presentStatus =
status();
574 realProduct_.setProvenance(provRetriever, ph, pid);
589 <<
"AliasProductResolver::putProduct_() not implemented and should never be called.\n" 590 <<
"Contact a Framework developer\n";
596 <<
"AliasProductResolver::putOrMergeProduct_(std::unique_ptr<WrapperBase> edp, MergeableRunProductMetadata " 597 "const*) not implemented and should never be called.\n" 598 <<
"Contact a Framework developer\n";
613 <<
"SwitchBaseProductResolver::connectTo() not implemented and should never be called.\n" 614 <<
"Contact a Framework developer\n";
622 if (res.
data() ==
nullptr)
641 <<
"This makes no sense for SwitchBaseProductResolver.\nContact a Framework developer";
646 bool expected =
false;
655 <<
"SwitchBaseProductResolver::putOrMergeProduct_(std::unique_ptr<WrapperBase> edp, " 656 "MergeableRunProductMetadata const*) not implemented and should never be called.\n" 657 <<
"Contact a Framework developer\n";
671 if (not deleteEarly) {
677 bool skipCurrentProcess,
688 bool skipCurrentProcess,
692 if (skipCurrentProcess) {
700 bool expected =
false;
706 auto waiting =
make_waiting_task(tbb::task::allocate_root(), [
this](std::exception_ptr
const* iException) {
707 if (
nullptr != iException) {
726 bool skipCurrentProcess,
734 bool skipCurrentProcess,
738 if (skipCurrentProcess) {
747 provRetriever_ = provRetriever;
753 return provRetriever_ ? provRetriever_->branchIDToProvenance(bd_->originalBranchID()) :
nullptr;
762 <<
"ParentProcessProductResolver::putProduct_() not implemented and should never be called.\n" 763 <<
"Contact a Framework developer\n";
769 <<
"ParentProcessProductResolver::putOrMergeProduct_(std::unique_ptr<WrapperBase> edp, " 770 "MergeableRunProductMetadata const*) not implemented and should never be called.\n" 771 <<
"Contact a Framework developer\n";
791 <<
"ParentProcessProductResolver::throwNullRealProduct RealProduct pointer not set in this context.\n" 792 <<
"Contact a Framework developer\n";
796 std::vector<bool>
const& ambiguous,
798 : matchingHolders_(matchingHolders),
799 ambiguous_(ambiguous),
801 lastSkipCurrentCheckIndex_(lastCheckIndex_.
load()),
803 skippingPrefetchRequested_(
false),
804 madeAtEnd_{madeAtEnd} {
810 bool skipCurrentProcess,
814 return productResolver->
resolveProduct(principal, skipCurrentProcess, sra, mcc);
818 bool skipCurrentProcess,
823 const unsigned int choiceSize =
ambiguous_.size();
826 if ((not skipCurrentProcess) and (
madeAtEnd_ and mcc)) {
827 skipCurrentProcess = not mcc->parent().isAtEndTransition();
837 return tryResolver(checkCacheIndex, principal, skipCurrentProcess, sra, mcc);
843 for (
unsigned int k : lookupProcessOrder) {
854 updateCacheIndex =
k;
866 bool skipCurrentProcess,
870 bool timeToMakeAtEnd =
true;
872 timeToMakeAtEnd = mcc->parent().isAtEndTransition();
876 if (not skipCurrentProcess and timeToMakeAtEnd) {
879 bool expected =
false;
886 bool expected =
false;
896 std::exception_ptr iExceptPtr)
const {
897 if (not iSkipCurrentProcess) {
910 unsigned int iResolverIndex,
914 bool iSkipCurrentProcess,
916 : resolver_(iResolver),
917 principal_(iPrincipal),
920 serviceToken_(iToken),
921 index_(iResolverIndex),
922 skipCurrentProcess_(iSkipCurrentProcess) {}
925 auto exceptPtr = exceptionPtr();
927 resolver_->prefetchFailed(index_, *principal_, skipCurrentProcess_, *exceptPtr);
929 if (not resolver_->dataValidFromResolver(index_, *principal_, skipCurrentProcess_)) {
930 resolver_->tryPrefetchResolverAsync(
931 index_ + 1, *principal_, skipCurrentProcess_, sra_, mcc_, serviceToken_);
944 bool skipCurrentProcess_;
950 bool iSkipCurrentProcess,
951 std::exception_ptr iExceptPtr)
const {
953 auto k = lookupProcessOrder[iProcessingIndex];
955 setCache(iSkipCurrentProcess,
k, iExceptPtr);
960 bool iSkipCurrentProcess)
const {
962 auto k = lookupProcessOrder[iProcessingIndex];
966 setCache(iSkipCurrentProcess, k,
nullptr);
974 bool skipCurrentProcess,
979 auto index = iProcessingIndex;
981 const unsigned int choiceSize =
ambiguous_.size();
983 while (
index < lookupProcessOrder.size()) {
984 auto k = lookupProcessOrder[
index];
996 auto task =
new (tbb::task::allocate_root())
997 TryNextResolverWaitingTask(
this,
index, &principal, sra, mcc, skipCurrentProcess, token);
998 task->increment_ref_count();
1004 productResolver->
prefetchAsync(
task, principal, skipCurrentProcess, token, sra, mcc);
1005 if (0 ==
task->decrement_ref_count()) {
1006 tbb::task::spawn(*
task);
1013 setCache(skipCurrentProcess, newCacheIndex,
nullptr);
1040 <<
"NoProcessProductResolver::unscheduledWasNotRun_() not implemented and should never be called.\n" 1041 <<
"Contact a Framework developer\n";
1046 <<
"NoProcessProductResolver::productUnavailable_() not implemented and should never be called.\n" 1047 <<
"Contact a Framework developer\n";
1052 <<
"NoProcessProductResolver::productResolved_() not implemented and should never be called.\n" 1053 <<
"Contact a Framework developer\n";
1058 <<
"NoProcessProductResolver::productWasDeleted_() not implemented and should never be called.\n" 1059 <<
"Contact a Framework developer\n";
1064 <<
"NoProcessProductResolver::productWasFetchedAndIsValid_() not implemented and should never be called.\n" 1065 <<
"Contact a Framework developer\n";
1070 <<
"NoProcessProductResolver::putProduct_() not implemented and should never be called.\n" 1071 <<
"Contact a Framework developer\n";
1077 <<
"NoProcessProductResolver::putOrMergeProduct_(std::unique_ptr<WrapperBase> edp, MergeableRunProductMetadata " 1078 "const*) not implemented and should never be called.\n" 1079 <<
"Contact a Framework developer\n";
1084 <<
"NoProcessProductResolver::branchDescription_() not implemented and should never be called.\n" 1085 <<
"Contact a Framework developer\n";
1090 <<
"NoProcessProductResolver::resetBranchDescription_() not implemented and should never be called.\n" 1091 <<
"Contact a Framework developer\n";
1096 <<
"NoProcessProductResolver::provenance_() not implemented and should never be called.\n" 1097 <<
"Contact a Framework developer\n";
1102 <<
"NoProcessProductResolver::connectTo() not implemented and should never be called.\n" 1103 <<
"Contact a Framework developer\n";
1109 bool skipCurrentProcess,
1115 ->resolveProduct(principal, skipCurrentProcess, sra, mcc);
1120 bool skipCurrentProcess,
1125 ->prefetchAsync(waitTask, principal, skipCurrentProcess, token, sra, mcc);
1142 <<
"SingleChoiceNoProcessProductResolver::unscheduledWasNotRun_() not implemented and should never be called.\n" 1143 <<
"Contact a Framework developer\n";
1148 <<
"SingleChoiceNoProcessProductResolver::productUnavailable_() not implemented and should never be called.\n" 1149 <<
"Contact a Framework developer\n";
1154 <<
"SingleChoiceNoProcessProductResolver::productResolved_() not implemented and should never be called.\n" 1155 <<
"Contact a Framework developer\n";
1160 <<
"SingleChoiceNoProcessProductResolver::productWasDeleted_() not implemented and should never be called.\n" 1161 <<
"Contact a Framework developer\n";
1166 "implemented and should never be called.\n" 1167 <<
"Contact a Framework developer\n";
1172 <<
"SingleChoiceNoProcessProductResolver::putProduct_() not implemented and should never be called.\n" 1173 <<
"Contact a Framework developer\n";
1179 <<
"SingleChoiceNoProcessProductResolver::putOrMergeProduct_(std::unique_ptr<WrapperBase> edp, " 1180 "MergeableRunProductMetadata const*) not implemented and should never be called.\n" 1181 <<
"Contact a Framework developer\n";
1186 <<
"SingleChoiceNoProcessProductResolver::branchDescription_() not implemented and should never be called.\n" 1187 <<
"Contact a Framework developer\n";
1192 "implemented and should never be called.\n" 1193 <<
"Contact a Framework developer\n";
1198 <<
"SingleChoiceNoProcessProductResolver::provenance_() not implemented and should never be called.\n" 1199 <<
"Contact a Framework developer\n";
1204 <<
"SingleChoiceNoProcessProductResolver::connectTo() not implemented and should never be called.\n" 1205 <<
"Contact a Framework developer\n";
static constexpr const ProductStatus defaultStatus_
void connectTo(ProductResolverBase const &iOther, Principal const *) final
Resolution resolveProduct_(Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
std::string const & branchName() const
std::string const & productInstanceName() const
bool isAtEndTransition() const
std::unique_ptr< T, F > make_sentry(T *iObject, F iFunc)
NOTE: if iObject is null, then iFunc will not be called.
DataManagingOrAliasProductResolver const & realProduct() const
void resetProductData_(bool deleteEarly) override
Provenance const * provenance() const
BranchType const & branchType() const
void callWhenDoneAsync(WaitingTask *task)
WaitingTaskList & waitingTasks() const
void setProcessHistory(ProcessHistory const &ph)
unsigned int ProductResolverIndex
StreamContext const * getStreamContext() const
void prefetchAsync_(WaitingTask *waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const final
SwitchBaseProductResolver(std::shared_ptr< BranchDescription const > bd, DataManagingOrAliasProductResolver &realProduct)
void checkType(WrapperBase const &prod) const
std::type_info const & dynamicTypeInfo() const
Resolution resolveProductImpl(Resolution) const
void prefetchAsync(WaitingTask *waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const
unsigned int unsetIndexValue() const
void putOrMergeProduct_(std::unique_ptr< WrapperBase > edp, MergeableRunProductMetadata const *mergeableRunProductMetadata) const final
ProductStatus const defaultStatus_
void resetProductData_(bool deleteEarly) override
void prefetchAsync_(WaitingTask *waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
FunctorTask< F > * make_functor_task(ALLOC &&iAlloc, F f)
void setCache(bool skipCurrentProcess, ProductResolverIndex index, std::exception_ptr exceptionPtr) const
void add(WaitingTask *)
Adds task to the waiting list.
std::atomic< bool > prefetchRequested_
void setProcessHistory_(ProcessHistory const &ph) override
ProductStatus status() const
bool singleProduct_() const override
bool unscheduledWasNotRun_() const override
void putProduct_(std::unique_ptr< WrapperBase > edp) const override
void throwNullRealProduct() const
Resolution tryResolver(unsigned int index, Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const
WrapperBase const * wrapper() const
WaitingTaskList skippingWaitingTasks_
std::string const & processName() const
bool singleProduct_() const override
void resetProductData_(bool deleteEarly) override
ProductProvenance const * productProvenance() const
void push(T &&iAction)
asynchronously pushes functor iAction into queue
Resolution resolveProduct(Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const
void setProvenance_(ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid) override
void reset()
Resets access to the resource so that added tasks will wait.
void setProvenance_(ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid) final
Resolution resolveProduct_(Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const final
void connectTo(ProductResolverBase const &iOther, Principal const *) final
BranchDescription const & branchDescription_() const override
bool productResolved_() const final
static unsigned int kUnsetOffset
bool productUnavailable() const
void resetProductData_(bool deleteEarly) override
std::string const & processName() const
void putOrMergeProduct_(std::unique_ptr< WrapperBase > prod, MergeableRunProductMetadata const *mergeableRunProductMetadata) const final
WaitingTaskList waitingTasks_
void connectTo(ProductResolverBase const &, Principal const *) final
ParentContext const & parent() const
void putProduct_(std::unique_ptr< WrapperBase > edp) const final
DataManagingOrAliasProductResolver & realProduct_
void putProduct_(std::unique_ptr< WrapperBase > edp) const override
void putOrMergeProduct(std::unique_ptr< WrapperBase > edp, MergeableRunProductMetadata const *mergeableRunProductMetadata=nullptr) const
void throwProductDeletedException() const
bool productUnavailable_() const final
ProductProvenance const * productProvenancePtr_() const override
static unsigned int kMissingOffset
bool productResolved_() const final
void putProduct(std::unique_ptr< WrapperBase > edp) const
WaitingTaskList waitingTasks_
void unsafe_setWrapper(std::unique_ptr< WrapperBase > iValue) const
void prefetchFailed(unsigned int iProcessingIndex, Principal const &principal, bool iSkipCurrentProcess, std::exception_ptr iExceptPtr) const
void doneWaiting(std::exception_ptr iPtr)
Signals that the resource is now available and tasks should be spawned.
void setParents(std::vector< BranchID > const &parents)
void setProvenance(ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid)
void setupUnscheduled(UnscheduledConfigurator const &iConfigure) final
void mergeProduct(std::unique_ptr< WrapperBase > edp, MergeableRunProductMetadata const *) const
std::string const & className() const
std::atomic< unsigned int > lastSkipCurrentCheckIndex_
bool singleProduct_() const override
bool productWasDeleted() const
Provenance const * provenance_() const override
std::vector< unsigned int > const & lookupProcessOrder() const
The Signals That Services Can Subscribe To This is based on ActivityRegistry and is current per Services can connect to the signals distributed by the ActivityRegistry in order to monitor the activity of the application Each possible callback has some defined which we here list in angle e< void, edm::EventID const &, edm::Timestamp const & > We also list in braces which AR_WATCH_USING_METHOD_ is used for those or
void putOrMergeProduct_(std::unique_ptr< WrapperBase > prod, MergeableRunProductMetadata const *mergeableRunProductMetadata) const final
Provenance const * provenance_() const override
bool dataValidFromResolver(unsigned int iProcessingIndex, Principal const &principal, bool iSkipCurrentProcess) const
void setProvenance_(ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid) override
TypeID unwrappedTypeID() const
void resetProductData_(bool deleteEarly) override
BranchDescription const & branchDescription_() const final
void setProcessHistory_(ProcessHistory const &ph) override
ProductProvenance const * productProvenancePtr_() const override
bool isFromCurrentProcess() const final
bool productUnavailable_() const override
void resetBranchDescription_(std::shared_ptr< BranchDescription const > bd) override
ProductProvenance const * productProvenancePtr_() const override
std::atomic< bool > prefetchRequested_
ProductProvenance const * productProvenancePtr_() const override
bool productResolved_() const final
std::atomic< unsigned int > lastCheckIndex_
TypeWithDict const & unwrappedType() const
void setProvenance_(ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid) final
How EventSelector::AcceptEvent() decides whether to accept an event for output otherwise it is excluding the probing of A single or multiple positive and the trigger will pass if any such matching triggers are PASS or EXCEPTION[A criterion thatmatches no triggers at all is detected and causes a throw.] A single negative with an expectation of appropriate bit checking in the decision and the trigger will pass if any such matching triggers are FAIL or EXCEPTION A wildcarded negative criterion that matches more than one trigger in the trigger but the state exists so we define the behavior If all triggers are the negative crieriion will lead to accepting the event(this again matches the behavior of"!*"before the partial wildcard feature was incorporated).The per-event"cost"of each negative criterion with multiple relevant triggers is about the same as!*was in the past
void setMergeableRunProductMetadata(MergeableRunProductMetadataBase const *mrpm)
std::type_info const & unvalidatedTypeInfo() const
bool productWasDeleted_() const override
void putProduct_(std::unique_ptr< WrapperBase > edp) const override
virtual bool isFromCurrentProcess() const =0
void prefetchAsync_(WaitingTask *waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const final
Resolution resolveProduct_(Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
Resolution resolveProduct_(Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const final
void setProduct(std::unique_ptr< WrapperBase > edp) const
void prefetchAsync_(WaitingTask *waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
NoProcessProductResolver(std::vector< ProductResolverIndex > const &matchingHolders, std::vector< bool > const &ambiguous, bool madeAtEnd)
std::atomic< bool > skippingPrefetchRequested_
std::atomic< ProductStatus > theStatus_
std::vector< bool > ambiguous_
Resolution resolveProductImpl(FUNC resolver) const
DelayedReader * reader() const
void putOrMergeProduct_(std::unique_ptr< WrapperBase > prod, MergeableRunProductMetadata const *mergeableRunProductMetadata) const final
UnscheduledAuxiliary const * auxiliary() const
void resetProductData_(bool deleteEarly) override
void insertIntoSet(ProductProvenance provenanceProduct) const
static unsigned int kAmbiguousOffset
void setupUnscheduled(UnscheduledConfigurator const &) final
ProductStatus defaultStatus() const
ProductData const * data() const
bool productWasFetchedAndIsValid(bool iSkipCurrentProcess) const
void setMergeableRunProductMetadataInProductData(MergeableRunProductMetadata const *)
bool productWasFetchedAndIsValid_(bool iSkipCurrentProcess) const final
void setProcessHistory_(ProcessHistory const &ph) override
BranchDescription const & branchDescription_() const override
bool productWasDeleted_() const final
FunctorWaitingTask< F > * make_waiting_task(ALLOC &&iAlloc, F f)
void resetProductData_(bool deleteEarly) final
void putOrMergeProduct_(std::unique_ptr< WrapperBase > prod, MergeableRunProductMetadata const *mergeableRunProductMetadata) const final
bool productUnavailable_() const override
void addContext(std::string const &context)
bool singleProduct_() const override
std::vector< ProductResolverIndex > matchingHolders_
void resetBranchDescription_(std::shared_ptr< BranchDescription const > bd) override
void tryPrefetchResolverAsync(unsigned int iProcessingIndex, Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc, ServiceToken token) const
BranchDescription const & branchDescription() const
Worker * findWorker(std::string const &iLabel) const
ConstProductResolverPtr getProductResolverByIndex(ProductResolverIndex const &oid) const
ProductProvenance const * productProvenancePtr_() const final
void putProduct_(std::unique_ptr< WrapperBase > edp) const override
bool productResolved_() const final
void setProvenance_(ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid) override
void setProcessHistory_(ProcessHistory const &ph) override
void putProduct_(std::unique_ptr< WrapperBase > edp) const override
void setupUnscheduled(UnscheduledConfigurator const &) final
bool productResolved() const
ProductData const & getProductData() const
void resetProductData_(bool deleteEarly) override=0
BranchID const & originalBranchID() const
bool productWasDeleted_() const override
Resolution resolveProduct_(Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
bool productWasFetchedAndIsValid_(bool iSkipCurrentProcess) const override
void resetProductData_(bool deleteEarly) override
std::atomic< bool > & prefetchRequested() const
bool productUnavailable_() const final
bool unscheduledWasNotRun_() const override
bool singleProduct_() const final
Resolution resolveProduct_(Principal const &principal, bool skipCurrentProcess, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
static ParentageRegistry * instance()
void setProvenance_(ProductProvenanceRetriever const *provRetriever, ProcessHistory const &ph, ProductID const &pid) override
static Resolution makeAmbiguous()
void setProcessHistory_(ProcessHistory const &ph) final
bool productWasFetchedAndIsValid_(bool iSkipCurrentProcess) const override
bool insertMapped(value_type const &v)
std::shared_ptr< WrapperBase const > sharedConstWrapper() const
void prefetchAsync_(WaitingTask *waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
void putOrMergeProduct_(std::unique_ptr< WrapperBase > prod, MergeableRunProductMetadata const *mergeableRunProductMetadata) const final
def branchType(schema, name)
std::string const & moduleLabel() const
ProductStatus status() const
WrapperBase * unsafe_wrapper() const
static HepMC::HEPEVT_Wrapper wrapper
void putProduct_(std::unique_ptr< WrapperBase > edp) const override
void prefetchAsync_(WaitingTask *waitTask, Principal const &principal, bool skipCurrentProcess, ServiceToken const &token, SharedResourcesAcquirer *sra, ModuleCallingContext const *mcc) const override
void setFailedStatus() const
void connectTo(ProductResolverBase const &iOther, Principal const *iParentPrincipal) final