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