Line data Source code
1 : #include "TRACE/tracemf.h"
2 : #include "artdaq/DAQdata/Globals.hh" // include these 2 first -
3 : #define TRACE_NAME (app_name + "_DispatcherCore").c_str()
4 :
5 : #include "artdaq/Application/DispatcherCore.hh"
6 :
7 : #include "fhiclcpp/ParameterSet.h"
8 :
9 : #include <boost/algorithm/string.hpp>
10 : #include <boost/exception/all.hpp>
11 : #include <boost/throw_exception.hpp>
12 : #include <boost/tokenizer.hpp>
13 :
14 : #include <bitset>
15 : #include <cerrno>
16 : #include <csignal>
17 : #include <iomanip>
18 : #include <set>
19 : #include <sstream>
20 : #include <string>
21 : #include <unordered_map>
22 : #include <utility>
23 : #include <vector>
24 :
25 0 : bool artdaq::DispatcherCore::initialize(fhicl::ParameterSet const& pset)
26 : {
27 0 : TLOG(TLVL_DEBUG + 32) << "initialize method called with DAQ "
28 0 : << "ParameterSet = \"" << pset.to_string() << "\".";
29 :
30 0 : pset_ = pset;
31 : // 04-Apr-2019, KAB: added support for art config params to be in an "art" block
32 0 : if (pset_.has_key("art"))
33 : {
34 0 : pset_ = pset_.get<fhicl::ParameterSet>("art");
35 : }
36 0 : pset_.erase("outputs");
37 0 : pset_.erase("physics");
38 0 : pset_.erase("daq");
39 :
40 0 : TLOG(TLVL_DEBUG + 32) << "Pieces of the input pset that are saved for later: \"" << pset_.to_string() << "\".";
41 :
42 : // pull out the relevant parts of the ParameterSet
43 0 : fhicl::ParameterSet daq_pset;
44 : try
45 : {
46 0 : daq_pset = pset.get<fhicl::ParameterSet>("daq");
47 : }
48 0 : catch (...)
49 : {
50 0 : TLOG(TLVL_ERROR)
51 0 : << "Unable to find the DAQ parameters in the initialization "
52 0 : << "ParameterSet: \"" + pset.to_string() + "\".";
53 0 : return false;
54 0 : }
55 0 : fhicl::ParameterSet agg_pset;
56 : try
57 : {
58 0 : agg_pset = daq_pset.get<fhicl::ParameterSet>("dispatcher", daq_pset.get<fhicl::ParameterSet>("aggregator"));
59 : }
60 0 : catch (...)
61 : {
62 0 : TLOG(TLVL_ERROR)
63 0 : << "Unable to find the Dispatcher parameters in the DAQ "
64 0 : << "initialization ParameterSet: \"" + daq_pset.to_string() + "\".";
65 0 : return false;
66 0 : }
67 :
68 0 : if (!agg_pset.has_key("init_fragment_count"))
69 : {
70 0 : TLOG(TLVL_ERROR) << "ERROR: init_fragment_count must be set for the Dispatcher to function properly. Please set `init_fragment_count: 0` in the Dispatcher configuration. (Will be book-kept by DAQInterface)";
71 0 : return false;
72 : }
73 0 : broadcast_mode_ = agg_pset.get<bool>("broadcast_mode", true);
74 0 : allow_label_overwrites_ = agg_pset.get<bool>("allow_label_overwrites", true);
75 0 : if (broadcast_mode_ && !agg_pset.has_key("broadcast_mode"))
76 : {
77 0 : agg_pset.put<bool>("broadcast_mode", true);
78 0 : if (!agg_pset.has_key("non_reliable_mode"))
79 : {
80 0 : agg_pset.put<bool>("non_reliable_mode", true);
81 : }
82 0 : if (!agg_pset.has_key("non_reliable_mode_retry_count"))
83 : {
84 0 : agg_pset.put<int>("non_reliable_mode_retry_count", 2);
85 : }
86 : }
87 :
88 0 : agg_pset.erase("restart_crashed_art_processes");
89 0 : agg_pset.put<bool>("restart_crashed_art_processes", false);
90 :
91 0 : agg_pset.erase("art_analyzer_count");
92 0 : agg_pset.put<int>("art_analyzer_count", 0);
93 :
94 : // initialize the MetricManager and the names of our metrics
95 0 : fhicl::ParameterSet metric_pset = daq_pset.get<fhicl::ParameterSet>("metrics", fhicl::ParameterSet());
96 :
97 0 : return initializeDataReceiver(pset, agg_pset, metric_pset);
98 0 : }
99 :
100 0 : std::string artdaq::DispatcherCore::register_monitor(fhicl::ParameterSet const& pset)
101 : {
102 0 : TLOG(TLVL_DEBUG + 32) << "DispatcherCore::register_monitor called with argument \"" << pset.to_string() << "\"";
103 0 : check_filters_();
104 :
105 0 : if (stop_requested_.load())
106 : {
107 0 : TLOG(TLVL_WARNING) << "DispatcherCore::register_monitor called after stop. NOT registering new monitor!";
108 0 : return "Stopped";
109 : }
110 :
111 : try
112 : {
113 0 : TLOG(TLVL_DEBUG + 32) << "Getting unique_label from input ParameterSet";
114 0 : auto const& label = pset.get<std::string>("unique_label");
115 0 : TLOG(TLVL_DEBUG + 32) << "Unique label is " << label;
116 0 : TLOG(TLVL_INFO) << "Registering monitor with unique_label \"" << label << "\"";
117 :
118 0 : bool overwriting_label = false;
119 :
120 : {
121 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
122 0 : if (registered_monitors_.count(label) != 0u)
123 : {
124 0 : if (allow_label_overwrites_)
125 : {
126 0 : TLOG(TLVL_WARNING) << "Restarting companion art process with label " << label << "!";
127 0 : overwriting_label = true;
128 : }
129 : else
130 : {
131 0 : TLOG(TLVL_ERROR) << "Unique label already exists!";
132 0 : throw cet::exception("DispatcherCore") << "Unique label already exists!"; // NOLINT(cert-err60-cpp)
133 : }
134 : }
135 :
136 0 : registered_monitors_[label] = pset;
137 0 : }
138 :
139 0 : if (overwriting_label)
140 : {
141 : // ELF, Jul 21, 2020: This can take a long time, and we don't want to block the XMLRPC thread
142 0 : boost::thread thread([this, label] { restart_art_process_(label); });
143 0 : thread.detach();
144 0 : }
145 : else
146 : {
147 : // ELF, Jul 21, 2020: This can take a long time, and we don't want to block the XMLRPC thread
148 0 : boost::thread thread([this, label] { start_art_process_(label); });
149 0 : thread.detach();
150 0 : }
151 0 : }
152 0 : catch (const cet::exception& e)
153 : {
154 0 : std::stringstream errmsg;
155 0 : errmsg << "Unable to create a Transfer plugin with the FHiCL code \"" << pset.to_string() << "\", a new monitor has not been registered" << std::endl;
156 0 : errmsg << "Exception: " << e.what();
157 0 : TLOG(TLVL_ERROR) << errmsg.str();
158 0 : return errmsg.str();
159 0 : }
160 0 : catch (const boost::exception& e)
161 : {
162 0 : std::stringstream errmsg;
163 0 : errmsg << "Unable to create a Transfer plugin with the FHiCL code \"" << pset.to_string() << "\", a new monitor has not been registered" << std::endl;
164 0 : errmsg << "Exception: " << boost::diagnostic_information(e);
165 0 : TLOG(TLVL_ERROR) << errmsg.str();
166 0 : return errmsg.str();
167 0 : }
168 0 : catch (const std::exception& e)
169 : {
170 0 : std::stringstream errmsg;
171 0 : errmsg << "Unable to create a Transfer plugin with the FHiCL code \"" << pset.to_string() << "\", a new monitor has not been registered" << std::endl;
172 0 : errmsg << "Exception: " << e.what();
173 0 : TLOG(TLVL_ERROR) << errmsg.str();
174 0 : return errmsg.str();
175 0 : }
176 0 : catch (...)
177 : {
178 0 : std::stringstream errmsg;
179 0 : errmsg << "Unable to create a Transfer plugin with the FHiCL code \"" << pset.to_string() << "\", a new monitor has not been registered";
180 0 : TLOG(TLVL_ERROR) << errmsg.str();
181 0 : return errmsg.str();
182 0 : }
183 :
184 0 : TLOG(TLVL_DEBUG + 32) << "Successfully registered monitor";
185 0 : return "Success";
186 : }
187 :
188 0 : std::string artdaq::DispatcherCore::unregister_monitor(std::string const& label)
189 : {
190 0 : TLOG(TLVL_INFO) << "DispatcherCore::unregister_monitor called for unique_label \"" << label << "\"";
191 0 : check_filters_();
192 :
193 : try
194 : {
195 : {
196 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
197 0 : if (registered_monitors_.count(label) == 0)
198 : {
199 0 : std::stringstream errmsg;
200 : errmsg << "Warning in DispatcherCore::unregister_monitor: unable to find requested transfer plugin with "
201 0 : << "label \"" << label << "\"";
202 0 : TLOG(TLVL_WARNING) << errmsg.str();
203 0 : return errmsg.str();
204 0 : }
205 :
206 0 : registered_monitors_.erase(label);
207 0 : }
208 :
209 : // ELF, Jul 21, 2020: This can take a long time, and we don't want to block the XMLRPC thread
210 0 : boost::thread thread([this, label] { stop_art_process_(label); });
211 0 : thread.detach();
212 0 : }
213 0 : catch (...)
214 : {
215 0 : std::stringstream errmsg;
216 0 : errmsg << "Unable to unregister transfer plugin with label \"" << label << "\"";
217 0 : TLOG(TLVL_ERROR) << errmsg.str();
218 0 : return errmsg.str();
219 0 : }
220 :
221 0 : TLOG(TLVL_DEBUG + 32) << "unregister_monitor completed successfully";
222 0 : return "Success";
223 : }
224 :
225 0 : fhicl::ParameterSet artdaq::DispatcherCore::merge_parameter_sets_(fhicl::ParameterSet const& skel, const std::string& label, const fhicl::ParameterSet& pset)
226 : {
227 0 : fhicl::ParameterSet generated_pset(skel);
228 0 : fhicl::ParameterSet generated_outputs;
229 0 : fhicl::ParameterSet generated_physics;
230 0 : fhicl::ParameterSet generated_physics_analyzers;
231 0 : fhicl::ParameterSet generated_physics_producers;
232 0 : fhicl::ParameterSet generated_physics_filters;
233 0 : std::unordered_map<std::string, std::vector<std::string>> generated_physics_filter_paths;
234 :
235 0 : TLOG(TLVL_DEBUG + 32) << "merge_parameter_sets_: Generating fhicl for monitor " << label;
236 :
237 : try
238 : {
239 0 : auto path = pset.get<std::vector<std::string>>("path");
240 :
241 0 : auto filters = pset.get<std::vector<fhicl::ParameterSet>>("filter_paths", std::vector<fhicl::ParameterSet>());
242 0 : for (auto& filter : filters)
243 : {
244 : try
245 : {
246 0 : auto name = filter.get<std::string>("name");
247 0 : auto filter_path = filter.get<std::vector<std::string>>("path");
248 0 : if (generated_physics_filter_paths.count(name) != 0u)
249 : {
250 0 : bool matched = generated_physics_filter_paths[name].size() == filter_path.size();
251 0 : for (size_t ii = 0; matched && ii < generated_physics_filter_paths[name].size(); ++ii)
252 : {
253 0 : matched = matched && filter_path[ii] == generated_physics_filter_paths[name][ii];
254 : }
255 :
256 0 : if (matched)
257 : {
258 : // Path is already configured
259 0 : continue;
260 : }
261 :
262 0 : auto newname = label + name;
263 0 : generated_physics_filter_paths[newname] = filter_path;
264 0 : }
265 : else
266 : {
267 0 : generated_physics_filter_paths[name] = filter_path;
268 : }
269 0 : }
270 0 : catch (...)
271 : {
272 0 : }
273 : }
274 :
275 : // outputs section
276 0 : auto outputs = pset.get<fhicl::ParameterSet>("outputs");
277 0 : if (outputs.get_pset_names().size() > 1 || outputs.get_pset_names().empty())
278 : {
279 : // Only one output allowed per monitor
280 : }
281 0 : auto output_name = outputs.get_pset_names()[0];
282 0 : auto output_pset = outputs.get<fhicl::ParameterSet>(output_name);
283 0 : generated_outputs.put(label + output_name, output_pset);
284 0 : bool outputInPath = false;
285 0 : for (auto& ii : path)
286 : {
287 0 : if (ii == output_name)
288 : {
289 0 : ii = label + output_name;
290 0 : outputInPath = true;
291 : }
292 : }
293 0 : if (!outputInPath)
294 : {
295 0 : path.push_back(label + output_name);
296 : }
297 :
298 : // physics section
299 0 : auto physics_pset = pset.get<fhicl::ParameterSet>("physics");
300 :
301 0 : if (physics_pset.has_key("analyzers"))
302 : {
303 0 : auto analyzers = physics_pset.get<fhicl::ParameterSet>("analyzers");
304 0 : for (const auto& key : analyzers.get_pset_names())
305 : {
306 0 : if (generated_physics_analyzers.has_key(key) && analyzers.get<fhicl::ParameterSet>(key) == generated_physics_analyzers.get<fhicl::ParameterSet>(key))
307 : {
308 : // Module is already configured
309 0 : continue;
310 : }
311 0 : if (generated_physics_analyzers.has_key(key))
312 : {
313 : // Module already exists with name, rename
314 0 : auto newkey = label + key;
315 0 : generated_physics_analyzers.put<fhicl::ParameterSet>(newkey, analyzers.get<fhicl::ParameterSet>(key));
316 0 : for (auto& ii : path)
317 : {
318 0 : if (ii == key)
319 : {
320 0 : ii = newkey;
321 : }
322 : }
323 0 : }
324 : else
325 : {
326 0 : generated_physics_analyzers.put<fhicl::ParameterSet>(key, analyzers.get<fhicl::ParameterSet>(key));
327 : }
328 0 : }
329 0 : }
330 0 : if (physics_pset.has_key("producers"))
331 : {
332 0 : auto producers = physics_pset.get<fhicl::ParameterSet>("producers");
333 0 : for (const auto& key : producers.get_pset_names())
334 : {
335 0 : if (generated_physics_producers.has_key(key) && producers.get<fhicl::ParameterSet>(key) == generated_physics_producers.get<fhicl::ParameterSet>(key))
336 : {
337 : // Module is already configured
338 0 : continue;
339 : }
340 0 : if (generated_physics_producers.has_key(key))
341 : {
342 : // Module already exists with name, rename
343 0 : auto newkey = label + key;
344 0 : generated_physics_producers.put<fhicl::ParameterSet>(newkey, producers.get<fhicl::ParameterSet>(key));
345 0 : for (auto& ii : path)
346 : {
347 0 : if (ii == key)
348 : {
349 0 : ii = newkey;
350 : }
351 : }
352 0 : }
353 : else
354 : {
355 0 : generated_physics_producers.put<fhicl::ParameterSet>(key, producers.get<fhicl::ParameterSet>(key));
356 : }
357 0 : }
358 0 : }
359 0 : if (physics_pset.has_key("filters"))
360 : {
361 0 : auto filters = physics_pset.get<fhicl::ParameterSet>("filters");
362 0 : for (const auto& key : filters.get_pset_names())
363 : {
364 0 : if (generated_physics_filters.has_key(key) && filters.get<fhicl::ParameterSet>(key) == generated_physics_filters.get<fhicl::ParameterSet>(key))
365 : {
366 : // Module is already configured
367 0 : continue;
368 : }
369 0 : if (generated_physics_filters.has_key(key))
370 : {
371 : // Module already exists with name, rename
372 0 : auto newkey = label + key;
373 0 : generated_physics_filters.put<fhicl::ParameterSet>(newkey, filters.get<fhicl::ParameterSet>(key));
374 0 : for (auto& ii : path)
375 : {
376 0 : if (ii == key)
377 : {
378 0 : ii = newkey;
379 : }
380 : }
381 0 : }
382 : else
383 : {
384 0 : generated_physics_filters.put<fhicl::ParameterSet>(key, filters.get<fhicl::ParameterSet>(key));
385 : }
386 0 : }
387 0 : }
388 0 : generated_physics.put<std::vector<std::string>>(label, path);
389 0 : }
390 0 : catch (cet::exception& e)
391 : {
392 : // Error in parsing input fhicl
393 0 : TLOG(TLVL_ERROR) << "merge_parameter_sets_: Error processing input fhicl: " << e.what();
394 0 : }
395 :
396 0 : TLOG(TLVL_DEBUG + 32) << "merge_parameter_sets_: Building final ParameterSet";
397 0 : generated_pset.put("outputs", generated_outputs);
398 :
399 0 : generated_physics.put("analyzers", generated_physics_analyzers);
400 0 : generated_physics.put("producers", generated_physics_producers);
401 0 : generated_physics.put("filters", generated_physics_filters);
402 :
403 0 : for (auto& filter_path : generated_physics_filter_paths)
404 : {
405 0 : generated_physics.put(filter_path.first, filter_path.second);
406 : }
407 :
408 0 : generated_pset.put("physics", generated_physics);
409 :
410 0 : return generated_pset;
411 0 : }
412 :
413 0 : fhicl::ParameterSet artdaq::DispatcherCore::generate_filter_fhicl_()
414 : {
415 0 : TLOG(TLVL_DEBUG + 32) << "generate_filter_fhicl_ BEGIN";
416 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
417 0 : fhicl::ParameterSet generated_pset = pset_;
418 :
419 0 : for (auto& monitor : registered_monitors_)
420 : {
421 0 : auto label = monitor.first;
422 0 : auto pset = monitor.second;
423 0 : generated_pset = merge_parameter_sets_(generated_pset, label, pset);
424 0 : }
425 :
426 0 : TLOG(TLVL_DEBUG + 32) << "generate_filter_fhicl_ returning ParameterSet: " << generated_pset.to_string();
427 0 : return generated_pset;
428 0 : }
429 :
430 0 : void artdaq::DispatcherCore::check_filters_()
431 : {
432 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
433 0 : auto it = registered_monitors_.begin();
434 0 : while (it != registered_monitors_.end())
435 : {
436 0 : if (registered_monitor_pids_.count(it->first) != 0u)
437 : {
438 0 : if (!event_store_ptr_)
439 : {
440 0 : registered_monitor_pids_.erase(it->first);
441 : }
442 : else
443 : {
444 0 : auto pid = registered_monitor_pids_[it->first];
445 0 : auto sts = kill(pid, 0);
446 0 : if (sts < 0)
447 : {
448 0 : registered_monitor_pids_.erase(it->first);
449 : }
450 : }
451 : }
452 0 : ++it;
453 : }
454 0 : }
455 :
456 0 : void artdaq::DispatcherCore::restart_art_process_(std::string const& label)
457 : {
458 0 : stop_art_process_(label);
459 0 : start_art_process_(label);
460 0 : }
461 :
462 0 : void artdaq::DispatcherCore::start_art_process_(std::string const& label)
463 : {
464 0 : if (event_store_ptr_ != nullptr)
465 : {
466 0 : if (broadcast_mode_)
467 : {
468 0 : fhicl::ParameterSet pset;
469 : {
470 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
471 0 : pset = registered_monitors_[label];
472 0 : }
473 0 : fhicl::ParameterSet ps = merge_parameter_sets_(pset_, label, pset);
474 0 : TLOG(TLVL_DEBUG + 32) << "Starting art process with received fhicl";
475 0 : auto pid = event_store_ptr_->StartArtProcess(ps, registered_monitors_.size() - 1);
476 : {
477 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
478 0 : registered_monitor_pids_[label] = pid;
479 0 : }
480 0 : }
481 : else
482 : {
483 0 : TLOG(TLVL_DEBUG + 32) << "Calling ReconfigureArt";
484 0 : fhicl::ParameterSet generated_fhicl;
485 : {
486 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
487 0 : generated_fhicl = generate_filter_fhicl_();
488 0 : }
489 0 : event_store_ptr_->ReconfigureArt(generated_fhicl);
490 0 : TLOG(TLVL_DEBUG + 32) << "Done with ReconfigureArt";
491 0 : }
492 : }
493 : else
494 : {
495 0 : TLOG(TLVL_ERROR) << "Unable to add monitor as there is no SharedMemoryEventManager instance!";
496 : }
497 0 : }
498 :
499 0 : void artdaq::DispatcherCore::stop_art_process_(std::string const& label)
500 : {
501 0 : if (event_store_ptr_ != nullptr)
502 : {
503 0 : if (broadcast_mode_)
504 : {
505 0 : if (registered_monitor_pids_.count(label) != 0u)
506 : {
507 0 : std::set<pid_t> pids;
508 : {
509 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
510 0 : pids.insert(registered_monitor_pids_[label]);
511 0 : registered_monitor_pids_.erase(label);
512 0 : }
513 0 : TLOG(TLVL_DEBUG + 32) << "Calling ShutdownArtProcesses";
514 0 : event_store_ptr_->ShutdownArtProcesses(pids);
515 0 : TLOG(TLVL_DEBUG + 32) << "Done with ShutdownArtProcesses";
516 0 : }
517 : }
518 : else
519 : {
520 0 : TLOG(TLVL_DEBUG + 32) << "Calling ReconfigureArt";
521 0 : fhicl::ParameterSet generated_fhicl;
522 : {
523 0 : std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
524 0 : generated_fhicl = generate_filter_fhicl_();
525 0 : }
526 0 : event_store_ptr_->ReconfigureArt(generated_fhicl);
527 0 : TLOG(TLVL_DEBUG + 32) << "Done with ReconfigureArt";
528 0 : }
529 : }
530 0 : }
|
