Integration/plugins/TestESConcurrentSource.cc

0001 // -*- C++ -*-
0002 //
0003 // Package:     FWCore/Integration
0004 // Class  :     TestESConcurrentSource
0005 //
0006 // Implementation:
0007 //     ESSource used for tests of Framework support for
0008 //     ESSources and ESProducers. This is primarily focused
0009 //     on the infrastructure used by CondDBESSource.
0010 //
0011 // Original Author:  C Jones
0012 //         Created:  16 Dec 2021
0013
0014 #include "DataFormats/Provenance/interface/EventID.h"
0015 #include "FWCore/Framework/interface/DataKey.h"
0016 #include "FWCore/Framework/interface/ESSourceProductResolverConcurrentBase.h"
0017 #include "FWCore/Framework/interface/ESProductResolverProvider.h"
0018 #include "FWCore/Framework/interface/EventSetupRecordIntervalFinder.h"
0019 #include "FWCore/Framework/interface/EventSetupRecordKey.h"
0020 #include "FWCore/Framework/interface/IOVSyncValue.h"
0021 #include "FWCore/Framework/interface/SourceFactory.h"
0022 #include "FWCore/Framework/interface/ValidityInterval.h"
0023 #include "FWCore/Framework/interface/ESModuleProducesInfo.h"
0024 #include "FWCore/Integration/interface/ESTestRecords.h"
0025 #include "FWCore/Integration/interface/IOVTestInfo.h"
0026 #include "FWCore/MessageLogger/interface/MessageLogger.h"
0027 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
0028 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0029 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
0030 #include "FWCore/Utilities/interface/Exception.h"
0031
0032 #include <atomic>
0033 #include <cmath>
0034 #include <limits>
0035 #include <set>
0036 #include <utility>
0037 #include <vector>
0038
0039 namespace edmtest {
0040
0041   class TestESConcurrentSource;
0042
0043   class TestESConcurrentSourceTestResolver : public edm::eventsetup::ESSourceProductResolverConcurrentBase {
0044   public:
0045     TestESConcurrentSourceTestResolver(TestESConcurrentSource* TestESConcurrentSource);
0046
0047   private:
0048     void prefetch(edm::eventsetup::DataKey const&, edm::EventSetupRecordDetails) override;
0049     void initializeForNewIOV() override;
0050     void const* getAfterPrefetchImpl() const override;
0051
0052     IOVTestInfo iovTestInfo_;
0053     TestESConcurrentSource* testESConcurrentSource_;
0054   };
0055
0056   class TestESConcurrentSource : public edm::eventsetup::ESProductResolverProvider,
0057                                  public edm::EventSetupRecordIntervalFinder {
0058   public:
0059     using EventSetupRecordKey = edm::eventsetup::EventSetupRecordKey;
0060     explicit TestESConcurrentSource(edm::ParameterSet const&);
0061     ~TestESConcurrentSource() override;
0062
0063     static void fillDescriptions(edm::ConfigurationDescriptions& descriptions);
0064
0065     void busyWait(char const* msg) const;
0066
0067     void incrementCount() {
0068       auto const v = ++count_;
0069       auto m = maxCount_.load();
0070       while (m < v) {
0071         maxCount_.compare_exchange_strong(m, v);
0072       }
0073     }
0074     std::atomic<unsigned int> count_;
0075     std::atomic<unsigned int> maxCount_;
0076     std::atomic<unsigned int> count_setIntervalFor_;
0077     std::atomic<unsigned int> count_initializeForNewIOV_;
0078
0079     std::vector<edm::eventsetup::ESModuleProducesInfo> producesInfo() const override;
0080
0081   private:
0082     bool isConcurrentFinder() const override { return true; }
0083     void setIntervalFor(EventSetupRecordKey const&, edm::IOVSyncValue const&, edm::ValidityInterval&) override;
0084     KeyedResolversVector registerResolvers(EventSetupRecordKey const&, unsigned int iovIndex) override;
0085     void initConcurrentIOVs(EventSetupRecordKey const&, unsigned int nConcurrentIOVs) override;
0086
0087     std::set<edm::IOVSyncValue> setOfIOV_;
0088     const unsigned int iterations_;
0089     const double pi_;
0090     unsigned int expectedNumberOfConcurrentIOVs_;
0091     unsigned int nConcurrentIOVs_ = 0;
0092     bool checkIOVInitialization_;
0093   };
0094
0095   TestESConcurrentSourceTestResolver::TestESConcurrentSourceTestResolver(TestESConcurrentSource* testESConcurrentSource)
0096       : edm::eventsetup::ESSourceProductResolverConcurrentBase(), testESConcurrentSource_(testESConcurrentSource) {}
0097
0098   void TestESConcurrentSourceTestResolver::prefetch(edm::eventsetup::DataKey const& iKey,
0099                                                     edm::EventSetupRecordDetails iRecord) {
0100     testESConcurrentSource_->incrementCount();
0101     testESConcurrentSource_->busyWait((std::string("getImpl ") + iKey.name().value()).c_str());
0102
0103     edm::ValidityInterval iov = iRecord.validityInterval();
0104     edm::LogAbsolute("TestESConcurrentSourceTestResolver")
0105         << "TestESConcurrentSourceTestResolver::getImpl '" << iKey.name().value()
0106         << "' startIOV = " << iov.first().luminosityBlockNumber() << " endIOV = " << iov.last().luminosityBlockNumber()
0107         << " IOV index = " << iRecord.iovIndex() << " cache identifier = " << iRecord.cacheIdentifier();
0108
0109     iovTestInfo_.iovStartLumi_ = iov.first().luminosityBlockNumber();
0110     iovTestInfo_.iovEndLumi_ = iov.last().luminosityBlockNumber();
0111     iovTestInfo_.iovIndex_ = iRecord.iovIndex();
0112     iovTestInfo_.cacheIdentifier_ = iRecord.cacheIdentifier();
0113
0114     --testESConcurrentSource_->count_;
0115   }
0116
0117   void const* TestESConcurrentSourceTestResolver::getAfterPrefetchImpl() const { return &iovTestInfo_; }
0118
0119   void TestESConcurrentSourceTestResolver::initializeForNewIOV() {
0120     edm::LogAbsolute("TestESConcurrentSourceTestResolver::initializeForNewIOV")
0121         << "TestESConcurrentSourceTestResolver::initializeForNewIOV";
0122     ++testESConcurrentSource_->count_initializeForNewIOV_;
0123   }
0124
0125   TestESConcurrentSource::TestESConcurrentSource(edm::ParameterSet const& pset)
0126       : count_(0),
0127         maxCount_(0),
0128         count_setIntervalFor_(0),
0129         count_initializeForNewIOV_(0),
0130         iterations_(pset.getParameter<unsigned int>("iterations")),
0131         pi_(std::acos(-1)),
0132         expectedNumberOfConcurrentIOVs_(pset.getParameter<unsigned int>("expectedNumberOfConcurrentIOVs")),
0133         checkIOVInitialization_(pset.getParameter<bool>("checkIOVInitialization")) {
0134     std::vector<unsigned int> temp(pset.getParameter<std::vector<unsigned int>>("firstValidLumis"));
0135     for (auto val : temp) {
0136       setOfIOV_.insert(edm::IOVSyncValue(edm::EventID(1, val, 0)));
0137     }
0138
0139     findingRecord<ESTestRecordI>();
0140     usingRecord<ESTestRecordI>();
0141   }
0142
0143   TestESConcurrentSource::~TestESConcurrentSource() {
0144     edm::LogAbsolute("TestESConcurrentSource") << "max concurrency seen " << maxCount_.load();
0145   }
0146
0147   void TestESConcurrentSource::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
0148     edm::ParameterSetDescription desc;
0149     std::vector<unsigned int> emptyVector;
0150     desc.add<unsigned int>("iterations", 10 * 1000 * 1000);
0151     desc.add<bool>("checkIOVInitialization", false);
0152     desc.add<unsigned int>("expectedNumberOfConcurrentIOVs", 0);
0153     desc.add<std::vector<unsigned int>>("firstValidLumis", emptyVector);
0154     descriptions.addDefault(desc);
0155   }
0156
0157   void TestESConcurrentSource::setIntervalFor(EventSetupRecordKey const&,
0158                                               edm::IOVSyncValue const& syncValue,
0159                                               edm::ValidityInterval& iov) {
0160     if (checkIOVInitialization_) {
0161       // Note that this check should pass with the specific configuration where I enable
0162       // the check, but in general it does not have to be true. The counts are offset
0163       // by 1 because the beginRun IOV is invalid (no IOV initialization).
0164       if (count_setIntervalFor_ > 0 && count_initializeForNewIOV_ != 2 * (count_setIntervalFor_ - 1)) {
0165         throw cms::Exception("TestFailure") << "TestESConcurrentSource::setIntervalFor,"
0166                                             << " unexpected number of IOV initializations";
0167       }
0168     }
0169     incrementCount();
0170     ++count_setIntervalFor_;
0171     busyWait("setIntervalFor");
0172     iov = edm::ValidityInterval::invalidInterval();
0173
0174     if (setOfIOV_.empty()) {
0175       --count_;
0176       return;
0177     }
0178
0179     std::pair<std::set<edm::IOVSyncValue>::iterator, std::set<edm::IOVSyncValue>::iterator> itFound =
0180         setOfIOV_.equal_range(syncValue);
0181
0182     if (itFound.first == itFound.second) {
0183       if (itFound.first == setOfIOV_.begin()) {
0184         //request is before first valid interval, so fail
0185         --count_;
0186         return;
0187       }
0188       //go back one step
0189       --itFound.first;
0190     }
0191
0192     edm::IOVSyncValue endOfInterval = edm::IOVSyncValue::endOfTime();
0193     if (itFound.second != setOfIOV_.end()) {
0194       endOfInterval = edm::IOVSyncValue(
0195           edm::EventID(1, itFound.second->eventID().luminosityBlock() - 1, edm::EventID::maxEventNumber()));
0196     }
0197     iov = edm::ValidityInterval(*(itFound.first), endOfInterval);
0198     --count_;
0199   }
0200
0201   std::vector<edm::eventsetup::ESModuleProducesInfo> TestESConcurrentSource::producesInfo() const {
0202     std::vector<edm::eventsetup::ESModuleProducesInfo> producesInfo;
0203     producesInfo.emplace_back(ESTestRecordI::keyForClass(),
0204                               edm::eventsetup::DataKey(edm::eventsetup::DataKey::makeTypeTag<IOVTestInfo>(), ""),
0205                               0);
0206     producesInfo.emplace_back(ESTestRecordI::keyForClass(),
0207                               edm::eventsetup::DataKey(edm::eventsetup::DataKey::makeTypeTag<IOVTestInfo>(), "other"),
0208                               1);
0209     return producesInfo;
0210   }
0211
0212   edm::eventsetup::ESProductResolverProvider::KeyedResolversVector TestESConcurrentSource::registerResolvers(
0213       EventSetupRecordKey const&, unsigned int iovIndex) {
0214     if (expectedNumberOfConcurrentIOVs_ != 0 && nConcurrentIOVs_ != expectedNumberOfConcurrentIOVs_) {
0215       throw cms::Exception("TestFailure") << "TestESConcurrentSource::registerResolvers,"
0216                                           << " unexpected number of concurrent IOVs";
0217     }
0218     KeyedResolversVector keyedResolversVector;
0219
0220     {
0221       edm::eventsetup::DataKey dataKey(edm::eventsetup::DataKey::makeTypeTag<IOVTestInfo>(),
0222                                        edm::eventsetup::IdTags(""));
0223       keyedResolversVector.emplace_back(dataKey, std::make_shared<TestESConcurrentSourceTestResolver>(this));
0224     }
0225     {
0226       edm::eventsetup::DataKey dataKey(edm::eventsetup::DataKey::makeTypeTag<IOVTestInfo>(),
0227                                        edm::eventsetup::IdTags("other"));
0228       keyedResolversVector.emplace_back(dataKey, std::make_shared<TestESConcurrentSourceTestResolver>(this));
0229     }
0230
0231     return keyedResolversVector;
0232   }
0233
0234   void TestESConcurrentSource::initConcurrentIOVs(EventSetupRecordKey const& key, unsigned int nConcurrentIOVs) {
0235     edm::LogAbsolute("TestESConcurrentSource::initConcurrentIOVs")
0236         << "Start TestESConcurrentSource::initConcurrentIOVs " << nConcurrentIOVs << " " << key.name();
0237     if (EventSetupRecordKey::makeKey<ESTestRecordI>() != key) {
0238       throw cms::Exception("TestFailure") << "TestESConcurrentSource::initConcurrentIOVs,"
0239                                           << " unexpected EventSetupRecordKey";
0240     }
0241     if (expectedNumberOfConcurrentIOVs_ != 0 && nConcurrentIOVs != expectedNumberOfConcurrentIOVs_) {
0242       throw cms::Exception("TestFailure") << "TestESConcurrentSource::initConcurrentIOVs,"
0243                                           << " unexpected number of concurrent IOVs";
0244     }
0245     nConcurrentIOVs_ = nConcurrentIOVs;
0246   }
0247
0248   void TestESConcurrentSource::busyWait(char const* msg) const {
0249     edm::LogAbsolute("TestESConcurrentSource::busyWait") << "Start TestESConcurrentSource::busyWait " << msg;
0250     double sum = 0.;
0251     const double stepSize = pi_ / iterations_;
0252     for (unsigned int i = 0; i < iterations_; ++i) {
0253       sum += stepSize * cos(i * stepSize);
0254     }
0255     edm::LogAbsolute("TestESConcurrentSource::busyWait")
0256         << "Stop TestESConcurrentSource::busyWait " << msg << " " << sum;
0257   }
0258 }  // namespace edmtest
0259 using namespace edmtest;
0260 DEFINE_FWK_EVENTSETUP_SOURCE(TestESConcurrentSource);