Line data Source code
1 : /**
2 : * @file generate_modules.cpp
3 : *
4 : * Implementation of ReadoutApplication's generate_modules dal method
5 : *
6 : * This is part of the DUNE DAQ Software Suite, copyright 2023.
7 : * Licensing/copyright details are in the COPYING file that you should have
8 : * received with this code.
9 : */
10 :
11 : #include "appmodel/ReadoutApplication.hpp"
12 : #include "ConfigObjectFactory.hpp"
13 : #include "appmodel/ConfigurationHelper.hpp"
14 : #include "conffwk/Configuration.hpp"
15 : #include "confmodel/DetDataReceiver.hpp"
16 : #include "confmodel/DetDataSender.hpp"
17 : #include "confmodel/DetectorStream.hpp"
18 : #include "confmodel/DetectorToDaqConnection.hpp"
19 : #include "confmodel/Session.hpp"
20 :
21 : #include "appmodel/DPDKReceiver.hpp"
22 : #include "appmodel/NWDetDataReceiver.hpp"
23 : #include "appmodel/NWDetDataSender.hpp"
24 :
25 : #include "appmodel/FelixDataReceiver.hpp"
26 : #include "appmodel/FelixDataSender.hpp"
27 :
28 : #include "appmodel/SocketReceiver.hpp"
29 : #include "confmodel/QueueWithSourceId.hpp"
30 :
31 : #include "confmodel/Connection.hpp"
32 : #include "confmodel/ExcludableEntitySet.hpp"
33 : #include "confmodel/GeoId.hpp"
34 : #include "confmodel/NetworkConnection.hpp"
35 : #include "confmodel/Service.hpp"
36 :
37 : #include "appmodel/DataMoveCallbackConf.hpp"
38 : #include "appmodel/DataReaderConf.hpp"
39 : #include "appmodel/DataReaderModule.hpp"
40 : #include "appmodel/DataRecorderConf.hpp"
41 : #include "appmodel/DataRecorderModule.hpp"
42 : #include "appmodel/SourceIDConf.hpp"
43 :
44 : #include "appmodel/DataHandlerConf.hpp"
45 : #include "appmodel/DataHandlerModule.hpp"
46 : #include "appmodel/FragmentAggregatorConf.hpp"
47 : #include "appmodel/FragmentAggregatorModule.hpp"
48 : #include "appmodel/NetworkConnectionDescriptor.hpp"
49 : #include "appmodel/NetworkConnectionRule.hpp"
50 : #include "appmodel/QueueConnectionRule.hpp"
51 : #include "appmodel/QueueDescriptor.hpp"
52 : #include "appmodel/RequestHandler.hpp"
53 :
54 : #include "appmodel/appmodelIssues.hpp"
55 :
56 : #include "logging/Logging.hpp"
57 : #include <fmt/core.h>
58 :
59 : #include <string>
60 : #include <vector>
61 :
62 : // using namespace dunedaq;
63 : // using namespace dunedaq::appmodel;
64 :
65 : namespace dunedaq {
66 : namespace appmodel {
67 :
68 : //-----------------------------------------------------------------------------
69 :
70 : std::vector<const confmodel::ExcludableEntity*>
71 12 : ReadoutApplication::contained_excludable_entities() const
72 : {
73 12 : return to_resources(get_detector_connections());
74 : }
75 :
76 : void
77 0 : ReadoutApplication::generate_modules(std::shared_ptr<ConfigurationHelper> helper) const
78 : {
79 :
80 0 : TLOG_DEBUG(6) << "Generating modules for application " << this->UID();
81 :
82 0 : ConfigObjectFactory obj_fac(this);
83 : // conffwk::Configuration& confdb = this->configuration();
84 :
85 : //
86 : // Extract basic configuration objects
87 : //
88 :
89 : // Data reader
90 0 : auto reader_conf = get_data_reader();
91 0 : if (reader_conf == 0) {
92 : throw(BadConf(ERS_HERE, "No DataReaderModule configuration given"));
93 : }
94 0 : std::string reader_class = reader_conf->get_template_for();
95 :
96 : // Link handler
97 0 : auto dlh_conf = get_link_handler();
98 : // What is template for?
99 0 : auto dlh_class = dlh_conf->get_template_for();
100 :
101 0 : auto tph_conf = get_tp_handler();
102 0 : if (tph_conf == nullptr && get_tp_generation_enabled()) {
103 0 : throw(BadConf(ERS_HERE, "TP generation is enabled but there is no TP data handler configuration"));
104 : }
105 :
106 0 : std::string tph_class = "";
107 0 : if (tph_conf != nullptr && get_tp_generation_enabled()) {
108 0 : tph_class = tph_conf->get_template_for();
109 : }
110 :
111 : //
112 : // Process the queue rules looking for inputs to our DL/TP handler modules
113 : //
114 0 : const QueueDescriptor* dlh_reqinput_qdesc = nullptr;
115 0 : const QueueDescriptor* tp_input_qdesc = nullptr;
116 : // const QueueDescriptor* tpReqInputQDesc = nullptr;
117 0 : const QueueDescriptor* fa_output_qdesc = nullptr;
118 :
119 0 : for (auto rule : get_queue_rules()) {
120 0 : auto destination_class = rule->get_destination_class();
121 0 : auto data_type = rule->get_descriptor()->get_data_type();
122 : // Why datahander here? It is the base class for several DataHandler types (e.g. FDDataHandlerModule,
123 : // SNBDataHandlerModule)
124 0 : if (destination_class == "DataHandlerModule" || destination_class == dlh_class || destination_class == tph_class) {
125 0 : if (data_type == "DataRequest") {
126 0 : dlh_reqinput_qdesc = rule->get_descriptor();
127 0 : } else if ((data_type == "TriggerPrimitive" || data_type == "TriggerPrimitiveVector") &&
128 0 : get_tp_generation_enabled()) {
129 0 : tp_input_qdesc = rule->get_descriptor();
130 : }
131 0 : } else if (destination_class == "FragmentAggregatorModule") {
132 0 : fa_output_qdesc = rule->get_descriptor();
133 : }
134 0 : }
135 :
136 0 : if (dlh_reqinput_qdesc == nullptr) {
137 0 : throw(BadConf(ERS_HERE, "No data link handler request input queue descriptor given"));
138 : }
139 :
140 : //
141 : // Process the network rules looking for the Fragment Aggregator and TP handler data reuest inputs
142 : //
143 0 : const NetworkConnectionDescriptor* fa_net_desc = nullptr;
144 0 : const NetworkConnectionDescriptor* tp_net_desc = nullptr;
145 0 : const NetworkConnectionDescriptor* ta_net_desc = nullptr;
146 0 : const NetworkConnectionDescriptor* ts_net_desc = nullptr;
147 0 : for (auto rule : get_network_rules()) {
148 0 : auto endpoint_class = rule->get_endpoint_class();
149 0 : auto data_type = rule->get_descriptor()->get_data_type();
150 :
151 0 : if (endpoint_class == "FragmentAggregatorModule") {
152 0 : fa_net_desc = rule->get_descriptor();
153 0 : } else if (data_type == "TPSet") {
154 0 : tp_net_desc = rule->get_descriptor();
155 0 : } else if (data_type == "TriggerActivity") {
156 0 : ta_net_desc = rule->get_descriptor();
157 0 : } else if (data_type == "TimeSync") {
158 0 : ts_net_desc = rule->get_descriptor();
159 : }
160 0 : }
161 :
162 0 : if (fa_net_desc == nullptr) {
163 0 : throw(BadConf(ERS_HERE, "No Fragment Aggregator network descriptor given"));
164 : }
165 0 : if (ts_net_desc == nullptr && dlh_conf->get_generate_timesync()) {
166 0 : throw(BadConf(ERS_HERE, "No Time Sync network descriptor given but time sync generation is enabled"));
167 : }
168 :
169 : //
170 : // Get the callback descriptor
171 : //
172 0 : const DataMoveCallbackDescriptor* raw_data_callback_desc = get_callback_desc();
173 :
174 0 : if (raw_data_callback_desc == nullptr) {
175 : throw(BadConf(ERS_HERE, "No Raw Data Callback descriptor given"));
176 : }
177 :
178 : // Create here the Queue on which all data fragments are forwarded to the fragment aggregator
179 : // and a container for the queues of data request to TP handler and DLH
180 0 : if (fa_output_qdesc == nullptr) {
181 0 : throw(BadConf(ERS_HERE, "No fragment output queue descriptor given"));
182 : }
183 0 : std::vector<const confmodel::Connection*> req_queues;
184 0 : conffwk::ConfigObject frag_queue_obj = obj_fac.create_queue_obj(fa_output_qdesc);
185 :
186 : //
187 : // Scan Detector 2 DAQ connections to extract sender, receiver and stream information
188 : //
189 :
190 0 : std::vector<const confmodel::DaqModule*> modules;
191 :
192 : // Loop over the detector to daq connections and generate one data reader per connection
193 : // and the cooresponding datalink handlers
194 :
195 : // Collect all streams
196 0 : std::vector<const confmodel::DetectorStream*> all_enabled_det_streams;
197 0 : std::map<uint32_t, const appmodel::DataMoveCallbackConf*> callback_confs_by_sid;
198 :
199 : // std::vector<const conffwk::ConfigObject*> d2d_conn_objs;
200 0 : uint16_t conn_idx = 0;
201 :
202 0 : for (auto d2d_conn : get_detector_connections()) {
203 0 : if (helper->is_excluded(d2d_conn)) {
204 0 : TLOG_DEBUG(7) << "Ignoring excluded DetectorToDaqConnection " << d2d_conn->UID();
205 0 : continue;
206 0 : }
207 :
208 0 : TLOG_DEBUG(6) << "Processing DetectorToDaqConnection " << d2d_conn->UID();
209 :
210 : // Are these tests necessary? Schema does not allow 0 cardinality
211 : // for these relationships!! TODO
212 0 : if (d2d_conn->senders().empty()) {
213 0 : throw(BadConf(ERS_HERE, "DetectorToDaqConnection does not contain senders"));
214 : }
215 0 : if (d2d_conn->receiver() == nullptr) {
216 0 : throw(BadConf(ERS_HERE, "DetectorToDaqConnection does not contain a receiver"));
217 : }
218 :
219 : // Find senders and receiver
220 0 : auto det_senders = d2d_conn->senders();
221 0 : auto det_receiver = d2d_conn->receiver();
222 :
223 0 : std::vector<const confmodel::DetectorStream*> enabled_det_streams;
224 : // Loop over streams
225 0 : for (auto stream : d2d_conn->streams()) {
226 :
227 : // Are we sure?
228 0 : if (helper->is_excluded(stream)) {
229 0 : TLOG_DEBUG(7) << "Ignoring excluded DetectorStream " << stream->UID();
230 0 : continue;
231 0 : }
232 :
233 0 : all_enabled_det_streams.push_back(stream);
234 0 : enabled_det_streams.push_back(stream);
235 0 : }
236 :
237 : // Here I want to resolve the type of connection (network, felix, or?)
238 : // Rules of engagement: if the receiver interface is network or felix, the receivers should be castable to the
239 : // counterpart
240 0 : if (reader_class == "DPDKReaderModule") {
241 0 : if (!d2d_conn->castable("NetworkDetectorToDaqConnection")) {
242 0 : throw(BadConf(ERS_HERE,
243 0 : fmt::format("{} requires NetworkDetectorToDaqConnection, found {} of class {}",
244 : reader_class,
245 0 : d2d_conn->UID(),
246 0 : d2d_conn->class_name())));
247 : }
248 0 : if (!det_receiver->cast<appmodel::DPDKReceiver>()) {
249 0 : throw(BadConf(ERS_HERE,
250 0 : fmt::format("{} requires NWDetDataReceiver, found {} of class {}",
251 : reader_class,
252 0 : det_receiver->UID(),
253 0 : det_receiver->class_name())));
254 : }
255 0 : } else if (reader_class == "SocketReaderModule") {
256 0 : if (!d2d_conn->castable("SocketDetectorToDaqConnection")) {
257 0 : throw(BadConf(ERS_HERE,
258 0 : fmt::format("{} requires SocketDetectorToDaqConnection, found {} of class {}",
259 : reader_class,
260 0 : d2d_conn->UID(),
261 0 : d2d_conn->class_name())));
262 : }
263 0 : if (!det_receiver->cast<appmodel::SocketReceiver>()) {
264 0 : throw(BadConf(ERS_HERE,
265 0 : fmt::format("{} requires SocketReceiver, found {} of class {}",
266 : reader_class,
267 0 : det_receiver->UID(),
268 0 : det_receiver->class_name())));
269 : }
270 0 : } else if (reader_class == "FelixReaderModule") {
271 0 : if (!d2d_conn->castable("FelixDetectorToDaqConnection")) {
272 0 : throw(BadConf(ERS_HERE,
273 0 : fmt::format("{} requires FelixDetectorToDaqConnection, found {} of class {}",
274 : reader_class,
275 0 : d2d_conn->UID(),
276 0 : d2d_conn->class_name())));
277 : }
278 0 : if (!det_receiver->cast<appmodel::FelixDataReceiver>()) {
279 0 : throw(BadConf(ERS_HERE,
280 0 : fmt::format("FelixReaderModule requires FelixDataReceiver, found {} of class {}",
281 0 : det_receiver->UID(),
282 0 : det_receiver->class_name())));
283 : }
284 : }
285 : // }
286 :
287 : //-----------------------------------------------------------------
288 : //
289 : // Create DataReaderModule object
290 : //
291 :
292 : //
293 : // Instantiate DataReaderModule of type DPDKReaderModule
294 : //
295 :
296 : // Create the Data reader object
297 :
298 0 : std::string reader_uid(fmt::format("datareader-{}-{}", this->UID(), std::to_string(conn_idx++)));
299 0 : TLOG_DEBUG(6) << fmt::format(
300 0 : "creating OKS configuration object for Data reader class {} with id {}", reader_class, reader_uid);
301 0 : auto reader_obj = obj_fac.create(reader_class, reader_uid);
302 :
303 : // Populate configuration and interfaces (leave output queues for later)
304 0 : reader_obj.set_obj("configuration", &reader_conf->config_object());
305 0 : reader_obj.set_objs("connections", { &d2d_conn->config_object() });
306 :
307 : // Create the raw data callbacks
308 0 : std::vector<const conffwk::ConfigObject*> raw_data_callback_objs;
309 :
310 : // Create data queues
311 0 : for (auto ds : enabled_det_streams) {
312 0 : conffwk::ConfigObject callback_obj = obj_fac.create_callback_sid_obj(raw_data_callback_desc, ds->get_source_id());
313 0 : const auto* callback_conf = obj_fac.get_dal<DataMoveCallbackConf>(callback_obj.UID());
314 0 : raw_data_callback_objs.push_back(&callback_conf->config_object());
315 0 : callback_confs_by_sid[ds->get_source_id()] = callback_conf;
316 0 : }
317 :
318 0 : reader_obj.set_objs("raw_data_callbacks", raw_data_callback_objs);
319 :
320 0 : modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(reader_obj.UID()));
321 0 : }
322 :
323 : //-----------------------------------------------------------------
324 : //
325 : // Prepare the tp handlers and related queues
326 : //
327 0 : std::vector<const confmodel::Connection*> tp_queues;
328 0 : if (get_tp_generation_enabled()) {
329 0 : if (tp_input_qdesc == nullptr) {
330 0 : throw(BadConf(ERS_HERE, "TP generation is enabled but no TP input queue descriptor given"));
331 : }
332 0 : if (tp_net_desc == nullptr) {
333 0 : throw(BadConf(ERS_HERE, "TP generation is enabled but no TPSet network descriptor given"));
334 : }
335 0 : if (ta_net_desc == nullptr) {
336 0 : throw(BadConf(ERS_HERE, "TP generation is enabled but no TriggerActivity network descriptor given"));
337 : }
338 : // Create TP handler object
339 0 : auto tph_conf_obj = tph_conf->config_object();
340 0 : auto tpsrc_ids = get_tp_source_ids();
341 :
342 0 : for (auto sid : tpsrc_ids) {
343 0 : conffwk::ConfigObject tp_queue_obj;
344 0 : conffwk::ConfigObject tpreq_queue_obj;
345 0 : std::string tp_uid("tphandler-" + std::to_string(sid->get_sid()));
346 0 : auto tph_obj = obj_fac.create(tph_class, tp_uid);
347 0 : tph_obj.set_by_val<uint32_t>("source_id", sid->get_sid());
348 0 : tph_obj.set_by_val<uint32_t>("detector_id", 1); // 1 == kDAQ
349 0 : tph_obj.set_by_val<bool>("post_processing_enabled", get_ta_generation_enabled());
350 0 : tph_obj.set_obj("module_configuration", &tph_conf_obj);
351 :
352 : // Create the TPs aggregator queue (from RawData Handlers to TP handlers)
353 0 : tp_queue_obj = obj_fac.create_queue_sid_obj(tp_input_qdesc, sid->get_sid());
354 0 : tp_queue_obj.set_by_val<uint32_t>("recv_timeout_ms", 50);
355 0 : tp_queue_obj.set_by_val<uint32_t>("send_timeout_ms", 1);
356 :
357 0 : tp_queues.push_back(obj_fac.get_dal<confmodel::Connection>(tp_queue_obj.UID()));
358 : // Create tp data requests queue from Fragment Aggregator
359 0 : tpreq_queue_obj = obj_fac.create_queue_sid_obj(dlh_reqinput_qdesc, sid->get_sid());
360 0 : req_queues.push_back(obj_fac.get_dal<confmodel::Connection>(tpreq_queue_obj.UID()));
361 :
362 : // Create the tp(set) publishing service
363 0 : conffwk::ConfigObject tp_net_obj = obj_fac.create_net_obj(tp_net_desc, tp_uid);
364 :
365 : // Create the ta(set) publishing service
366 0 : conffwk::ConfigObject ta_net_obj = obj_fac.create_net_obj(ta_net_desc, tp_uid);
367 :
368 : // Register queues with tp hankder
369 0 : tph_obj.set_objs("inputs", { &tp_queue_obj, &tpreq_queue_obj });
370 0 : tph_obj.set_objs("outputs", { &tp_net_obj, &ta_net_obj, &frag_queue_obj });
371 :
372 0 : modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(tph_obj.UID()));
373 0 : }
374 0 : }
375 :
376 : // Add output queueus of tps
377 0 : std::vector<const conffwk::ConfigObject*> tp_queue_objs;
378 0 : for (auto q : tp_queues) {
379 0 : tp_queue_objs.push_back(&q->config_object());
380 : }
381 :
382 : //-----------------------------------------------------------------
383 : //
384 : // Create datalink handlers
385 : //
386 : // Recover the emulation flag
387 0 : auto emulation_mode = reader_conf->get_emulation_mode();
388 0 : for (auto ds : all_enabled_det_streams) {
389 :
390 0 : uint32_t sid = ds->get_source_id();
391 0 : TLOG_DEBUG(6) << fmt::format(
392 0 : "Processing stream {}, id {}, det id {}", ds->UID(), ds->get_source_id(), ds->get_geo_id()->get_detector_id());
393 0 : std::string uid(fmt::format("DLH-{}", sid));
394 0 : TLOG_DEBUG(6) << fmt::format(
395 0 : "creating OKS configuration object for Data Link Handler class {}, if {}", dlh_class, sid);
396 0 : auto dlh_obj = obj_fac.create(dlh_class, uid);
397 0 : dlh_obj.set_by_val<uint32_t>("source_id", sid);
398 0 : dlh_obj.set_by_val<uint32_t>("detector_id", ds->get_geo_id()->get_detector_id());
399 0 : dlh_obj.set_by_val<bool>("post_processing_enabled", get_tp_generation_enabled());
400 0 : dlh_obj.set_by_val<bool>("emulation_mode", emulation_mode);
401 0 : dlh_obj.set_obj("geo_id", &ds->get_geo_id()->config_object());
402 0 : dlh_obj.set_obj("module_configuration", &dlh_conf->config_object());
403 0 : dlh_obj.set_obj("raw_data_callback", &callback_confs_by_sid[sid]->config_object());
404 :
405 0 : std::vector<const conffwk::ConfigObject*> dlh_ins, dlh_outs;
406 :
407 : // Create request queue
408 0 : conffwk::ConfigObject req_queue_obj = obj_fac.create_queue_sid_obj(dlh_reqinput_qdesc, ds);
409 :
410 : // Add the requessts queue dal pointer to the outputs of the FragmentAggregatorModule
411 0 : req_queues.push_back(obj_fac.get_dal<confmodel::Connection>(req_queue_obj.UID()));
412 0 : dlh_ins.push_back(&req_queue_obj);
413 0 : dlh_outs.push_back(&frag_queue_obj);
414 :
415 : // Time Sync network connection
416 0 : if (dlh_conf->get_generate_timesync()) {
417 : // Add timestamp endpoint
418 0 : conffwk::ConfigObject ts_net_obj = obj_fac.create_net_obj(ts_net_desc, std::to_string(sid));
419 0 : dlh_outs.push_back(&ts_net_obj);
420 0 : }
421 :
422 0 : for (auto tpq : tp_queue_objs) {
423 0 : dlh_outs.push_back(tpq);
424 : }
425 0 : dlh_obj.set_objs("inputs", dlh_ins);
426 0 : dlh_obj.set_objs("outputs", dlh_outs);
427 :
428 0 : modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(dlh_obj.UID()));
429 0 : }
430 :
431 : // Finally create Fragment Aggregator
432 0 : auto aggregator_conf = get_fragment_aggregator();
433 0 : if (aggregator_conf == 0) {
434 : throw(BadConf(ERS_HERE, "No FragmentAggregatorModule configuration given"));
435 : }
436 0 : std::string faUid("fragmentaggregator-" + UID());
437 : // conffwk::ConfigObject frag_aggr;
438 0 : TLOG_DEBUG(7) << "creating OKS configuration object for Fragment Aggregator class ";
439 0 : auto frag_aggr = obj_fac.create("FragmentAggregatorModule", faUid);
440 0 : conffwk::ConfigObject fa_net_obj = obj_fac.create_net_obj(fa_net_desc);
441 :
442 : // Process special Network rules!
443 : // Looking for Fragment rules from DFAppplications in current Session
444 0 : std::vector<conffwk::ConfigObject> fragOutObjs;
445 0 : for (auto [uid, descriptor] : helper->get_netdescriptors("Fragment", "DFApplication")) {
446 0 : std::string dreqNetUid(descriptor->get_uid_base() + uid);
447 0 : auto frag_conn = obj_fac.create("NetworkConnection", dreqNetUid);
448 :
449 0 : frag_conn.set_by_val<std::string>("data_type", descriptor->get_data_type());
450 0 : frag_conn.set_by_val<std::string>("connection_type", descriptor->get_connection_type());
451 : // Override capacity, set to 2x expected number of Fragments
452 0 : frag_conn.set_by_val<int>("capacity", all_enabled_det_streams.size() * 2);
453 :
454 0 : auto serviceObj = descriptor->get_associated_service()->config_object();
455 0 : frag_conn.set_obj("associated_service", &serviceObj);
456 0 : fragOutObjs.push_back(frag_conn);
457 0 : }
458 :
459 : // Add output queueus of data requests and Fragments
460 0 : std::vector<const conffwk::ConfigObject*> fa_output_objs;
461 0 : for (auto& fNet : fragOutObjs) {
462 0 : fa_output_objs.push_back(&fNet);
463 : }
464 :
465 0 : for (auto& q : req_queues) {
466 0 : fa_output_objs.push_back(&q->config_object());
467 : }
468 :
469 0 : frag_aggr.set_obj("configuration", &aggregator_conf->config_object());
470 0 : frag_aggr.set_objs("inputs", { &fa_net_obj, &frag_queue_obj });
471 0 : frag_aggr.set_objs("outputs", fa_output_objs);
472 0 : modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(frag_aggr.UID()));
473 :
474 0 : obj_fac.update_modules(modules);
475 0 : }
476 :
477 : } // namespace appmodel
478 : } // namespace dunedaq
|
