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