Integration/plugins/TransformAsyncIntProducer.cc

0001 #include "FWCore/Framework/interface/global/EDProducer.h"
0002 #include "FWCore/Framework/interface/Event.h"
0003 #include "FWCore/Framework/interface/MakerMacros.h"
0004 #include "DataFormats/TestObjects/interface/ToyProducts.h"
0005 #include "FWCore/Utilities/interface/Exception.h"
0006
0007 #include <thread>
0008 #include <chrono>
0009 namespace edmtest {
0010   class TransformAsyncIntProducer : public edm::global::EDProducer<edm::Transformer> {
0011   public:
0012     struct WorkCache {
0013       std::shared_ptr<std::thread> thread_;
0014       IntProduct value_;
0015     };
0016
0017     TransformAsyncIntProducer(edm::ParameterSet const& iPSet)
0018         : getToken_(consumes(iPSet.getParameter<edm::InputTag>("get"))),
0019           offset_(iPSet.getParameter<unsigned int>("offset")),
0020           transformOffset_(iPSet.getParameter<unsigned int>("transformOffset")),
0021           noPut_(iPSet.getParameter<bool>("noPut")) {
0022       putToken_ = produces<IntProduct>();
0023       bool check = iPSet.getUntrackedParameter<bool>("checkTransformNotCalled");
0024       registerTransformAsync(
0025           putToken_,
0026           [offset = transformOffset_, check](edm::StreamID, auto const& iFrom, auto iTask) {
0027             if (check) {
0028               throw cms::Exception("TransformShouldNotBeCalled");
0029             }
0030             WorkCache ret;
0031             using namespace std::chrono_literals;
0032             ret.thread_ = std::make_shared<std::thread>([iTask] { std::this_thread::sleep_for(100ms); });
0033             ret.value_ = IntProduct(iFrom.value + offset);
0034             return ret;
0035           },
0036           [](edm::StreamID, auto const& iFrom) {
0037             iFrom.thread_->join();
0038             return iFrom.value_;
0039           },
0040           "transform");
0041     }
0042
0043     void produce(edm::StreamID, edm::Event& iEvent, edm::EventSetup const&) const override {
0044       if (not noPut_) {
0045         iEvent.emplace(putToken_, iEvent.get(getToken_).value + offset_);
0046       }
0047     }
0048     static void fillDescriptions(edm::ConfigurationDescriptions& desc) {
0049       edm::ParameterSetDescription pset;
0050       pset.add<edm::InputTag>("get");
0051       pset.add<unsigned int>("offset", 0);
0052       pset.add<unsigned int>("transformOffset", 1);
0053       pset.addUntracked<bool>("checkTransformNotCalled", false);
0054       pset.add<bool>("noPut", false);
0055
0056       desc.addDefault(pset);
0057     }
0058
0059   private:
0060     const edm::EDGetTokenT<IntProduct> getToken_;
0061     edm::EDPutTokenT<IntProduct> putToken_;
0062     const unsigned int offset_;
0063     const unsigned int transformOffset_;
0064     const bool noPut_;
0065   };
0066 }  // namespace edmtest
0067
0068 using edmtest::TransformAsyncIntProducer;
0069 DEFINE_FWK_MODULE(TransformAsyncIntProducer);