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