Core/bin/externalGenerator.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 <memory>
0009 #include <filesystem>
0010 #include <ctime>
0011
0012 #include "FWCore/TestProcessor/interface/TestProcessor.h"
0013
0014 #include "SimDataFormats/GeneratorProducts/interface/HepMCProduct.h"
0015 #include "SimDataFormats/GeneratorProducts/interface/GenRunInfoProduct.h"
0016 #include "SimDataFormats/GeneratorProducts/interface/GenLumiInfoHeader.h"
0017 #include "SimDataFormats/GeneratorProducts/interface/GenLumiInfoProduct.h"
0018 #include "SimDataFormats/GeneratorProducts/interface/GenEventInfoProduct.h"
0019 #include "SimDataFormats/GeneratorProducts/interface/ExternalGeneratorEventInfo.h"
0020 #include "SimDataFormats/GeneratorProducts/interface/ExternalGeneratorLumiInfo.h"
0021
0022 #include "FWCore/Services/interface/ExternalRandomNumberGeneratorService.h"
0023
0024 #include "FWCore/SharedMemory/interface/WriteBuffer.h"
0025 #include "FWCore/SharedMemory/interface/ReadBuffer.h"
0026 #include "FWCore/SharedMemory/interface/WorkerChannel.h"
0027 #include "FWCore/SharedMemory/interface/ROOTSerializer.h"
0028 #include "FWCore/SharedMemory/interface/ROOTDeserializer.h"
0029 #include "FWCore/SharedMemory/interface/WorkerMonitorThread.h"
0030
0031 #include "FWCore/Utilities/interface/thread_safety_macros.h"
0032
0033 static char const* const kMemoryNameOpt = "memory-name";
0034 static char const* const kMemoryNameCommandOpt = "memory-name,m";
0035 static char const* const kUniqueIDOpt = "unique-id";
0036 static char const* const kUniqueIDCommandOpt = "unique-id,i";
0037 static char const* const kHelpOpt = "help";
0038 static char const* const kHelpCommandOpt = "help,h";
0039 static char const* const kVerboseOpt = "verbose";
0040 static char const* const kVerboseCommandOpt = "verbose,v";
0041
0042 //This application only uses 1 thread
0043 CMS_THREAD_SAFE static std::string s_uniqueID;
0044
0045 //NOTE: Can use TestProcessor as the harness for the worker
0046
0047 namespace {
0048   //Based on MessageLogger time handling
0049   constexpr char timeFormat[] = "dd-Mon-yyyy hh:mm:ss TZN     ";
0050   constexpr size_t kTimeSize = sizeof(timeFormat);
0051   std::array<char, kTimeSize> formattedTime() {
0052     auto t = time(nullptr);
0053     std::array<char, kTimeSize> ts;
0054
0055     struct tm timebuf;
0056     std::strftime(ts.data(), ts.size(), "%d-%b-%Y %H:%M:%S %Z", localtime_r(&t, &timebuf));
0057     return ts;
0058   }
0059 }  // namespace
0060
0061 using namespace edm::shared_memory;
0062 class Harness {
0063 public:
0064   Harness(std::string const& iConfig, edm::ServiceToken iToken)
0065       : tester_(edm::test::TestProcessor::Config{iConfig}, iToken) {}
0066
0067   ExternalGeneratorLumiInfo getBeginLumiValue(unsigned int iLumi) {
0068     auto lumi = tester_.testBeginLuminosityBlock(iLumi);
0069     ExternalGeneratorLumiInfo returnValue;
0070     returnValue.header_ = *lumi.get<GenLumiInfoHeader>();
0071     return returnValue;
0072   }
0073
0074   ExternalGeneratorEventInfo getEventValue() {
0075     ExternalGeneratorEventInfo returnValue;
0076     auto event = tester_.test();
0077     returnValue.hepmc_ = *event.get<edm::HepMCProduct>("unsmeared");
0078     returnValue.eventInfo_ = *event.get<GenEventInfoProduct>();
0079     returnValue.keepEvent_ = event.modulePassed();
0080     return returnValue;
0081   }
0082
0083   GenLumiInfoProduct getEndLumiValue() {
0084     auto lumi = tester_.testEndLuminosityBlock();
0085     return *lumi.get<GenLumiInfoProduct>();
0086   }
0087
0088   GenRunInfoProduct getEndRunValue() {
0089     auto run = tester_.testEndRun();
0090     return *run.get<GenRunInfoProduct>();
0091   }
0092
0093 private:
0094   edm::test::TestProcessor tester_;
0095 };
0096
0097 template <typename T>
0098 using Serializer = ROOTSerializer<T, WriteBuffer>;
0099
0100 namespace {
0101   //needed for atexit handling
0102   CMS_THREAD_SAFE boost::interprocess::scoped_lock<boost::interprocess::named_mutex>* s_sharedLock = nullptr;
0103
0104   void atexit_handler() {
0105     if (s_sharedLock) {
0106       std::cerr << s_uniqueID << " process: early exit called: unlock " << formattedTime().data() << "\n";
0107       s_sharedLock->unlock();
0108     }
0109   }
0110 }  // namespace
0111
0112 int main(int argc, char* argv[]) {
0113   std::string descString(argv[0]);
0114   descString += " [--";
0115   descString += kMemoryNameOpt;
0116   descString += "] memory_name";
0117   boost::program_options::options_description desc(descString);
0118
0119   desc.add_options()(kHelpCommandOpt, "produce help message")(
0120       kMemoryNameCommandOpt, boost::program_options::value<std::string>(), "memory name")(
0121       kUniqueIDCommandOpt, boost::program_options::value<std::string>(), "unique id")(kVerboseCommandOpt,
0122                                                                                       "verbose output");
0123
0124   boost::program_options::positional_options_description p;
0125   p.add(kMemoryNameOpt, 1);
0126   p.add(kUniqueIDOpt, 2);
0127
0128   boost::program_options::options_description all_options("All Options");
0129   all_options.add(desc);
0130
0131   boost::program_options::variables_map vm;
0132   try {
0133     store(boost::program_options::command_line_parser(argc, argv).options(all_options).positional(p).run(), vm);
0134     notify(vm);
0135   } catch (boost::program_options::error const& iException) {
0136     std::cout << argv[0] << ": Error while trying to process command line arguments:\n"
0137               << iException.what() << "\nFor usage and an options list, please do 'cmsRun --help'.";
0138     return 1;
0139   }
0140
0141   if (vm.count(kHelpOpt)) {
0142     std::cout << desc << std::endl;
0143     return 0;
0144   }
0145
0146   bool verbose = false;
0147   if (vm.count(kVerboseOpt)) {
0148     verbose = true;
0149   }
0150
0151   if (!vm.count(kMemoryNameOpt)) {
0152     std::cout << " no argument given" << std::endl;
0153     return 1;
0154   }
0155
0156   if (!vm.count(kUniqueIDOpt)) {
0157     std::cout << " no second argument given" << std::endl;
0158     return 1;
0159   }
0160
0161   using namespace std::string_literals;
0162   using namespace std::filesystem;
0163
0164   auto newDir = path("thread"s + vm[kUniqueIDOpt].as<std::string>());
0165   create_directory(newDir);
0166   current_path(newDir);
0167
0168   WorkerMonitorThread monitorThread;
0169
0170   monitorThread.startThread();
0171
0172   std::string presentState = "setting up communicationChannel";
0173
0174   CMS_SA_ALLOW try {
0175     std::string const memoryName(vm[kMemoryNameOpt].as<std::string>());
0176     std::string const uniqueID(vm[kUniqueIDOpt].as<std::string>());
0177     s_uniqueID = uniqueID;
0178     {
0179       //This class is holding the lock
0180       WorkerChannel communicationChannel(memoryName, uniqueID);
0181
0182       presentState = "setting up read/write buffers";
0183       WriteBuffer sm_buffer{memoryName, communicationChannel.fromWorkerBufferInfo()};
0184       ReadBuffer sm_readbuffer{std::string("Rand") + memoryName, communicationChannel.toWorkerBufferInfo()};
0185       int counter = 0;
0186
0187       presentState = "setting up monitor thread";
0188       //The lock must be released if there is a catastrophic signal
0189       auto lockPtr = communicationChannel.accessLock();
0190
0191       monitorThread.setAction([lockPtr]() {
0192         if (lockPtr) {
0193           std::cerr << s_uniqueID << " process: SIGNAL CAUGHT: unlock " << formattedTime().data() << "\n";
0194           lockPtr->unlock();
0195         }
0196       });
0197
0198       presentState = "setting up termination handler";
0199       //be sure to unset the address of the shared lock before the lock goes away
0200       s_sharedLock = lockPtr;
0201       auto unsetLockPtr = [](void*) { s_sharedLock = nullptr; };
0202       std::unique_ptr<decltype(s_sharedLock), decltype(unsetLockPtr)> sharedLockGuard{&s_sharedLock, unsetLockPtr};
0203       std::atexit(atexit_handler);
0204       auto releaseLock = []() {
0205         if (s_sharedLock) {
0206           std::cerr << s_uniqueID << " process: terminate called: unlock " << formattedTime().data() << "\n";
0207           s_sharedLock->unlock();
0208           s_sharedLock = nullptr;
0209           //deactivate the abort signal
0210
0211           struct sigaction act;
0212           act.sa_sigaction = nullptr;
0213           act.sa_flags = SA_SIGINFO;
0214           sigemptyset(&act.sa_mask);
0215           sigaction(SIGABRT, &act, nullptr);
0216           std::abort();
0217         }
0218       };
0219       std::set_terminate(releaseLock);
0220
0221       presentState = "setting up serializers";
0222       Serializer<ExternalGeneratorEventInfo> serializer(sm_buffer);
0223       Serializer<ExternalGeneratorLumiInfo> bl_serializer(sm_buffer);
0224       Serializer<GenLumiInfoProduct> el_serializer(sm_buffer);
0225       Serializer<GenRunInfoProduct> er_serializer(sm_buffer);
0226
0227       ROOTDeserializer<edm::RandomNumberGeneratorState, ReadBuffer> random_deserializer(sm_readbuffer);
0228
0229       presentState = "reading configuration";
0230       std::cerr << uniqueID << " process: initializing " << formattedTime().data() << std::endl;
0231       int nlines;
0232       std::cin >> nlines;
0233
0234       std::string configuration;
0235       for (int i = 0; i < nlines; ++i) {
0236         std::string c;
0237         std::getline(std::cin, c);
0238         if (verbose) {
0239           std::cerr << c << "\n";
0240         }
0241         configuration += c + "\n";
0242       }
0243
0244       presentState = "setting up random number generator";
0245       edm::ExternalRandomNumberGeneratorService* randomService = new edm::ExternalRandomNumberGeneratorService;
0246       auto serviceToken =
0247           edm::ServiceRegistry::createContaining(std::unique_ptr<edm::RandomNumberGenerator>(randomService));
0248       Harness harness(configuration, serviceToken);
0249
0250       //Some generator libraries override the signal handlers
0251       monitorThread.setupSignalHandling();
0252       std::set_terminate(releaseLock);
0253
0254       if (verbose) {
0255         std::cerr << uniqueID << " process: done initializing " << formattedTime().data() << std::endl;
0256       }
0257       presentState = "finished initialization";
0258       communicationChannel.workerSetupDone();
0259
0260       presentState = "waiting for transition";
0261       if (verbose)
0262         std::cerr << uniqueID << " process: waiting " << counter << " " << formattedTime().data() << std::endl;
0263       communicationChannel.handleTransitions([&](edm::Transition iTransition, unsigned long long iTransitionID) {
0264         ++counter;
0265         switch (iTransition) {
0266           case edm::Transition::BeginRun: {
0267             presentState = "beginRun transition";
0268             if (verbose)
0269               std::cerr << uniqueID << " process: start beginRun " << formattedTime().data() << std::endl;
0270             if (verbose)
0271               std::cerr << uniqueID << " process: end beginRun " << formattedTime().data() << std::endl;
0272
0273             break;
0274           }
0275           case edm::Transition::BeginLuminosityBlock: {
0276             presentState = "begin lumi";
0277             if (verbose)
0278               std::cerr << uniqueID << " process: start beginLumi " << formattedTime().data() << std::endl;
0279             auto randState = random_deserializer.deserialize();
0280             presentState = "deserialized random state in begin lumi";
0281             if (verbose)
0282               std::cerr << uniqueID << " random " << randState.state_.size() << " " << randState.seed_ << std::endl;
0283             randomService->setState(randState.state_, randState.seed_);
0284             presentState = "processing begin lumi";
0285             auto value = harness.getBeginLumiValue(iTransitionID);
0286             value.randomState_.state_ = randomService->getState();
0287             value.randomState_.seed_ = randomService->mySeed();
0288
0289             presentState = "serialize lumi";
0290             bl_serializer.serialize(value);
0291             if (verbose)
0292               std::cerr << uniqueID << " process: end beginLumi " << formattedTime().data() << std::endl;
0293             if (verbose)
0294               std::cerr << uniqueID << "   rand " << value.randomState_.state_.size() << " " << value.randomState_.seed_
0295                         << std::endl;
0296             break;
0297           }
0298           case edm::Transition::Event: {
0299             presentState = "begin event";
0300             if (verbose)
0301               std::cerr << uniqueID << " process: event " << counter << " " << formattedTime().data() << std::endl;
0302             presentState = "deserialized random state in event";
0303             auto randState = random_deserializer.deserialize();
0304             randomService->setState(randState.state_, randState.seed_);
0305             presentState = "processing event";
0306             auto value = harness.getEventValue();
0307             value.randomState_.state_ = randomService->getState();
0308             value.randomState_.seed_ = randomService->mySeed();
0309
0310             if (verbose)
0311               std::cerr << uniqueID << " process: event " << counter << " " << formattedTime().data() << std::endl;
0312
0313             presentState = "serialize event";
0314             serializer.serialize(value);
0315             if (verbose)
0316               std::cerr << uniqueID << " process: "
0317                         << " " << counter << std::endl;
0318             //usleep(10000000);
0319             break;
0320           }
0321           case edm::Transition::EndLuminosityBlock: {
0322             presentState = "begin end lumi";
0323             if (verbose)
0324               std::cerr << uniqueID << " process: start endLumi " << formattedTime().data() << std::endl;
0325             presentState = "processing end lumi";
0326             auto value = harness.getEndLumiValue();
0327
0328             presentState = "serialize end lumi";
0329             el_serializer.serialize(value);
0330             if (verbose)
0331               std::cerr << uniqueID << " process: end endLumi " << formattedTime().data() << std::endl;
0332
0333             break;
0334           }
0335           case edm::Transition::EndRun: {
0336             presentState = "begin end run";
0337             if (verbose)
0338               std::cerr << uniqueID << " process: start endRun " << formattedTime().data() << std::endl;
0339             presentState = "process end run";
0340             auto value = harness.getEndRunValue();
0341
0342             presentState = "serialize end run";
0343             er_serializer.serialize(value);
0344             if (verbose)
0345               std::cerr << uniqueID << " process: end endRun " << formattedTime().data() << std::endl;
0346
0347             break;
0348           }
0349           default: {
0350             assert(false);
0351           }
0352         }
0353         presentState = "notifying and waiting after " + presentState;
0354         if (verbose)
0355           std::cerr << uniqueID << " process: notifying and waiting " << counter << " " << std::endl;
0356       });
0357     }
0358   } catch (std::exception const& iExcept) {
0359     std::cerr << "%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%\n"
0360               << s_uniqueID << " process: caught exception \n"
0361               << iExcept.what() << " " << formattedTime().data() << "\n"
0362               << "  while " << presentState << "\n"
0363               << "%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%\n";
0364     return 1;
0365   } catch (...) {
0366     std::cerr << "%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%\n"
0367               << s_uniqueID << " process: caught unknown exception " << formattedTime().data() << "\n  while "
0368               << presentState << "\n"
0369               << "%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%\n";
0370     return 1;
0371   }
0372   return 0;
0373 }