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