Base/src/SecondaryEventProvider.cc

0001 #include "FWCore/Concurrency/interface/include_first_syncWait.h"
0002 #include "Mixing/Base/src/SecondaryEventProvider.h"
0003 #include "FWCore/Common/interface/ProcessBlockHelper.h"
0004 #include "FWCore/Framework/interface/ExceptionActions.h"
0005 #include "FWCore/Framework/interface/PreallocationConfiguration.h"
0006 #include "FWCore/Framework/interface/TransitionInfoTypes.h"
0007 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0008 #include "FWCore/Utilities/interface/StreamID.h"
0009 #include "DataFormats/Provenance/interface/ProductRegistry.h"
0010 #include "oneapi/tbb/task_arena.h"
0011
0012 namespace {
0013   template <typename T, typename U>
0014   void processOneOccurrence(edm::WorkerManager& manager,
0015                             typename T::TransitionInfoType& info,
0016                             edm::StreamID streamID,
0017                             typename T::Context const* topContext,
0018                             U const* context,
0019                             bool cleaningUpAfterException = false) {
0020     manager.resetAll();
0021
0022     if (manager.allWorkers().empty())
0023       return;
0024
0025     auto token = edm::ServiceRegistry::instance().presentToken();
0026     //we need the arena to guarantee that the syncWait will return to this thread
0027     // and not cause this callstack to possibly be moved to a new thread
0028     tbb::task_arena localArena{tbb::this_task_arena::max_concurrency()};
0029     std::exception_ptr exceptPtr = localArena.execute([&]() {
0030       return edm::syncWait([&](edm::WaitingTaskHolder&& iHolder) {
0031         manager.processOneOccurrenceAsync<T, U>(std::move(iHolder), info, token, streamID, topContext, context);
0032       });
0033     });
0034
0035     if (exceptPtr) {
0036       try {
0037         edm::convertException::wrap([&]() { std::rethrow_exception(exceptPtr); });
0038       } catch (cms::Exception& ex) {
0039         if (ex.context().empty()) {
0040           edm::addContextAndPrintException("Calling SecondaryEventProvider", ex, cleaningUpAfterException);
0041         } else {
0042           edm::addContextAndPrintException("", ex, cleaningUpAfterException);
0043         }
0044         throw;
0045       }
0046     }
0047   }
0048 }  // namespace
0049
0050 namespace edm {
0051   SecondaryEventProvider::SecondaryEventProvider(std::vector<ParameterSet>& psets,
0052                                                  ProductRegistry& preg,
0053                                                  std::shared_ptr<ProcessConfiguration> processConfiguration)
0054       : exceptionToActionTable_(new ExceptionToActionTable),
0055         workerManager_(std::make_shared<ActivityRegistry>(), *exceptionToActionTable_) {
0056     std::vector<std::string> shouldBeUsedLabels;
0057     std::set<std::string> unscheduledLabels;
0058     const PreallocationConfiguration preallocConfig;
0059     for (auto& pset : psets) {
0060       std::string label = pset.getParameter<std::string>("@module_label");
0061       workerManager_.addToUnscheduledWorkers(
0062           pset, preg, &preallocConfig, processConfiguration, label, unscheduledLabels, shouldBeUsedLabels);
0063     }
0064     if (!unscheduledLabels.empty()) {
0065       preg.setUnscheduledProducts(unscheduledLabels);
0066     }
0067   }  // SecondaryEventProvider::SecondaryEventProvider
0068
0069   void SecondaryEventProvider::beginJob(ProductRegistry const& iRegistry,
0070                                         eventsetup::ESRecordsToProxyIndices const& iIndices) {
0071     ProcessBlockHelper dummyProcessBlockHelper;
0072     workerManager_.beginJob(iRegistry, iIndices, dummyProcessBlockHelper);
0073   }
0074
0075   //NOTE: When the Stream interfaces are propagated to the modules, this code must be updated
0076   // to also send the stream based transitions
0077   void SecondaryEventProvider::beginRun(RunPrincipal& run,
0078                                         const EventSetupImpl& setup,
0079                                         ModuleCallingContext const* mcc,
0080                                         StreamContext& sContext) {
0081     RunTransitionInfo info(run, setup);
0082     processOneOccurrence<OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin> >(
0083         workerManager_, info, StreamID::invalidStreamID(), nullptr, mcc);
0084     processOneOccurrence<OccurrenceTraits<RunPrincipal, BranchActionStreamBegin> >(
0085         workerManager_, info, sContext.streamID(), &sContext, mcc);
0086   }
0087
0088   void SecondaryEventProvider::beginLuminosityBlock(LuminosityBlockPrincipal& lumi,
0089                                                     const EventSetupImpl& setup,
0090                                                     ModuleCallingContext const* mcc,
0091                                                     StreamContext& sContext) {
0092     LumiTransitionInfo info(lumi, setup);
0093     processOneOccurrence<OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin> >(
0094         workerManager_, info, StreamID::invalidStreamID(), nullptr, mcc);
0095     processOneOccurrence<OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin> >(
0096         workerManager_, info, sContext.streamID(), &sContext, mcc);
0097   }
0098
0099   void SecondaryEventProvider::endRun(RunPrincipal& run,
0100                                       const EventSetupImpl& setup,
0101                                       ModuleCallingContext const* mcc,
0102                                       StreamContext& sContext) {
0103     RunTransitionInfo info(run, setup);
0104     processOneOccurrence<OccurrenceTraits<RunPrincipal, BranchActionStreamEnd> >(
0105         workerManager_, info, sContext.streamID(), &sContext, mcc);
0106     processOneOccurrence<OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd> >(
0107         workerManager_, info, StreamID::invalidStreamID(), nullptr, mcc);
0108   }
0109
0110   void SecondaryEventProvider::endLuminosityBlock(LuminosityBlockPrincipal& lumi,
0111                                                   const EventSetupImpl& setup,
0112                                                   ModuleCallingContext const* mcc,
0113                                                   StreamContext& sContext) {
0114     LumiTransitionInfo info(lumi, setup);
0115     processOneOccurrence<OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd> >(
0116         workerManager_, info, sContext.streamID(), &sContext, mcc);
0117     processOneOccurrence<OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd> >(
0118         workerManager_, info, StreamID::invalidStreamID(), nullptr, mcc);
0119   }
0120
0121   void SecondaryEventProvider::setupPileUpEvent(EventPrincipal& ep,
0122                                                 const EventSetupImpl& setup,
0123                                                 StreamContext& sContext) {
0124     workerManager_.setupResolvers(ep);
0125     EventTransitionInfo info(ep, setup);
0126     workerManager_.setupOnDemandSystem(info);
0127   }
0128   void SecondaryEventProvider::beginStream(edm::StreamID iID, StreamContext& sContext) {
0129     workerManager_.beginStream(iID, sContext);
0130   }
0131
0132   void SecondaryEventProvider::endStream(edm::StreamID iID, StreamContext& sContext) {
0133     workerManager_.endStream(iID, sContext);
0134   }
0135 }  // namespace edm