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