Line data Source code
1 : /**
2 : * @file CTBModule.cpp CTBModule class
3 : * implementation
4 : *
5 : * This is part of the DUNE DAQ Software Suite, copyright 2020.
6 : * Licensing/copyright details are in the COPYING file that you should have
7 : * received with this code.
8 : */
9 :
10 : #include "confmodel/GeoId.hpp"
11 :
12 : #include "appmodel/CTBCalibrationStream.hpp"
13 : #include "appmodel/CTBConf.hpp"
14 : #include "appmodel/CTBCountLLT.hpp"
15 : #include "appmodel/CTBHLT.hpp"
16 : #include "appmodel/CTBLLT.hpp"
17 : #include "appmodel/CTBMisc.hpp"
18 : #include "appmodel/CTBRandomTrigger.hpp"
19 : #include "appmodel/CTBReceiverSocket.hpp"
20 : #include "appmodel/CTBSockets.hpp"
21 : #include "appmodel/CTBoardConf.hpp"
22 :
23 : #include "CTBModule.hpp"
24 : #include "CTBModuleIssues.hpp"
25 :
26 : #include "iomanager/IOManager.hpp"
27 : #include "logging/Logging.hpp"
28 :
29 : #include <chrono>
30 : #include <map>
31 : #include <memory>
32 : #include <queue>
33 : #include <string>
34 : #include <thread>
35 : #include <utility>
36 : #include <vector>
37 :
38 : /**
39 : * @brief Name used by TRACE TLOG calls from this source file
40 : */
41 : #define TRACE_NAME "CTBModule" // NOLINT
42 : enum
43 : {
44 : TLVL_ENTER_EXIT_METHODS = 10,
45 : TLVL_CTB_MODULE = 15
46 : };
47 :
48 : constexpr uint16_t CTB_HSI_FRAME_VERSION = 0x1; // NOLINT
49 :
50 : using namespace dunedaq;
51 : using namespace ctbmodules;
52 :
53 0 : CTBModule::CTBModule(const std::string& name)
54 : : hsilibs::HSIEventSender(name)
55 0 : , m_is_running(false)
56 0 : , m_stop_requested(false)
57 0 : , m_is_configured(false)
58 0 : , m_error_state(false)
59 0 : , m_total_hlt_counter(0)
60 0 : , m_ts_word_counter(0)
61 0 : , m_hlt_trigger_counter()
62 0 : , m_llt_trigger_counter()
63 0 : , m_control_ios()
64 0 : , m_receiver_ios()
65 0 : , m_control_socket(m_control_ios)
66 0 : , m_receiver_socket(m_receiver_ios)
67 0 : , m_thread_(std::bind(&CTBModule::do_hsi_work, this, std::placeholders::_1))
68 0 : , m_has_calibration_stream(false)
69 0 : , m_run_HLT_counter(0)
70 0 : , m_run_LLT_counter(0)
71 0 : , m_run_channel_status_counter(0)
72 0 : , m_num_control_messages_sent(0)
73 0 : , m_num_control_responses_received(0)
74 0 : , m_last_readout_hlt_timestamp(0)
75 : {
76 0 : register_command("conf", &CTBModule::do_configure);
77 0 : register_command("start", &CTBModule::do_start);
78 0 : register_command("stop", &CTBModule::do_stop);
79 0 : }
80 :
81 0 : CTBModule::~CTBModule()
82 : {
83 : // check if running. and in case stop the run
84 0 : if (m_is_running) {
85 0 : const CommandData_t stopobj;
86 0 : do_stop(stopobj);
87 0 : }
88 0 : m_control_socket.close();
89 0 : }
90 :
91 : void
92 0 : CTBModule::init(std::shared_ptr<appfwk::ConfigurationManager> cfgMgr)
93 : {
94 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering init() method";
95 :
96 0 : HSIEventSender::init(cfgMgr);
97 :
98 0 : m_cfg = cfgMgr;
99 :
100 0 : auto mdal = cfgMgr->get_dal<appmodel::CTBModule>(get_name());
101 :
102 0 : if (!mdal) {
103 0 : throw ctbmodules::CTBConfigFailure(ERS_HERE, "Missing Module configuration for " + get_name());
104 : }
105 :
106 0 : m_module = mdal;
107 :
108 : // setting up connections
109 0 : auto iom = iomanager::IOManager::get();
110 :
111 0 : using hsi_frame_t = dunedaq::hsilibs::HSI_FRAME_STRUCT;
112 0 : for (auto con : m_module->get_outputs()) {
113 0 : if (con->get_data_type() == datatype_to_string<hsi_frame_t>()) {
114 0 : if (con->UID().find("HLT") != std::string::npos || con->UID().find("hlt") != std::string::npos) {
115 0 : m_hlt_hsi_data_sender = iom->get_sender<hsi_frame_t>(con->UID());
116 : }
117 0 : if (con->UID().find("LLT") != std::string::npos || con->UID().find("llt") != std::string::npos) {
118 0 : m_llt_hsi_data_sender = iom->get_sender<hsi_frame_t>(con->UID());
119 : }
120 : } // if data type is HSI Frame
121 : } // loop over outputs
122 :
123 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting init() method";
124 0 : }
125 :
126 : void
127 0 : CTBModule::do_configure(const CommandData_t&)
128 : {
129 :
130 0 : TLOG_DEBUG(0) << get_name() << ": Configuring CTB";
131 :
132 0 : auto conf = m_module->get_configuration();
133 :
134 0 : m_receiver_port = m_module->get_board()->get_sockets()->get_receiver()->get_port();
135 0 : m_timeout = std::chrono::milliseconds(conf->get_connection_timeout_ms());
136 :
137 0 : auto hostname = conf->get_hostname();
138 0 : TLOG() << get_name() << ": Board receiver network location " << hostname << ':' << m_receiver_port << std::endl;
139 :
140 : // Initialise monitoring variables
141 0 : m_num_control_messages_sent = 0;
142 0 : m_num_control_responses_received = 0;
143 0 : m_ts_word_counter = 0;
144 :
145 0 : std::map<std::string, size_t> id_to_idx;
146 0 : for (size_t i = 0; i < m_hlt_range; i++)
147 0 : id_to_idx["HLT_" + std::to_string(i)] = i;
148 0 : for (size_t i = 0; i < m_llt_range; i++)
149 0 : id_to_idx["LLT_" + std::to_string(i)] = i;
150 :
151 : // configuring the board
152 0 : auto board = m_module->get_board();
153 0 : auto geo_id = board->get_geo_id();
154 0 : m_det = geo_id->get_detector_id();
155 0 : m_crate = geo_id->get_crate_id();
156 0 : m_slot = geo_id->get_slot_id();
157 :
158 0 : const auto& misc = board->get_misc();
159 0 : auto session = m_cfg->get_session();
160 : // HLTs
161 : // 0th HLT is random trigger that's not in HLT array
162 0 : if (!misc->get_randomtrigger_1()->is_excluded(*session))
163 0 : m_hlt_trigger_counter[0] = 0;
164 :
165 0 : auto hlts = board->get_HLTs();
166 0 : for (const auto& hlt : hlts) {
167 0 : if (!hlt->is_excluded(*session))
168 0 : m_hlt_trigger_counter[id_to_idx[hlt->UID()]] = 0;
169 : }
170 :
171 : // LLTs: Beam and CRT
172 : // 0th LLT is random trigger that's not in HLT array
173 0 : if (!misc->get_randomtrigger_2()->is_excluded(*session))
174 0 : m_llt_trigger_counter[0] = 0;
175 :
176 0 : auto beam_llts = board->get_beam_LLTs();
177 0 : for (const auto& llt : beam_llts) {
178 0 : if (!llt->is_excluded(*session))
179 0 : m_llt_trigger_counter[id_to_idx[llt->UID()]] = 0;
180 : }
181 :
182 0 : auto crt_llts = board->get_CRT_LLTs();
183 0 : for (const auto& llt : crt_llts) {
184 0 : if (!llt->is_excluded(*session))
185 0 : m_llt_trigger_counter[id_to_idx[llt->UID()]] = 0;
186 : }
187 :
188 : // network connection to ctb hardware control
189 0 : boost::asio::ip::tcp::resolver resolver(m_control_ios);
190 0 : boost::asio::ip::tcp::resolver::query query(hostname,
191 0 : std::to_string(conf->get_control_connection_port())); //"np04-ctb-1", 8991
192 0 : boost::asio::ip::tcp::resolver::iterator iter = resolver.resolve(query);
193 :
194 0 : m_endpoint = iter->endpoint();
195 0 : m_control_socket.connect(m_endpoint);
196 :
197 : // if necessary, set the calibration stream
198 0 : auto stream_conf = conf->get_calibration_stream();
199 0 : if (stream_conf) {
200 0 : m_has_calibration_stream = true;
201 0 : m_calibration_dir = stream_conf->get_directory();
202 0 : m_calibration_file_interval = std::chrono::duration_cast<decltype(m_calibration_file_interval)>(
203 0 : std::chrono::seconds(stream_conf->get_update_period_s()));
204 0 : ;
205 : }
206 :
207 : // at this point we have to find the hostname to tell the board what to get
208 0 : boost::asio::ip::tcp::resolver::query query_for_local(boost::asio::ip::host_name(), "");
209 0 : iter = resolver.resolve(query_for_local);
210 :
211 : // create the json string
212 0 : auto json_conf = m_module->get_board()->get_ctb_json(*session, iter->endpoint().address().to_string());
213 :
214 0 : auto json_dump = json_conf.dump();
215 :
216 0 : TLOG() << "Sending configuration: " << json_dump;
217 :
218 0 : send_config(json_dump);
219 0 : }
220 :
221 : void
222 0 : CTBModule::do_start(const CommandData_t& startobj)
223 : {
224 :
225 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_start() method";
226 :
227 : // Set this to false early so it doesn't interfere with the start
228 0 : m_stop_requested.store(false);
229 :
230 0 : m_run_number.store(startobj.at("run").get<daqdataformats::run_number_t>());
231 :
232 0 : m_total_hlt_counter.store(0);
233 :
234 0 : TLOG_DEBUG(0) << get_name() << ": Sending start of run command";
235 0 : m_thread_.start_working_thread();
236 :
237 0 : if (m_has_calibration_stream) {
238 0 : std::stringstream run;
239 0 : run << "run" << m_run_number.load();
240 0 : SetCalibrationStream(run.str());
241 0 : }
242 :
243 0 : if (send_message("{\"command\":\"StartRun\"}")) {
244 0 : m_is_running.store(true);
245 0 : TLOG_DEBUG(1) << get_name() << ": successfully started";
246 : } else {
247 0 : throw CTBCommunicationError(ERS_HERE, "Unable to start CTB");
248 : }
249 :
250 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_start() method";
251 0 : }
252 :
253 : void
254 0 : CTBModule::do_stop(const CommandData_t& /*stopobj*/)
255 : {
256 :
257 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_stop() method";
258 :
259 0 : TLOG_DEBUG(0) << get_name() << ": Sending stop run command" << std::endl;
260 :
261 : // Give the do_work thread a chance to stop before stopping the CTB,
262 : // otherwise we end up reading from an empty buffer
263 0 : m_stop_requested.store(true);
264 0 : std::this_thread::sleep_for(std::chrono::milliseconds(2));
265 :
266 0 : if (send_message("{\"command\":\"StopRun\"}")) {
267 0 : TLOG_DEBUG(1) << get_name() << ": successfully stopped";
268 0 : m_is_running.store(false);
269 : } else {
270 0 : throw CTBCommunicationError(ERS_HERE, "Unable to stop CTB");
271 : }
272 0 : m_thread_.stop_working_thread();
273 :
274 0 : m_run_HLT_counter = 0;
275 0 : m_run_LLT_counter = 0;
276 0 : m_run_channel_status_counter = 0;
277 :
278 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_stop() method";
279 0 : }
280 :
281 : void
282 0 : CTBModule::do_hsi_work(std::atomic<bool>& running_flag)
283 : {
284 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_work() method";
285 :
286 0 : std::size_t n_bytes = 0;
287 0 : std::size_t n_words = 0;
288 :
289 0 : const size_t header_size = sizeof(content::tcp_header_t);
290 0 : const size_t word_size = content::word::word_t::size_bytes;
291 :
292 0 : TLOG_DEBUG(TLVL_CTB_MODULE) << get_name() << ": Header size: " << header_size << std::endl
293 0 : << "Word size: " << word_size << std::endl;
294 :
295 : // connect to socket
296 0 : boost::asio::ip::tcp::acceptor acceptor(m_receiver_ios,
297 0 : boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), m_receiver_port));
298 0 : TLOG_DEBUG(0) << get_name() << ": Waiting for an incoming connection on port " << m_receiver_port << std::endl;
299 :
300 0 : std::future<void> accepting = async(std::launch::async, [&] { acceptor.accept(m_receiver_socket); });
301 :
302 0 : while (running_flag.load() && !m_stop_requested.load()) {
303 0 : if (accepting.wait_for(m_timeout) == std::future_status::ready) {
304 : break;
305 : }
306 : }
307 :
308 0 : TLOG_DEBUG(0) << get_name() << ": Connection received: start reading" << std::endl;
309 :
310 0 : content::tcp_header_t head;
311 0 : head.packet_size = 0;
312 0 : content::word::word_t temp_word;
313 0 : bool connection_closed = false;
314 0 : uint64_t ch_stat_beam, ch_stat_crt, ch_stat_pds; // NOLINT
315 0 : uint64_t prev_timestamp = 0; // NOLINT
316 0 : ts_payload prev_hlt, prev_llt, prev_ch_stat;
317 0 : ts_payload curr_hlt, curr_llt, curr_ch_stat;
318 : // buffers for word matching. buf_a are the trigger words, buf_b are corresponding payloads
319 0 : std::queue<content::word::trigger_t> match_buf_a_hlts, match_buf_a_llts;
320 0 : std::queue<ts_payload> match_buf_b_llts, match_buf_b_chstatus;
321 :
322 0 : while (running_flag.load() && !m_stop_requested.load()) {
323 :
324 0 : update_calibration_file();
325 :
326 0 : if (!read(head)) {
327 0 : connection_closed = true;
328 : break;
329 : }
330 :
331 0 : n_bytes = head.packet_size;
332 : // extract n_words
333 :
334 0 : n_words = n_bytes / word_size;
335 : // read n words as requested from the header
336 :
337 0 : update_buffer_counts(n_words);
338 :
339 0 : for (unsigned int i = 0; i < n_words; ++i) {
340 :
341 0 : if (!running_flag.load() || m_stop_requested.load()) {
342 : break;
343 : }
344 :
345 : // read a word
346 0 : if (!read(temp_word)) {
347 : connection_closed = true;
348 : break;
349 : }
350 : // put it in the calibration stream
351 0 : if (m_has_calibration_stream) {
352 0 : m_calibration_file.write(reinterpret_cast<const char*>(&temp_word), word_size); // NOLINT
353 0 : m_calibration_file.flush();
354 : } // word printing in calibration stream
355 :
356 : // check if it is a TS word and increment the counter
357 0 : if (IsTSWord(temp_word)) {
358 0 : ++m_ts_word_counter;
359 0 : TLOG_DEBUG(9) << "Received timestamp word! TS: " + temp_word.timestamp;
360 0 : prev_timestamp = temp_word.timestamp;
361 0 : } else if (IsFeedbackWord(temp_word)) {
362 0 : m_error_state.store(true);
363 0 : content::word::feedback_t* feedback = reinterpret_cast<content::word::feedback_t*>(&temp_word); // NOLINT
364 0 : TLOG_DEBUG(7) << "Received feedback word!";
365 :
366 0 : TLOG_DEBUG(8) << get_name() << ": Feedback word: " << std::endl
367 0 : << std::hex << " \t Type -> " << feedback->word_type << std::endl
368 0 : << " \t TS -> " << feedback->timestamp << std::endl
369 0 : << " \t Code -> " << feedback->code << std::endl
370 0 : << " \t Source -> " << feedback->source << std::endl
371 0 : << " \t Padding -> " << feedback->padding << std::dec << std::endl;
372 0 : } else if (temp_word.word_type == content::word::t_gt) {
373 0 : TLOG_DEBUG(3) << "Received HLT word! TS: " + temp_word.timestamp;
374 0 : content::word::trigger_t* hlt_word = reinterpret_cast<content::word::trigger_t*>(&temp_word); // NOLINT
375 0 : curr_hlt = { hlt_word->timestamp, (hlt_word->trigger_word & 0x1FFFFFFFFFFFFFFF) };
376 0 : if (check_repeated_word(curr_hlt, prev_hlt, temp_word.word_type))
377 0 : continue;
378 0 : match_buf_a_hlts.push(*hlt_word);
379 : // Count the total HLTs and each specific one
380 0 : ++m_run_HLT_counter;
381 0 : ++m_total_hlt_counter;
382 0 : for (auto& hlt : m_hlt_trigger_counter) {
383 0 : if ((hlt_word->trigger_word >> hlt.first) & 0x1)
384 0 : ++hlt.second;
385 : }
386 0 : m_last_readout_hlt_timestamp = temp_word.timestamp;
387 0 : prev_hlt = curr_hlt;
388 0 : } else if (temp_word.word_type == content::word::t_lt) {
389 0 : TLOG_DEBUG(5) << "Received LLT word! TS: " + temp_word.timestamp;
390 0 : content::word::trigger_t* llt_word = reinterpret_cast<content::word::trigger_t*>(&temp_word); // NOLINT
391 0 : curr_llt = { llt_word->timestamp, (llt_word->trigger_word & 0xFFFFFFFF) };
392 0 : if (check_repeated_word(curr_llt, prev_llt, temp_word.word_type))
393 0 : continue;
394 0 : match_buf_a_llts.push(*llt_word);
395 0 : match_buf_b_llts.push(curr_llt);
396 :
397 0 : ++m_run_LLT_counter;
398 0 : for (auto& llt : m_llt_trigger_counter) {
399 0 : if ((llt_word->trigger_word >> llt.first) & 0x1)
400 0 : ++llt.second;
401 : }
402 0 : prev_llt = curr_llt;
403 0 : } else if (temp_word.word_type == content::word::t_ch) {
404 :
405 0 : content::word::ch_status_t* ch_stat_word = reinterpret_cast<content::word::ch_status_t*>(&temp_word); // NOLINT
406 : // The channel status only has 60b TS so complete the upper 4b from the TS Word. (fyi 60b rolls over >500yr
407 : // @ 62.5MHz)
408 0 : uint64_t corrected_ts = ((prev_timestamp & 0xF000000000000000) | ch_stat_word->timestamp); // NOLINT
409 0 : TLOG_DEBUG(6) << "Received Channel Status word! TS: " + corrected_ts;
410 0 : ch_stat_beam = ch_stat_word->get_beam();
411 0 : ch_stat_crt = ch_stat_word->get_crt();
412 0 : ch_stat_pds = ch_stat_word->get_pds();
413 0 : curr_ch_stat = { corrected_ts, ((ch_stat_pds << 48) | (ch_stat_crt << 16) | ch_stat_beam) };
414 0 : if (check_repeated_word(curr_ch_stat, prev_ch_stat, temp_word.word_type))
415 0 : continue;
416 0 : match_buf_b_chstatus.push(curr_ch_stat);
417 0 : prev_ch_stat = curr_ch_stat;
418 :
419 0 : ++m_run_channel_status_counter;
420 : } // else if on word types
421 : // do matching
422 0 : match_between_buffers(match_buf_a_hlts, match_buf_b_llts, prev_hlt.first, content::word::word_type::t_gt);
423 0 : match_between_buffers(match_buf_a_llts, match_buf_b_chstatus, prev_llt.first, content::word::word_type::t_lt);
424 :
425 : } // n_words loop
426 :
427 0 : if (connection_closed) {
428 : break;
429 : }
430 : }
431 :
432 : // Make sure CTB run stops before closing socket
433 0 : while (m_is_running.load()) {
434 0 : std::this_thread::sleep_for(std::chrono::microseconds(100));
435 : }
436 :
437 0 : boost::system::error_code closing_error;
438 :
439 0 : if (m_error_state.load()) {
440 :
441 0 : m_receiver_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_send, closing_error);
442 :
443 0 : if (closing_error) {
444 0 : std::stringstream msg;
445 0 : msg << "Error in shutdown " << closing_error.message();
446 0 : ers::error(CTBCommunicationError(ERS_HERE, msg.str()));
447 0 : }
448 : }
449 :
450 0 : m_receiver_socket.close(closing_error);
451 :
452 0 : if (closing_error) {
453 0 : std::stringstream msg;
454 0 : msg << "Socket closing failed:: " << closing_error.message();
455 0 : ers::error(CTBCommunicationError(ERS_HERE, msg.str()));
456 0 : }
457 :
458 0 : TLOG_DEBUG(TLVL_CTB_MODULE) << get_name() << ": End of do_work loop: stop receiving data from the CTB";
459 :
460 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_work() method";
461 :
462 0 : } // NOLINT
463 :
464 : bool
465 0 : CTBModule::check_repeated_word(ts_payload& curr_word, ts_payload& prev_word, uint64_t wtype)
466 : { // NOLINT
467 0 : if (curr_word.first == prev_word.first) { // words with repeated timestamp. Not good!
468 0 : std::stringstream msg;
469 0 : msg << "Multiple words have the same timestamp, Using the first one. Word type: ";
470 0 : if (wtype == content::word::word_type::t_gt)
471 0 : msg << "HLT";
472 0 : else if (wtype == content::word::word_type::t_lt)
473 0 : msg << "LLT";
474 0 : else if (wtype == content::word::word_type::t_ch)
475 0 : msg << "Channel Status";
476 : else
477 0 : msg << wtype;
478 0 : msg << ", TS: " << curr_word.first << ".";
479 0 : if (curr_word.second !=
480 0 : prev_word.second) { // not only do we have repeated timestamp, they have different payload...
481 0 : msg << " Different payload!! Previous payload: 0x" << std::hex << prev_word.second << " Current payload: 0x"
482 0 : << curr_word.second;
483 0 : ers::warning(CTBRepeatedTimestampWarning(ERS_HERE, msg.str()));
484 : } else {
485 0 : msg << " Both have payload 0x" << std::hex << prev_word.second;
486 0 : TLOG() << msg.str();
487 : }
488 0 : return true;
489 0 : }
490 : return false;
491 : }
492 :
493 : void
494 0 : CTBModule::send_matched_trigger_word(const content::word::trigger_t& word, uint64_t payload)
495 : { // NOLINT
496 : // Send HSI data to a DLH
497 0 : std::array<uint32_t, 7> hsi_struct; // NOLINT
498 0 : bool is_hlt = word.IsHLT();
499 0 : hsi_struct[0] = (is_hlt << 26) | // link
500 0 : (m_slot << 22) | (m_crate << 12) | (m_det << 6) | CTB_HSI_FRAME_VERSION;
501 0 : hsi_struct[1] = word.timestamp; // ts low
502 0 : hsi_struct[2] = word.timestamp >> 32; // ts high
503 0 : hsi_struct[3] = payload; // lower 32b
504 0 : hsi_struct[4] = payload >> 32; // upper 32b (will be 0x0 for llt payloads)
505 0 : hsi_struct[5] = word.trigger_word; // trigger_map;
506 0 : hsi_struct[6] = is_hlt ? m_run_HLT_counter : m_run_LLT_counter; // m_generated_counter;
507 0 : int dbg_lvl = is_hlt ? 4 : 6;
508 0 : TLOG_DEBUG(dbg_lvl) << get_name() << ": Formed HSI_FRAME_STRUCT for " << (is_hlt ? "HLT" : "LLT") << std::hex << "0x"
509 0 : << hsi_struct[0] << ", 0x" << hsi_struct[1] << ", 0x" << hsi_struct[2] << ", 0x" << hsi_struct[3]
510 0 : << ", 0x" << hsi_struct[4] << ", 0x" << hsi_struct[5] << ", 0x" << hsi_struct[6] << "\n";
511 0 : if (is_hlt) {
512 0 : send_raw_hsi_data(hsi_struct, m_hlt_hsi_data_sender.get());
513 0 : dfmessages::HSIEvent event(m_det, word.trigger_word, word.timestamp, m_run_HLT_counter, m_run_number);
514 0 : send_hsi_event(event);
515 : } else {
516 0 : send_raw_hsi_data(hsi_struct, m_llt_hsi_data_sender.get());
517 : }
518 0 : }
519 :
520 : void
521 0 : CTBModule::match_between_buffers(std::queue<content::word::trigger_t>& buf_a,
522 : std::queue<ts_payload>& buf_b,
523 : uint64_t timeout_reference,
524 : content::word::word_type buf_a_wtype)
525 : { // NOLINT
526 0 : bool is_hlt = (buf_a_wtype == content::word::word_type::t_gt);
527 0 : while (buf_a.size() > 0) {
528 0 : content::word::trigger_t trigger = buf_a.front();
529 0 : auto trigger_ts = trigger.timestamp;
530 0 : auto trigger_word = trigger.trigger_word;
531 0 : if (timeout_reference > (trigger_ts + 100)) {
532 0 : std::stringstream msg;
533 0 : msg << "Time out while waiting for a match for the " << (is_hlt ? "HLT" : "LLT") << ": TS = " << trigger_ts
534 0 : << ", trigger word = " << std::hex << "0x" << trigger_word << " Timeout reference: " << std::dec
535 0 : << timeout_reference;
536 0 : ers::warning(CTBWordMatchWarning(ERS_HERE, msg.str()));
537 0 : buf_a.pop();
538 0 : continue;
539 0 : }
540 0 : if (buf_b.size() == 0)
541 : break; // no input to match yet. Return for now, wait for matching input to come
542 0 : while (buf_b.size() > 0) {
543 0 : auto input_ts = buf_b.front().first;
544 0 : if (input_ts < (trigger_ts - 1)) { // word is too early. No longer needed
545 0 : if (is_hlt)
546 0 : last_popped_llt = buf_b.front();
547 : else
548 0 : last_popped_chstatus = buf_b.front();
549 0 : buf_b.pop();
550 0 : } else if (input_ts == (trigger_ts - 1)) { // match is found
551 0 : send_matched_trigger_word(trigger, buf_b.front().second);
552 0 : buf_a.pop();
553 0 : if (is_hlt)
554 0 : last_popped_llt = buf_b.front();
555 : else
556 0 : last_popped_chstatus = buf_b.front();
557 0 : buf_b.pop();
558 0 : break;
559 : } else { // buf_b is already past the match window. No matching is found, error!
560 0 : if (is_hlt && (trigger_word == 0x1 || trigger_word == (0x1 << 16))) { // Fake HLTs, no matching is OK
561 0 : send_matched_trigger_word(trigger, 0);
562 : } else {
563 0 : std::stringstream msg;
564 0 : msg << "No match found for " << (is_hlt ? "HLT" : "LLT") << ": TS = " << trigger_ts << ", trigger word = 0x"
565 0 : << std::hex << trigger_word << " Adjacent input ts: " << std::dec
566 0 : << (is_hlt ? last_popped_llt.first : last_popped_chstatus.first) << " " << input_ts;
567 0 : ers::warning(CTBWordMatchWarning(ERS_HERE, msg.str()));
568 0 : }
569 0 : buf_a.pop();
570 0 : break;
571 : }
572 : } // end loop buf_b
573 : } // end loop buf_a
574 : // Don't let buf_b get too long (e.g. when LLT rate is high but HLT rate is low)
575 0 : while (buf_b.size() > 32) {
576 0 : if (is_hlt)
577 0 : last_popped_llt = buf_b.front();
578 : else
579 0 : last_popped_chstatus = buf_b.front();
580 0 : buf_b.pop();
581 : }
582 0 : }
583 :
584 : template<typename T>
585 : bool
586 0 : CTBModule::read(T& obj)
587 : {
588 :
589 0 : boost::system::error_code receiving_error;
590 0 : boost::asio::read(m_receiver_socket, boost::asio::buffer(&obj, sizeof(T)), receiving_error);
591 :
592 0 : if (!receiving_error) {
593 : return true;
594 : }
595 :
596 0 : if (receiving_error == boost::asio::error::eof) {
597 0 : std::string error_message = "Socket closed: " + receiving_error.message();
598 0 : ers::error(CTBCommunicationError(ERS_HERE, error_message));
599 : return false;
600 0 : }
601 :
602 : if (receiving_error) {
603 0 : std::string error_message = "Read failure: " + receiving_error.message();
604 0 : ers::error(CTBCommunicationError(ERS_HERE, error_message));
605 : return false;
606 0 : }
607 :
608 : return true;
609 : }
610 :
611 : bool
612 0 : CTBModule::IsTSWord(const content::word::word_t& w) noexcept
613 : {
614 :
615 0 : if (w.word_type == content::word::t_ts) {
616 0 : return true;
617 : }
618 : return false;
619 : }
620 :
621 : bool
622 0 : CTBModule::IsFeedbackWord(const content::word::word_t& w) noexcept
623 : {
624 :
625 0 : if (w.word_type == content::word::t_fback) {
626 0 : return true;
627 : }
628 : return false;
629 : }
630 :
631 : void
632 0 : CTBModule::init_calibration_file()
633 : {
634 :
635 0 : if (!m_has_calibration_stream) {
636 0 : return;
637 : }
638 0 : std::array<char, 200> file_name;
639 0 : time_t rawtime = time(nullptr);
640 0 : struct tm timeinfo;
641 0 : localtime_r(&rawtime, &timeinfo);
642 0 : strftime(file_name.data(), file_name.size(), "%F_%H.%M.%S.calib", &timeinfo);
643 0 : std::string global_name = m_calibration_dir + m_calibration_prefix + file_name.data();
644 0 : m_calibration_file.open(global_name, std::ofstream::binary);
645 0 : m_last_calibration_file_update = std::chrono::steady_clock::now();
646 : // _calibration_file.setf ( std::ios::hex, std::ios::basefield );
647 : // _calibration_file.unsetf ( std::ios::showbase );
648 0 : TLOG_DEBUG(0) << get_name() << ": New Calibration Stream file: " << global_name << std::endl;
649 0 : }
650 :
651 : void
652 0 : CTBModule::update_calibration_file()
653 : {
654 :
655 0 : if (!m_has_calibration_stream) {
656 0 : return;
657 : }
658 :
659 0 : std::chrono::steady_clock::time_point check_point = std::chrono::steady_clock::now();
660 :
661 0 : if (check_point - m_last_calibration_file_update < m_calibration_file_interval) {
662 : return;
663 : }
664 :
665 0 : m_calibration_file.close();
666 0 : init_calibration_file();
667 : }
668 :
669 : bool
670 0 : CTBModule::SetCalibrationStream(const std::string& prefix)
671 : {
672 :
673 0 : if (m_calibration_dir.back() != '/') {
674 0 : m_calibration_dir += '/';
675 : }
676 0 : m_calibration_prefix = prefix;
677 0 : if (prefix.size() > 0) {
678 0 : m_calibration_prefix += '_';
679 : }
680 : // possibly we could check here if the directory is valid and writable before assuming the calibration stream is
681 : // valid
682 0 : return true;
683 : }
684 :
685 : void
686 0 : CTBModule::send_config(const std::string& config)
687 : {
688 :
689 0 : if (m_is_configured.load()) {
690 :
691 0 : TLOG_DEBUG(1) << get_name() << ": Resetting before configuring" << std::endl;
692 0 : send_reset();
693 : }
694 :
695 0 : TLOG_DEBUG(1) << get_name() << ": Sending config" << std::endl;
696 :
697 0 : if (send_message(config)) {
698 :
699 0 : m_is_configured.store(true);
700 :
701 : } else {
702 0 : throw CTBCommunicationError(ERS_HERE, "Unable to configure CTB");
703 : }
704 0 : }
705 :
706 : void
707 0 : CTBModule::send_reset()
708 : {
709 :
710 0 : TLOG_DEBUG(1) << get_name() << ": Sending a reset" << std::endl;
711 :
712 0 : if (send_message("{\"command\":\"HardReset\"}")) {
713 :
714 0 : m_is_running.store(false);
715 0 : m_is_configured.store(false);
716 :
717 : } else {
718 0 : ers::error(CTBCommunicationError(ERS_HERE, "Unable to reset CTB"));
719 : }
720 0 : }
721 :
722 : bool
723 0 : CTBModule::send_message(const std::string& msg)
724 : {
725 :
726 : // add error options
727 :
728 0 : boost::system::error_code error;
729 0 : TLOG_DEBUG(1) << get_name() << ": Sending message: " << msg;
730 :
731 0 : m_num_control_messages_sent++;
732 :
733 0 : boost::asio::write(m_control_socket, boost::asio::buffer(msg), error);
734 0 : boost::array<char, 4096> reply_buf{ " " };
735 0 : m_control_socket.read_some(boost::asio::buffer(reply_buf), error);
736 0 : std::stringstream raw_answer(std::string(reply_buf.begin(), reply_buf.end()));
737 0 : TLOG_DEBUG(1) << get_name() << ": Unformatted answer: " << raw_answer.str();
738 :
739 0 : nlohmann::json answer;
740 0 : raw_answer >> answer;
741 0 : nlohmann::json& messages = answer["feedback"];
742 0 : TLOG_DEBUG(1) << get_name() << ": Received messages: " << messages.size();
743 :
744 0 : bool ret = true;
745 0 : for (nlohmann::json::size_type i = 0; i != messages.size(); ++i) {
746 :
747 0 : m_num_control_responses_received++;
748 :
749 0 : std::string type = messages[i]["type"].dump();
750 0 : if (type.find("error") != std::string::npos || type.find("Error") != std::string::npos ||
751 0 : type.find("ERROR") != std::string::npos) {
752 0 : ers::error(CTBMessage(ERS_HERE, messages[i]["message"].dump()));
753 0 : ret = false;
754 0 : } else if (type.find("warning") != std::string::npos || type.find("Warning") != std::string::npos ||
755 0 : type.find("WARNING") != std::string::npos) {
756 0 : ers::warning(CTBMessage(ERS_HERE, messages[i]["message"].dump()));
757 0 : } else if (type.find("info") != std::string::npos || type.find("Info") != std::string::npos ||
758 0 : type.find("INFO") != std::string::npos) {
759 0 : TLOG() << "Message from the board: " << messages[i]["message"].dump();
760 : } else {
761 0 : std::stringstream blob;
762 0 : blob << messages[i];
763 0 : TLOG() << get_name() << ": Unformatted from the board: " << blob.str();
764 0 : }
765 0 : }
766 :
767 0 : return ret;
768 0 : }
769 :
770 : void
771 0 : CTBModule::update_buffer_counts(uint new_count) // NOLINT(build/unsigned)
772 : {
773 0 : std::unique_lock mon_data_lock(m_buffer_counts_mutex);
774 0 : if (m_buffer_counts.size() > 1000)
775 0 : m_buffer_counts.pop_front();
776 0 : m_buffer_counts.push_back(new_count);
777 0 : }
778 :
779 : double
780 0 : CTBModule::read_average_buffer_counts()
781 : {
782 0 : std::unique_lock mon_data_lock(m_buffer_counts_mutex);
783 :
784 0 : double total_counts = 0;
785 0 : ;
786 0 : uint32_t number_of_counts = m_buffer_counts.size(); // NOLINT(build/unsigned)
787 :
788 0 : if (number_of_counts) {
789 0 : for (uint i = 0; i < number_of_counts; ++i) { // NOLINT(build/unsigned)
790 0 : total_counts = total_counts + m_buffer_counts.at(i);
791 : }
792 0 : return total_counts / number_of_counts;
793 : } else {
794 : return 0;
795 : }
796 0 : }
797 :
798 : void
799 0 : CTBModule::generate_opmon_data()
800 : {
801 0 : dunedaq::ctbmodules::opmon::CTBModuleInfo module_info;
802 :
803 0 : module_info.set_num_control_messages_sent(m_num_control_messages_sent.load());
804 0 : module_info.set_num_control_responses_received(m_num_control_responses_received.load());
805 0 : module_info.set_ctb_hardware_running(m_is_running.load());
806 0 : module_info.set_ctb_hardware_configured(m_is_configured.load());
807 :
808 0 : module_info.set_last_readout_timestamp(m_last_readout_hlt_timestamp.load());
809 0 : module_info.set_failed_to_send_hsi_events_counter(m_failed_to_send_counter.load());
810 0 : module_info.set_last_sent_timestamp(m_last_sent_timestamp.load());
811 0 : module_info.set_average_buffer_occupancy(read_average_buffer_counts());
812 :
813 0 : module_info.set_total_hlt_count(m_total_hlt_counter.load());
814 0 : module_info.set_ts_word_count(m_ts_word_counter.exchange(0));
815 :
816 0 : publish(std::move(module_info));
817 :
818 0 : for (auto& hlt : m_hlt_trigger_counter) {
819 0 : dunedaq::ctbmodules::opmon::TriggerInfo ti;
820 0 : ti.set_count(hlt.second.exchange(0));
821 0 : publish(std::move(ti), { { "trigger", "HLT_" + std::to_string(hlt.first) } });
822 0 : }
823 :
824 0 : for (auto& llt : m_llt_trigger_counter) {
825 0 : dunedaq::ctbmodules::opmon::TriggerInfo ti;
826 0 : ti.set_count(llt.second.exchange(0));
827 0 : publish(std::move(ti), { { "trigger", "LLT_" + std::to_string(llt.first) } });
828 0 : }
829 0 : }
830 :
831 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::ctbmodules::CTBModule)
832 :
833 : // Local Variables:
834 : // c-basic-offset: 2
835 : // End:
|
