test/stubs/TestStreamProducers.cc

0001
0002 /*----------------------------------------------------------------------
0003
0004 Toy edm::stream::EDProducer modules of
0005 edm::*Cache templates and edm::*Producer classes
0006 for testing purposes only.
0007
0008 ----------------------------------------------------------------------*/
0009
0010 #include <atomic>
0011 #include <functional>
0012 #include <iostream>
0013 #include <map>
0014 #include <tuple>
0015 #include <unistd.h>
0016 #include <vector>
0017
0018 #include "FWCore/Framework/interface/CacheHandle.h"
0019 #include "FWCore/Framework/interface/stream/EDProducer.h"
0020 #include "FWCore/Framework/interface/maker/WorkerT.h"
0021 #include "FWCore/Framework/interface/HistoryAppender.h"
0022 #include "FWCore/MessageLogger/interface/MessageLogger.h"
0023 #include "FWCore/ServiceRegistry/interface/ParentContext.h"
0024 #include "FWCore/ServiceRegistry/interface/StreamContext.h"
0025 #include "FWCore/Utilities/interface/GlobalIdentifier.h"
0026 #include "FWCore/Framework/interface/Event.h"
0027 #include "FWCore/Framework/interface/MakerMacros.h"
0028 #include "FWCore/Framework/interface/ProcessBlock.h"
0029 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0030 #include "FWCore/Utilities/interface/EDMException.h"
0031 #include "DataFormats/TestObjects/interface/ToyProducts.h"
0032
0033 namespace edmtest {
0034   namespace stream {
0035
0036     // anonymous namespace here causes build warnings
0037     namespace cache {
0038       struct Cache {
0039         Cache() : value(0), run(0), lumi(0) {}
0040         //Using mutable since we want to update the value.
0041         mutable std::atomic<unsigned int> value;
0042         mutable std::atomic<unsigned int> run;
0043         mutable std::atomic<unsigned int> lumi;
0044       };
0045
0046       struct SummaryCache {
0047         // Intentionally not thread safe, not atomic
0048         unsigned int value = 0;
0049       };
0050
0051       struct TestGlobalCache {
0052         CMS_THREAD_SAFE mutable edm::EDPutTokenT<unsigned int> token_;
0053         CMS_THREAD_SAFE mutable edm::EDGetTokenT<unsigned int> getTokenBegin_;
0054         CMS_THREAD_SAFE mutable edm::EDGetTokenT<unsigned int> getTokenEnd_;
0055         unsigned int trans_{0};
0056         mutable std::atomic<unsigned int> m_count{0};
0057       };
0058
0059     }  // namespace cache
0060
0061     using Cache = cache::Cache;
0062     using SummaryCache = cache::SummaryCache;
0063     using TestGlobalCache = cache::TestGlobalCache;
0064
0065     class GlobalIntProducer : public edm::stream::EDProducer<edm::GlobalCache<Cache>> {
0066     public:
0067       static std::atomic<unsigned int> m_count;
0068       unsigned int trans_;
0069       static std::atomic<unsigned int> cvalue_;
0070
0071       static std::unique_ptr<Cache> initializeGlobalCache(edm::ParameterSet const&) {
0072         ++m_count;
0073         return std::make_unique<Cache>();
0074       }
0075
0076       GlobalIntProducer(edm::ParameterSet const& p, const Cache* iGlobal) {
0077         trans_ = p.getParameter<int>("transitions");
0078         cvalue_ = p.getParameter<int>("cachevalue");
0079         produces<unsigned int>();
0080       }
0081
0082       static void globalBeginJob(Cache* iGlobal) {
0083         ++m_count;
0084         if (iGlobal->value != 0) {
0085           throw cms::Exception("cache value") << iGlobal->value << " but it was supposed to be 0";
0086         }
0087       }
0088
0089       void produce(edm::Event&, edm::EventSetup const&) override {
0090         ++m_count;
0091         ++((globalCache())->value);
0092       }
0093
0094       static void globalEndJob(Cache* iGlobal) {
0095         ++m_count;
0096         if (iGlobal->value != cvalue_) {
0097           throw cms::Exception("cache value") << iGlobal->value << " but it was supposed to be " << cvalue_;
0098         }
0099       }
0100
0101       ~GlobalIntProducer() {
0102         if (m_count != trans_) {
0103           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0104         }
0105       }
0106     };
0107
0108     class RunIntProducer : public edm::stream::EDProducer<edm::RunCache<Cache>, edm::stream::WatchRuns> {
0109     public:
0110       static std::atomic<unsigned int> m_count;
0111       unsigned int trans_;
0112       static std::atomic<unsigned int> cvalue_;
0113
0114       RunIntProducer(edm::ParameterSet const& p) {
0115         trans_ = p.getParameter<int>("transitions");
0116         cvalue_ = p.getParameter<int>("cachevalue");
0117         produces<unsigned int>();
0118       }
0119
0120       void produce(edm::Event&, edm::EventSetup const&) override {
0121         ++m_count;
0122         ++(runCache()->value);
0123       }
0124
0125       static std::shared_ptr<Cache> globalBeginRun(edm::Run const& iRun, edm::EventSetup const&, GlobalCache const*) {
0126         ++m_count;
0127         auto pCache = std::make_shared<Cache>();
0128         pCache->run = iRun.runAuxiliary().run();
0129         return pCache;
0130       }
0131
0132       void beginRun(edm::Run const& iRun, edm::EventSetup const&) override {
0133         if (runCache()->run != iRun.runAuxiliary().run()) {
0134           throw cms::Exception("begin out of sequence") << "beginRun seen before globalBeginRun";
0135         }
0136       }
0137
0138       static void globalEndRun(edm::Run const& iRun, edm::EventSetup const&, RunContext const* iContext) {
0139         ++m_count;
0140         auto pCache = iContext->run();
0141         if (pCache->run != iRun.runAuxiliary().run()) {
0142           throw cms::Exception("end out of sequence") << "globalEndRun seen before globalBeginRun in Run" << iRun.run();
0143         }
0144         pCache->run = 0;
0145         if (iContext->run()->value != cvalue_) {
0146           throw cms::Exception("cache value") << iContext->run()->value << " but it was supposed to be " << cvalue_;
0147         }
0148       }
0149
0150       void endRun(edm::Run const& iRun, edm::EventSetup const&) override {
0151         if (runCache()->run != iRun.runAuxiliary().run()) {
0152           throw cms::Exception("end out of sequence") << "globalEndRun seen before endRun";
0153         }
0154       }
0155
0156       ~RunIntProducer() {
0157         if (m_count != trans_) {
0158           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0159         }
0160       }
0161     };
0162
0163     class LumiIntProducer
0164         : public edm::stream::EDProducer<edm::LuminosityBlockCache<Cache>, edm::stream::WatchLuminosityBlocks> {
0165     public:
0166       static std::atomic<unsigned int> m_count;
0167       unsigned int trans_;
0168       static std::atomic<unsigned int> cvalue_;
0169
0170       LumiIntProducer(edm::ParameterSet const& p) {
0171         trans_ = p.getParameter<int>("transitions");
0172         cvalue_ = p.getParameter<int>("cachevalue");
0173         produces<unsigned int>();
0174       }
0175
0176       void produce(edm::Event&, edm::EventSetup const&) override {
0177         ++m_count;
0178         ++(luminosityBlockCache()->value);
0179       }
0180
0181       static std::shared_ptr<Cache> globalBeginLuminosityBlock(edm::LuminosityBlock const& iLB,
0182                                                                edm::EventSetup const&,
0183                                                                RunContext const*) {
0184         ++m_count;
0185         auto pCache = std::make_shared<Cache>();
0186         pCache->run = iLB.luminosityBlockAuxiliary().run();
0187         pCache->lumi = iLB.luminosityBlockAuxiliary().luminosityBlock();
0188         return pCache;
0189       }
0190
0191       void beginLuminosityBlock(edm::LuminosityBlock const& iLB, edm::EventSetup const&) override {
0192         if (luminosityBlockCache()->run != iLB.luminosityBlockAuxiliary().run() ||
0193             luminosityBlockCache()->lumi != iLB.luminosityBlockAuxiliary().luminosityBlock()) {
0194           throw cms::Exception("begin out of sequence")
0195               << "beginLuminosityBlock seen before globalBeginLuminosityBlock " << luminosityBlockCache()->run << " "
0196               << iLB.luminosityBlockAuxiliary().run();
0197         }
0198       }
0199
0200       static void globalEndLuminosityBlock(edm::LuminosityBlock const& iLB,
0201                                            edm::EventSetup const&,
0202                                            LuminosityBlockContext const* iLBContext) {
0203         ++m_count;
0204         auto pCache = iLBContext->luminosityBlock();
0205         if (pCache->run != iLB.luminosityBlockAuxiliary().run() ||
0206             pCache->lumi != iLB.luminosityBlockAuxiliary().luminosityBlock()) {
0207           throw cms::Exception("end out of sequence")
0208               << "globalEndLuminosityBlock seen before globalBeginLuminosityBlock in LuminosityBlock"
0209               << iLB.luminosityBlock();
0210         }
0211         pCache->run = 0;
0212         pCache->lumi = 0;
0213         if (iLBContext->luminosityBlock()->value != cvalue_) {
0214           throw cms::Exception("cache value")
0215               << iLBContext->luminosityBlock()->value << " but it was supposed to be " << cvalue_;
0216         }
0217       }
0218
0219       void endLuminosityBlock(edm::LuminosityBlock const& iLB, edm::EventSetup const&) override {
0220         if (luminosityBlockCache()->run != iLB.luminosityBlockAuxiliary().run() ||
0221             luminosityBlockCache()->lumi != iLB.luminosityBlockAuxiliary().luminosityBlock()) {
0222           throw cms::Exception("end out of sequence") << "globalEndLuminosityBlock seen before endLuminosityBlock";
0223         }
0224       }
0225
0226       ~LumiIntProducer() {
0227         if (m_count != trans_) {
0228           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0229         }
0230       }
0231     };
0232
0233     class RunSummaryIntProducer : public edm::stream::EDProducer<edm::RunCache<Cache>,
0234                                                                  edm::RunSummaryCache<SummaryCache>,
0235                                                                  edm::stream::WatchRuns> {
0236     public:
0237       static std::atomic<unsigned int> m_count;
0238       unsigned int trans_;
0239       static std::atomic<unsigned int> cvalue_;
0240       static std::atomic<bool> globalBeginRunCalled_;
0241       unsigned int valueAccumulatedForStream_ = 0;
0242       bool endRunWasCalled_ = false;
0243
0244       RunSummaryIntProducer(edm::ParameterSet const& p) {
0245         trans_ = p.getParameter<int>("transitions");
0246         cvalue_ = p.getParameter<int>("cachevalue");
0247         produces<unsigned int>();
0248       }
0249
0250       void beginRun(edm::Run const&, edm::EventSetup const&) override {
0251         valueAccumulatedForStream_ = 0;
0252         endRunWasCalled_ = false;
0253       }
0254
0255       void produce(edm::Event&, edm::EventSetup const&) override {
0256         ++m_count;
0257         ++(runCache()->value);
0258         ++valueAccumulatedForStream_;
0259       }
0260
0261       static std::shared_ptr<Cache> globalBeginRun(edm::Run const& iRun, edm::EventSetup const&, GlobalCache const*) {
0262         ++m_count;
0263         globalBeginRunCalled_ = true;
0264         auto pCache = std::make_shared<Cache>();
0265         ++(pCache->run);
0266         return pCache;
0267       }
0268
0269       static std::shared_ptr<SummaryCache> globalBeginRunSummary(edm::Run const&,
0270                                                                  edm::EventSetup const&,
0271                                                                  GlobalCache const*) {
0272         ++m_count;
0273         if (!globalBeginRunCalled_) {
0274           throw cms::Exception("begin out of sequence") << "globalBeginRunSummary seen before globalBeginRun";
0275         }
0276         globalBeginRunCalled_ = false;
0277         return std::make_shared<SummaryCache>();
0278       }
0279
0280       void endRunSummary(edm::Run const&, edm::EventSetup const&, SummaryCache* runSummaryCache) const override {
0281         runSummaryCache->value += valueAccumulatedForStream_;
0282         if (!endRunWasCalled_) {
0283           throw cms::Exception("end out of sequence") << "endRunSummary seen before endRun";
0284         }
0285       }
0286
0287       static void globalEndRunSummary(edm::Run const&,
0288                                       edm::EventSetup const&,
0289                                       RunContext const*,
0290                                       SummaryCache* runSummaryCache) {
0291         ++m_count;
0292         if (runSummaryCache->value != cvalue_) {
0293           throw cms::Exception("unexpectedValue")
0294               << "run summary cache value = " << runSummaryCache->value << " but it was supposed to be " << cvalue_;
0295         }
0296       }
0297
0298       static void globalEndRun(edm::Run const& iRun, edm::EventSetup const&, RunContext const* iContext) {
0299         ++m_count;
0300         auto pCache = iContext->run();
0301         if (pCache->value != cvalue_) {
0302           throw cms::Exception("unExpectedValue")
0303               << "run cache value " << pCache->value << " but it was supposed to be " << cvalue_;
0304         }
0305         if (pCache->run != 1) {
0306           throw cms::Exception("end out of sequence") << "globalEndRun seen before globalBeginRun in Run" << iRun.run();
0307         }
0308       }
0309
0310       void endRun(edm::Run const&, edm::EventSetup const&) override { endRunWasCalled_ = true; }
0311
0312       ~RunSummaryIntProducer() {
0313         if (m_count != trans_) {
0314           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0315         }
0316       }
0317     };
0318
0319     class LumiSummaryIntProducer : public edm::stream::EDProducer<edm::LuminosityBlockCache<Cache>,
0320                                                                   edm::LuminosityBlockSummaryCache<SummaryCache>,
0321                                                                   edm::stream::WatchLuminosityBlocks> {
0322     public:
0323       static std::atomic<unsigned int> m_count;
0324       unsigned int trans_;
0325       static std::atomic<unsigned int> cvalue_;
0326       static std::atomic<bool> globalBeginLumiCalled_;
0327       unsigned int valueAccumulatedForStream_ = 0;
0328       bool endLumiWasCalled_ = false;
0329
0330       LumiSummaryIntProducer(edm::ParameterSet const& p) {
0331         trans_ = p.getParameter<int>("transitions");
0332         cvalue_ = p.getParameter<int>("cachevalue");
0333         produces<unsigned int>();
0334       }
0335
0336       void produce(edm::Event&, edm::EventSetup const&) override {
0337         ++m_count;
0338         ++(luminosityBlockCache()->value);
0339         ++valueAccumulatedForStream_;
0340       }
0341
0342       void beginLuminosityBlock(edm::LuminosityBlock const&, edm::EventSetup const&) override {
0343         valueAccumulatedForStream_ = 0;
0344         endLumiWasCalled_ = false;
0345       }
0346
0347       static std::shared_ptr<Cache> globalBeginLuminosityBlock(edm::LuminosityBlock const& iLB,
0348                                                                edm::EventSetup const&,
0349                                                                RunContext const*) {
0350         ++m_count;
0351         globalBeginLumiCalled_ = true;
0352         auto pCache = std::make_shared<Cache>();
0353         ++(pCache->lumi);
0354         return pCache;
0355       }
0356
0357       static std::shared_ptr<SummaryCache> globalBeginLuminosityBlockSummary(edm::LuminosityBlock const&,
0358                                                                              edm::EventSetup const&,
0359                                                                              LuminosityBlockContext const*) {
0360         ++m_count;
0361         if (!globalBeginLumiCalled_) {
0362           throw cms::Exception("begin out of sequence")
0363               << "globalBeginLuminosityBlockSummary seen before globalBeginLuminosityBlock";
0364         }
0365         globalBeginLumiCalled_ = false;
0366         return std::make_shared<SummaryCache>();
0367       }
0368
0369       void endLuminosityBlockSummary(edm::LuminosityBlock const&,
0370                                      edm::EventSetup const&,
0371                                      SummaryCache* lumiSummaryCache) const override {
0372         lumiSummaryCache->value += valueAccumulatedForStream_;
0373         if (!endLumiWasCalled_) {
0374           throw cms::Exception("end out of sequence") << "endLuminosityBlockSummary seen before endLuminosityBlock";
0375         }
0376       }
0377
0378       static void globalEndLuminosityBlockSummary(edm::LuminosityBlock const&,
0379                                                   edm::EventSetup const&,
0380                                                   LuminosityBlockContext const* iLBContext,
0381                                                   SummaryCache* lumiSummaryCache) {
0382         ++m_count;
0383         if (lumiSummaryCache->value != cvalue_) {
0384           throw cms::Exception("unexpectedValue")
0385               << "lumi summary cache value = " << lumiSummaryCache->value << " but it was supposed to be " << cvalue_;
0386         }
0387         auto pCache = iLBContext->luminosityBlock();
0388         // Add one so globalEndLuminosityBlock can check this function was called first
0389         ++pCache->value;
0390       }
0391
0392       static void globalEndLuminosityBlock(edm::LuminosityBlock const& iLB,
0393                                            edm::EventSetup const&,
0394                                            LuminosityBlockContext const* iLBContext) {
0395         ++m_count;
0396         auto pCache = iLBContext->luminosityBlock();
0397         if (pCache->value != cvalue_ + 1) {
0398           throw cms::Exception("unexpectedValue")
0399               << "lumi cache value " << pCache->value << " but it was supposed to be " << cvalue_ + 1;
0400         }
0401         if (pCache->lumi != 1) {
0402           throw cms::Exception("end out of sequence")
0403               << "globalEndLuminosityBlock seen before globalBeginLuminosityBlock in LuminosityBlock"
0404               << iLB.luminosityBlock();
0405         }
0406       }
0407
0408       void endLuminosityBlock(edm::LuminosityBlock const&, edm::EventSetup const&) override {
0409         endLumiWasCalled_ = true;
0410       }
0411
0412       ~LumiSummaryIntProducer() {
0413         if (m_count != trans_) {
0414           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0415         }
0416       }
0417     };
0418
0419     class ProcessBlockIntProducer
0420         : public edm::stream::EDProducer<edm::WatchProcessBlock, edm::GlobalCache<TestGlobalCache>> {
0421     public:
0422       explicit ProcessBlockIntProducer(edm::ParameterSet const& pset, TestGlobalCache const* testGlobalCache) {
0423         {
0424           auto tag = pset.getParameter<edm::InputTag>("consumesBeginProcessBlock");
0425           if (not tag.label().empty()) {
0426             testGlobalCache->getTokenBegin_ = consumes<unsigned int, edm::InProcess>(tag);
0427           }
0428         }
0429         {
0430           auto tag = pset.getParameter<edm::InputTag>("consumesEndProcessBlock");
0431           if (not tag.label().empty()) {
0432             testGlobalCache->getTokenEnd_ = consumes<unsigned int, edm::InProcess>(tag);
0433           }
0434         }
0435       }
0436
0437       static std::unique_ptr<TestGlobalCache> initializeGlobalCache(edm::ParameterSet const& pset) {
0438         auto testGlobalCache = std::make_unique<TestGlobalCache>();
0439         testGlobalCache->trans_ = pset.getParameter<int>("transitions");
0440         return testGlobalCache;
0441       }
0442
0443       static void beginProcessBlock(edm::ProcessBlock const& processBlock, TestGlobalCache* testGlobalCache) {
0444         if (testGlobalCache->m_count != 0) {
0445           throw cms::Exception("transitions") << "ProcessBlockIntProducer::begin transitions "
0446                                               << testGlobalCache->m_count << " but it was supposed to be " << 0;
0447         }
0448         ++testGlobalCache->m_count;
0449
0450         const unsigned int valueToGet = 51;
0451         if (not testGlobalCache->getTokenBegin_.isUninitialized()) {
0452           if (processBlock.get(testGlobalCache->getTokenBegin_) != valueToGet) {
0453             throw cms::Exception("BadValue")
0454                 << "expected " << valueToGet << " but got " << processBlock.get(testGlobalCache->getTokenBegin_);
0455           }
0456         }
0457       }
0458
0459       void produce(edm::Event&, edm::EventSetup const&) override {
0460         TestGlobalCache const* testGlobalCache = globalCache();
0461         if (testGlobalCache->m_count < 1u) {
0462           throw cms::Exception("out of sequence") << "produce before beginProcessBlock " << testGlobalCache->m_count;
0463         }
0464         ++testGlobalCache->m_count;
0465       }
0466
0467       static void endProcessBlock(edm::ProcessBlock const& processBlock, TestGlobalCache* testGlobalCache) {
0468         ++testGlobalCache->m_count;
0469         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0470           throw cms::Exception("transitions") << "ProcessBlockIntProducer::end transitions " << testGlobalCache->m_count
0471                                               << " but it was supposed to be " << testGlobalCache->trans_;
0472         }
0473         {
0474           const unsigned int valueToGet = 51;
0475           if (not testGlobalCache->getTokenBegin_.isUninitialized()) {
0476             if (processBlock.get(testGlobalCache->getTokenBegin_) != valueToGet) {
0477               throw cms::Exception("BadValue")
0478                   << "expected " << valueToGet << " but got " << processBlock.get(testGlobalCache->getTokenBegin_);
0479             }
0480           }
0481         }
0482         {
0483           const unsigned int valueToGet = 61;
0484           if (not testGlobalCache->getTokenEnd_.isUninitialized()) {
0485             if (processBlock.get(testGlobalCache->getTokenEnd_) != valueToGet) {
0486               throw cms::Exception("BadValue")
0487                   << "expected " << valueToGet << " but got " << processBlock.get(testGlobalCache->getTokenEnd_);
0488             }
0489           }
0490         }
0491       }
0492
0493       static void globalEndJob(TestGlobalCache* testGlobalCache) {
0494         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0495           throw cms::Exception("transitions")
0496               << "TestBeginProcessBlockProducer transitions " << testGlobalCache->m_count
0497               << " but it was supposed to be " << testGlobalCache->trans_;
0498         }
0499       }
0500
0501       ~ProcessBlockIntProducer() {
0502         TestGlobalCache const* testGlobalCache = globalCache();
0503         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0504           throw cms::Exception("transitions") << "ProcessBlockIntProducer transitions " << testGlobalCache->m_count
0505                                               << " but it was supposed to be " << testGlobalCache->trans_;
0506         }
0507       }
0508     };
0509
0510     class TestBeginProcessBlockProducer
0511         : public edm::stream::EDProducer<edm::BeginProcessBlockProducer, edm::GlobalCache<TestGlobalCache>> {
0512     public:
0513       explicit TestBeginProcessBlockProducer(edm::ParameterSet const& pset, TestGlobalCache const* testGlobalCache) {
0514         testGlobalCache->token_ = produces<unsigned int, edm::Transition::BeginProcessBlock>("begin");
0515
0516         auto tag = pset.getParameter<edm::InputTag>("consumesBeginProcessBlock");
0517         if (not tag.label().empty()) {
0518           testGlobalCache->getTokenBegin_ = consumes<unsigned int, edm::InProcess>(tag);
0519         }
0520       }
0521
0522       static std::unique_ptr<TestGlobalCache> initializeGlobalCache(edm::ParameterSet const& pset) {
0523         auto testGlobalCache = std::make_unique<TestGlobalCache>();
0524         testGlobalCache->trans_ = pset.getParameter<int>("transitions");
0525         return testGlobalCache;
0526       }
0527
0528       static void beginProcessBlockProduce(edm::ProcessBlock& processBlock, TestGlobalCache const* testGlobalCache) {
0529         if (testGlobalCache->m_count != 0) {
0530           throw cms::Exception("transitions") << "TestBeginProcessBlockProducer transitions "
0531                                               << testGlobalCache->m_count << " but it was supposed to be " << 0;
0532         }
0533         ++testGlobalCache->m_count;
0534
0535         const unsigned int valueToPutAndGet = 51;
0536         processBlock.emplace(testGlobalCache->token_, valueToPutAndGet);
0537
0538         if (not testGlobalCache->getTokenBegin_.isUninitialized()) {
0539           if (processBlock.get(testGlobalCache->getTokenBegin_) != valueToPutAndGet) {
0540             throw cms::Exception("BadValue")
0541                 << "expected " << valueToPutAndGet << " but got " << processBlock.get(testGlobalCache->getTokenBegin_);
0542           }
0543         }
0544       }
0545
0546       void produce(edm::Event&, edm::EventSetup const&) override {
0547         TestGlobalCache const* testGlobalCache = globalCache();
0548         if (testGlobalCache->m_count < 1u) {
0549           throw cms::Exception("out of sequence")
0550               << "produce before beginProcessBlockProduce " << testGlobalCache->m_count;
0551         }
0552         ++testGlobalCache->m_count;
0553       }
0554
0555       static void globalEndJob(TestGlobalCache* testGlobalCache) {
0556         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0557           throw cms::Exception("transitions")
0558               << "TestBeginProcessBlockProducer transitions " << testGlobalCache->m_count
0559               << " but it was supposed to be " << testGlobalCache->trans_;
0560         }
0561       }
0562
0563       ~TestBeginProcessBlockProducer() {
0564         TestGlobalCache const* testGlobalCache = globalCache();
0565         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0566           throw cms::Exception("transitions")
0567               << "TestBeginProcessBlockProducer transitions " << testGlobalCache->m_count
0568               << " but it was supposed to be " << testGlobalCache->trans_;
0569         }
0570       }
0571     };
0572
0573     class TestEndProcessBlockProducer
0574         : public edm::stream::EDProducer<edm::EndProcessBlockProducer, edm::GlobalCache<TestGlobalCache>> {
0575     public:
0576       explicit TestEndProcessBlockProducer(edm::ParameterSet const& pset, TestGlobalCache const* testGlobalCache) {
0577         testGlobalCache->token_ = produces<unsigned int, edm::Transition::EndProcessBlock>("end");
0578
0579         auto tag = pset.getParameter<edm::InputTag>("consumesEndProcessBlock");
0580         if (not tag.label().empty()) {
0581           testGlobalCache->getTokenEnd_ = consumes<unsigned int, edm::InProcess>(tag);
0582         }
0583       }
0584
0585       static std::unique_ptr<TestGlobalCache> initializeGlobalCache(edm::ParameterSet const& pset) {
0586         auto testGlobalCache = std::make_unique<TestGlobalCache>();
0587         testGlobalCache->trans_ = pset.getParameter<int>("transitions");
0588         return testGlobalCache;
0589       }
0590
0591       void produce(edm::Event&, edm::EventSetup const&) override {
0592         TestGlobalCache const* testGlobalCache = globalCache();
0593         ++testGlobalCache->m_count;
0594       }
0595
0596       static void endProcessBlockProduce(edm::ProcessBlock& processBlock, TestGlobalCache const* testGlobalCache) {
0597         ++testGlobalCache->m_count;
0598         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0599           throw cms::Exception("transitions") << "TestEndProcessBlockProducer transitions " << testGlobalCache->m_count
0600                                               << " but it was supposed to be " << testGlobalCache->trans_;
0601         }
0602         const unsigned int valueToPutAndGet = 61;
0603         processBlock.emplace(testGlobalCache->token_, valueToPutAndGet);
0604
0605         if (not testGlobalCache->getTokenEnd_.isUninitialized()) {
0606           if (processBlock.get(testGlobalCache->getTokenEnd_) != valueToPutAndGet) {
0607             throw cms::Exception("BadValue")
0608                 << "expected " << valueToPutAndGet << " but got " << processBlock.get(testGlobalCache->getTokenEnd_);
0609           }
0610         }
0611       }
0612
0613       static void globalEndJob(TestGlobalCache* testGlobalCache) {
0614         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0615           throw cms::Exception("transitions")
0616               << "TestBeginProcessBlockProducer transitions " << testGlobalCache->m_count
0617               << " but it was supposed to be " << testGlobalCache->trans_;
0618         }
0619       }
0620
0621       ~TestEndProcessBlockProducer() {
0622         TestGlobalCache const* testGlobalCache = globalCache();
0623         if (testGlobalCache->m_count != testGlobalCache->trans_) {
0624           throw cms::Exception("transitions") << "~TestEndProcessBlockProducer transitions " << testGlobalCache->m_count
0625                                               << " but it was supposed to be " << testGlobalCache->trans_;
0626         }
0627       }
0628     };
0629
0630     class ProcessBlockIntProducerNoGlobalCache : public edm::stream::EDProducer<edm::WatchProcessBlock> {
0631     public:
0632       explicit ProcessBlockIntProducerNoGlobalCache(edm::ParameterSet const&) {}
0633
0634       static void beginProcessBlock(edm::ProcessBlock const&) {}
0635
0636       void produce(edm::Event&, edm::EventSetup const&) override {}
0637
0638       static void endProcessBlock(edm::ProcessBlock const&) {}
0639     };
0640
0641     class TestBeginProcessBlockProducerNoGlobalCache : public edm::stream::EDProducer<edm::BeginProcessBlockProducer> {
0642     public:
0643       explicit TestBeginProcessBlockProducerNoGlobalCache(edm::ParameterSet const&) {}
0644
0645       static void beginProcessBlockProduce(edm::ProcessBlock&) {}
0646
0647       void produce(edm::Event&, edm::EventSetup const&) override {}
0648     };
0649
0650     class TestEndProcessBlockProducerNoGlobalCache : public edm::stream::EDProducer<edm::EndProcessBlockProducer> {
0651     public:
0652       explicit TestEndProcessBlockProducerNoGlobalCache(edm::ParameterSet const&) {}
0653
0654       void produce(edm::Event&, edm::EventSetup const&) override {}
0655
0656       static void endProcessBlockProduce(edm::ProcessBlock&) {}
0657     };
0658
0659     class TestBeginRunProducer : public edm::stream::EDProducer<edm::RunCache<bool>, edm::BeginRunProducer> {
0660     public:
0661       static std::atomic<unsigned int> m_count;
0662       unsigned int trans_;
0663       static std::atomic<unsigned int> cvalue_;
0664       static std::atomic<bool> gbr;
0665       static std::atomic<bool> ger;
0666       static std::atomic<bool> gbrp;
0667
0668       TestBeginRunProducer(edm::ParameterSet const& p) {
0669         trans_ = p.getParameter<int>("transitions");
0670         cvalue_ = p.getParameter<int>("cachevalue");
0671         produces<unsigned int>();
0672         produces<unsigned int, edm::Transition::BeginRun>("a");
0673       }
0674
0675       static std::shared_ptr<bool> globalBeginRun(edm::Run const& iRun, edm::EventSetup const&, GlobalCache const*) {
0676         gbr = true;
0677         ger = false;
0678         gbrp = false;
0679         return std::shared_ptr<bool>{};
0680       }
0681
0682       void produce(edm::Event&, edm::EventSetup const&) override {
0683         if (!gbrp) {
0684           throw cms::Exception("out of sequence") << "produce before globalBeginRunProduce";
0685         }
0686       }
0687
0688       static void globalBeginRunProduce(edm::Run& iRun, edm::EventSetup const&, RunContext const*) {
0689         gbrp = true;
0690         ++m_count;
0691         if (!gbr) {
0692           throw cms::Exception("begin out of sequence") << "globalBeginRunProduce seen before globalBeginRun";
0693         }
0694       }
0695
0696       static void globalEndRun(edm::Run const& iRun, edm::EventSetup const&, RunContext const* iContext) {
0697         if (!gbr) {
0698           throw cms::Exception("end out of sequence") << "globalEndRun seen before globalBeginRun in Run" << iRun.run();
0699         }
0700         gbr = false;
0701         ger = true;
0702       }
0703
0704       ~TestBeginRunProducer() {
0705         if (m_count != trans_) {
0706           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0707         }
0708       }
0709     };
0710
0711     class TestEndRunProducer : public edm::stream::EDProducer<edm::RunCache<bool>, edm::EndRunProducer> {
0712     public:
0713       static std::atomic<unsigned int> m_count;
0714       unsigned int trans_;
0715       static std::atomic<unsigned int> cvalue_;
0716       static std::atomic<bool> gbr;
0717       static std::atomic<bool> ger;
0718       static std::atomic<bool> p;
0719
0720       static std::shared_ptr<bool> globalBeginRun(edm::Run const& iRun, edm::EventSetup const&, GlobalCache const*) {
0721         gbr = true;
0722         ger = false;
0723         p = false;
0724         return std::shared_ptr<bool>{};
0725       }
0726
0727       TestEndRunProducer(edm::ParameterSet const& p) {
0728         trans_ = p.getParameter<int>("transitions");
0729         cvalue_ = p.getParameter<int>("cachevalue");
0730         produces<unsigned int>();
0731         produces<unsigned int, edm::Transition::EndRun>("a");
0732       }
0733
0734       void produce(edm::Event&, edm::EventSetup const&) override { p = true; }
0735
0736       static void globalEndRunProduce(edm::Run& iRun, edm::EventSetup const&, RunContext const*) {
0737         ++m_count;
0738         if (!p) {
0739           throw cms::Exception("out of sequence") << "globalEndRunProduce seen before produce";
0740         }
0741         if (ger) {
0742           throw cms::Exception("out of sequence") << "globalEndRun seen before globalEndRunProduce";
0743         }
0744       }
0745
0746       static void globalEndRun(edm::Run const& iRun, edm::EventSetup const&, RunContext const* iContext) {
0747         if (!gbr) {
0748           throw cms::Exception("out of sequence") << "globalEndRun seen before globalBeginRun in Run" << iRun.run();
0749         }
0750         gbr = false;
0751         ger = true;
0752       }
0753
0754       ~TestEndRunProducer() {
0755         if (m_count != trans_) {
0756           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0757         }
0758       }
0759     };
0760
0761     class TestBeginLumiBlockProducer
0762         : public edm::stream::EDProducer<edm::LuminosityBlockCache<bool>, edm::BeginLuminosityBlockProducer> {
0763     public:
0764       static std::atomic<unsigned int> m_count;
0765       unsigned int trans_;
0766       static std::atomic<unsigned int> cvalue_;
0767       static std::atomic<bool> gbl;
0768       static std::atomic<bool> gel;
0769       static std::atomic<bool> gblp;
0770
0771       TestBeginLumiBlockProducer(edm::ParameterSet const& p) {
0772         trans_ = p.getParameter<int>("transitions");
0773         cvalue_ = p.getParameter<int>("cachevalue");
0774         produces<unsigned int>();
0775         produces<unsigned int, edm::Transition::BeginLuminosityBlock>("a");
0776       }
0777
0778       void produce(edm::Event&, edm::EventSetup const&) override {
0779         if (!gblp) {
0780           throw cms::Exception("begin out of sequence") << "produce seen before globalBeginLumiBlockProduce";
0781         }
0782       }
0783
0784       static void globalBeginLuminosityBlockProduce(edm::LuminosityBlock&,
0785                                                     edm::EventSetup const&,
0786                                                     LuminosityBlockContext const*) {
0787         ++m_count;
0788         if (!gbl) {
0789           throw cms::Exception("begin out of sequence")
0790               << "globalBeginLumiBlockProduce seen before globalBeginLumiBlock";
0791         }
0792         gblp = true;
0793       }
0794
0795       static std::shared_ptr<bool> globalBeginLuminosityBlock(edm::LuminosityBlock const& iLB,
0796                                                               edm::EventSetup const&,
0797                                                               RunContext const*) {
0798         gbl = true;
0799         gel = false;
0800         gblp = false;
0801         return std::shared_ptr<bool>();
0802       }
0803
0804       static void globalEndLuminosityBlock(edm::LuminosityBlock const& iLB,
0805                                            edm::EventSetup const&,
0806                                            LuminosityBlockContext const* iLBContext) {
0807         if (!gbl) {
0808           throw cms::Exception("end out of sequence")
0809               << "globalEndLuminosityBlock seen before globalBeginLuminosityBlock in LuminosityBlock"
0810               << iLB.luminosityBlock();
0811         }
0812         gel = true;
0813         gbl = false;
0814       }
0815
0816       ~TestBeginLumiBlockProducer() {
0817         if (m_count != trans_) {
0818           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0819         }
0820       }
0821     };
0822
0823     class TestEndLumiBlockProducer
0824         : public edm::stream::EDProducer<edm::LuminosityBlockCache<bool>, edm::EndLuminosityBlockProducer> {
0825     public:
0826       static std::atomic<unsigned int> m_count;
0827       unsigned int trans_;
0828       static std::atomic<unsigned int> cvalue_;
0829       static std::atomic<bool> gbl;
0830       static std::atomic<bool> gel;
0831       static std::atomic<bool> p;
0832
0833       TestEndLumiBlockProducer(edm::ParameterSet const& p) {
0834         trans_ = p.getParameter<int>("transitions");
0835         cvalue_ = p.getParameter<int>("cachevalue");
0836         produces<unsigned int>();
0837         produces<unsigned int, edm::Transition::EndLuminosityBlock>("a");
0838       }
0839
0840       void produce(edm::Event&, edm::EventSetup const&) override { p = true; }
0841
0842       static void globalEndLuminosityBlockProduce(edm::LuminosityBlock&,
0843                                                   edm::EventSetup const&,
0844                                                   LuminosityBlockContext const*) {
0845         ++m_count;
0846         if (!p) {
0847           throw cms::Exception("out of sequence") << "globalEndLumiBlockProduce seen before produce";
0848         }
0849       }
0850
0851       static std::shared_ptr<bool> globalBeginLuminosityBlock(edm::LuminosityBlock const& iLB,
0852                                                               edm::EventSetup const&,
0853                                                               RunContext const*) {
0854         gbl = true;
0855         gel = false;
0856         p = false;
0857         return std::shared_ptr<bool>{};
0858       }
0859
0860       static void globalEndLuminosityBlock(edm::LuminosityBlock const& iLB,
0861                                            edm::EventSetup const&,
0862                                            LuminosityBlockContext const* iLBContext) {
0863         if (!gbl) {
0864           throw cms::Exception("end out of sequence")
0865               << "globalEndLuminosityBlock seen before globalBeginLuminosityBlock in LuminosityBlock"
0866               << iLB.luminosityBlock();
0867         }
0868       }
0869
0870       ~TestEndLumiBlockProducer() {
0871         if (m_count != trans_) {
0872           throw cms::Exception("transitions") << m_count << " but it was supposed to be " << trans_;
0873         }
0874       }
0875     };
0876
0877     struct Count {
0878       Count() : m_value(0), m_expectedValue(0) {}
0879       //Using mutable since we want to update the value.
0880       mutable std::atomic<unsigned int> m_value;
0881       mutable std::atomic<unsigned int> m_expectedValue;
0882       // Set only in constructor, framework runs constructors serially
0883       CMS_THREAD_SAFE mutable edm::EDPutTokenT<unsigned int> m_putToken;
0884     };
0885
0886     class TestAccumulator
0887         : public edm::stream::EDProducer<edm::GlobalCache<Count>, edm::Accumulator, edm::EndLuminosityBlockProducer> {
0888     public:
0889       static std::atomic<unsigned int> m_expectedCount;
0890
0891       explicit TestAccumulator(edm::ParameterSet const& p, Count const* iCount) {
0892         iCount->m_expectedValue = p.getParameter<unsigned int>("expectedCount");
0893         if (iCount->m_putToken.isUninitialized()) {
0894           iCount->m_putToken = produces<unsigned int, edm::Transition::EndLuminosityBlock>();
0895         }
0896       }
0897
0898       static std::unique_ptr<Count> initializeGlobalCache(edm::ParameterSet const&) {
0899         return std::unique_ptr<Count>(new Count());
0900       }
0901
0902       void accumulate(edm::Event const&, edm::EventSetup const&) override { ++(globalCache()->m_value); }
0903
0904       static void globalEndLuminosityBlockProduce(edm::LuminosityBlock& l,
0905                                                   edm::EventSetup const&,
0906                                                   LuminosityBlockContext const* ctx) {
0907         Count const* count = ctx->global();
0908         l.emplace(count->m_putToken, count->m_value.load());
0909       }
0910
0911       static void globalEndJob(Count const* iCount) {
0912         if (iCount->m_value != iCount->m_expectedValue) {
0913           throw cms::Exception("CountEvents") << "Number of events seen = " << iCount->m_value
0914                                               << " but it was supposed to be " << iCount->m_expectedValue;
0915         }
0916       }
0917
0918       ~TestAccumulator() {}
0919     };
0920
0921     class TestInputProcessBlockCache {
0922     public:
0923       long long int value_ = 0;
0924     };
0925
0926     class TestInputProcessBlockCache1 {
0927     public:
0928       long long int value_ = 0;
0929     };
0930
0931     class InputProcessBlockIntProducer
0932         : public edm::stream::EDProducer<
0933               edm::InputProcessBlockCache<int, TestInputProcessBlockCache, TestInputProcessBlockCache1>> {
0934     public:
0935       explicit InputProcessBlockIntProducer(edm::ParameterSet const& pset) {
0936         {
0937           expectedByRun_ = pset.getParameter<std::vector<int>>("expectedByRun");
0938           sleepTime_ = pset.getParameter<unsigned int>("sleepTime");
0939           auto tag = pset.getParameter<edm::InputTag>("consumesBeginProcessBlock");
0940           if (not tag.label().empty()) {
0941             getTokenBegin_ = consumes<IntProduct, edm::InProcess>(tag);
0942           }
0943         }
0944         {
0945           auto tag = pset.getParameter<edm::InputTag>("consumesEndProcessBlock");
0946           if (not tag.label().empty()) {
0947             getTokenEnd_ = consumes<IntProduct, edm::InProcess>(tag);
0948           }
0949         }
0950         registerProcessBlockCacheFiller<TestInputProcessBlockCache1>(
0951             getTokenBegin_,
0952             [this](edm::ProcessBlock const& processBlock,
0953                    std::shared_ptr<TestInputProcessBlockCache1> const& previousCache) {
0954               auto returnValue = std::make_shared<TestInputProcessBlockCache1>();
0955               returnValue->value_ += processBlock.get(getTokenBegin_).value;
0956               returnValue->value_ += processBlock.get(getTokenEnd_).value;
0957               return returnValue;
0958             });
0959       }
0960
0961       static void accessInputProcessBlock(edm::ProcessBlock const&) {
0962         edm::LogAbsolute("InputProcessBlockIntProducer") << "InputProcessBlockIntProducer::accessInputProcessBlock";
0963       }
0964
0965       void produce(edm::Event& event, edm::EventSetup const&) override {
0966         auto cacheTuple = processBlockCaches(event);
0967         if (!expectedByRun_.empty()) {
0968           if (expectedByRun_.at(event.run() - 1) !=
0969               std::get<edm::CacheHandle<TestInputProcessBlockCache1>>(cacheTuple)->value_) {
0970             throw cms::Exception("UnexpectedValue")
0971                 << "InputProcessBlockIntProducer::produce cached value was "
0972                 << std::get<edm::CacheHandle<TestInputProcessBlockCache1>>(cacheTuple)->value_
0973                 << " but it was supposed to be " << expectedByRun_.at(event.run() - 1);
0974           }
0975         }
0976         // Force events to be processed concurrently
0977         if (sleepTime_ > 0) {
0978           usleep(sleepTime_);
0979         }
0980       }
0981
0982     private:
0983       edm::EDGetTokenT<IntProduct> getTokenBegin_;
0984       edm::EDGetTokenT<IntProduct> getTokenEnd_;
0985       std::vector<int> expectedByRun_;
0986       unsigned int sleepTime_{0};
0987     };
0988
0989     struct InputProcessBlockGlobalCacheAn {
0990       // The tokens are duplicated in this test module to prove that they
0991       // work both as GlobalCache members and module data members.
0992       // We need them as GlobalCache members for accessInputProcessBlock.
0993       // In registerProcessBlockCacheFiller we use tokens that are member
0994       // variables of the class and because the lambda captures the "this"
0995       // pointer of the zeroth stream module instance. We always
0996       // use the zeroth EDConsumer. In the case of registerProcessBlockCacheFiller,
0997       // either set of tokens would work. Note that in the GlobalCache case
0998       // there is a slight weirdness that the zeroth consumer is used but
0999       // the token comes from the last consumer instance. It works because
1000       // all the stream module instances have EDConsumer base classes with
1001       // containers with the same contents in the same order (not 100% guaranteed,
1002       // but it would be difficult to implement a module where this isn't true).
1003       CMS_THREAD_SAFE mutable edm::EDGetTokenT<IntProduct> getTokenBegin_;
1004       CMS_THREAD_SAFE mutable edm::EDGetTokenT<IntProduct> getTokenEnd_;
1005       CMS_THREAD_SAFE mutable edm::EDGetTokenT<IntProduct> getTokenBeginM_;
1006       CMS_THREAD_SAFE mutable edm::EDGetTokenT<IntProduct> getTokenEndM_;
1007       mutable std::atomic<unsigned int> transitions_{0};
1008       int sum_{0};
1009       unsigned int expectedTransitions_{0};
1010       std::vector<int> expectedByRun_;
1011       int expectedSum_{0};
1012       unsigned int sleepTime_{0};
1013     };
1014
1015     // Same thing as previous class except with a GlobalCache added
1016     class InputProcessBlockIntProducerG
1017         : public edm::stream::EDProducer<
1018               edm::InputProcessBlockCache<int, TestInputProcessBlockCache, TestInputProcessBlockCache1>,
1019               edm::GlobalCache<InputProcessBlockGlobalCacheAn>> {
1020     public:
1021       explicit InputProcessBlockIntProducerG(edm::ParameterSet const& pset,
1022                                              InputProcessBlockGlobalCacheAn const* testGlobalCache) {
1023         {
1024           auto tag = pset.getParameter<edm::InputTag>("consumesBeginProcessBlock");
1025           if (not tag.label().empty()) {
1026             getTokenBegin_ = consumes<IntProduct, edm::InProcess>(tag);
1027             testGlobalCache->getTokenBegin_ = getTokenBegin_;
1028           }
1029         }
1030         {
1031           auto tag = pset.getParameter<edm::InputTag>("consumesEndProcessBlock");
1032           if (not tag.label().empty()) {
1033             getTokenEnd_ = consumes<IntProduct, edm::InProcess>(tag);
1034             testGlobalCache->getTokenEnd_ = getTokenEnd_;
1035           }
1036         }
1037         {
1038           auto tag = pset.getParameter<edm::InputTag>("consumesBeginProcessBlockM");
1039           if (not tag.label().empty()) {
1040             getTokenBeginM_ = consumes<IntProduct, edm::InProcess>(tag);
1041             testGlobalCache->getTokenBeginM_ = getTokenBeginM_;
1042           }
1043         }
1044         {
1045           auto tag = pset.getParameter<edm::InputTag>("consumesEndProcessBlockM");
1046           if (not tag.label().empty()) {
1047             getTokenEndM_ = consumes<IntProduct, edm::InProcess>(tag);
1048             testGlobalCache->getTokenEndM_ = getTokenEndM_;
1049           }
1050         }
1051         registerProcessBlockCacheFiller<int>(
1052             getTokenBegin_, [this](edm::ProcessBlock const& processBlock, std::shared_ptr<int> const& previousCache) {
1053               auto returnValue = std::make_shared<int>(0);
1054               *returnValue += processBlock.get(getTokenBegin_).value;
1055               *returnValue += processBlock.get(getTokenEnd_).value;
1056               ++globalCache()->transitions_;
1057               return returnValue;
1058             });
1059         registerProcessBlockCacheFiller<1>(getTokenBegin_,
1060                                            [this](edm::ProcessBlock const& processBlock,
1061                                                   std::shared_ptr<TestInputProcessBlockCache> const& previousCache) {
1062                                              auto returnValue = std::make_shared<TestInputProcessBlockCache>();
1063                                              returnValue->value_ += processBlock.get(getTokenBegin_).value;
1064                                              returnValue->value_ += processBlock.get(getTokenEnd_).value;
1065                                              ++globalCache()->transitions_;
1066                                              return returnValue;
1067                                            });
1068         registerProcessBlockCacheFiller<TestInputProcessBlockCache1>(
1069             getTokenBegin_,
1070             [this](edm::ProcessBlock const& processBlock,
1071                    std::shared_ptr<TestInputProcessBlockCache1> const& previousCache) {
1072               auto returnValue = std::make_shared<TestInputProcessBlockCache1>();
1073               returnValue->value_ += processBlock.get(getTokenBegin_).value;
1074               returnValue->value_ += processBlock.get(getTokenEnd_).value;
1075               ++globalCache()->transitions_;
1076               return returnValue;
1077             });
1078       }
1079
1080       static std::unique_ptr<InputProcessBlockGlobalCacheAn> initializeGlobalCache(edm::ParameterSet const& pset) {
1081         auto testGlobalCache = std::make_unique<InputProcessBlockGlobalCacheAn>();
1082         testGlobalCache->expectedTransitions_ = pset.getParameter<int>("transitions");
1083         testGlobalCache->expectedByRun_ = pset.getParameter<std::vector<int>>("expectedByRun");
1084         testGlobalCache->expectedSum_ = pset.getParameter<int>("expectedSum");
1085         testGlobalCache->sleepTime_ = pset.getParameter<unsigned int>("sleepTime");
1086         return testGlobalCache;
1087       }
1088
1089       static void accessInputProcessBlock(edm::ProcessBlock const& processBlock,
1090                                           InputProcessBlockGlobalCacheAn* testGlobalCache) {
1091         if (processBlock.processName() == "PROD1") {
1092           testGlobalCache->sum_ += processBlock.get(testGlobalCache->getTokenBegin_).value;
1093           testGlobalCache->sum_ += processBlock.get(testGlobalCache->getTokenEnd_).value;
1094         }
1095         if (processBlock.processName() == "MERGE") {
1096           testGlobalCache->sum_ += processBlock.get(testGlobalCache->getTokenBeginM_).value;
1097           testGlobalCache->sum_ += processBlock.get(testGlobalCache->getTokenEndM_).value;
1098         }
1099         ++testGlobalCache->transitions_;
1100       }
1101
1102       void produce(edm::Event& event, edm::EventSetup const&) override {
1103         auto cacheTuple = processBlockCaches(event);
1104         auto testGlobalCache = globalCache();
1105         if (!testGlobalCache->expectedByRun_.empty()) {
1106           if (testGlobalCache->expectedByRun_.at(event.run() - 1) != *std::get<edm::CacheHandle<int>>(cacheTuple)) {
1107             throw cms::Exception("UnexpectedValue")
1108                 << "InputProcessBlockIntProducerG::produce cached value was "
1109                 << *std::get<edm::CacheHandle<int>>(cacheTuple) << " but it was supposed to be "
1110                 << testGlobalCache->expectedByRun_.at(event.run() - 1);
1111           }
1112           if (testGlobalCache->expectedByRun_.at(event.run() - 1) != std::get<1>(cacheTuple)->value_) {
1113             throw cms::Exception("UnexpectedValue")
1114                 << "InputProcessBlockIntProducerG::produce second cached value was " << std::get<1>(cacheTuple)->value_
1115                 << " but it was supposed to be " << testGlobalCache->expectedByRun_.at(event.run() - 1);
1116           }
1117           if (testGlobalCache->expectedByRun_.at(event.run() - 1) !=
1118               std::get<edm::CacheHandle<TestInputProcessBlockCache1>>(cacheTuple)->value_) {
1119             throw cms::Exception("UnexpectedValue")
1120                 << "InputProcessBlockIntProducerG::produce third cached value was "
1121                 << std::get<edm::CacheHandle<TestInputProcessBlockCache1>>(cacheTuple)->value_
1122                 << " but it was supposed to be " << testGlobalCache->expectedByRun_.at(event.run() - 1);
1123           }
1124         }
1125         ++testGlobalCache->transitions_;
1126
1127         // Force events to be processed concurrently
1128         if (testGlobalCache->sleepTime_ > 0) {
1129           usleep(testGlobalCache->sleepTime_);
1130         }
1131       }
1132
1133       static void globalEndJob(InputProcessBlockGlobalCacheAn* testGlobalCache) {
1134         if (testGlobalCache->transitions_ != testGlobalCache->expectedTransitions_) {
1135           throw cms::Exception("transitions")
1136               << "InputProcessBlockIntProducerG transitions " << testGlobalCache->transitions_
1137               << " but it was supposed to be " << testGlobalCache->expectedTransitions_;
1138         }
1139
1140         if (testGlobalCache->sum_ != testGlobalCache->expectedSum_) {
1141           throw cms::Exception("UnexpectedValue") << "InputProcessBlockIntProducerG sum " << testGlobalCache->sum_
1142                                                   << " but it was supposed to be " << testGlobalCache->expectedSum_;
1143         }
1144       }
1145
1146     private:
1147       edm::EDGetTokenT<IntProduct> getTokenBegin_;
1148       edm::EDGetTokenT<IntProduct> getTokenEnd_;
1149       edm::EDGetTokenT<IntProduct> getTokenBeginM_;
1150       edm::EDGetTokenT<IntProduct> getTokenEndM_;
1151     };
1152
1153   }  // namespace stream
1154 }  // namespace edmtest
1155 std::atomic<unsigned int> edmtest::stream::GlobalIntProducer::m_count{0};
1156 std::atomic<unsigned int> edmtest::stream::RunIntProducer::m_count{0};
1157 std::atomic<unsigned int> edmtest::stream::LumiIntProducer::m_count{0};
1158 std::atomic<unsigned int> edmtest::stream::RunSummaryIntProducer::m_count{0};
1159 std::atomic<unsigned int> edmtest::stream::LumiSummaryIntProducer::m_count{0};
1160 std::atomic<unsigned int> edmtest::stream::TestBeginRunProducer::m_count{0};
1161 std::atomic<unsigned int> edmtest::stream::TestEndRunProducer::m_count{0};
1162 std::atomic<unsigned int> edmtest::stream::TestBeginLumiBlockProducer::m_count{0};
1163 std::atomic<unsigned int> edmtest::stream::TestEndLumiBlockProducer::m_count{0};
1164 std::atomic<unsigned int> edmtest::stream::GlobalIntProducer::cvalue_{0};
1165 std::atomic<unsigned int> edmtest::stream::RunIntProducer::cvalue_{0};
1166 std::atomic<unsigned int> edmtest::stream::LumiIntProducer::cvalue_{0};
1167 std::atomic<unsigned int> edmtest::stream::RunSummaryIntProducer::cvalue_{0};
1168 std::atomic<unsigned int> edmtest::stream::LumiSummaryIntProducer::cvalue_{0};
1169 std::atomic<unsigned int> edmtest::stream::TestBeginRunProducer::cvalue_{0};
1170 std::atomic<unsigned int> edmtest::stream::TestEndRunProducer::cvalue_{0};
1171 std::atomic<unsigned int> edmtest::stream::TestBeginLumiBlockProducer::cvalue_{0};
1172 std::atomic<unsigned int> edmtest::stream::TestEndLumiBlockProducer::cvalue_{0};
1173 std::atomic<bool> edmtest::stream::RunSummaryIntProducer::globalBeginRunCalled_{false};
1174 std::atomic<bool> edmtest::stream::LumiSummaryIntProducer::globalBeginLumiCalled_{false};
1175 std::atomic<bool> edmtest::stream::TestBeginRunProducer::gbr{false};
1176 std::atomic<bool> edmtest::stream::TestBeginRunProducer::gbrp{false};
1177 std::atomic<bool> edmtest::stream::TestBeginRunProducer::ger{false};
1178 std::atomic<bool> edmtest::stream::TestEndRunProducer::gbr{false};
1179 std::atomic<bool> edmtest::stream::TestEndRunProducer::ger{false};
1180 std::atomic<bool> edmtest::stream::TestEndRunProducer::p{false};
1181 std::atomic<bool> edmtest::stream::TestBeginLumiBlockProducer::gbl{false};
1182 std::atomic<bool> edmtest::stream::TestBeginLumiBlockProducer::gblp{false};
1183 std::atomic<bool> edmtest::stream::TestBeginLumiBlockProducer::gel{false};
1184 std::atomic<bool> edmtest::stream::TestEndLumiBlockProducer::gbl{false};
1185 std::atomic<bool> edmtest::stream::TestEndLumiBlockProducer::gel{false};
1186 std::atomic<bool> edmtest::stream::TestEndLumiBlockProducer::p{false};
1187 DEFINE_FWK_MODULE(edmtest::stream::GlobalIntProducer);
1188 DEFINE_FWK_MODULE(edmtest::stream::RunIntProducer);
1189 DEFINE_FWK_MODULE(edmtest::stream::LumiIntProducer);
1190 DEFINE_FWK_MODULE(edmtest::stream::RunSummaryIntProducer);
1191 DEFINE_FWK_MODULE(edmtest::stream::LumiSummaryIntProducer);
1192 DEFINE_FWK_MODULE(edmtest::stream::ProcessBlockIntProducer);
1193 DEFINE_FWK_MODULE(edmtest::stream::TestBeginProcessBlockProducer);
1194 DEFINE_FWK_MODULE(edmtest::stream::TestEndProcessBlockProducer);
1195 DEFINE_FWK_MODULE(edmtest::stream::ProcessBlockIntProducerNoGlobalCache);
1196 DEFINE_FWK_MODULE(edmtest::stream::TestBeginProcessBlockProducerNoGlobalCache);
1197 DEFINE_FWK_MODULE(edmtest::stream::TestEndProcessBlockProducerNoGlobalCache);
1198 DEFINE_FWK_MODULE(edmtest::stream::TestBeginRunProducer);
1199 DEFINE_FWK_MODULE(edmtest::stream::TestEndRunProducer);
1200 DEFINE_FWK_MODULE(edmtest::stream::TestBeginLumiBlockProducer);
1201 DEFINE_FWK_MODULE(edmtest::stream::TestEndLumiBlockProducer);
1202 DEFINE_FWK_MODULE(edmtest::stream::TestAccumulator);
1203 DEFINE_FWK_MODULE(edmtest::stream::InputProcessBlockIntProducer);
1204 DEFINE_FWK_MODULE(edmtest::stream::InputProcessBlockIntProducerG);