Integration/bin/interprocess_random.cc

0001 #include "boost/program_options.hpp"
0002
0003 #include <atomic>
0004 #include <csignal>
0005 #include <iostream>
0006 #include <string>
0007 #include <thread>
0008 #include <chrono>
0009
0010 #include "FWCore/TestProcessor/interface/TestProcessor.h"
0011 #include "DataFormats/TestObjects/interface/ToyProducts.h"
0012 #include "DataFormats/Common/interface/RandomNumberGeneratorState.h"
0013
0014 #include "FWCore/Services/interface/ExternalRandomNumberGeneratorService.h"
0015
0016 #include "FWCore/SharedMemory/interface/WriteBuffer.h"
0017 #include "FWCore/SharedMemory/interface/WorkerChannel.h"
0018 #include "FWCore/SharedMemory/interface/ROOTSerializer.h"
0019 #include "FWCore/SharedMemory/interface/ReadBuffer.h"
0020 #include "FWCore/SharedMemory/interface/ROOTDeserializer.h"
0021 #include "FWCore/SharedMemory/interface/WorkerMonitorThread.h"
0022
0023 static char const* const kMemoryNameOpt = "memory-name";
0024 static char const* const kMemoryNameCommandOpt = "memory-name,m";
0025 static char const* const kUniqueIDOpt = "unique-id";
0026 static char const* const kUniqueIDCommandOpt = "unique-id,i";
0027 static char const* const kHelpOpt = "help";
0028 static char const* const kHelpCommandOpt = "help,h";
0029
0030 //NOTE: Can use TestProcessor as the harness for the worker
0031
0032 using SentType = std::pair<edmtest::IntProduct, edm::RandomNumberGeneratorState>;
0033
0034 using namespace edm::shared_memory;
0035 class Harness {
0036 public:
0037   Harness(std::string const& iConfig, edm::ServiceToken iToken)
0038       : tester_(edm::test::TestProcessor::Config{iConfig}, iToken) {}
0039
0040   edmtest::IntProduct getBeginLumiValue(unsigned int iLumi) {
0041     auto lumi = tester_.testBeginLuminosityBlock(iLumi);
0042     return *lumi.get<edmtest::IntProduct>("lumi");
0043   }
0044
0045   edmtest::IntProduct getEventValue() {
0046     auto event = tester_.test();
0047     return *event.get<edmtest::IntProduct>();
0048   }
0049
0050 private:
0051   edm::test::TestProcessor tester_;
0052 };
0053
0054 int main(int argc, char* argv[]) {
0055   std::string descString(argv[0]);
0056   descString += " [--";
0057   descString += kMemoryNameOpt;
0058   descString += "] memory_name";
0059   boost::program_options::options_description desc(descString);
0060
0061   desc.add_options()(kHelpCommandOpt, "produce help message")(
0062       kMemoryNameCommandOpt, boost::program_options::value<std::string>(), "memory name")(
0063       kUniqueIDCommandOpt, boost::program_options::value<std::string>(), "unique id");
0064
0065   boost::program_options::positional_options_description p;
0066   p.add(kMemoryNameOpt, 1);
0067   p.add(kUniqueIDOpt, 2);
0068
0069   boost::program_options::options_description all_options("All Options");
0070   all_options.add(desc);
0071
0072   boost::program_options::variables_map vm;
0073   try {
0074     store(boost::program_options::command_line_parser(argc, argv).options(all_options).positional(p).run(), vm);
0075     notify(vm);
0076   } catch (boost::program_options::error const& iException) {
0077     std::cout << argv[0] << ": Error while trying to process command line arguments:\n"
0078               << iException.what() << "\nFor usage and an options list, please do 'cmsRun --help'.";
0079     return 1;
0080   }
0081
0082   if (vm.count(kHelpOpt)) {
0083     std::cout << desc << std::endl;
0084     return 0;
0085   }
0086
0087   if (!vm.count(kMemoryNameOpt)) {
0088     std::cout << " no argument given" << std::endl;
0089     return 1;
0090   }
0091
0092   if (!vm.count(kUniqueIDOpt)) {
0093     std::cout << " no second argument given" << std::endl;
0094     return 1;
0095   }
0096
0097   WorkerMonitorThread monitorThread;
0098
0099   monitorThread.startThread();
0100
0101   CMS_SA_ALLOW try {
0102     std::string const memoryName(vm[kMemoryNameOpt].as<std::string>());
0103     std::string const uniqueID(vm[kUniqueIDOpt].as<std::string>());
0104     {
0105       //This class is holding the lock
0106       WorkerChannel communicationChannel(memoryName, uniqueID);
0107
0108       WriteBuffer sm_buffer{memoryName, communicationChannel.fromWorkerBufferInfo()};
0109       ReadBuffer sm_readbuffer{std::string("Rand") + memoryName, communicationChannel.toWorkerBufferInfo()};
0110       int counter = 0;
0111
0112       //The lock must be released if there is a catastrophic signal
0113       auto lockPtr = communicationChannel.accessLock();
0114       monitorThread.setAction([lockPtr]() {
0115         if (lockPtr) {
0116           std::cerr << "SIGNAL CAUGHT: unlock\n";
0117           lockPtr->unlock();
0118         }
0119       });
0120
0121       using TCSerializer = ROOTSerializer<SentType, WriteBuffer>;
0122       TCSerializer serializer(sm_buffer);
0123       TCSerializer bl_serializer(sm_buffer);
0124
0125       using TCDeserializer = ROOTDeserializer<edm::RandomNumberGeneratorState, ReadBuffer>;
0126       TCDeserializer random_deserializer(sm_readbuffer);
0127
0128       std::cerr << uniqueID << " process: initializing " << std::endl;
0129       int nlines;
0130       std::cin >> nlines;
0131
0132       std::string configuration;
0133       for (int i = 0; i < nlines; ++i) {
0134         std::string c;
0135         std::getline(std::cin, c);
0136         std::cerr << c << "\n";
0137         configuration += c + "\n";
0138       }
0139
0140       edm::ExternalRandomNumberGeneratorService* randomService = new edm::ExternalRandomNumberGeneratorService;
0141       auto serviceToken =
0142           edm::ServiceRegistry::createContaining(std::unique_ptr<edm::RandomNumberGenerator>(randomService));
0143
0144       Harness harness(configuration, serviceToken);
0145
0146       //Either ROOT or the Framework are overriding the signal handlers
0147       monitorThread.setupSignalHandling();
0148
0149       std::cerr << uniqueID << " process: done initializing" << std::endl;
0150       communicationChannel.workerSetupDone();
0151
0152       std::cerr << uniqueID << " process: waiting " << counter << std::endl;
0153       communicationChannel.handleTransitions([&](edm::Transition iTransition, unsigned long long iTransitionID) {
0154         ++counter;
0155         switch (iTransition) {
0156           case edm::Transition::BeginLuminosityBlock: {
0157             std::cerr << uniqueID << " process: start beginLumi " << std::endl;
0158             auto randState = random_deserializer.deserialize();
0159             std::cerr << " state " << randState.seed_ << std::endl;
0160             randomService->setState(randState.state_, randState.seed_);
0161             SentType toSend;
0162             toSend.first = harness.getBeginLumiValue(iTransitionID);
0163             toSend.second.state_ = randomService->getState();
0164             toSend.second.seed_ = randomService->mySeed();
0165             bl_serializer.serialize(toSend);
0166             std::cerr << uniqueID << " process: end beginLumi " << toSend.first.value << std::endl;
0167
0168             break;
0169           }
0170           case edm::Transition::Event: {
0171             std::cerr << uniqueID << " process: begin event " << counter << std::endl;
0172             auto randState = random_deserializer.deserialize();
0173             randomService->setState(randState.state_, randState.seed_);
0174             SentType toSend;
0175             toSend.first = harness.getEventValue();
0176             toSend.second.state_ = randomService->getState();
0177             toSend.second.seed_ = randomService->mySeed();
0178             std::cerr << uniqueID << " process: end event " << counter << std::endl;
0179
0180             serializer.serialize(toSend);
0181             std::cerr << uniqueID << " process: " << toSend.first.value << " " << counter << std::endl;
0182             //std::this_thread::sleep_for(std::chrono::microseconds(10000000));
0183             break;
0184           }
0185           default: {
0186             assert(false);
0187           }
0188         }
0189         std::cerr << uniqueID << " process: notifying and waiting" << counter << std::endl;
0190       });
0191     }
0192   } catch (std::exception const& iExcept) {
0193     std::cerr << "caught exception \n" << iExcept.what() << "\n";
0194     return 1;
0195   } catch (...) {
0196     std::cerr << "caught unknown exception";
0197     return 1;
0198   }
0199   return 0;
0200 }