Framework/src/OutputModuleCore.cc

0001 // -*- C++ -*-
0002 //
0003 // Package:     FWCore/Framework
0004 // Class  :     OutputModuleCore
0005 //
0006 // Implementation:
0007 //     [Notes on implementation]
0008 //
0009 // Original Author:  Chris Jones
0010 //         Created:  Wed, 31 Jul 2013 15:59:19 GMT
0011 //
0012
0013 // system include files
0014 #include <cassert>
0015
0016 // user include files
0017 #include "FWCore/Framework/interface/OutputModuleCore.h"
0018
0019 #include "DataFormats/Common/interface/Handle.h"
0020 #include "DataFormats/Common/interface/ThinnedAssociation.h"
0021 #include "DataFormats/Common/interface/EndPathStatus.h"
0022 #include "DataFormats/Provenance/interface/ProductDescription.h"
0023 #include "DataFormats/Provenance/interface/BranchKey.h"
0024 #include "DataFormats/Provenance/interface/ProductRegistry.h"
0025 #include "DataFormats/Provenance/interface/ThinnedAssociationsHelper.h"
0026 #include "FWCore/Framework/interface/SignallingProductRegistryFiller.h"
0027 #include "FWCore/Framework/interface/EventForOutput.h"
0028 #include "FWCore/Framework/interface/EventPrincipal.h"
0029 #include "FWCore/Framework/src/insertSelectedProcesses.h"
0030 #include "FWCore/Framework/interface/LuminosityBlockForOutput.h"
0031 #include "FWCore/Framework/interface/ProcessBlockForOutput.h"
0032 #include "FWCore/Framework/interface/RunForOutput.h"
0033 #include "FWCore/Framework/src/OutputModuleDescription.h"
0034 #include "FWCore/Framework/interface/TriggerNamesService.h"
0035 #include "FWCore/Framework/src/EventSignalsSentry.h"
0036 #include "FWCore/Framework/interface/PreallocationConfiguration.h"
0037 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
0038 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0039 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
0040 #include "FWCore/ServiceRegistry/interface/Service.h"
0041 #include "FWCore/Utilities/interface/DebugMacros.h"
0042 #include "FWCore/Reflection/interface/DictionaryTools.h"
0043
0044 namespace edm {
0045   namespace core {
0046
0047     // -------------------------------------------------------
0048     OutputModuleCore::OutputModuleCore(ParameterSet const& pset)
0049         : remainingEvents_(-1),
0050           maxEvents_(-1),
0051           keptProducts_(),
0052           hasNewlyDroppedBranch_(),
0053           process_name_(),
0054           productSelectorRules_(pset, "outputCommands", "OutputModule"),
0055           productSelector_(),
0056           moduleDescription_(),
0057           wantAllEvents_(false),
0058           selectors_(),
0059           selector_config_id_(),
0060           droppedBranchIDToKeptBranchID_(),
0061           branchIDLists_(new BranchIDLists),
0062           origBranchIDLists_(nullptr),
0063           thinnedAssociationsHelper_(new ThinnedAssociationsHelper) {
0064       hasNewlyDroppedBranch_.fill(false);
0065
0066       Service<service::TriggerNamesService> tns;
0067       process_name_ = tns->getProcessName();
0068
0069       selectEvents_ = pset.getUntrackedParameterSet("SelectEvents", ParameterSet());
0070
0071       selectEvents_.registerIt();  // Just in case this PSet is not registered
0072
0073       selector_config_id_ = selectEvents_.id();
0074
0075       //need to set wantAllEvents_ in constructor
0076       // we will make the remaining selectors once we know how many streams
0077       selectors_.resize(1);
0078       wantAllEvents_ = detail::configureEventSelector(
0079           selectEvents_, process_name_, getAllTriggerNames(), selectors_[0], consumesCollector());
0080     }
0081
0082     void OutputModuleCore::configure(OutputModuleDescription const& desc) {
0083       remainingEvents_ = maxEvents_ = desc.maxEvents_;
0084       origBranchIDLists_ = desc.branchIDLists_;
0085     }
0086
0087     void OutputModuleCore::selectProducts(ProductRegistry const& preg,
0088                                           ThinnedAssociationsHelper const& thinnedAssociationsHelper,
0089                                           ProcessBlockHelperBase const& processBlockHelper) {
0090       if (productSelector_.initialized())
0091         return;
0092       productSelector_.initialize(productSelectorRules_, preg.allProductDescriptions());
0093
0094       // TODO: See if we can collapse keptProducts_ and productSelector_ into a
0095       // single object. See the notes in the header for ProductSelector
0096       // for more information.
0097
0098       std::map<BranchID, ProductDescription const*> trueBranchIDToKeptBranchDesc;
0099       std::vector<ProductDescription const*> associationDescriptions;
0100       std::set<BranchID> keptProductsInEvent;
0101       std::set<std::string> processesWithSelectedMergeableRunProducts;
0102       std::set<std::string> processesWithKeptProcessBlockProducts;
0103
0104       for (auto const& it : preg.productList()) {
0105         ProductDescription const& desc = it.second;
0106         if (desc.transient()) {
0107           // if the class of the branch is marked transient, output nothing
0108         } else if (!desc.present() && !desc.produced()) {
0109           // else if the branch containing the product has been previously dropped,
0110           // output nothing
0111         } else if (desc.unwrappedType() == typeid(ThinnedAssociation)) {
0112           associationDescriptions.push_back(&desc);
0113         } else if (selected(desc)) {
0114           keepThisBranch(desc, trueBranchIDToKeptBranchDesc, keptProductsInEvent);
0115           insertSelectedProcesses(
0116               desc, processesWithSelectedMergeableRunProducts, processesWithKeptProcessBlockProducts);
0117         } else {
0118           // otherwise, output nothing,
0119           // and mark the fact that there is a newly dropped branch of this type.
0120           hasNewlyDroppedBranch_[desc.branchType()] = true;
0121         }
0122       }
0123
0124       setProcessesWithSelectedMergeableRunProducts(processesWithSelectedMergeableRunProducts);
0125
0126       thinnedAssociationsHelper.selectAssociationProducts(
0127           associationDescriptions, keptProductsInEvent, keepAssociation_);
0128
0129       for (auto association : associationDescriptions) {
0130         if (keepAssociation_[association->branchID()]) {
0131           keepThisBranch(*association, trueBranchIDToKeptBranchDesc, keptProductsInEvent);
0132         } else {
0133           hasNewlyDroppedBranch_[association->branchType()] = true;
0134         }
0135       }
0136
0137       // Now fill in a mapping needed in the case that a branch was dropped while its EDAlias was kept.
0138       ProductSelector::fillDroppedToKept(preg, trueBranchIDToKeptBranchDesc, droppedBranchIDToKeptBranchID_);
0139
0140       thinnedAssociationsHelper_->updateFromParentProcess(
0141           thinnedAssociationsHelper, keepAssociation_, droppedBranchIDToKeptBranchID_);
0142       outputProcessBlockHelper_.updateAfterProductSelection(processesWithKeptProcessBlockProducts, processBlockHelper);
0143
0144       initialRegistry(preg);
0145     }
0146
0147     void OutputModuleCore::updateBranchIDListsWithKeptAliases() {
0148       if (!droppedBranchIDToKeptBranchID_.empty()) {
0149         // Make a private copy of the BranchIDLists.
0150         *branchIDLists_ = *origBranchIDLists_;
0151         // Check for branches dropped while an EDAlias was kept.
0152         for (BranchIDList& branchIDList : *branchIDLists_) {
0153           for (BranchID::value_type& branchID : branchIDList) {
0154             // Replace BranchID of each dropped branch with that of the kept
0155             // alias, so the alias branch will have the product ID of the original branch.
0156             std::map<BranchID::value_type, BranchID::value_type>::const_iterator iter =
0157                 droppedBranchIDToKeptBranchID_.find(branchID);
0158             if (iter != droppedBranchIDToKeptBranchID_.end()) {
0159               branchID = iter->second;
0160             }
0161           }
0162         }
0163       }
0164     }
0165
0166     void OutputModuleCore::keepThisBranch(ProductDescription const& desc,
0167                                           std::map<BranchID, ProductDescription const*>& trueBranchIDToKeptBranchDesc,
0168                                           std::set<BranchID>& keptProductsInEvent) {
0169       ProductSelector::checkForDuplicateKeptBranch(desc, trueBranchIDToKeptBranchDesc);
0170
0171       EDGetToken token;
0172
0173       std::vector<std::string> missingDictionaries;
0174       if (!checkDictionary(missingDictionaries, desc.className(), desc.unwrappedType())) {
0175         std::string context("Calling OutputModuleCore::keepThisBranch, checking dictionaries for kept types");
0176         throwMissingDictionariesException(missingDictionaries, context);
0177       }
0178
0179       switch (desc.branchType()) {
0180         case InEvent: {
0181           if (desc.produced()) {
0182             keptProductsInEvent.insert(desc.originalBranchID());
0183           } else {
0184             keptProductsInEvent.insert(desc.branchID());
0185           }
0186           token = consumes(TypeToGet{desc.unwrappedTypeID(), PRODUCT_TYPE},
0187                            InputTag{desc.moduleLabel(), desc.productInstanceName(), desc.processName()});
0188           break;
0189         }
0190         case InLumi: {
0191           token = consumes<InLumi>(TypeToGet{desc.unwrappedTypeID(), PRODUCT_TYPE},
0192                                    InputTag(desc.moduleLabel(), desc.productInstanceName(), desc.processName()));
0193           break;
0194         }
0195         case InRun: {
0196           token = consumes<InRun>(TypeToGet{desc.unwrappedTypeID(), PRODUCT_TYPE},
0197                                   InputTag(desc.moduleLabel(), desc.productInstanceName(), desc.processName()));
0198           break;
0199         }
0200         case InProcess: {
0201           token = consumes<InProcess>(TypeToGet{desc.unwrappedTypeID(), PRODUCT_TYPE},
0202                                       InputTag(desc.moduleLabel(), desc.productInstanceName(), desc.processName()));
0203           break;
0204         }
0205         default:
0206           assert(false);
0207           break;
0208       }
0209       // Now put it in the list of selected branches.
0210       keptProducts_[desc.branchType()].push_back(std::make_pair(&desc, token));
0211     }
0212
0213     OutputModuleCore::~OutputModuleCore() {}
0214
0215     void OutputModuleCore::doPreallocate_(PreallocationConfiguration const& iPC) {
0216       auto nstreams = iPC.numberOfStreams();
0217       selectors_.resize(nstreams);
0218
0219       preallocLumis(iPC.numberOfLuminosityBlocks());
0220
0221       bool seenFirst = false;
0222       for (auto& s : selectors_) {
0223         if (seenFirst) {
0224           detail::configureEventSelector(selectEvents_, process_name_, getAllTriggerNames(), s, consumesCollector());
0225         } else {
0226           seenFirst = true;
0227         }
0228       }
0229     }
0230
0231     void OutputModuleCore::preallocLumis(unsigned int) {}
0232
0233     void OutputModuleCore::doBeginJob_() { this->beginJob(); }
0234
0235     void OutputModuleCore::doEndJob() { endJob(); }
0236
0237     void OutputModuleCore::registerProductsAndCallbacks(OutputModuleCore const*, SignallingProductRegistryFiller* reg) {
0238       if (callWhenNewProductsRegistered_) {
0239         reg->callForEachBranch(callWhenNewProductsRegistered_);
0240
0241         reg->watchProductAdditions(callWhenNewProductsRegistered_);
0242       }
0243     }
0244
0245     bool OutputModuleCore::needToRunSelection() const noexcept { return !wantAllEvents_; }
0246
0247     std::vector<ProductResolverIndexAndSkipBit> OutputModuleCore::productsUsedBySelection() const noexcept {
0248       std::vector<ProductResolverIndexAndSkipBit> returnValue;
0249       auto const& s = selectors_[0];
0250       auto const n = s.numberOfTokens();
0251       returnValue.reserve(n);
0252
0253       for (unsigned int i = 0; i < n; ++i) {
0254         returnValue.emplace_back(uncheckedIndexFrom(s.token(i)));
0255       }
0256       return returnValue;
0257     }
0258
0259     bool OutputModuleCore::prePrefetchSelection(StreamID id,
0260                                                 EventPrincipal const& ep,
0261                                                 ModuleCallingContext const* mcc) {
0262       if (wantAllEvents_)
0263         return true;
0264       auto& s = selectors_[id.value()];
0265       EventForOutput e(ep, moduleDescription_, mcc);
0266       e.setConsumer(this);
0267       return s.wantEvent(e);
0268     }
0269
0270     bool OutputModuleCore::doEvent_(EventTransitionInfo const& info,
0271                                     ActivityRegistry* act,
0272                                     ModuleCallingContext const* mcc) {
0273       {
0274         EventSignalsSentry sentry(act, mcc);
0275         EventForOutput e(info, moduleDescription_, mcc);
0276         e.setConsumer(this);
0277         write(e);
0278       }
0279       //remainingEvents_ is decremented by inheriting classes
0280       return true;
0281     }
0282
0283     bool OutputModuleCore::doBeginRun(RunTransitionInfo const& info, ModuleCallingContext const* mcc) {
0284       RunForOutput r(info, moduleDescription_, mcc, false);
0285       r.setConsumer(this);
0286       doBeginRun_(r);
0287       return true;
0288     }
0289
0290     bool OutputModuleCore::doEndRun(RunTransitionInfo const& info, ModuleCallingContext const* mcc) {
0291       RunForOutput r(info, moduleDescription_, mcc, true);
0292       r.setConsumer(this);
0293       doEndRun_(r);
0294       return true;
0295     }
0296
0297     void OutputModuleCore::doWriteProcessBlock(ProcessBlockPrincipal const& pbp, ModuleCallingContext const* mcc) {
0298       ProcessBlockForOutput pb(pbp, moduleDescription_, mcc, true);
0299       pb.setConsumer(this);
0300       writeProcessBlock(pb);
0301     }
0302
0303     void OutputModuleCore::doWriteRun(RunPrincipal const& rp,
0304                                       ModuleCallingContext const* mcc,
0305                                       MergeableRunProductMetadata const* mrpm) {
0306       RunForOutput r(rp, moduleDescription_, mcc, true, mrpm);
0307       r.setConsumer(this);
0308       writeRun(r);
0309     }
0310
0311     bool OutputModuleCore::doBeginLuminosityBlock(LumiTransitionInfo const& info, ModuleCallingContext const* mcc) {
0312       LuminosityBlockForOutput lb(info, moduleDescription_, mcc, false);
0313       lb.setConsumer(this);
0314       doBeginLuminosityBlock_(lb);
0315       return true;
0316     }
0317
0318     bool OutputModuleCore::doEndLuminosityBlock(LumiTransitionInfo const& info, ModuleCallingContext const* mcc) {
0319       LuminosityBlockForOutput lb(info, moduleDescription_, mcc, true);
0320       lb.setConsumer(this);
0321       doEndLuminosityBlock_(lb);
0322
0323       return true;
0324     }
0325
0326     void OutputModuleCore::doWriteLuminosityBlock(LuminosityBlockPrincipal const& lbp,
0327                                                   ModuleCallingContext const* mcc) {
0328       LuminosityBlockForOutput lb(lbp, moduleDescription_, mcc, true);
0329       lb.setConsumer(this);
0330       writeLuminosityBlock(lb);
0331     }
0332
0333     void OutputModuleCore::doOpenFile(FileBlock const& fb) { openFile(fb); }
0334
0335     void OutputModuleCore::doRespondToOpenInputFile(FileBlock const& fb) {
0336       updateBranchIDListsWithKeptAliases();
0337       doRespondToOpenInputFile_(fb);
0338     }
0339
0340     void OutputModuleCore::doRespondToCloseInputFile(FileBlock const& fb) { doRespondToCloseInputFile_(fb); }
0341
0342     void OutputModuleCore::doCloseFile() {
0343       if (isFileOpen()) {
0344         reallyCloseFile();
0345       }
0346     }
0347
0348     void OutputModuleCore::reallyCloseFile() {}
0349
0350     BranchIDLists const* OutputModuleCore::branchIDLists() const {
0351       if (!droppedBranchIDToKeptBranchID_.empty()) {
0352         return branchIDLists_.get();
0353       }
0354       return origBranchIDLists_;
0355     }
0356
0357     ThinnedAssociationsHelper const* OutputModuleCore::thinnedAssociationsHelper() const {
0358       return thinnedAssociationsHelper_.get();
0359     }
0360
0361     ModuleDescription const& OutputModuleCore::description() const { return moduleDescription_; }
0362
0363     bool OutputModuleCore::selected(ProductDescription const& desc) const { return productSelector_.selected(desc); }
0364
0365     void OutputModuleCore::fillDescriptions(ConfigurationDescriptions& descriptions) {
0366       ParameterSetDescription desc;
0367       desc.setUnknown();
0368       descriptions.addDefault(desc);
0369     }
0370
0371     void OutputModuleCore::fillDescription(ParameterSetDescription& desc,
0372                                            std::vector<std::string> const& defaultOutputCommands) {
0373       ProductSelectorRules::fillDescription(desc, "outputCommands", defaultOutputCommands);
0374       EventSelector::fillDescription(desc);
0375     }
0376
0377     void OutputModuleCore::prevalidate(ConfigurationDescriptions&) {}
0378
0379     static const std::string kBaseType("OutputModule");
0380     const std::string& OutputModuleCore::baseType() { return kBaseType; }
0381
0382     void OutputModuleCore::setEventSelectionInfo(
0383         std::map<std::string, std::vector<std::pair<std::string, int>>> const& outputModulePathPositions,
0384         bool anyProductProduced) {
0385       selector_config_id_ = detail::registerProperSelectionInfo(getParameterSet(selector_config_id_),
0386                                                                 description().moduleLabel(),
0387                                                                 outputModulePathPositions,
0388                                                                 anyProductProduced);
0389     }
0390   }  // namespace core
0391 }  // namespace edm