Integration/plugins/AcquireIntESProducer.cc

0001 // -*- C++ -*-
0002 //
0003 // Package:    FWCore/Integration
0004 // Class:      AcquireIntESProducer
0005 //
0006 /**\class edmtest::AcquireIntESProducer
0007
0008   Description: Used in tests of the asynchronous ESProducer.
0009 */
0010 // Original Author:  W. David Dagenhart
0011 //         Created:  12 January 2023
0012
0013 #include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
0014 #include "FWCore/Framework/interface/ESHandle.h"
0015 #include "FWCore/Framework/interface/ESProducerExternalWork.h"
0016 #include "FWCore/Framework/interface/EventSetupRecordKey.h"
0017 #include "FWCore/Framework/interface/ModuleFactory.h"
0018 #include "FWCore/Framework/interface/ValidityInterval.h"
0019 #include "FWCore/Integration/interface/ESTestData.h"
0020 #include "FWCore/Integration/interface/ESTestRecords.h"
0021 #include "FWCore/Integration/interface/IOVTestInfo.h"
0022 #include "WaitingServer.h"
0023 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
0024 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0025 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
0026 #include "FWCore/Utilities/interface/ESGetToken.h"
0027 #include "FWCore/Utilities/interface/ESInputTag.h"
0028 #include "FWCore/Utilities/interface/Exception.h"
0029
0030 #include <memory>
0031 #include <optional>
0032 #include <unistd.h>
0033 #include <vector>
0034
0035 namespace {
0036   constexpr int kAcquireTestValue = 11;
0037   constexpr int kAcquireTestValueUniquePtr1 = 101;
0038   constexpr int kAcquireTestValueUniquePtr2 = 102;
0039   constexpr int kAcquireTestValueOptional1 = 201;
0040   constexpr int kAcquireTestValueOptional2 = 202;
0041 }  // namespace
0042
0043 namespace edmtest {
0044
0045   class AcquireIntESProducer : public edm::ESProducerExternalWork {
0046   public:
0047     AcquireIntESProducer(edm::ParameterSet const&);
0048
0049     ~AcquireIntESProducer() override;
0050     AcquireIntESProducer(const AcquireIntESProducer&) = delete;
0051     AcquireIntESProducer& operator=(const AcquireIntESProducer&) = delete;
0052     AcquireIntESProducer(AcquireIntESProducer&&) = delete;
0053     AcquireIntESProducer& operator=(AcquireIntESProducer&&) = delete;
0054
0055     void initConcurrentIOVs(EventSetupRecordKey const&, unsigned int nConcurrentIOVs) override;
0056
0057     int acquire(ESTestRecordI const&, edm::WaitingTaskWithArenaHolder);
0058
0059     std::unique_ptr<ESTestDataI> produce(ESTestRecordI const&, int);
0060
0061     std::unique_ptr<ESTestDataB> produceESTestDataB(ESTestRecordB const&);
0062
0063     class TestValue {
0064     public:
0065       TestValue(int value) : value_(value) {}
0066       int value_;
0067     };
0068
0069     std::unique_ptr<TestValue> acquireUniquePtr(ESTestRecordI const&, edm::WaitingTaskWithArenaHolder);
0070
0071     std::unique_ptr<ESTestDataI> produceUniquePtr(ESTestRecordI const&, std::unique_ptr<TestValue>);
0072
0073     std::optional<std::vector<TestValue>> acquireOptional(ESTestRecordI const&, edm::WaitingTaskWithArenaHolder);
0074
0075     std::unique_ptr<ESTestDataI> produceOptional(ESTestRecordI const&, std::optional<std::vector<TestValue>>);
0076
0077     static void fillDescriptions(edm::ConfigurationDescriptions&);
0078
0079   private:
0080     edm::ESGetToken<IOVTestInfo, ESTestRecordI> token_;
0081     std::vector<test_acquire::Cache> caches_;
0082     std::unique_ptr<test_acquire::WaitingServer> server_;
0083     const unsigned int numberOfIOVsToAccumulate_;
0084     const unsigned int secondsToWaitForWork_;
0085     std::vector<TestValue*> uniqueTestPointers_;
0086     std::vector<TestValue*> optionalTestPointers_;
0087     std::vector<TestValue*> lambdaUniqueTestPointers_;
0088     std::vector<TestValue*> lambdaOptionalTestPointers_;
0089   };
0090
0091   AcquireIntESProducer::AcquireIntESProducer(edm::ParameterSet const& pset)
0092       : numberOfIOVsToAccumulate_(pset.getUntrackedParameter<unsigned int>("numberOfIOVsToAccumulate")),
0093         secondsToWaitForWork_(pset.getUntrackedParameter<unsigned int>("secondsToWaitForWork")) {
0094     auto collector = setWhatAcquiredProduced(this, "fromAcquireIntESProducer");
0095     token_ = collector.consumes<IOVTestInfo>(edm::ESInputTag{"", ""});
0096
0097     setWhatProduced(this, &edmtest::AcquireIntESProducer::produceESTestDataB);
0098
0099     setWhatAcquiredProduced(this,
0100                             &AcquireIntESProducer::acquireUniquePtr,
0101                             &AcquireIntESProducer::produceUniquePtr,
0102                             edm::es::Label("uniquePtr"));
0103
0104     setWhatAcquiredProduced(this,
0105                             &AcquireIntESProducer::acquireOptional,
0106                             &AcquireIntESProducer::produceOptional,
0107                             edm::es::Label("optional"));
0108
0109     setWhatAcquiredProducedWithLambda(
0110         [this](ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0111           usleep(200000);
0112           auto returnValue = std::make_unique<TestValue>(kAcquireTestValueUniquePtr1);
0113           lambdaUniqueTestPointers_[record.iovIndex()] = returnValue.get();
0114           return returnValue;
0115         },
0116         [this](ESTestRecordI const& record, auto testValue) {
0117           usleep(200000);
0118           if (testValue.get() != lambdaUniqueTestPointers_[record.iovIndex()]) {
0119             throw cms::Exception("TestFailure") << "AcquireIntESProducer::<lambda produceUniquePtr>"
0120                                                 << " unexpected value passed in as argument";
0121           }
0122           return std::make_unique<ESTestDataI>(kAcquireTestValueUniquePtr2);
0123         },
0124         edm::es::Label("uniquePtrLambda"));
0125
0126     setWhatAcquiredProducedWithLambda(
0127         [this](ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0128           usleep(200000);
0129           std::vector<TestValue> testVector;
0130           testVector.push_back(kAcquireTestValueOptional1);
0131           auto returnValue = std::make_optional<std::vector<TestValue>>(std::move(testVector));
0132           lambdaOptionalTestPointers_[record.iovIndex()] = &returnValue.value()[0];
0133           return returnValue;
0134         },
0135         [this](ESTestRecordI const& record, std::optional<std::vector<TestValue>> testValue) {
0136           usleep(200000);
0137           if (&testValue.value()[0] != lambdaOptionalTestPointers_[record.iovIndex()]) {
0138             throw cms::Exception("TestFailure") << "AcquireIntESProducer::<lambda produceOptional>"
0139                                                 << " unexpected value passed in as argument";
0140           }
0141           return std::make_unique<ESTestDataI>(kAcquireTestValueOptional2);
0142         },
0143         edm::es::Label("optionalLambda"));
0144   }
0145
0146   AcquireIntESProducer::~AcquireIntESProducer() {
0147     if (server_) {
0148       server_->stop();
0149     }
0150     server_.reset();
0151   }
0152
0153   void AcquireIntESProducer::initConcurrentIOVs(EventSetupRecordKey const& key, unsigned int nConcurrentIOVs) {
0154     if (key == EventSetupRecordKey::makeKey<ESTestRecordI>()) {
0155       caches_.resize(nConcurrentIOVs);
0156       server_ = std::make_unique<test_acquire::WaitingServer>(
0157           nConcurrentIOVs, numberOfIOVsToAccumulate_, secondsToWaitForWork_);
0158       server_->start();
0159       uniqueTestPointers_.resize(nConcurrentIOVs);
0160       optionalTestPointers_.resize(nConcurrentIOVs);
0161       lambdaUniqueTestPointers_.resize(nConcurrentIOVs);
0162       lambdaOptionalTestPointers_.resize(nConcurrentIOVs);
0163     }
0164   }
0165
0166   int AcquireIntESProducer::acquire(ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0167     usleep(200000);
0168
0169     test_acquire::Cache& iovCache = caches_[record.iovIndex()];
0170     iovCache.retrieved().clear();
0171     iovCache.processed().clear();
0172
0173     // Get some data and push it into the input cache for the ExternalWork.
0174     // There is no significance to the particular data we are using.
0175     // Using anything from the EventSetup would be good enough for the test.
0176     // I already had test modules around that would make IOVTestInfo
0177     // data, so that was easy to use. We put in known values and later
0178     // check that we get the expected result (they get incremented by one
0179     // to simulate some "external work", then summed in the produce method
0180     // calculate a result we can check easily).
0181     IOVTestInfo const& iovTestInfo = record.get(token_);
0182     std::vector<int>& retrieved = iovCache.retrieved();
0183     retrieved.push_back(iovTestInfo.iovStartRun_);
0184     retrieved.push_back(iovTestInfo.iovStartLumi_);
0185     retrieved.push_back(iovTestInfo.iovEndRun_);
0186     retrieved.push_back(iovTestInfo.iovEndLumi_);
0187     retrieved.push_back(iovTestInfo.cacheIdentifier_);
0188
0189     server_->requestValuesAsync(record.iovIndex(), &iovCache.retrieved(), &iovCache.processed(), holder);
0190
0191     edm::ValidityInterval iov = record.validityInterval();
0192     if (iovTestInfo.iovStartLumi_ != iov.first().luminosityBlockNumber() ||
0193         iovTestInfo.iovEndLumi_ != iov.last().luminosityBlockNumber() || iovTestInfo.iovIndex_ != record.iovIndex() ||
0194         iovTestInfo.cacheIdentifier_ != record.cacheIdentifier()) {
0195       throw cms::Exception("TestFailure") << "AcquireIntESProducer::acquire"
0196                                           << "read values do not agree with record";
0197     }
0198     return kAcquireTestValue;
0199   }
0200
0201   std::unique_ptr<ESTestDataI> AcquireIntESProducer::produce(ESTestRecordI const& record, int valueReturnedByAcquire) {
0202     usleep(200000);
0203
0204     if (valueReturnedByAcquire != kAcquireTestValue) {
0205       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produce"
0206                                           << " unexpected value passed in as argument";
0207     }
0208
0209     edm::ESHandle<IOVTestInfo> iovTestInfo = record.getHandle(token_);
0210     edm::ValidityInterval iov = record.validityInterval();
0211     if (iovTestInfo->iovStartLumi_ != iov.first().luminosityBlockNumber() ||
0212         iovTestInfo->iovEndLumi_ != iov.last().luminosityBlockNumber() || iovTestInfo->iovIndex_ != record.iovIndex() ||
0213         iovTestInfo->cacheIdentifier_ != record.cacheIdentifier()) {
0214       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produce"
0215                                           << "read values do not agree with record";
0216     }
0217
0218     test_acquire::Cache& iovCache = caches_[record.iovIndex()];
0219     int sum = 0;
0220     for (auto v : iovCache.processed()) {
0221       sum += v;
0222     }
0223     return std::make_unique<ESTestDataI>(sum);
0224   }
0225
0226   std::unique_ptr<ESTestDataB> AcquireIntESProducer::produceESTestDataB(ESTestRecordB const&) {
0227     return std::make_unique<ESTestDataB>(11);
0228   }
0229
0230   std::unique_ptr<AcquireIntESProducer::TestValue> AcquireIntESProducer::acquireUniquePtr(
0231       ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0232     usleep(200000);
0233     auto returnValue = std::make_unique<TestValue>(kAcquireTestValueUniquePtr1);
0234     uniqueTestPointers_[record.iovIndex()] = returnValue.get();
0235     return returnValue;
0236   }
0237
0238   std::unique_ptr<ESTestDataI> AcquireIntESProducer::produceUniquePtr(ESTestRecordI const& record,
0239                                                                       std::unique_ptr<TestValue> testValue) {
0240     usleep(200000);
0241     if (testValue.get() != uniqueTestPointers_[record.iovIndex()]) {
0242       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produceUniquePtr"
0243                                           << " unexpected value passed in as argument";
0244     }
0245     return std::make_unique<ESTestDataI>(kAcquireTestValueUniquePtr2);
0246   }
0247
0248   std::optional<std::vector<AcquireIntESProducer::TestValue>> AcquireIntESProducer::acquireOptional(
0249       ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0250     usleep(200000);
0251     std::vector<TestValue> testVector;
0252     testVector.push_back(kAcquireTestValueOptional1);
0253     auto returnValue = std::make_optional<std::vector<TestValue>>(std::move(testVector));
0254     optionalTestPointers_[record.iovIndex()] = &returnValue.value()[0];
0255     return returnValue;
0256   }
0257
0258   std::unique_ptr<ESTestDataI> AcquireIntESProducer::produceOptional(ESTestRecordI const& record,
0259                                                                      std::optional<std::vector<TestValue>> testValue) {
0260     usleep(200000);
0261     if (&testValue.value()[0] != optionalTestPointers_[record.iovIndex()]) {
0262       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produceOptional"
0263                                           << " unexpected value passed in as argument";
0264     }
0265     return std::make_unique<ESTestDataI>(kAcquireTestValueOptional2);
0266   }
0267
0268   void AcquireIntESProducer::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
0269     edm::ParameterSetDescription desc;
0270     desc.addUntracked<unsigned int>("numberOfIOVsToAccumulate", 8);
0271     desc.addUntracked<unsigned int>("secondsToWaitForWork", 1);
0272     descriptions.addDefault(desc);
0273   }
0274
0275 }  // namespace edmtest
0276 using namespace edmtest;
0277 DEFINE_FWK_EVENTSETUP_MODULE(AcquireIntESProducer);