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   };
0088
0089   AcquireIntESProducer::AcquireIntESProducer(edm::ParameterSet const& pset)
0090       : numberOfIOVsToAccumulate_(pset.getUntrackedParameter<unsigned int>("numberOfIOVsToAccumulate")),
0091         secondsToWaitForWork_(pset.getUntrackedParameter<unsigned int>("secondsToWaitForWork")) {
0092     auto collector = setWhatAcquiredProduced(this, "fromAcquireIntESProducer");
0093     token_ = collector.consumes<IOVTestInfo>(edm::ESInputTag{"", ""});
0094
0095     setWhatProduced(this, &edmtest::AcquireIntESProducer::produceESTestDataB);
0096
0097     setWhatAcquiredProduced(this,
0098                             &AcquireIntESProducer::acquireUniquePtr,
0099                             &AcquireIntESProducer::produceUniquePtr,
0100                             edm::es::Label("uniquePtr"));
0101
0102     setWhatAcquiredProduced(this,
0103                             &AcquireIntESProducer::acquireOptional,
0104                             &AcquireIntESProducer::produceOptional,
0105                             edm::es::Label("optional"));
0106   }
0107
0108   AcquireIntESProducer::~AcquireIntESProducer() {
0109     if (server_) {
0110       server_->stop();
0111     }
0112     server_.reset();
0113   }
0114
0115   void AcquireIntESProducer::initConcurrentIOVs(EventSetupRecordKey const& key, unsigned int nConcurrentIOVs) {
0116     if (key == EventSetupRecordKey::makeKey<ESTestRecordI>()) {
0117       caches_.resize(nConcurrentIOVs);
0118       server_ = std::make_unique<test_acquire::WaitingServer>(
0119           nConcurrentIOVs, numberOfIOVsToAccumulate_, secondsToWaitForWork_);
0120       server_->start();
0121       uniqueTestPointers_.resize(nConcurrentIOVs);
0122       optionalTestPointers_.resize(nConcurrentIOVs);
0123     }
0124   }
0125
0126   int AcquireIntESProducer::acquire(ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0127     usleep(200000);
0128
0129     test_acquire::Cache& iovCache = caches_[record.iovIndex()];
0130     iovCache.retrieved().clear();
0131     iovCache.processed().clear();
0132
0133     // Get some data and push it into the input cache for the ExternalWork.
0134     // There is no significance to the particular data we are using.
0135     // Using anything from the EventSetup would be good enough for the test.
0136     // I already had test modules around that would make IOVTestInfo
0137     // data, so that was easy to use. We put in known values and later
0138     // check that we get the expected result (they get incremented by one
0139     // to simulate some "external work", then summed in the produce method
0140     // calculate a result we can check easily).
0141     IOVTestInfo const& iovTestInfo = record.get(token_);
0142     std::vector<int>& retrieved = iovCache.retrieved();
0143     retrieved.push_back(iovTestInfo.iovStartRun_);
0144     retrieved.push_back(iovTestInfo.iovStartLumi_);
0145     retrieved.push_back(iovTestInfo.iovEndRun_);
0146     retrieved.push_back(iovTestInfo.iovEndLumi_);
0147     retrieved.push_back(iovTestInfo.cacheIdentifier_);
0148
0149     server_->requestValuesAsync(record.iovIndex(), &iovCache.retrieved(), &iovCache.processed(), holder);
0150
0151     edm::ValidityInterval iov = record.validityInterval();
0152     if (iovTestInfo.iovStartLumi_ != iov.first().luminosityBlockNumber() ||
0153         iovTestInfo.iovEndLumi_ != iov.last().luminosityBlockNumber() || iovTestInfo.iovIndex_ != record.iovIndex() ||
0154         iovTestInfo.cacheIdentifier_ != record.cacheIdentifier()) {
0155       throw cms::Exception("TestFailure") << "AcquireIntESProducer::acquire"
0156                                           << "read values do not agree with record";
0157     }
0158     return kAcquireTestValue;
0159   }
0160
0161   std::unique_ptr<ESTestDataI> AcquireIntESProducer::produce(ESTestRecordI const& record, int valueReturnedByAcquire) {
0162     usleep(200000);
0163
0164     if (valueReturnedByAcquire != kAcquireTestValue) {
0165       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produce"
0166                                           << " unexpected value passed in as argument";
0167     }
0168
0169     edm::ESHandle<IOVTestInfo> iovTestInfo = record.getHandle(token_);
0170     edm::ValidityInterval iov = record.validityInterval();
0171     if (iovTestInfo->iovStartLumi_ != iov.first().luminosityBlockNumber() ||
0172         iovTestInfo->iovEndLumi_ != iov.last().luminosityBlockNumber() || iovTestInfo->iovIndex_ != record.iovIndex() ||
0173         iovTestInfo->cacheIdentifier_ != record.cacheIdentifier()) {
0174       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produce"
0175                                           << "read values do not agree with record";
0176     }
0177
0178     test_acquire::Cache& iovCache = caches_[record.iovIndex()];
0179     int sum = 0;
0180     for (auto v : iovCache.processed()) {
0181       sum += v;
0182     }
0183     return std::make_unique<ESTestDataI>(sum);
0184   }
0185
0186   std::unique_ptr<ESTestDataB> AcquireIntESProducer::produceESTestDataB(ESTestRecordB const&) {
0187     return std::make_unique<ESTestDataB>(11);
0188   }
0189
0190   std::unique_ptr<AcquireIntESProducer::TestValue> AcquireIntESProducer::acquireUniquePtr(
0191       ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0192     usleep(200000);
0193     auto returnValue = std::make_unique<TestValue>(kAcquireTestValueUniquePtr1);
0194     uniqueTestPointers_[record.iovIndex()] = returnValue.get();
0195     return returnValue;
0196   }
0197
0198   std::unique_ptr<ESTestDataI> AcquireIntESProducer::produceUniquePtr(ESTestRecordI const& record,
0199                                                                       std::unique_ptr<TestValue> testValue) {
0200     usleep(200000);
0201     if (testValue.get() != uniqueTestPointers_[record.iovIndex()]) {
0202       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produceUniquePtr"
0203                                           << " unexpected value passed in as argument";
0204     }
0205     return std::make_unique<ESTestDataI>(kAcquireTestValueUniquePtr2);
0206   }
0207
0208   std::optional<std::vector<AcquireIntESProducer::TestValue>> AcquireIntESProducer::acquireOptional(
0209       ESTestRecordI const& record, edm::WaitingTaskWithArenaHolder holder) {
0210     usleep(200000);
0211     std::vector<TestValue> testVector;
0212     testVector.push_back(kAcquireTestValueOptional1);
0213     auto returnValue = std::make_optional<std::vector<TestValue>>(std::move(testVector));
0214     optionalTestPointers_[record.iovIndex()] = &returnValue.value()[0];
0215     return returnValue;
0216   }
0217
0218   std::unique_ptr<ESTestDataI> AcquireIntESProducer::produceOptional(ESTestRecordI const& record,
0219                                                                      std::optional<std::vector<TestValue>> testValue) {
0220     usleep(200000);
0221     if (&testValue.value()[0] != optionalTestPointers_[record.iovIndex()]) {
0222       throw cms::Exception("TestFailure") << "AcquireIntESProducer::produceOptional"
0223                                           << " unexpected value passed in as argument";
0224     }
0225     return std::make_unique<ESTestDataI>(kAcquireTestValueOptional2);
0226   }
0227
0228   void AcquireIntESProducer::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
0229     edm::ParameterSetDescription desc;
0230     desc.addUntracked<unsigned int>("numberOfIOVsToAccumulate", 8);
0231     desc.addUntracked<unsigned int>("secondsToWaitForWork", 1);
0232     descriptions.addDefault(desc);
0233   }
0234
0235 }  // namespace edmtest
0236 using namespace edmtest;
0237 DEFINE_FWK_EVENTSETUP_MODULE(AcquireIntESProducer);