Test inter process random state transfers

DataFormats/Common/interface/RandomNumberGeneratorState.h

@@ -0,0 +1,26 @@
+#ifndef DataFormats_Common_RandomNumberGeneratorState_h
+#define DataFormats_Common_RandomNumberGeneratorState_h
+
+/*----------------------------------------------------------------------
+
+RandomNumberGeneratorState is used to communicate with an external process
+----------------------------------------------------------------------*/
+
+#include <vector>
+namespace edm {
+  struct RandomNumberGeneratorState {
+    RandomNumberGeneratorState() = default;
+    RandomNumberGeneratorState(std::vector<unsigned long> iState, long iSeed)
+        : state_(std::move(iState)), seed_{iSeed} {}
+
+    RandomNumberGeneratorState(RandomNumberGeneratorState const&) = default;
+    RandomNumberGeneratorState(RandomNumberGeneratorState&&) = default;
+
+    RandomNumberGeneratorState& operator=(RandomNumberGeneratorState const&) = default;
+    RandomNumberGeneratorState& operator=(RandomNumberGeneratorState&&) = default;
+
+    std::vector<unsigned long> state_;
+    long seed_;
+  };
+}  // namespace edm
+#endif

DataFormats/Common/src/classes.h

@@ -31,5 +31,6 @@
 #include "DataFormats/Common/interface/SecondaryEventIDAndFileInfo.h"
 #include "DataFormats/Provenance/interface/EventAuxiliary.h"
 #include "FWCore/MessageLogger/interface/ErrorSummaryEntry.h"
+#include "DataFormats/Common/interface/RandomNumberGeneratorState.h"
 
 #include <vector>

DataFormats/Common/src/classes_def.xml

@@ -187,5 +187,8 @@
 
  <class name="edm::RefProd<std::vector<int> >"/>
  <class name="edm::RefToBaseProd<int>"/>
+ <class name="edm::RandomNumberGeneratorState" ClassVersion="3">
+  <version ClassVersion="3" checksum="2237762164"/>
+ </class>
 
 </lcgdict>

DataFormats/TestObjects/src/classes.h

@@ -30,6 +30,7 @@
 #include "DataFormats/Common/interface/Holder.h"
 #include "DataFormats/Common/interface/RefToBaseProd.h"
 #include "DataFormats/Common/interface/RefToBaseVector.h"
+#include "DataFormats/Common/interface/RandomNumberGeneratorState.h"
 
 #include "DataFormats/Provenance/interface/EventID.h"
 #include "DataFormats/Provenance/interface/ProductID.h"

DataFormats/TestObjects/src/classes_def.xml

@@ -5,6 +5,7 @@
  <class name="edmtest::IntProduct" ClassVersion="10">
   <version ClassVersion="10" checksum="4286676158"/>
  </class>
+ <class name="std::pair<edmtest::IntProduct, edm::RandomNumberGeneratorState>"/>
  <class name="edmtest::UInt64Product" ClassVersion="11">
   <version ClassVersion="11" checksum="3240540910"/>
   <version ClassVersion="10" checksum="380152127"/>

FWCore/Integration/bin/BuildFile.xml

@@ -4,3 +4,12 @@
   <use   name="FWCore/TestProcessor"/>
   <use   name="FWCore/SharedMemory"/>
 </bin>
+
+<bin   name="cmsTestInterProcessRandom" file="interprocess_random.cc">
+  <use   name="boost"/>
+  <use   name="boost_program_options"/>
+  <use   name="FWCore/TestProcessor"/>
+  <use   name="FWCore/SharedMemory"/>
+  <use   name="FWCore/Services"/>
+  <use   name="FWCore/Utilities"/>
+</bin>

FWCore/Integration/bin/interprocess_random.cc

@@ -0,0 +1,207 @@
+#include "boost/program_options.hpp"
+
+#include <atomic>
+#include <csignal>
+#include <iostream>
+#include <string>
+#include <thread>
+
+//TEMP
+//#include "CLHEP/Random/JamesRandom.h"
+
+#include "FWCore/TestProcessor/interface/TestProcessor.h"
+#include "DataFormats/TestObjects/interface/ToyProducts.h"
+#include "DataFormats/Common/interface/RandomNumberGeneratorState.h"
+
+#include "FWCore/Services/interface/ExternalRandomNumberGeneratorService.h"
+
+#include "FWCore/SharedMemory/interface/WriteBuffer.h"
+#include "FWCore/SharedMemory/interface/WorkerChannel.h"
+#include "FWCore/SharedMemory/interface/ROOTSerializer.h"
+#include "FWCore/SharedMemory/interface/ReadBuffer.h"
+#include "FWCore/SharedMemory/interface/ROOTDeserializer.h"
+#include "FWCore/SharedMemory/interface/WorkerMonitorThread.h"
+
+static char const* const kMemoryNameOpt = "memory-name";
+static char const* const kMemoryNameCommandOpt = "memory-name,m";
+static char const* const kUniqueIDOpt = "unique-id";
+static char const* const kUniqueIDCommandOpt = "unique-id,i";
+static char const* const kHelpOpt = "help";
+static char const* const kHelpCommandOpt = "help,h";
+
+//NOTE: Can use TestProcessor as the harness for the worker
+
+using SentType = std::pair<edmtest::IntProduct, edm::RandomNumberGeneratorState>;
+
+using namespace edm::shared_memory;
+class Harness {
+public:
+  Harness(std::string const& iConfig, edm::ServiceToken iToken)
+      : tester_(edm::test::TestProcessor::Config{iConfig}, iToken) {}
+
+  edmtest::IntProduct getBeginLumiValue(unsigned int iLumi) {
+    auto lumi = tester_.testBeginLuminosityBlock(iLumi);
+    return *lumi.get<edmtest::IntProduct>("lumi");
+  }
+
+  edmtest::IntProduct getEventValue() {
+    auto event = tester_.test();
+    return *event.get<edmtest::IntProduct>();
+  }
+
+private:
+  edm::test::TestProcessor tester_;
+};
+
+int main(int argc, char* argv[]) {
+  std::string descString(argv[0]);
+  descString += " [--";
+  descString += kMemoryNameOpt;
+  descString += "] memory_name";
+  boost::program_options::options_description desc(descString);
+
+  desc.add_options()(kHelpCommandOpt, "produce help message")(
+      kMemoryNameCommandOpt, boost::program_options::value<std::string>(), "memory name")(
+      kUniqueIDCommandOpt, boost::program_options::value<std::string>(), "unique id");
+
+  boost::program_options::positional_options_description p;
+  p.add(kMemoryNameOpt, 1);
+  p.add(kUniqueIDOpt, 2);
+
+  boost::program_options::options_description all_options("All Options");
+  all_options.add(desc);
+
+  boost::program_options::variables_map vm;
+  try {
+    store(boost::program_options::command_line_parser(argc, argv).options(all_options).positional(p).run(), vm);
+    notify(vm);
+  } catch (boost::program_options::error const& iException) {
+    std::cout << argv[0] << ": Error while trying to process command line arguments:\n"
+              << iException.what() << "\nFor usage and an options list, please do 'cmsRun --help'.";
+    return 1;
+  }
+
+  if (vm.count(kHelpOpt)) {
+    std::cout << desc << std::endl;
+    return 0;
+  }
+
+  if (!vm.count(kMemoryNameOpt)) {
+    std::cout << " no argument given" << std::endl;
+    return 1;
+  }
+
+  if (!vm.count(kUniqueIDOpt)) {
+    std::cout << " no second argument given" << std::endl;
+    return 1;
+  }
+
+  WorkerMonitorThread monitorThread;
+
+  monitorThread.startThread();
+
+  CMS_SA_ALLOW try {
+    std::string const memoryName(vm[kMemoryNameOpt].as<std::string>());
+    std::string const uniqueID(vm[kUniqueIDOpt].as<std::string>());
+    {
+      //using namespace boost::interprocess;
+      //auto controlNameUnique = unique_name(memoryName, uniqueID);
+
+      //This class is holding the lock
+      WorkerChannel communicationChannel(memoryName, uniqueID);
+
+      WriteBuffer sm_buffer{memoryName, communicationChannel.fromWorkerBufferIndex()};
+      ReadBuffer sm_readbuffer{std::string("Rand") + memoryName, communicationChannel.toWorkerBufferIndex()};
+      int counter = 0;
+
+      //The lock must be released if there is a catastrophic signal
+      auto lockPtr = communicationChannel.accessLock();
+      monitorThread.setAction([lockPtr]() {
+        if (lockPtr) {
+          std::cerr << "SIGNAL CAUGHT: unlock\n";
+          lockPtr->unlock();
+        }
+      });
+
+      using TCSerializer = ROOTSerializer<SentType, WriteBuffer>;
+      TCSerializer serializer(sm_buffer);
+      TCSerializer bl_serializer(sm_buffer);
+
+      using TCDeserializer = ROOTDeserializer<edm::RandomNumberGeneratorState, ReadBuffer>;
+      TCDeserializer random_deserializer(sm_readbuffer);
+
+      std::cerr << uniqueID << " process: initializing " << std::endl;
+      int nlines;
+      std::cin >> nlines;
+
+      std::string configuration;
+      for (int i = 0; i < nlines; ++i) {
+        std::string c;
+        std::getline(std::cin, c);
+        std::cerr << c << "\n";
+        configuration += c + "\n";
+      }
+
+      edm::ExternalRandomNumberGeneratorService* randomService = new edm::ExternalRandomNumberGeneratorService;
+      auto serviceToken =
+          edm::ServiceRegistry::createContaining(std::unique_ptr<edm::RandomNumberGenerator>(randomService));
+
+      //CLHEP::HepJamesRandom rng(12345);
+      //randomService->setState(rng.put(), rng.getSeed());
+      Harness harness(configuration, serviceToken);
+
+      //Either ROOT or the Framework are overriding the signal handlers
+      monitorThread.setupSignalHandling();
+
+      std::cerr << uniqueID << " process: done initializing" << std::endl;
+      communicationChannel.workerSetupDone();
+
+      std::cerr << uniqueID << " process: waiting " << counter << std::endl;
+      communicationChannel.handleTransitions([&](edm::Transition iTransition, unsigned long long iTransitionID) {
+        ++counter;
+        switch (iTransition) {
+          case edm::Transition::BeginLuminosityBlock: {
+            std::cerr << uniqueID << " process: start beginLumi " << std::endl;
+            auto randState = random_deserializer.deserialize();
+            std::cerr << " state " << randState.seed_ << std::endl;
+            randomService->setState(randState.state_, randState.seed_);
+            SentType toSend;
+            toSend.first = harness.getBeginLumiValue(iTransitionID);
+            toSend.second.state_ = randomService->getState();
+            toSend.second.seed_ = randomService->mySeed();
+            bl_serializer.serialize(toSend);
+            std::cerr << uniqueID << " process: end beginLumi " << toSend.first.value << std::endl;
+
+            break;
+          }
+          case edm::Transition::Event: {
+            std::cerr << uniqueID << " process: begin event " << counter << std::endl;
+            auto randState = random_deserializer.deserialize();
+            randomService->setState(randState.state_, randState.seed_);
+            SentType toSend;
+            toSend.first = harness.getEventValue();
+            toSend.second.state_ = randomService->getState();
+            toSend.second.seed_ = randomService->mySeed();
+            std::cerr << uniqueID << " process: end event " << counter << std::endl;
+
+            serializer.serialize(toSend);
+            std::cerr << uniqueID << " process: " << toSend.first.value << " " << counter << std::endl;
+            //usleep(10000000);
+            break;
+          }
+          default: {
+            assert(false);
+          }
+        }
+        std::cerr << uniqueID << " process: notifying and waiting" << counter << std::endl;
+      });
+    }
+  } catch (std::exception const& iExcept) {
+    std::cerr << "caught exception \n" << iExcept.what() << "\n";
+    return 1;
+  } catch (...) {
+    std::cerr << "caught unknown exception";
+    return 1;
+  }
+  return 0;
+}

FWCore/Integration/test/BuildFile.xml

@@ -331,5 +331,29 @@
     <use   name="FWCore/SharedMemory"/>
     <use   name="boost"/>
   </library>
+  <library   file="RandomIntProducer.cc" name="RandomIntProducer">
+    <flags   EDM_PLUGIN="1"/>
+    <use   name="FWCore/ParameterSet"/>
+    <use   name="FWCore/Framework"/>
+    <use   name="FWCore/Utilities"/>
+    <use   name="DataFormats/TestObjects"/>
+    <use   name="clhep"/>
+  </library>
+  <library   file="TestInterProcessRandomProd.cc" name="TestInterProcessRandomProd">
+    <flags   EDM_PLUGIN="1"/>
+    <use   name="FWCore/Framework"/>
+    <use   name="FWCore/ParameterSet"/>
+    <use   name="FWCore/Sources"/>
+    <use   name="FWCore/SharedMemory"/>
+    <use   name="boost"/>
+    <use   name="clhep"/>
+  </library>
+  <bin   file="RandomIntProducer_t.cpp">
+    <use   name="FWCore/Framework"/>
+    <use   name="FWCore/ParameterSet"/>
+    <use   name="FWCore/TestProcessor"/>
+    <use   name="DataFormats/Provenance"/>
+    <use   name="catch2"/>
+  </bin>
   <test name="TestFWCoreIntegrationInterProcess" command="cmsRun ${LOCALTOP}/src/FWCore/Integration/test/test_TestInterProcessProd_cfg.py"/>
 </environment>

FWCore/Integration/test/RandomIntProducer.cc

@@ -0,0 +1,46 @@
+#include "FWCore/Framework/interface/one/EDProducer.h"
+#include "FWCore/Framework/interface/MakerMacros.h"
+#include "FWCore/Framework/interface/Event.h"
+#include "FWCore/Framework/interface/LuminosityBlock.h"
+
+#include "FWCore/Utilities/interface/RandomNumberGenerator.h"
+#include "FWCore/Utilities/interface/EDPutToken.h"
+#include "FWCore/Utilities/interface/Transition.h"
+#include "FWCore/ServiceRegistry/interface/Service.h"
+
+#include "DataFormats/TestObjects/interface/ToyProducts.h"
+
+#include "CLHEP/Random/RandFlat.h"
+
+namespace edmtest {
+  class RandomIntProducer : public edm::one::EDProducer<edm::BeginLuminosityBlockProducer> {
+  public:
+    RandomIntProducer(edm::ParameterSet const& iPSet);
+
+    void produce(edm::Event&, edm::EventSetup const&) final;
+
+    void beginLuminosityBlockProduce(edm::LuminosityBlock&, edm::EventSetup const&) final;
+
+  private:
+    edm::EDPutTokenT<IntProduct> const evToken_;
+    edm::EDPutTokenT<IntProduct> const lumiToken_;
+  };
+  RandomIntProducer::RandomIntProducer(edm::ParameterSet const&)
+      : evToken_{produces<IntProduct>()},
+        lumiToken_{produces<IntProduct, edm::Transition::BeginLuminosityBlock>("lumi")} {}
+
+  void RandomIntProducer::produce(edm::Event& iEvent, edm::EventSetup const&) {
+    edm::Service<edm::RandomNumberGenerator> gen;
+    std::cout << gen->getEngine(iEvent.streamID()).name() << std::endl;
+    iEvent.emplace(evToken_, CLHEP::RandFlat::shootInt(&gen->getEngine(iEvent.streamID()), 10));
+  }
+
+  void RandomIntProducer::beginLuminosityBlockProduce(edm::LuminosityBlock& iLumi, edm::EventSetup const&) {
+    edm::Service<edm::RandomNumberGenerator> gen;
+    iLumi.emplace(lumiToken_, CLHEP::RandFlat::shootInt(&gen->getEngine(iLumi.index()), 10));
+  }
+
+}  // namespace edmtest
+
+using namespace edmtest;
+DEFINE_FWK_MODULE(RandomIntProducer);