CMS 3D CMS Logo

interprocess.cc
Go to the documentation of this file.
1 #include "boost/program_options.hpp"
2 
3 #include <atomic>
4 #include <csignal>
5 #include <iostream>
6 #include <string>
7 #include <thread>
8 #include <chrono>
9 
11 #include "DataFormats/TestObjects/interface/ToyProducts.h"
12 #include "DataFormats/TestObjects/interface/ThingCollection.h"
13 
18 
19 static char const* const kMemoryNameOpt = "memory-name";
20 static char const* const kMemoryNameCommandOpt = "memory-name,m";
21 static char const* const kUniqueIDOpt = "unique-id";
22 static char const* const kUniqueIDCommandOpt = "unique-id,i";
23 static char const* const kHelpOpt = "help";
24 static char const* const kHelpCommandOpt = "help,h";
25 
26 //NOTE: Can use TestProcessor as the harness for the worker
27 
28 using namespace edm::shared_memory;
29 class Harness {
30 public:
31  Harness(std::string const& iConfig) : tester_(edm::test::TestProcessor::Config{iConfig}) {}
32 
33  edmtest::ThingCollection getBeginRunValue(unsigned int iRun) {
34  auto run = tester_.testBeginRun(iRun);
35  return *run.get<edmtest::ThingCollection>("beginRun");
36  }
37 
38  edmtest::ThingCollection getBeginLumiValue(unsigned int iLumi) {
39  auto lumi = tester_.testBeginLuminosityBlock(iLumi);
40  return *lumi.get<edmtest::ThingCollection>("beginLumi");
41  }
42 
43  edmtest::ThingCollection getEventValue() {
44  auto event = tester_.test();
45  return *event.get<edmtest::ThingCollection>();
46  }
47 
48  edmtest::ThingCollection getEndLumiValue() {
49  auto lumi = tester_.testEndLuminosityBlock();
50  return *lumi.get<edmtest::ThingCollection>("endLumi");
51  }
52 
53  edmtest::ThingCollection getEndRunValue() {
54  auto run = tester_.testEndRun();
55  return *run.get<edmtest::ThingCollection>("endRun");
56  }
57 
58 private:
60 };
61 
62 int main(int argc, char* argv[]) {
63  std::string descString(argv[0]);
64  descString += " [--";
65  descString += kMemoryNameOpt;
66  descString += "] memory_name";
67  boost::program_options::options_description desc(descString);
68 
69  desc.add_options()(kHelpCommandOpt, "produce help message")(
70  kMemoryNameCommandOpt, boost::program_options::value<std::string>(), "memory name")(
71  kUniqueIDCommandOpt, boost::program_options::value<std::string>(), "unique id");
72 
73  boost::program_options::positional_options_description p;
74  p.add(kMemoryNameOpt, 1);
75  p.add(kUniqueIDOpt, 2);
76 
77  boost::program_options::options_description all_options("All Options");
78  all_options.add(desc);
79 
80  boost::program_options::variables_map vm;
81  try {
82  store(boost::program_options::command_line_parser(argc, argv).options(all_options).positional(p).run(), vm);
83  notify(vm);
84  } catch (boost::program_options::error const& iException) {
85  std::cout << argv[0] << ": Error while trying to process command line arguments:\n"
86  << iException.what() << "\nFor usage and an options list, please do 'cmsRun --help'.";
87  return 1;
88  }
89 
90  if (vm.count(kHelpOpt)) {
91  std::cout << desc << std::endl;
92  return 0;
93  }
94 
95  if (!vm.count(kMemoryNameOpt)) {
96  std::cout << " no argument given" << std::endl;
97  return 1;
98  }
99 
100  if (!vm.count(kUniqueIDOpt)) {
101  std::cout << " no second argument given" << std::endl;
102  return 1;
103  }
104 
105  WorkerMonitorThread monitorThread;
106 
107  monitorThread.startThread();
108 
109  CMS_SA_ALLOW try {
110  std::string const memoryName(vm[kMemoryNameOpt].as<std::string>());
111  std::string const uniqueID(vm[kUniqueIDOpt].as<std::string>());
112  {
113  //using namespace boost::interprocess;
114  //auto controlNameUnique = unique_name(memoryName, uniqueID);
115 
116  //This class is holding the lock
117  WorkerChannel communicationChannel(memoryName, uniqueID);
118 
119  WriteBuffer sm_buffer{memoryName, communicationChannel.fromWorkerBufferInfo()};
120  int counter = 0;
121 
122  //The lock must be released if there is a catastrophic signal
123  auto lockPtr = communicationChannel.accessLock();
124  monitorThread.setAction([lockPtr]() {
125  if (lockPtr) {
126  std::cerr << "SIGNAL CAUGHT: unlock\n";
127  lockPtr->unlock();
128  }
129  });
130 
132  TCSerializer serializer(sm_buffer);
133  TCSerializer br_serializer(sm_buffer);
134  TCSerializer bl_serializer(sm_buffer);
135  TCSerializer el_serializer(sm_buffer);
136  TCSerializer er_serializer(sm_buffer);
137 
138  std::cerr << uniqueID << " process: initializing " << std::endl;
139  int nlines;
140  std::cin >> nlines;
141 
143  for (int i = 0; i < nlines; ++i) {
144  std::string c;
145  std::getline(std::cin, c);
146  std::cerr << c << "\n";
147  configuration += c + "\n";
148  }
149 
150  Harness harness(configuration);
151 
152  //Either ROOT or the Framework are overriding the signal handlers
153  monitorThread.setupSignalHandling();
154 
155  std::cerr << uniqueID << " process: done initializing" << std::endl;
156  communicationChannel.workerSetupDone();
157 
158  std::cerr << uniqueID << " process: waiting " << counter << std::endl;
159  communicationChannel.handleTransitions([&](edm::Transition iTransition, unsigned long long iTransitionID) {
160  ++counter;
161  switch (iTransition) {
163  std::cerr << uniqueID << " process: start beginRun " << std::endl;
164  auto value = harness.getBeginRunValue(iTransitionID);
165 
166  br_serializer.serialize(value);
167  std::cerr << uniqueID << " process: end beginRun " << value.size() << std::endl;
168 
169  break;
170  }
172  std::cerr << uniqueID << " process: start beginLumi " << std::endl;
173  auto value = harness.getBeginLumiValue(iTransitionID);
174 
175  bl_serializer.serialize(value);
176  std::cerr << uniqueID << " process: end beginLumi " << value.size() << std::endl;
177 
178  break;
179  }
180  case edm::Transition::Event: {
181  std::cerr << uniqueID << " process: integrating " << counter << std::endl;
182  auto value = harness.getEventValue();
183 
184  std::cerr << uniqueID << " process: integrated " << counter << std::endl;
185 
186  serializer.serialize(value);
187  std::cerr << uniqueID << " process: " << value.size() << " " << counter << std::endl;
188  //std::this_thread::sleep_for(std::chrono::microseconds(10000000));
189  break;
190  }
192  std::cerr << uniqueID << " process: start endLumi " << std::endl;
193  auto value = harness.getEndLumiValue();
194 
195  el_serializer.serialize(value);
196  std::cerr << uniqueID << " process: end endLumi " << value.size() << std::endl;
197 
198  break;
199  }
201  std::cerr << uniqueID << " process: start endRun " << std::endl;
202  auto value = harness.getEndRunValue();
203 
204  er_serializer.serialize(value);
205  std::cerr << uniqueID << " process: end endRun " << value.size() << std::endl;
206 
207  break;
208  }
209  default: {
210  assert(false);
211  }
212  }
213  std::cerr << uniqueID << " process: notifying and waiting" << counter << std::endl;
214  });
215  }
216  } catch (std::exception const& iExcept) {
217  std::cerr << "caught exception \n" << iExcept.what() << "\n";
218  return 1;
219  } catch (...) {
220  std::cerr << "caught unknown exception";
221  return 1;
222  }
223  return 0;
224 }
edmtest::ThingCollection getEndLumiValue()
Definition: interprocess.cc:48
#define CMS_SA_ALLOW
int main(int argc, char *argv[])
Definition: interprocess.cc:62
edmtest::ThingCollection getEventValue()
Definition: interprocess.cc:43
boost::interprocess::scoped_lock< boost::interprocess::named_mutex > * accessLock()
the lock is made accessible so that the WorkerMonitorThread can be used to unlock it in the event of ...
Definition: WorkerChannel.h:48
edm::test::TestProcessor tester_
Definition: interprocess.cc:59
edmtest::ThingCollection getBeginLumiValue(unsigned int iLumi)
Definition: interprocess.cc:38
void setupSignalHandling()
Sets the unix signal handler which communicates with the thread.
void setAction(std::function< void()> iFunc)
assert(be >=bs)
static char const *const kUniqueIDCommandOpt
Definition: interprocess.cc:22
static char const *const kUniqueIDOpt
Definition: interprocess.cc:21
void workerSetupDone()
Matches the ControllerChannel::setupWorker call.
Definition: WorkerChannel.h:56
dictionary configuration
Definition: JetHT_cfg.py:38
Transition
Definition: Transition.h:12
Harness(std::string const &iConfig)
Definition: interprocess.cc:31
Definition: value.py:1
static char const *const kMemoryNameOpt
Definition: interprocess.cc:19
BufferInfo * fromWorkerBufferInfo()
This can be used with WriteBuffer to keep Controller and Worker in sync.
Definition: WorkerChannel.h:53
edmtest::ThingCollection getEndRunValue()
Definition: interprocess.cc:53
HLT enums.
static std::atomic< unsigned int > counter
Definition: Config.py:1
static char const *const kMemoryNameCommandOpt
Definition: interprocess.cc:20
static char const *const kHelpOpt
Definition: interprocess.cc:23
static char const *const kHelpCommandOpt
Definition: interprocess.cc:24
edmtest::ThingCollection getBeginRunValue(unsigned int iRun)
Definition: interprocess.cc:33