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