Framework/src/TransformerBase.cc

0001 #include "FWCore/Framework/interface/TransformerBase.h"
0002 #include "FWCore/Framework/interface/ProducerBase.h"
0003 #include "FWCore/Framework/interface/EventForTransformer.h"
0004 #include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
0005 #include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
0006 #include "DataFormats/Provenance/interface/ProductResolverIndexHelper.h"
0007 #include "DataFormats/Provenance/interface/ProductDescription.h"
0008 #include "DataFormats/Provenance/interface/ModuleDescription.h"
0009
0010 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
0011 #include "FWCore/ServiceRegistry/interface/ModuleCallingContext.h"
0012 #include "FWCore/ServiceRegistry/interface/StreamContext.h"
0013
0014 #include <optional>
0015
0016 namespace {
0017   class TransformSignalSentry {
0018   public:
0019     TransformSignalSentry(edm::ActivityRegistry* a, edm::StreamContext const& sc, edm::ModuleCallingContext const& mcc)
0020         : a_(a), sc_(sc), mcc_(mcc) {
0021       if (a_)
0022         a_->preModuleTransformSignal_(sc_, mcc_);
0023     }
0024     ~TransformSignalSentry() {
0025       if (a_)
0026         a_->postModuleTransformSignal_(sc_, mcc_);
0027     }
0028
0029   private:
0030     edm::ActivityRegistry* a_;  // We do not use propagate_const because the registry itself is mutable.
0031     edm::StreamContext const& sc_;
0032     edm::ModuleCallingContext const& mcc_;
0033   };
0034
0035   class TransformAcquiringSignalSentry {
0036   public:
0037     TransformAcquiringSignalSentry(edm::ActivityRegistry* a,
0038                                    edm::StreamContext const& sc,
0039                                    edm::ModuleCallingContext const& mcc)
0040         : a_(a), sc_(sc), mcc_(mcc) {
0041       if (a_)
0042         a_->preModuleTransformAcquiringSignal_(sc_, mcc_);
0043     }
0044     ~TransformAcquiringSignalSentry() {
0045       if (a_)
0046         a_->postModuleTransformAcquiringSignal_(sc_, mcc_);
0047     }
0048
0049   private:
0050     edm::ActivityRegistry* a_;  // We do not use propagate_const because the registry itself is mutable.
0051     edm::StreamContext const& sc_;
0052     edm::ModuleCallingContext const& mcc_;
0053   };
0054
0055 }  // namespace
0056
0057 namespace edm {
0058   void TransformerBase::registerTransformImp(
0059       ProducerBase& iBase, EDPutToken iToken, const TypeID& id, std::string instanceName, TransformFunction iFunc) {
0060     auto transformPut = iBase.transforms(id, std::move(instanceName));
0061     PreTransformFunction ptf;
0062     transformInfo_.emplace_back(iToken.index(), id, transformPut, std::move(ptf), std::move(iFunc));
0063   }
0064
0065   void TransformerBase::registerTransformAsyncImp(ProducerBase& iBase,
0066                                                   EDPutToken iToken,
0067                                                   const TypeID& id,
0068                                                   std::string instanceName,
0069                                                   PreTransformFunction iPreFunc,
0070                                                   TransformFunction iFunc) {
0071     auto transformPut = iBase.transforms(id, std::move(instanceName));
0072     transformInfo_.emplace_back(iToken.index(), id, transformPut, std::move(iPreFunc), std::move(iFunc));
0073   }
0074
0075   std::size_t TransformerBase::findMatchingIndex(ProducerBase const& iBase,
0076                                                  edm::ProductDescription const& iBranch) const noexcept {
0077     auto const& list = iBase.typeLabelList();
0078
0079     std::size_t index = 0;
0080     bool found = false;
0081     for (auto const& element : list) {
0082       if (not element.isTransform_) {
0083         continue;
0084       }
0085       if (element.typeID_ == iBranch.unwrappedTypeID() &&
0086           element.productInstanceName_ == iBranch.productInstanceName()) {
0087         found = true;
0088         break;
0089       }
0090       ++index;
0091     }
0092     assert(found);
0093     return index;
0094   }
0095
0096   void TransformerBase::extendUpdateLookup(ProducerBase const& iBase,
0097                                            ModuleDescription const& iModuleDesc,
0098                                            ProductResolverIndexHelper const& iHelper) {
0099     auto const& list = iBase.typeLabelList();
0100
0101     for (auto it = transformInfo_.begin<0>(); it != transformInfo_.end<0>(); ++it) {
0102       auto const& putInfo = list[*it];
0103       *it = iHelper.index(PRODUCT_TYPE,
0104                           putInfo.typeID_,
0105                           iModuleDesc.moduleLabel().c_str(),
0106                           putInfo.productInstanceName_.c_str(),
0107                           iModuleDesc.processName().c_str());
0108     }
0109   }
0110
0111   void TransformerBase::transformImpAsync(edm::WaitingTaskHolder iHolder,
0112                                           std::size_t iIndex,
0113                                           edm::ActivityRegistry* iAct,
0114                                           ProducerBase const& iBase,
0115                                           edm::EventForTransformer& iEvent) const noexcept {
0116     auto const& mcc = iEvent.moduleCallingContext();
0117     if (transformInfo_.get<kPreTransform>(iIndex)) {
0118       std::optional<decltype(iEvent.get(transformInfo_.get<kType>(iIndex), transformInfo_.get<kResolverIndex>(iIndex)))>
0119           handle;
0120       //transform acquiring signal
0121       auto const& streamContext = *mcc.getStreamContext();
0122       TransformAcquiringSignalSentry sentry(iAct, streamContext, mcc);
0123       CMS_SA_ALLOW try {
0124         handle = iEvent.get(transformInfo_.get<kType>(iIndex), transformInfo_.get<kResolverIndex>(iIndex));
0125       } catch (...) {
0126         iHolder.doneWaiting(std::current_exception());
0127         return;
0128       }
0129       if (handle->wrapper()) {
0130         auto cache = std::make_shared<std::any>();
0131         auto nextTask =
0132             edm::make_waiting_task([holder = iHolder, cache, iIndex, this, &iBase, handle = *handle, iEvent, iAct](
0133                                        std::exception_ptr const* iPtr) mutable {
0134               if (iPtr) {
0135                 holder.doneWaiting(*iPtr);
0136               } else {
0137                 //transform signal
0138                 auto mcc = iEvent.moduleCallingContext();
0139                 auto const& streamContext = *mcc.getStreamContext();
0140                 TransformSignalSentry sentry(iAct, streamContext, mcc);
0141                 iEvent.put(iBase.putTokenIndexToProductResolverIndex()[transformInfo_.get<kToken>(iIndex).index()],
0142                            transformInfo_.get<kTransform>(iIndex)(streamContext.streamID(), std::move(*cache)),
0143                            handle);
0144               }
0145             });
0146         WaitingTaskHolder wth(*iHolder.group(), nextTask);
0147         CMS_SA_ALLOW try {
0148           // wth must be copied into wta below so that the
0149           // wth.doneWaiting() is called after the pre-transform
0150           // function has finished
0151           WaitingTaskWithArenaHolder wta(wth);
0152           *cache =
0153               transformInfo_.get<kPreTransform>(iIndex)(streamContext.streamID(), *(handle->wrapper()), std::move(wta));
0154         } catch (...) {
0155           wth.doneWaiting(std::current_exception());
0156         }
0157       }
0158     } else {
0159       CMS_SA_ALLOW try {
0160         auto handle = iEvent.get(transformInfo_.get<kType>(iIndex), transformInfo_.get<kResolverIndex>(iIndex));
0161
0162         if (handle.wrapper()) {
0163           std::any v = handle.wrapper();
0164           //transform signal
0165           auto const& streamContext = *mcc.getStreamContext();
0166           TransformSignalSentry sentry(iAct, streamContext, mcc);
0167           iEvent.put(iBase.putTokenIndexToProductResolverIndex()[transformInfo_.get<kToken>(iIndex).index()],
0168                      transformInfo_.get<kTransform>(iIndex)(streamContext.streamID(), std::move(v)),
0169                      handle);
0170         }
0171       } catch (...) {
0172         iHolder.doneWaiting(std::current_exception());
0173       }
0174     }
0175   }
0176
0177 }  // namespace edm