Line data Source code
1 : /**
2 : * @file CRTGrenobleFrameBuilderModule.cpp
3 : *
4 : * Reads data from the HW then puts it in a queue
5 : *
6 : * This is part of the DUNE DAQ Software Suite, copyright 2020.
7 : * Licensing/copyright details are in the COPYING file that you should have
8 : * received with this code.
9 : */
10 :
11 : #include "CRTGrenobleFrameBuilderModule.hpp"
12 :
13 : #include "crtmodules/opmon/CRTGrenobleFrameBuilderModule.pb.h"
14 :
15 : #include "datahandlinglibs/utils/RateLimiter.hpp"
16 :
17 : #include "appmodel/DetectorFrameBuilderModule.hpp"
18 : #include "appmodel/SocketDetectorToDaqConnection.hpp"
19 : #include "appmodel/NWDetDataSender.hpp"
20 :
21 : #include "confmodel/QueueWithSourceId.hpp"
22 : #include "confmodel/DetectorStream.hpp"
23 : #include "confmodel/GeoId.hpp"
24 :
25 : #include "datahandlinglibs/DataHandlingIssues.hpp"
26 :
27 : #include "detdataformats/DetID.hpp"
28 :
29 : #include <utility>
30 : #include <memory>
31 : #include <string>
32 :
33 0 : DUNE_DAQ_TYPESTRING(dunedaq::fddetdataformats::CRTGrenobleFrame, "CRTGrenobleFrame")
34 :
35 : namespace dunedaq::crtmodules {
36 :
37 : /**
38 : * @brief Maximum packet sequence ID before reset
39 : */
40 : constexpr uint64_t max_seq_id = 4095; // NOLINT(build/unsigned)
41 :
42 : /**
43 : * @brief Fake packet detector ID
44 : */
45 : constexpr uint8_t fake_det_id = static_cast<uint8_t>(detdataformats::DetID::Subdetector::kVD_GrenobleCRT); // NOLINT(build/unsigned)
46 :
47 : /**
48 : * @brief Fake packet block length
49 : */
50 : constexpr uint64_t fake_block_length = 0x382; // NOLINT(build/unsigned)
51 :
52 : /**
53 : * @brief Calculate the next fake sequence ID for a packet
54 : * @param seq_id Fake packet sequence ID
55 : */
56 : void
57 0 : fake_sequence_id(uint64_t& seq_id) // NOLINT(build/unsigned)
58 : {
59 0 : seq_id = (seq_id == max_seq_id ? 0 : seq_id+1);
60 0 : }
61 :
62 : /**
63 : * @brief Calculate the next fake timestamp for a packet
64 : * @param timestamp Fake packet timestamp
65 : */
66 : void
67 0 : fake_timestamp(uint64_t& timestamp) // NOLINT(build/unsigned)
68 : {
69 0 : auto time_now = std::chrono::steady_clock::now().time_since_epoch();
70 0 : uint64_t current_time = // NOLINT (build/unsigned)
71 0 : std::chrono::duration_cast<std::chrono::nanoseconds>(time_now).count();
72 0 : timestamp = current_time / 16; // 625/10000 (same as 625*us/10)
73 0 : }
74 :
75 : /**
76 : * @brief Fake ADC of the given packet
77 : * @param frame Fake packet
78 : */
79 : void
80 0 : fake_adc(fddetdataformats::CRTGrenobleFrame& frame)
81 : {
82 0 : for (int channel = 0; channel < fddetdataformats::CRTGrenobleFrame::s_num_channels; ++channel) {
83 0 : frame.set_adc(channel, 0);
84 : }
85 0 : }
86 :
87 : /**
88 : * @brief Create a fake packet
89 : * @param frame Fake packet
90 : * @param seq_id Fake packet sequence ID
91 : * @param timestamp Fake packet timestamp
92 : * @param stream_id Fake packet stream ID
93 : */
94 : void
95 0 : fake_data(fddetdataformats::CRTGrenobleFrame& frame, uint64_t& seq_id, uint64_t& timestamp, uint32_t stream_id) // NOLINT(build/unsigned)
96 : {
97 0 : frame.daq_header.det_id = fake_det_id & 0x3f; //6 bits for det id
98 0 : frame.daq_header.crate_id = 1;
99 0 : frame.daq_header.slot_id = 1;
100 0 : frame.daq_header.stream_id = stream_id;
101 0 : fake_sequence_id(seq_id);
102 0 : frame.daq_header.seq_id = seq_id;
103 0 : frame.daq_header.block_length = fake_block_length;
104 0 : fake_timestamp(timestamp);
105 0 : frame.daq_header.timestamp = timestamp;
106 0 : fake_adc(frame);
107 0 : }
108 :
109 0 : CRTGrenobleFrameBuilderModule::CRTGrenobleFrameBuilderModule(const std::string& name)
110 0 : : DAQModule(name)
111 : {
112 0 : register_command("conf", &CRTGrenobleFrameBuilderModule::do_conf);
113 0 : register_command("start", &CRTGrenobleFrameBuilderModule::do_start);
114 0 : register_command("stop_trigger_sources", &CRTGrenobleFrameBuilderModule::do_stop);
115 0 : register_command("scrap", &CRTGrenobleFrameBuilderModule::do_scrap);
116 0 : }
117 :
118 : void
119 0 : CRTGrenobleFrameBuilderModule::init(const std::shared_ptr<appfwk::ConfigurationManager> mcfg)
120 : {
121 0 : auto* mdal = mcfg->get_dal<appmodel::DetectorFrameBuilderModule>(get_name());
122 :
123 0 : auto* d2d_conn = mdal->get_connection();
124 0 : auto* socket_d2d_conn = d2d_conn->cast<appmodel::SocketDetectorToDaqConnection>();
125 0 : if (socket_d2d_conn == nullptr) {
126 0 : auto err = datahandlinglibs::InitializationError(ERS_HERE, "Connection is not of type SocketDetectorToDaqConnection.");
127 0 : ers::fatal(err);
128 0 : throw err;
129 0 : }
130 :
131 0 : auto* nw_sender = socket_d2d_conn->get_net_senders()[0]; // there's only 1 sender
132 :
133 0 : for (auto det_stream : nw_sender->get_streams()) {
134 0 : if (det_stream->is_disabled(*(mcfg->get_session()))) {
135 0 : continue;
136 : }
137 :
138 0 : m_fake_stream_ids.push_back(det_stream->get_geo_id()->get_stream_id());
139 :
140 0 : m_producer_threads.emplace_back(std::make_unique<utilities::ReusableThread>());
141 : }
142 :
143 0 : auto* con = mdal->get_outputs()[0]; // there's only 1 output
144 0 : auto* queue = con->cast<confmodel::Queue>();
145 0 : if (queue == nullptr) {
146 0 : auto err = datahandlinglibs::InitializationError(ERS_HERE, "Output is not of type Queue.");
147 0 : ers::fatal(err);
148 0 : throw err;
149 0 : }
150 :
151 0 : auto connection_name = queue->UID();
152 0 : m_sender = get_iom_sender<fddetdataformats::CRTGrenobleFrame>(connection_name);
153 0 : }
154 :
155 : void
156 0 : CRTGrenobleFrameBuilderModule::do_conf(const CommandData_t& /*obj*/)
157 : {
158 : // Configure HW interface?
159 0 : if (!m_run_marker.load()) {
160 0 : set_running(true);
161 : } else {
162 0 : TLOG_DEBUG(5) << "Already running!";
163 : }
164 0 : }
165 :
166 : void
167 0 : CRTGrenobleFrameBuilderModule::do_scrap(const CommandData_t& /*obj*/)
168 : {
169 0 : if (m_run_marker.load()) {
170 0 : TLOG() << "Raising stop through variables!";
171 0 : set_running(false);
172 0 : for (const auto& producer : m_producer_threads) {
173 0 : while (!producer->get_readiness()) {
174 0 : std::this_thread::sleep_for(std::chrono::milliseconds(10));
175 : }
176 : }
177 : } else {
178 0 : TLOG_DEBUG(5) << "Already stopped!";
179 : }
180 0 : }
181 :
182 : void
183 0 : CRTGrenobleFrameBuilderModule::do_start(const CommandData_t& /*startobj*/)
184 : {
185 0 : m_packet_count = 0;
186 :
187 0 : m_t0 = std::chrono::steady_clock::now();
188 :
189 0 : enable_flow();
190 :
191 0 : uint32_t i = 0; // NOLINT(build/unsigned)
192 0 : for (auto& producer : m_producer_threads) {
193 0 : producer->set_work(&CRTGrenobleFrameBuilderModule::run_produce, this, m_fake_stream_ids[i++]);
194 : }
195 0 : }
196 :
197 : void
198 0 : CRTGrenobleFrameBuilderModule::do_stop(const CommandData_t& /*stopobj*/)
199 : {
200 0 : disable_flow();
201 0 : }
202 :
203 : void
204 0 : CRTGrenobleFrameBuilderModule::generate_opmon_data()
205 : {
206 0 : opmon::CRTGrenobleFrameBuilderInfo i;
207 :
208 0 : auto now = std::chrono::steady_clock::now();
209 0 : int new_packets = m_packet_count.exchange(0);
210 0 : double seconds = std::chrono::duration_cast<std::chrono::microseconds>(now - m_t0).count() / 1000000.;
211 0 : m_t0 = now;
212 :
213 0 : i.set_packet_rate_khz(new_packets / seconds / 1000.);
214 :
215 0 : publish(std::move(i));
216 0 : }
217 :
218 : void
219 0 : CRTGrenobleFrameBuilderModule::run_produce(uint32_t fake_stream_id) // NOLINT(build/unsigned)
220 : {
221 0 : TLOG() << "Producer thread started..."; // TODO (DTE): Debug log instead
222 :
223 0 : fddetdataformats::CRTGrenobleFrame frame;
224 0 : uint64_t seq_id = 0; // NOLINT(build/unsigned)
225 0 : uint64_t timestamp = 0; // NOLINT(build/unsigned)
226 :
227 0 : datahandlinglibs::RateLimiter rate_limiter(m_configured_packet_rate_khz);
228 :
229 0 : while (m_run_marker.load()) {
230 : // Create a fake packet for stream
231 0 : fake_data(frame, seq_id, timestamp, fake_stream_id); // TODO: To be filled by the CRT experts
232 :
233 0 : if (m_enable_flow.load()) [[likely]] {
234 0 : m_sender->try_send(std::move(frame), iomanager::Sender::s_no_block);
235 0 : ++m_packet_count;
236 : }
237 :
238 0 : rate_limiter.limit();
239 : }
240 :
241 0 : TLOG() << "Producer thread joins... "; // TODO (DTE): Debug log instead
242 0 : }
243 :
244 : void
245 0 : CRTGrenobleFrameBuilderModule::set_running(bool should_run)
246 : {
247 0 : bool was_running = m_run_marker.exchange(should_run);
248 0 : TLOG_DEBUG(5) << "Active state was toggled from " << was_running << " to " << should_run;
249 0 : }
250 :
251 : } // namespace dunedaq::crtmodules
252 :
253 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::crtmodules::CRTGrenobleFrameBuilderModule)
|
