Line data Source code
1 : /**
2 : * @file FakeDataApplication.cpp
3 : *
4 : * Implementation of FakeDataApplication's dal methods
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 :
12 : #include "ConfigObjectFactory.hpp"
13 :
14 : #include "conffwk/Configuration.hpp"
15 : #include "oks/kernel.hpp"
16 :
17 : #include "confmodel/Connection.hpp"
18 : #include "confmodel/NetworkConnection.hpp"
19 : // #include "confmodel/ReadoutGroup.hpp"
20 : #include "confmodel/ResourceSet.hpp"
21 : #include "confmodel/Service.hpp"
22 :
23 : #include "appmodel/FakeDataApplication.hpp"
24 : #include "appmodel/FakeDataProdConf.hpp"
25 : #include "appmodel/FakeDataProdModule.hpp"
26 : #include "appmodel/FragmentAggregatorModule.hpp"
27 : #include "appmodel/FragmentAggregatorConf.hpp"
28 : #include "appmodel/NetworkConnectionDescriptor.hpp"
29 : #include "appmodel/NetworkConnectionRule.hpp"
30 : #include "appmodel/QueueConnectionRule.hpp"
31 : #include "appmodel/QueueDescriptor.hpp"
32 :
33 : #include "appmodel/appmodelIssues.hpp"
34 :
35 : #include "logging/Logging.hpp"
36 :
37 : #include <string>
38 : #include <vector>
39 :
40 : namespace dunedaq {
41 : namespace appmodel {
42 :
43 : //-----------------------------------------------------------------------------
44 :
45 : std::vector<const confmodel::Resource*>
46 0 : FakeDataApplication::contained_resources() const {
47 0 : return to_resources(get_producers());
48 : }
49 :
50 : void
51 0 : FakeDataApplication::generate_modules(std::shared_ptr<appmodel::ConfigurationHelper> helper) const
52 : {
53 : // oks::OksFile::set_nolock_mode(true);
54 :
55 0 : std::vector<const confmodel::DaqModule*> modules;
56 :
57 0 : ConfigObjectFactory obj_fac(this);
58 :
59 :
60 : // Process the queue rules looking for inputs to our DL/TP handler modules
61 0 : const QueueDescriptor* dlhReqInputQDesc = nullptr;
62 0 : const QueueDescriptor* faOutputQDesc = nullptr;
63 :
64 0 : for (auto rule : get_queue_rules()) {
65 0 : auto destination_class = rule->get_destination_class();
66 0 : auto data_type = rule->get_descriptor()->get_data_type();
67 0 : if (destination_class == "FakeDataProdModule") {
68 0 : if (data_type == "DataRequest") {
69 0 : dlhReqInputQDesc = rule->get_descriptor();
70 : }
71 0 : } else if (destination_class == "FragmentAggregatorModule") {
72 0 : faOutputQDesc = rule->get_descriptor();
73 : }
74 0 : }
75 0 : if (faOutputQDesc == nullptr) {
76 0 : throw(BadConf(ERS_HERE, "No fragment output queue descriptor given"));
77 : }
78 0 : if (dlhReqInputQDesc == nullptr) {
79 0 : throw(BadConf(ERS_HERE, "No DLH request input queue descriptor given"));
80 : }
81 : // Process the network rules looking for the Fragment Aggregator and TP handler data reuest inputs
82 0 : const NetworkConnectionDescriptor* faNetDesc = nullptr;
83 0 : const NetworkConnectionDescriptor* tsNetDesc = nullptr;
84 0 : for (auto rule : get_network_rules()) {
85 0 : auto endpoint_class = rule->get_endpoint_class();
86 0 : if (endpoint_class == "FragmentAggregatorModule") {
87 0 : faNetDesc = rule->get_descriptor();
88 0 : } else if (endpoint_class == "FakeDataProdModule") {
89 0 : tsNetDesc = rule->get_descriptor();
90 : }
91 0 : }
92 0 : if (faNetDesc == nullptr) {
93 0 : throw(BadConf(ERS_HERE, "No Fragment output network descriptor given"));
94 : }
95 0 : if (tsNetDesc == nullptr) {
96 0 : throw(BadConf(ERS_HERE, "No TimeSync output network descriptor given"));
97 : }
98 :
99 : // Create here the Queue on which all data fragments are forwarded to the fragment aggregator
100 : // and a container for the queues of data request to TP handler and DLH
101 :
102 0 : std::vector<const confmodel::Connection*> faOutputQueues;
103 :
104 0 : conffwk::ConfigObject faQueueObj = obj_fac.create_queue_obj(faOutputQDesc, UID());
105 :
106 : // Create a FakeDataProdModule for each stream of this Readout Group
107 0 : for (auto fdpConf : get_producers()) {
108 0 : if (helper->is_disabled(fdpConf)) {
109 0 : TLOG_DEBUG(7) << "Ignoring disabled FakeDataProdConf " << fdpConf->UID();
110 0 : continue;
111 0 : }
112 :
113 0 : auto stream = fdpConf->cast<appmodel::FakeDataProdConf>();
114 0 : if (stream == nullptr) {
115 0 : throw(BadConf(ERS_HERE, "ReadoutGroup contains something other than FakeDataProdConf"));
116 : }
117 :
118 0 : auto id = stream->get_source_id();
119 0 : std::string uid("FakeDataProdModule-" + std::to_string(id));
120 0 : TLOG_DEBUG(7) << "creating OKS configuration object for FakeDataProdModule";
121 0 : conffwk::ConfigObject dlhObj = obj_fac.create("FakeDataProdModule", uid);
122 0 : dlhObj.set_obj("configuration", &stream->config_object());
123 :
124 : // Time Sync network connection
125 0 : auto tsNetObj = obj_fac.create_net_obj(tsNetDesc, std::to_string(id));
126 :
127 0 : dlhObj.set_objs("outputs", { &faQueueObj, &tsNetObj });
128 :
129 0 : auto reqQueueObj = obj_fac.create_queue_sid_obj(dlhReqInputQDesc, id);
130 :
131 : // Add the requessts queue dal pointer to the outputs of the FragmentAggregatorModule
132 0 : faOutputQueues.push_back(obj_fac.get_dal<confmodel::Connection>(
133 0 : dlhReqInputQDesc->get_uid_base() + std::to_string(id)));
134 :
135 0 : dlhObj.set_objs("inputs", { &reqQueueObj });
136 :
137 0 : modules.push_back(obj_fac.get_dal<FakeDataProdModule>(uid));
138 0 : }
139 :
140 : // Finally create Fragment Aggregator
141 0 : auto aggregator_conf = get_fragment_aggregator();
142 0 : if (aggregator_conf == 0) {
143 : throw(BadConf(ERS_HERE, "No FragmentAggregatorModule configuration given"));
144 : }
145 0 : std::string faUid("fragmentaggregator-" + UID());
146 0 : TLOG_DEBUG(7) << "creating OKS configuration object for Fragment Aggregator class ";
147 0 : conffwk::ConfigObject faObj = obj_fac.create("FragmentAggregatorModule", faUid);
148 :
149 : // Add network connection to TRBs
150 0 : conffwk::ConfigObject faNetObj = obj_fac.create_net_obj(faNetDesc, UID());
151 :
152 : // Add output queueus of data requests
153 0 : std::vector<const conffwk::ConfigObject*> qObjs;
154 0 : for (auto q : faOutputQueues) {
155 0 : qObjs.push_back(&q->config_object());
156 : }
157 0 : faObj.set_obj("configuration", &aggregator_conf->config_object());
158 0 : faObj.set_objs("inputs", { &faNetObj, &faQueueObj });
159 0 : faObj.set_objs("outputs", qObjs);
160 :
161 0 : modules.push_back(obj_fac.get_dal<FragmentAggregatorModule>(faUid));
162 :
163 0 : obj_fac.update_modules(modules);
164 0 : }
165 :
166 : } // namespace appmodel
167 : } // namespace dunedaq
|
