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