Framework/interface/Callback.h

0001 #ifndef FWCore_Framework_Callback_h
0002 #define FWCore_Framework_Callback_h
0003 // -*- C++ -*-
0004 //
0005 // Package:     Framework
0006 // Class  :     Callback
0007 //
0008 /**\class edm::eventsetup::Callback
0009
0010  Description: Functional object used as the 'callback' for the CallbackProxy
0011
0012  Usage:
0013     <usage>
0014
0015 */
0016 //
0017 // Author:      Chris Jones
0018 // Created:     Sun Apr 17 14:30:24 EDT 2005
0019 //
0020
0021 // system include files
0022 #include <array>
0023 #include <vector>
0024 #include <type_traits>
0025 #include <atomic>
0026 // user include files
0027 #include "FWCore/Framework/interface/produce_helpers.h"
0028 #include "FWCore/Framework/interface/EventSetupImpl.h"
0029 #include "FWCore/Utilities/interface/propagate_const.h"
0030 #include "FWCore/Utilities/interface/ESIndices.h"
0031 #include "FWCore/Utilities/interface/ConvertException.h"
0032 #include "FWCore/Concurrency/interface/WaitingTaskList.h"
0033 #include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
0034 #include "FWCore/ServiceRegistry/interface/ServiceToken.h"
0035 #include "FWCore/ServiceRegistry/interface/ServiceRegistry.h"
0036 #include "FWCore/ServiceRegistry/interface/ESParentContext.h"
0037 #include "FWCore/ServiceRegistry/interface/ESModuleCallingContext.h"
0038
0039 namespace edm {
0040   void exceptionContext(cms::Exception&, ESModuleCallingContext const&);
0041
0042   namespace eventsetup {
0043     class EventSetupRecordImpl;
0044
0045     // The default decorator that does nothing
0046     template <typename TRecord>
0047     struct CallbackSimpleDecorator {
0048       void pre(const TRecord&) {}
0049       void post(const TRecord&) {}
0050     };
0051
0052     template <typename T,          //producer's type
0053               typename TReturn,    //return type of the producer's method
0054               typename TRecord,    //the record passed in as an argument
0055               typename TDecorator  //allows customization using pre/post calls
0056               = CallbackSimpleDecorator<TRecord>>
0057     class Callback {
0058     public:
0059       using method_type = TReturn (T ::*)(const TRecord&);
0060
0061       Callback(T* iProd, method_type iMethod, unsigned int iID, const TDecorator& iDec = TDecorator())
0062           : proxyData_{},
0063             producer_(iProd),
0064             callingContext_(&iProd->description()),
0065             method_(iMethod),
0066             id_(iID),
0067             wasCalledForThisRecord_(false),
0068             decorator_(iDec) {}
0069
0070       Callback* clone() { return new Callback(producer_.get(), method_, id_, decorator_); }
0071
0072       Callback(const Callback&) = delete;
0073       const Callback& operator=(const Callback&) = delete;
0074
0075       void prefetchAsync(WaitingTaskHolder iTask,
0076                          EventSetupRecordImpl const* iRecord,
0077                          EventSetupImpl const* iEventSetupImpl,
0078                          ServiceToken const& token,
0079                          ESParentContext const& iParent) {
0080         bool expected = false;
0081         auto doPrefetch = wasCalledForThisRecord_.compare_exchange_strong(expected, true);
0082         taskList_.add(iTask);
0083         auto group = iTask.group();
0084         if (doPrefetch) {
0085           callingContext_.setContext(ESModuleCallingContext::State::kPrefetching, iParent);
0086           iRecord->activityRegistry()->preESModulePrefetchingSignal_.emit(iRecord->key(), callingContext_);
0087           if UNLIKELY (producer_->hasMayConsumes()) {
0088             //after prefetching need to do the mayGet
0089             ServiceWeakToken weakToken = token;
0090             auto mayGetTask = edm::make_waiting_task(
0091                 [this, iRecord, iEventSetupImpl, weakToken, group](std::exception_ptr const* iExcept) {
0092                   if (iExcept) {
0093                     runProducerAsync(group, iExcept, iRecord, iEventSetupImpl, weakToken.lock());
0094                     return;
0095                   }
0096                   if (handleMayGet(iRecord, iEventSetupImpl)) {
0097                     auto runTask = edm::make_waiting_task(
0098                         [this, group, iRecord, iEventSetupImpl, weakToken](std::exception_ptr const* iExcept) {
0099                           runProducerAsync(group, iExcept, iRecord, iEventSetupImpl, weakToken.lock());
0100                         });
0101                     prefetchNeededDataAsync(WaitingTaskHolder(*group, runTask),
0102                                             iEventSetupImpl,
0103                                             &((*postMayGetProxies_).front()),
0104                                             weakToken.lock());
0105                   } else {
0106                     runProducerAsync(group, iExcept, iRecord, iEventSetupImpl, weakToken.lock());
0107                   }
0108                 });
0109
0110             //Get everything we can before knowing about the mayGets
0111             prefetchNeededDataAsync(WaitingTaskHolder(*group, mayGetTask), iEventSetupImpl, getTokenIndices(), token);
0112           } else {
0113             ServiceWeakToken weakToken = token;
0114             auto task = edm::make_waiting_task(
0115                 [this, group, iRecord, iEventSetupImpl, weakToken](std::exception_ptr const* iExcept) {
0116                   runProducerAsync(group, iExcept, iRecord, iEventSetupImpl, weakToken.lock());
0117                 });
0118             prefetchNeededDataAsync(WaitingTaskHolder(*group, task), iEventSetupImpl, getTokenIndices(), token);
0119           }
0120         }
0121       }
0122
0123       template <class DataT>
0124       void holdOntoPointer(DataT* iData) {
0125         proxyData_[produce::find_index<TReturn, DataT>::value] = iData;
0126       }
0127
0128       void storeReturnedValues(TReturn iReturn) {
0129         using type = typename produce::product_traits<TReturn>::type;
0130         setData<typename type::head_type, typename type::tail_type>(iReturn);
0131       }
0132
0133       template <class RemainingContainerT, class DataT, class ProductsT>
0134       void setData(ProductsT& iProducts) {
0135         DataT* temp = reinterpret_cast<DataT*>(proxyData_[produce::find_index<TReturn, DataT>::value]);
0136         if (nullptr != temp) {
0137           moveFromTo(iProducts, *temp);
0138         }
0139         if constexpr (not std::is_same_v<produce::Null, RemainingContainerT>) {
0140           setData<typename RemainingContainerT::head_type, typename RemainingContainerT::tail_type>(iProducts);
0141         }
0142       }
0143       void newRecordComing() {
0144         wasCalledForThisRecord_ = false;
0145         taskList_.reset();
0146       }
0147
0148       unsigned int transitionID() const { return id_; }
0149       ESProxyIndex const* getTokenIndices() const { return producer_->getTokenIndices(id_); }
0150
0151     private:
0152       void prefetchNeededDataAsync(WaitingTaskHolder task,
0153                                    EventSetupImpl const* iImpl,
0154                                    ESProxyIndex const* proxies,
0155                                    edm::ServiceToken const& token) const {
0156         auto recs = producer_->getTokenRecordIndices(id_);
0157         auto n = producer_->numberOfTokenIndices(id_);
0158         for (size_t i = 0; i != n; ++i) {
0159           auto rec = iImpl->findImpl(recs[i]);
0160           if (rec) {
0161             rec->prefetchAsync(task, proxies[i], iImpl, token, edm::ESParentContext{&callingContext_});
0162           }
0163         }
0164       }
0165
0166       bool handleMayGet(EventSetupRecordImpl const* iRecord, EventSetupImpl const* iEventSetupImpl) {
0167         //Handle mayGets
0168         TRecord rec;
0169         edm::ESParentContext pc{&callingContext_};
0170         rec.setImpl(iRecord, transitionID(), getTokenIndices(), iEventSetupImpl, &pc);
0171         postMayGetProxies_ = producer_->updateFromMayConsumes(id_, rec);
0172         return static_cast<bool>(postMayGetProxies_);
0173       }
0174
0175       void runProducerAsync(oneapi::tbb::task_group* iGroup,
0176                             std::exception_ptr const* iExcept,
0177                             EventSetupRecordImpl const* iRecord,
0178                             EventSetupImpl const* iEventSetupImpl,
0179                             ServiceToken const& token) {
0180         if (iExcept) {
0181           //The cache held by the CallbackProxy was already set to invalid at the beginning of the IOV
0182           taskList_.doneWaiting(*iExcept);
0183           return;
0184         }
0185         iRecord->activityRegistry()->postESModulePrefetchingSignal_.emit(iRecord->key(), callingContext_);
0186         ServiceWeakToken weakToken = token;
0187         producer_->queue().push(*iGroup, [this, iRecord, iEventSetupImpl, weakToken]() {
0188           callingContext_.setState(ESModuleCallingContext::State::kRunning);
0189           std::exception_ptr exceptPtr;
0190           try {
0191             convertException::wrap([this, iRecord, iEventSetupImpl, weakToken] {
0192               auto proxies = getTokenIndices();
0193               if (postMayGetProxies_) {
0194                 proxies = &((*postMayGetProxies_).front());
0195               }
0196               TRecord rec;
0197               edm::ESParentContext pc{&callingContext_};
0198               rec.setImpl(iRecord, transitionID(), proxies, iEventSetupImpl, &pc);
0199               ServiceRegistry::Operate operate(weakToken.lock());
0200               iRecord->activityRegistry()->preESModuleSignal_.emit(iRecord->key(), callingContext_);
0201               struct EndGuard {
0202                 EndGuard(EventSetupRecordImpl const* iRecord, ESModuleCallingContext const& iContext)
0203                     : record_{iRecord}, context_{iContext} {}
0204                 ~EndGuard() { record_->activityRegistry()->postESModuleSignal_.emit(record_->key(), context_); }
0205                 EventSetupRecordImpl const* record_;
0206                 ESModuleCallingContext const& context_;
0207               };
0208               EndGuard guard(iRecord, callingContext_);
0209               decorator_.pre(rec);
0210               storeReturnedValues((producer_->*method_)(rec));
0211               decorator_.post(rec);
0212             });
0213           } catch (cms::Exception& iException) {
0214             edm::exceptionContext(iException, callingContext_);
0215             exceptPtr = std::current_exception();
0216           }
0217           taskList_.doneWaiting(exceptPtr);
0218         });
0219       }
0220
0221       std::array<void*, produce::size<TReturn>::value> proxyData_;
0222       std::optional<std::vector<ESProxyIndex>> postMayGetProxies_;
0223       edm::propagate_const<T*> producer_;
0224       ESModuleCallingContext callingContext_;
0225       edm::WaitingTaskList taskList_;
0226       method_type method_;
0227       // This transition id identifies which setWhatProduced call this Callback is associated with
0228       const unsigned int id_;
0229       std::atomic<bool> wasCalledForThisRecord_;
0230       TDecorator decorator_;
0231     };
0232   }  // namespace eventsetup
0233 }  // namespace edm
0234
0235 #endif