Line data Source code
1 : /**
2 : * @file DFApplication.cpp
3 : *
4 : * Implementation of DFApplication'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/DFApplication.hpp"
12 : #include "ConfigObjectFactory.hpp"
13 : #include "appmodel/ConfigurationHelper.hpp"
14 : #include "appmodel/DataStoreConf.hpp"
15 : #include "appmodel/DataWriterConf.hpp"
16 : #include "appmodel/DataWriterModule.hpp"
17 : #include "appmodel/FilenameParams.hpp"
18 : #include "appmodel/NetworkConnectionDescriptor.hpp"
19 : #include "appmodel/NetworkConnectionRule.hpp"
20 : #include "appmodel/QueueConnectionRule.hpp"
21 : #include "appmodel/QueueDescriptor.hpp"
22 : #include "appmodel/SourceIDConf.hpp"
23 : #include "appmodel/TRBConf.hpp"
24 : #include "appmodel/TRBModule.hpp"
25 : #include "appmodel/appmodelIssues.hpp"
26 :
27 : #include "confmodel/Connection.hpp"
28 : #include "confmodel/NetworkConnection.hpp"
29 :
30 : #include "logging/Logging.hpp"
31 : #include "oks/kernel.hpp"
32 :
33 : #include <fmt/core.h>
34 : #include <string>
35 : #include <vector>
36 :
37 : namespace dunedaq {
38 : namespace appmodel {
39 :
40 : static inline void
41 0 : fill_sourceid_object(const ConfigObjectFactory& obj_fac,
42 : const conffwk::ConfigObject* netConn,
43 : const std::string& uid,
44 : const std::vector<uint32_t>& stream_source_ids,
45 : const std::vector<const SourceIDConf*>& tp_source_ids,
46 : conffwk::ConfigObject& sidNetObj,
47 : std::vector<std::shared_ptr<conffwk::ConfigObject>> sidObjs)
48 : {
49 0 : sidNetObj.set_obj("netconn", netConn);
50 :
51 0 : std::vector<const conffwk::ConfigObject*> source_id_objs;
52 :
53 0 : for (auto& source_id : stream_source_ids) {
54 0 : std::string streamSidUid(uid + "SourceIDConf" + std::to_string(source_id));
55 0 : auto stream_sid_obj = std::make_shared<conffwk::ConfigObject>(obj_fac.create("SourceIDConf", streamSidUid));
56 0 : stream_sid_obj->set_by_val<uint32_t>("sid", source_id);
57 0 : stream_sid_obj->set_by_val<std::string>("subsystem", "Detector_Readout");
58 0 : sidObjs.push_back(stream_sid_obj);
59 0 : source_id_objs.push_back(sidObjs.back().get());
60 0 : }
61 :
62 0 : for (auto tp_sid : tp_source_ids) {
63 0 : sidObjs.push_back(std::make_shared<conffwk::ConfigObject>(tp_sid->config_object()));
64 0 : source_id_objs.push_back(sidObjs.back().get());
65 : }
66 : /*
67 : std::string trgSidUid(roapp->UID() + "TRGSourceIDConf" + std::to_string(roapp->get_tp_source_id()));
68 : auto trig_sid_obj = std::make_shared<conffwk::ConfigObject>(obj_fac.create("SourceIDConf", trgSidUid));
69 : trig_sid_obj->set_by_val<uint32_t>("sid", roapp->get_tp_source_id());
70 : trig_sid_obj->set_by_val<std::string>("subsystem", "Trigger");
71 : source_id_objs.push_back(sidObjs.back().get());
72 : */
73 :
74 0 : sidNetObj.set_objs("source_ids", source_id_objs);
75 0 : }
76 :
77 : inline void
78 0 : fill_replay_sourceid_object(const ConfigObjectFactory& obj_fac,
79 : const std::string& uid,
80 : const std::vector<const SourceIDConf*>& tp_source_ids,
81 : std::vector<conffwk::ConfigObject>* netConn,
82 : std::vector<conffwk::ConfigObject>* sidNetObj,
83 : const NetworkConnectionDescriptor* descriptor,
84 : std::vector<std::shared_ptr<conffwk::ConfigObject>> sidObjs)
85 : {
86 0 : std::vector<const conffwk::ConfigObject*> source_id_objs;
87 :
88 0 : for (auto tp_sid : tp_source_ids) {
89 : // get name extension
90 0 : std::string name = tp_sid->UID();
91 0 : size_t pos = name.find_last_of('-');
92 0 : std::string ext;
93 0 : if (pos != std::string::npos) {
94 0 : ext = name.substr(pos);
95 : }
96 :
97 : // set Network connections
98 0 : std::string dreqNetUid(uid + ext);
99 0 : netConn->emplace_back(obj_fac.create_net_obj(descriptor, dreqNetUid));
100 0 : netConn->back().set_by_val<std::string>("data_type", descriptor->get_data_type());
101 0 : netConn->back().set_by_val<std::string>("connection_type", descriptor->get_connection_type());
102 0 : auto serviceObj = descriptor->get_associated_service()->config_object();
103 0 : netConn->back().set_obj("associated_service", &serviceObj);
104 :
105 : // set SourceID to Network connections
106 0 : std::string sidToNetUid(uid + ext + "-sids");
107 0 : sidNetObj->emplace_back(obj_fac.create("SourceIDToNetworkConnection", sidToNetUid));
108 0 : sidNetObj->back().set_obj("netconn", &netConn->back());
109 :
110 : // set SourceID objs
111 0 : sidObjs.push_back(std::make_shared<conffwk::ConfigObject>(tp_sid->config_object()));
112 0 : sidNetObj->back().set_objs("source_ids", { sidObjs.back().get() });
113 0 : }
114 0 : }
115 :
116 : void
117 0 : DFApplication::generate_modules(std::shared_ptr<appmodel::ConfigurationHelper> helper) const
118 : {
119 :
120 0 : ConfigObjectFactory obj_fac(this);
121 :
122 0 : std::vector<const confmodel::DaqModule*> modules;
123 :
124 : // Containers for module specific config objects for output/input
125 : // Prepare TRB output objects
126 0 : std::vector<const conffwk::ConfigObject*> trbInputObjs;
127 0 : std::vector<const conffwk::ConfigObject*> trbOutputObjs;
128 0 : std::vector<const conffwk::ConfigObject*> trbSidNetObjs;
129 :
130 : // -- First, we process expected Queue and Network connections and create their objects.
131 :
132 : // Process the queue rules looking for the TriggerRecord queue between TRB and DataWriterModule
133 0 : const QueueDescriptor* trQDesc = nullptr;
134 0 : for (auto rule : get_queue_rules()) {
135 0 : auto destination_class = rule->get_destination_class();
136 0 : if (destination_class == "DataWriterModule") {
137 0 : trQDesc = rule->get_descriptor();
138 : }
139 0 : }
140 0 : if (trQDesc == nullptr) { // BadConf if no descriptor between TRB and DataWriterModule
141 0 : throw(BadConf(ERS_HERE, "Could not find queue descriptor rule for TriggerRecords!"));
142 : }
143 : // Create queue connection config object
144 0 : auto trQueueObj = obj_fac.create_queue_obj(trQDesc, UID());
145 :
146 : // Place trigger record queue object into vector of output objs of TRB module
147 0 : trbOutputObjs.push_back(&trQueueObj);
148 :
149 : // Process the network rules looking for the Fragments and TriggerDecision inputs for TRB
150 0 : const NetworkConnectionDescriptor* fragNetDesc = nullptr;
151 0 : const NetworkConnectionDescriptor* trigdecNetDesc = nullptr;
152 0 : const NetworkConnectionDescriptor* tokenNetDesc = nullptr;
153 0 : const NetworkConnectionDescriptor* trmonReqNetDesc = nullptr;
154 0 : const NetworkConnectionDescriptor* trmonTRNetDesc = nullptr;
155 0 : for (auto rule : get_network_rules()) {
156 0 : auto descriptor = rule->get_descriptor();
157 0 : auto data_type = descriptor->get_data_type();
158 0 : if (data_type == "Fragment") {
159 0 : fragNetDesc = rule->get_descriptor();
160 0 : } else if (data_type == "TriggerDecision") {
161 0 : trigdecNetDesc = rule->get_descriptor();
162 0 : } else if (data_type == "TriggerDecisionToken") {
163 0 : tokenNetDesc = rule->get_descriptor();
164 0 : } else if (data_type == "TRMonRequest") {
165 0 : trmonReqNetDesc = rule->get_descriptor();
166 0 : } else if (data_type == "TriggerRecord") {
167 0 : trmonTRNetDesc = rule->get_descriptor();
168 : }
169 0 : }
170 0 : if (fragNetDesc == nullptr) { // BadConf if no descriptor for Fragments into TRB
171 0 : throw(BadConf(ERS_HERE, "Could not find network descriptor rule for input Fragments!"));
172 : }
173 0 : if (trigdecNetDesc == nullptr) { // BadCond if no descriptor for TriggerDecisions into TRB
174 0 : throw(BadConf(ERS_HERE, "Could not find network descriptor rule for input TriggerDecisions!"));
175 : }
176 0 : if (tokenNetDesc == nullptr) { // BadCond if no descriptor for Tokens out of DataWriterModule
177 0 : throw(BadConf(ERS_HERE, "Could not find network descriptor rule for output TriggerDecisionTokens!"));
178 : }
179 0 : if (get_source_id() == nullptr) {
180 0 : throw(BadConf(ERS_HERE, "Could not retrieve SourceIDConf"));
181 : }
182 : // Create network connection config object
183 0 : auto fragNetObj = obj_fac.create_net_obj(fragNetDesc, UID());
184 0 : auto trigdecNetObj = obj_fac.create_net_obj(trigdecNetDesc, UID());
185 0 : auto tokenNetObj = obj_fac.create_net_obj(tokenNetDesc, "");
186 0 : conffwk::ConfigObject trmonReqNetObj;
187 0 : conffwk::ConfigObject trmonTRNetObj;
188 0 : if (trmonReqNetDesc != nullptr) {
189 0 : trmonReqNetObj = obj_fac.create_net_obj(trmonReqNetDesc, UID());
190 : }
191 0 : if (trmonTRNetDesc != nullptr) {
192 0 : trmonTRNetObj = obj_fac.create_net_obj(trmonTRNetDesc, "");
193 : }
194 :
195 : // Process special Network rules!
196 : // Looking for DataRequest rules from ReadoutAppplications in current Session
197 0 : std::vector<conffwk::ConfigObject> dreqNetObjs;
198 0 : std::vector<conffwk::ConfigObject> sidNetObjs;
199 0 : std::vector<std::shared_ptr<conffwk::ConfigObject>> sidObjs;
200 0 : std::set<std::string> processed_apps;
201 0 : for (auto uid : helper->get_app_uids("DFApplication")) {
202 0 : processed_apps.insert(uid);
203 0 : }
204 :
205 0 : auto stream_src_ids = helper->get_stream_source_ids();
206 0 : auto tp_src_ids = helper->get_tp_source_ids();
207 0 : for (auto [uid, descriptor] : helper->get_netdescriptors("DataRequest", "ReadoutApplication")) {
208 0 : dreqNetObjs.emplace_back(obj_fac.create_net_obj(descriptor, uid));
209 :
210 0 : std::string sidToNetUid(descriptor->get_uid_base() + uid + "-sids");
211 0 : sidNetObjs.emplace_back(obj_fac.create("SourceIDToNetworkConnection", sidToNetUid));
212 :
213 0 : fill_sourceid_object(
214 0 : obj_fac, &dreqNetObjs.back(), uid, stream_src_ids.at(uid), tp_src_ids.at(uid), sidNetObjs.back(), sidObjs);
215 0 : processed_apps.insert(uid);
216 0 : }
217 :
218 0 : for (auto [uid, descriptor] : helper->get_netdescriptors("DataRequest", "TPReplayApplication")) {
219 0 : fill_replay_sourceid_object(obj_fac, uid, tp_src_ids.at(uid), &dreqNetObjs, &sidNetObjs, descriptor, sidObjs);
220 0 : processed_apps.insert(uid);
221 0 : }
222 :
223 0 : for (auto [uid, descriptor] : helper->get_netdescriptors("DataRequest", "FakeDataApplication")) {
224 0 : dreqNetObjs.emplace_back(obj_fac.create_net_obj(descriptor, uid));
225 :
226 0 : std::string sidToNetUid(descriptor->get_uid_base() + uid + "-sids");
227 0 : sidNetObjs.emplace_back(obj_fac.create("SourceIDToNetworkConnection", sidToNetUid));
228 :
229 0 : fill_sourceid_object(obj_fac,
230 0 : &dreqNetObjs.back(),
231 : uid,
232 0 : stream_src_ids.at(uid),
233 : {}, // No tp src_ids for FakeDataApplication
234 0 : sidNetObjs.back(),
235 : sidObjs);
236 0 : processed_apps.insert(uid);
237 0 : }
238 :
239 : // now we treat the CTB which has 2 connections related to source IDs
240 0 : const auto ctb_type = "CTBApplication";
241 0 : for (auto [uid, descriptor] : helper->get_netdescriptors("DataRequest", ctb_type)) {
242 :
243 0 : if (processed_apps.contains(uid)) {
244 0 : continue;
245 : }
246 :
247 0 : for (const auto& [uid, rel_sources] : helper->get_all_app_source_ids(ctb_type)) {
248 0 : for (auto [rel, id] : rel_sources) {
249 0 : std::string local_uid = uid;
250 0 : local_uid += rel.find("LLT") != std::string::npos ? "_LLT" : "_HLT";
251 :
252 0 : dreqNetObjs.emplace_back(obj_fac.create_net_obj(descriptor, local_uid));
253 0 : sidObjs.push_back(std::make_shared<conffwk::ConfigObject>(id->config_object()));
254 :
255 0 : std::string sidToNetUid(descriptor->get_uid_base() + local_uid);
256 0 : sidNetObjs.emplace_back(obj_fac.create("SourceIDToNetworkConnection", sidToNetUid));
257 0 : sidNetObjs.back().set_objs("source_ids", { sidObjs.back().get() });
258 0 : sidNetObjs.back().set_obj("netconn", &dreqNetObjs.back());
259 :
260 0 : } // loop on relational sources
261 :
262 0 : processed_apps.insert(uid);
263 0 : } // loop over CTB apps
264 0 : } // loop over descriptors for the CTB apps
265 :
266 0 : auto app_sources = helper->get_app_source_ids();
267 : // Now look at all Smart apps that are not Readout, FakeData or DF
268 0 : for (auto [uid, descriptor] : helper->get_netdescriptors("DataRequest")) {
269 :
270 0 : if (processed_apps.contains(uid)) {
271 0 : continue;
272 : }
273 0 : if (app_sources.contains(uid)) {
274 0 : dreqNetObjs.emplace_back(obj_fac.create_net_obj(descriptor, uid));
275 :
276 0 : sidObjs.push_back(std::make_shared<conffwk::ConfigObject>(app_sources.at(uid)->config_object()));
277 :
278 0 : std::string sidToNetUid(descriptor->get_uid_base() + uid + "-sids");
279 0 : sidNetObjs.emplace_back(obj_fac.create("SourceIDToNetworkConnection", sidToNetUid));
280 0 : sidNetObjs.back().set_objs("source_ids", { sidObjs.back().get() });
281 0 : sidNetObjs.back().set_obj("netconn", &dreqNetObjs.back());
282 :
283 0 : processed_apps.insert(uid);
284 0 : }
285 0 : }
286 :
287 : // Get pointers to objects here, after vector has been filled so they don't move on us
288 0 : for (auto& obj : dreqNetObjs) {
289 0 : trbOutputObjs.push_back(&obj);
290 : }
291 0 : for (auto& obj : sidNetObjs) {
292 0 : trbSidNetObjs.push_back(&obj);
293 : }
294 :
295 : // -- Second, we create the Module objects and assign their configs, with the precreated
296 : // -- connection config objects above.
297 :
298 : // Get TRB Config Object
299 0 : auto trbConf = get_trb();
300 0 : if (trbConf == nullptr) {
301 : throw(BadConf(ERS_HERE, "No DataWriterModule or TRB configuration given"));
302 : }
303 0 : auto trbConfObj = trbConf->config_object();
304 0 : trbConfObj.set_by_val<uint32_t>("source_id", get_source_id()->get_sid());
305 0 : trbInputObjs = { &trigdecNetObj, &fragNetObj };
306 0 : if (trmonReqNetDesc != nullptr) {
307 0 : trbInputObjs.push_back(&trmonReqNetObj);
308 : }
309 0 : if (trmonTRNetDesc != nullptr) {
310 0 : trbOutputObjs.push_back(&trmonTRNetObj);
311 : }
312 : // Prepare TRB Module Object and assign its Config Object.
313 0 : std::string trbUid(UID() + "-trb");
314 0 : conffwk::ConfigObject trbObj = obj_fac.create("TRBModule", trbUid);
315 0 : trbObj.set_obj("configuration", &trbConfObj);
316 0 : trbObj.set_objs("inputs", trbInputObjs);
317 0 : trbObj.set_objs("outputs", trbOutputObjs);
318 0 : trbObj.set_obj("trigger_record_output", &trQueueObj);
319 0 : trbObj.set_objs("request_connections", trbSidNetObjs);
320 : // Push TRB Module Object from confdb
321 0 : modules.push_back(obj_fac.get_dal<TRBModule>(trbUid));
322 :
323 : // Get DataWriterModule Config Object (only one for now, maybe more later?)
324 0 : auto dwrConfs = get_data_writers();
325 0 : if (dwrConfs.size() == 0) {
326 0 : throw(BadConf(ERS_HERE, "No DataWriterModule or TRB configuration given"));
327 : }
328 0 : uint dw_idx = 0;
329 0 : for (auto dwrConf : dwrConfs) {
330 : // auto fnParamsObj = dwrConf->get_data_store_params()->get_filename_params()->config_object();
331 : // fnParamsObj.set_by_val<std::string>("writer_identifier", fmt::format("{}_datawriter-{}", UID(), dw_idx));
332 0 : auto dwrConfObj = dwrConf->config_object();
333 :
334 : // Prepare DataWriterModule Module Object and assign its Config Object.
335 0 : std::string dwrUid(fmt::format("{}-dw-{}", UID(), dw_idx));
336 0 : conffwk::ConfigObject dwrObj = obj_fac.create("DataWriterModule", dwrUid);
337 0 : dwrObj.set_by_val("writer_identifier", fmt::format("{}_dw_{}", UID(), dw_idx));
338 0 : dwrObj.set_obj("configuration", &dwrConfObj);
339 0 : dwrObj.set_objs("inputs", { &trQueueObj });
340 0 : dwrObj.set_objs("outputs", { &tokenNetObj });
341 : // Push DataWriterModule Module Object from confdb
342 0 : modules.push_back(obj_fac.get_dal<DataWriterModule>(dwrUid));
343 0 : ++dw_idx;
344 0 : }
345 :
346 0 : obj_fac.update_modules(modules);
347 0 : }
348 :
349 : } // namespace appmodel
350 : } // namespace dunedaq
|
