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