Framework/interface/SubProcess.h

0001 #ifndef FWCore_Framework_SubProcess_h
0002 #define FWCore_Framework_SubProcess_h
0003
0004 #include "DataFormats/Provenance/interface/BranchID.h"
0005 #include "FWCore/Common/interface/FWCoreCommonFwd.h"
0006 #include "FWCore/Framework/interface/EventSetupProvider.h"
0007 #include "FWCore/Framework/interface/EDConsumerBase.h"
0008 #include "FWCore/Framework/interface/PathsAndConsumesOfModules.h"
0009 #include "FWCore/Framework/interface/PrincipalCache.h"
0010 #include "FWCore/Framework/interface/Schedule.h"
0011 #include "FWCore/Framework/interface/TriggerResultsBasedEventSelector.h"
0012 #include "FWCore/Framework/interface/ProductSelectorRules.h"
0013 #include "FWCore/Framework/interface/ProductSelector.h"
0014 #include "FWCore/ServiceRegistry/interface/ProcessContext.h"
0015 #include "FWCore/ServiceRegistry/interface/ServiceLegacy.h"
0016 #include "FWCore/ServiceRegistry/interface/ServiceToken.h"
0017 #include "FWCore/Utilities/interface/Algorithms.h"
0018 #include "FWCore/Utilities/interface/BranchType.h"
0019 #include "FWCore/Utilities/interface/get_underlying_safe.h"
0020 #include "FWCore/Utilities/interface/propagate_const.h"
0021
0022 #include "DataFormats/Provenance/interface/SelectedProducts.h"
0023
0024 #include <map>
0025 #include <memory>
0026 #include <set>
0027 #include <vector>
0028
0029 namespace edm {
0030   class ActivityRegistry;
0031   class BranchDescription;
0032   class BranchIDListHelper;
0033   class EventPrincipal;
0034   class EventSetupImpl;
0035   class HistoryAppender;
0036   class LuminosityBlockPrincipal;
0037   class LumiTransitionInfo;
0038   class MergeableRunProductMetadata;
0039   class ModuleTypeResolverMaker;
0040   class ParameterSet;
0041   class Principal;
0042   class ProcessBlockTransitionInfo;
0043   class ProductRegistry;
0044   class PreallocationConfiguration;
0045   class RunTransitionInfo;
0046   class ThinnedAssociationsHelper;
0047   class SubProcessParentageHelper;
0048   class WaitingTaskHolder;
0049
0050   namespace eventsetup {
0051     class EventSetupsController;
0052   }
0053   class SubProcess : public EDConsumerBase {
0054   public:
0055     SubProcess(ParameterSet& parameterSet,
0056                ParameterSet const& topLevelParameterSet,
0057                std::shared_ptr<ProductRegistry const> parentProductRegistry,
0058                std::shared_ptr<BranchIDListHelper const> parentBranchIDListHelper,
0059                ProcessBlockHelperBase const& parentProcessBlockHelper,
0060                ThinnedAssociationsHelper const& parentThinnedAssociationsHelper,
0061                SubProcessParentageHelper const& parentSubProcessParentageHelper,
0062                eventsetup::EventSetupsController& esController,
0063                ActivityRegistry& parentActReg,
0064                ServiceToken const& token,
0065                serviceregistry::ServiceLegacy iLegacy,
0066                PreallocationConfiguration const& preallocConfig,
0067                ProcessContext const* parentProcessContext,
0068                ModuleTypeResolverMaker const* typeResolverMaker);
0069
0070     ~SubProcess() override;
0071
0072     SubProcess(SubProcess const&) = delete;             // Disallow copying
0073     SubProcess& operator=(SubProcess const&) = delete;  // Disallow copying
0074     SubProcess(SubProcess&&) = default;                 // Allow Moving
0075     SubProcess& operator=(SubProcess&&) = delete;       // Move not supported by PrincipalCache
0076
0077     //From OutputModule
0078     void selectProducts(ProductRegistry const& preg,
0079                         ThinnedAssociationsHelper const& parentThinnedAssociationsHelper,
0080                         std::map<BranchID, bool>& keepAssociation);
0081
0082     SelectedProductsForBranchType const& keptProducts() const { return keptProducts_; }
0083
0084     // Returns the set of modules whose products may be consumed by
0085     // modules in this SubProcess or its child SubProcesses
0086     std::vector<ModuleProcessName> keepOnlyConsumedUnscheduledModules(bool deleteModules);
0087
0088     void doBeginJob();
0089     void doEndJob();
0090
0091     void doEventAsync(WaitingTaskHolder iHolder,
0092                       EventPrincipal const& principal,
0093                       std::vector<std::shared_ptr<const EventSetupImpl>> const*);
0094
0095     template <typename Traits>
0096     void doBeginProcessBlockAsync(WaitingTaskHolder iHolder,
0097                                   ProcessBlockTransitionInfo const& iTransitionInfo,
0098                                   bool cleaningUpAfterException);
0099
0100     void doEndProcessBlockAsync(WaitingTaskHolder iHolder,
0101                                 ProcessBlockTransitionInfo const& iTransitionInfo,
0102                                 bool cleaningUpAfterException);
0103
0104     void doBeginRunAsync(WaitingTaskHolder iHolder, RunTransitionInfo const& iTransitionInfo);
0105
0106     void doEndRunAsync(WaitingTaskHolder iHolder,
0107                        RunTransitionInfo const& iTransitionInfo,
0108                        bool cleaningUpAfterException);
0109
0110     void doBeginLuminosityBlockAsync(WaitingTaskHolder iHolder, LumiTransitionInfo const& iTransitionInfo);
0111
0112     void doEndLuminosityBlockAsync(WaitingTaskHolder iHolder,
0113                                    LumiTransitionInfo const& iTransitionInfo,
0114                                    bool cleaningUpAfterException);
0115
0116     void doBeginStream(unsigned int);
0117     void doEndStream(unsigned int);
0118     void doStreamBeginRunAsync(WaitingTaskHolder iHolder, unsigned int iID, RunTransitionInfo const&);
0119
0120     void doStreamEndRunAsync(WaitingTaskHolder iHolder,
0121                              unsigned int iID,
0122                              RunTransitionInfo const&,
0123                              bool cleaningUpAfterException);
0124
0125     void doStreamBeginLuminosityBlockAsync(WaitingTaskHolder iHolder, unsigned int iID, LumiTransitionInfo const&);
0126
0127     void doStreamEndLuminosityBlockAsync(WaitingTaskHolder iHolder,
0128                                          unsigned int iID,
0129                                          LumiTransitionInfo const&,
0130                                          bool cleaningUpAfterException);
0131
0132     void writeLumiAsync(WaitingTaskHolder, LuminosityBlockPrincipal&);
0133
0134     void clearLumiPrincipal(LuminosityBlockPrincipal&);
0135
0136     using ProcessBlockType = PrincipalCache::ProcessBlockType;
0137     void writeProcessBlockAsync(edm::WaitingTaskHolder task, ProcessBlockType);
0138
0139     void writeRunAsync(WaitingTaskHolder, RunPrincipal const&, MergeableRunProductMetadata const*);
0140
0141     void clearRunPrincipal(RunPrincipal&);
0142
0143     void clearProcessBlockPrincipal(ProcessBlockType);
0144
0145     // Call closeFile() on all OutputModules.
0146     void closeOutputFiles() {
0147       ServiceRegistry::Operate operate(serviceToken_);
0148       schedule_->closeOutputFiles();
0149       for_all(subProcesses_, [](auto& subProcess) { subProcess.closeOutputFiles(); });
0150     }
0151
0152     // Call openFiles() on all OutputModules
0153     void openOutputFiles(FileBlock& fb) {
0154       ServiceRegistry::Operate operate(serviceToken_);
0155       schedule_->openOutputFiles(fb);
0156       for_all(subProcesses_, [&fb](auto& subProcess) { subProcess.openOutputFiles(fb); });
0157     }
0158
0159     void updateBranchIDListHelper(BranchIDLists const&);
0160
0161     // Call respondToOpenInputFile() on all Modules
0162     void respondToOpenInputFile(FileBlock const& fb);
0163
0164     // Call respondToCloseInputFile() on all Modules
0165     void respondToCloseInputFile(FileBlock const& fb) {
0166       ServiceRegistry::Operate operate(serviceToken_);
0167       schedule_->respondToCloseInputFile(fb);
0168       for_all(subProcesses_, [&fb](auto& subProcess) { subProcess.respondToCloseInputFile(fb); });
0169     }
0170
0171     // Call shouldWeCloseFile() on all OutputModules.
0172     bool shouldWeCloseOutput() const {
0173       ServiceRegistry::Operate operate(serviceToken_);
0174       if (schedule_->shouldWeCloseOutput()) {
0175         return true;
0176       }
0177       for (auto const& subProcess : subProcesses_) {
0178         if (subProcess.shouldWeCloseOutput()) {
0179           return true;
0180         }
0181       }
0182       return false;
0183     }
0184
0185     /// Return a vector allowing const access to all the ModuleDescriptions for this SubProcess
0186
0187     /// *** N.B. *** Ownership of the ModuleDescriptions is *not*
0188     /// *** passed to the caller. Do not call delete on these
0189     /// *** pointers!
0190     std::vector<ModuleDescription const*> getAllModuleDescriptions() const;
0191
0192     /// Return the number of events this SubProcess has tried to process
0193     /// (inclues both successes and failures, including failures due
0194     /// to exceptions during processing).
0195     int totalEvents() const { return schedule_->totalEvents(); }
0196
0197     /// Return the number of events which have been passed by one or more trigger paths.
0198     int totalEventsPassed() const {
0199       ServiceRegistry::Operate operate(serviceToken_);
0200       return schedule_->totalEventsPassed();
0201     }
0202
0203     /// Return the number of events that have not passed any trigger.
0204     /// (N.B. totalEventsFailed() + totalEventsPassed() == totalEvents()
0205     int totalEventsFailed() const {
0206       ServiceRegistry::Operate operate(serviceToken_);
0207       return schedule_->totalEventsFailed();
0208     }
0209
0210     /// Return the trigger report information on paths,
0211     /// modules-in-path, modules-in-endpath, and modules.
0212     void getTriggerReport(TriggerReport& rep) const {
0213       ServiceRegistry::Operate operate(serviceToken_);
0214       schedule_->getTriggerReport(rep);
0215     }
0216
0217     /// Return whether each output module has reached its maximum count.
0218     /// If there is a subprocess, get this information from the subprocess.
0219     bool terminate() const {
0220       ServiceRegistry::Operate operate(serviceToken_);
0221       if (schedule_->terminate()) {
0222         return true;
0223       }
0224       for (auto const& subProcess : subProcesses_) {
0225         if (subProcess.terminate()) {
0226           return true;
0227         }
0228       }
0229       return false;
0230     }
0231
0232     ///  Clear all the counters in the trigger report.
0233     void clearCounters() {
0234       ServiceRegistry::Operate operate(serviceToken_);
0235       schedule_->clearCounters();
0236       for_all(subProcesses_, [](auto& subProcess) { subProcess.clearCounters(); });
0237     }
0238
0239   private:
0240     void beginJob();
0241     void endJob();
0242     void processAsync(WaitingTaskHolder iHolder,
0243                       EventPrincipal const& e,
0244                       std::vector<std::shared_ptr<const EventSetupImpl>> const*);
0245
0246     void propagateProducts(BranchType type, Principal const& parentPrincipal, Principal& principal) const;
0247     bool parentProducedProductIsKept(Principal const& parentPrincipal, Principal& principal) const;
0248     void fixBranchIDListsForEDAliases(
0249         std::map<BranchID::value_type, BranchID::value_type> const& droppedBranchIDToKeptBranchID);
0250     void keepThisBranch(BranchDescription const& desc,
0251                         std::map<BranchID, BranchDescription const*>& trueBranchIDToKeptBranchDesc,
0252                         std::set<BranchID>& keptProductsInEvent);
0253
0254     std::map<BranchID::value_type, BranchID::value_type> const& droppedBranchIDToKeptBranchID() {
0255       return droppedBranchIDToKeptBranchID_;
0256     }
0257
0258     std::shared_ptr<BranchIDListHelper const> branchIDListHelper() const {
0259       return get_underlying_safe(branchIDListHelper_);
0260     }
0261     std::shared_ptr<BranchIDListHelper>& branchIDListHelper() { return get_underlying_safe(branchIDListHelper_); }
0262     std::shared_ptr<ThinnedAssociationsHelper const> thinnedAssociationsHelper() const {
0263       return get_underlying_safe(thinnedAssociationsHelper_);
0264     }
0265     std::shared_ptr<ThinnedAssociationsHelper> thinnedAssociationsHelper() {
0266       return get_underlying_safe(thinnedAssociationsHelper_);
0267     }
0268
0269     std::shared_ptr<ActivityRegistry> actReg_;  // We do not use propagate_const because the registry itself is mutable.
0270     ServiceToken serviceToken_;
0271     std::shared_ptr<ProductRegistry const> parentPreg_;
0272     std::shared_ptr<ProductRegistry const> preg_;
0273     edm::propagate_const<std::shared_ptr<BranchIDListHelper>> branchIDListHelper_;
0274     edm::propagate_const<std::shared_ptr<SubProcessBlockHelper>> processBlockHelper_;
0275     edm::propagate_const<std::shared_ptr<ThinnedAssociationsHelper>> thinnedAssociationsHelper_;
0276     edm::propagate_const<std::shared_ptr<SubProcessParentageHelper>> subProcessParentageHelper_;
0277     std::unique_ptr<ExceptionToActionTable const> act_table_;
0278     std::shared_ptr<ProcessConfiguration const> processConfiguration_;
0279     ProcessContext processContext_;
0280     PathsAndConsumesOfModules pathsAndConsumesOfModules_;
0281     //We require 1 history for each Run, Lumi and Stream
0282     // The vectors first hold Stream info, then Lumi then Run
0283     unsigned int historyLumiOffset_;
0284     unsigned int historyRunOffset_;
0285     std::vector<ProcessHistoryRegistry> processHistoryRegistries_;
0286     std::vector<HistoryAppender> historyAppenders_;
0287     PrincipalCache principalCache_;
0288     //vector index is principal's index value
0289     std::vector<std::shared_ptr<RunPrincipal>> inUseRunPrincipals_;
0290     std::vector<std::shared_ptr<LuminosityBlockPrincipal>> inUseLumiPrincipals_;
0291     edm::propagate_const<std::shared_ptr<eventsetup::EventSetupProvider>> esp_;
0292     edm::propagate_const<std::unique_ptr<Schedule>> schedule_;
0293     std::vector<SubProcess> subProcesses_;
0294     edm::propagate_const<std::unique_ptr<ParameterSet>> processParameterSet_;
0295
0296     // keptProducts_ are pointers to the BranchDescription objects describing
0297     // the branches we are to write.
0298     //
0299     // We do not own the BranchDescriptions to which we point.
0300     SelectedProductsForBranchType keptProducts_;
0301     ProductSelectorRules productSelectorRules_;
0302     ProductSelector productSelector_;
0303
0304     //EventSelection
0305     bool wantAllEvents_;
0306     ParameterSetID selector_config_id_;
0307     mutable detail::TriggerResultsBasedEventSelector selectors_;
0308
0309     // needed because of possible EDAliases.
0310     // filled in only if key and value are different.
0311     std::map<BranchID::value_type, BranchID::value_type> droppedBranchIDToKeptBranchID_;
0312   };
0313
0314   // free function
0315   std::vector<ParameterSet> popSubProcessVParameterSet(ParameterSet& parameterSet);
0316 }  // namespace edm
0317 #endif