Integration/plugins/TransformAsyncIntStreamProducer.cc

0001 #include "FWCore/Framework/interface/stream/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 namespace edmtest {
0009   class TransformAsyncIntStreamProducer : public edm::stream::EDProducer<edm::Transformer> {
0010   public:
0011     struct WorkCache {
0012       std::shared_ptr<std::thread> thread_;
0013       IntProduct value_;
0014     };
0015
0016     TransformAsyncIntStreamProducer(edm::ParameterSet const& iPSet)
0017         : getToken_(consumes(iPSet.getParameter<edm::InputTag>("get"))),
0018           offset_(iPSet.getParameter<unsigned int>("offset")),
0019           transformOffset_(iPSet.getParameter<unsigned int>("transformOffset")) {
0020       putToken_ = produces<IntProduct>();
0021       bool check = iPSet.getUntrackedParameter<bool>("checkTransformNotCalled");
0022       registerTransformAsync(
0023           putToken_,
0024           [offset = transformOffset_, check](auto const& iFrom, auto iTask) {
0025             if (check) {
0026               throw cms::Exception("TransformShouldNotBeCalled");
0027             }
0028             WorkCache ret;
0029             ret.thread_ = std::make_shared<std::thread>([iTask] { usleep(100000); });
0030             ret.value_ = IntProduct(iFrom.value + offset);
0031             return ret;
0032           },
0033           [](auto const& iFrom) {
0034             iFrom.thread_->join();
0035             return iFrom.value_;
0036           },
0037           "transform");
0038     }
0039
0040     void produce(edm::Event& iEvent, edm::EventSetup const&) override {
0041       iEvent.emplace(putToken_, iEvent.get(getToken_).value + offset_);
0042     }
0043     static void fillDescriptions(edm::ConfigurationDescriptions& desc) {
0044       edm::ParameterSetDescription pset;
0045       pset.add<edm::InputTag>("get");
0046       pset.add<unsigned int>("offset", 0);
0047       pset.add<unsigned int>("transformOffset", 1);
0048       pset.addUntracked<bool>("checkTransformNotCalled", false);
0049
0050       desc.addDefault(pset);
0051     }
0052
0053   private:
0054     const edm::EDGetTokenT<IntProduct> getToken_;
0055     edm::EDPutTokenT<IntProduct> putToken_;
0056     const unsigned int offset_;
0057     const unsigned int transformOffset_;
0058   };
0059 }  // namespace edmtest
0060
0061 using edmtest::TransformAsyncIntStreamProducer;
0062 DEFINE_FWK_MODULE(TransformAsyncIntStreamProducer);