Line data Source code
1 : /**
2 : * @file CIBModule.cpp
3 : *
4 : * Implementations of CIBModule's functions
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 : #include "confmodel/GeoId.hpp"
11 :
12 : // these are the data types in the application model
13 : #include "appmodel/CIBCalibrationStream.hpp"
14 : #include "appmodel/CIBConf.hpp"
15 : #include "appmodel/CIBTrigger.hpp"
16 : #include "appmodel/CIBoardConf.hpp"
17 :
18 : #include "CIBModule.hpp"
19 : #include "CIBModuleIssues.hpp"
20 :
21 : #include "iomanager/IOManager.hpp"
22 : #include "logging/Logging.hpp"
23 :
24 : #include <charconv>
25 : #include <chrono>
26 : #include <cstdint>
27 : #include <map>
28 : #include <queue>
29 : #include <string>
30 : #include <string_view>
31 : #include <thread>
32 : #include <utility>
33 : #include <vector>
34 :
35 : // since cib_data_fmt is shared with the board
36 : // we need this macro to avoid clashes with the namespaces
37 : // not the cleanest solution, but it works and avoids maintaining two separate versions of the same struct
38 : #define CIB_DUNEDAQ 1
39 : // we may need this to help parse the data format arriving from the CIB
40 : #include <cib_data_fmt.h>
41 : #include <cib_utilities.h>
42 : /**
43 : * @brief Name used by TRACE TLOG calls from this source file
44 : */
45 : #define TRACE_NAME "CIBModule" // NOLINT
46 : #define TLVL_ENTER_EXIT_METHODS 10
47 : #define TLVL_CIB_INFO 5
48 : #define TLVL_CIB_DEBUG 10
49 : #define TLVL_CIB_TRACE 15
50 :
51 : // as it happens, the frame structure remains unchanged since NP04
52 : constexpr uint16_t CIB_HSI_FRAME_VERSION = 0x1; // NOLINT
53 : namespace dunedaq::cibmodules {
54 :
55 0 : CIBModule::CIBModule(const std::string& name)
56 : : hsilibs::HSIEventSender(name)
57 0 : , m_is_running(false)
58 0 : , m_is_configured(false)
59 0 : , m_stop_requested(false)
60 0 : , m_receiver_port(8871)
61 0 : , m_receiver_timeout(70000) // 70 ms - we know that triggers will come at 10 Hz max
62 0 : , m_error_state(false)
63 0 : , m_control_ios()
64 0 : , m_control_socket(m_control_ios)
65 0 : , m_control_endpoint()
66 0 : , m_receiver_ios()
67 0 : , m_receiver_socket(m_receiver_ios)
68 0 : , m_thread_(std::bind(&CIBModule::do_hsi_work, this, std::placeholders::_1))
69 0 : , m_calibration_stream_enable(false)
70 0 : , m_calibration_dir("")
71 0 : , m_calibration_prefix("")
72 0 : , m_calibration_file_interval(std::chrono::minutes(15))
73 : // metric utilities
74 0 : , m_num_control_messages_sent(0)
75 0 : , m_num_control_responses_received(0)
76 0 : , m_num_total_triggers_received(0)
77 0 : , m_num_run_triggers_received(0)
78 :
79 0 : , m_trigger_bit(0)
80 0 : , m_receiver_ready(false)
81 : {
82 : // we can infer the instance from the name
83 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Instantiating a cibmodule with argument [" << name << "]";
84 0 : register_command("conf", &CIBModule::do_configure);
85 0 : register_command("start", &CIBModule::do_start);
86 0 : register_command("stop", &CIBModule::do_stop);
87 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Leaving [" << name << "]constructor.";
88 0 : }
89 :
90 0 : CIBModule::~CIBModule()
91 : {
92 0 : if (m_is_running.load()) {
93 0 : const CommandData_t stopobj;
94 :
95 : // const nlohmann::json stopobj;
96 : // this should also take care of closing the streaming socket
97 0 : do_stop(stopobj);
98 0 : }
99 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Closing the control socket " << std::endl;
100 0 : m_control_socket.close();
101 0 : }
102 :
103 : void
104 0 : CIBModule::init(std::shared_ptr<appfwk::ConfigurationManager> cfgMgr)
105 : {
106 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering init() method";
107 :
108 : // init the sender
109 0 : HSIEventSender::init(cfgMgr);
110 :
111 : // assign the local configuration manager to the argument
112 0 : m_cfg = cfgMgr;
113 :
114 : // get the configuration fragment for this module
115 0 : auto mdal = cfgMgr->get_dal<appmodel::CIBModule>(get_name());
116 0 : if (!mdal) {
117 0 : throw cibmodules::CIBConfigFailure(ERS_HERE, "Missing Module configuration for " + get_name());
118 : }
119 :
120 : // assign the module configuration
121 0 : m_module = mdal;
122 :
123 : // setting up connections
124 0 : auto iom = iomanager::IOManager::get();
125 :
126 0 : using hsi_frame_t = dunedaq::hsilibs::HSI_FRAME_STRUCT;
127 0 : for (auto con : m_module->get_outputs()) {
128 0 : if (con->get_data_type() == datatype_to_string<hsi_frame_t>()) {
129 : // Filter connections by UID: only process those with "CIB" or "cib" in the name
130 : // This string comes from the HSISignalWindow UID in the appmodel configuration
131 0 : if ((con->UID().find("CIB") != std::string::npos) || (con->UID().find("cib") != std::string::npos)) {
132 0 : TLOG() << get_name() << ": Setting up HSI Frame output : " << con->UID() << std::endl;
133 0 : m_cib_hsi_data_sender = iom->get_sender<hsi_frame_t>(con->UID());
134 : } else {
135 0 : TLOG_DEBUG(5) << get_name() << ": Skipping output : " << con->UID();
136 : }
137 :
138 : } // if data type is HSI Frame
139 : } // loop over outputs
140 :
141 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting init() method";
142 0 : }
143 :
144 : void
145 0 : CIBModule::do_configure(const CommandData_t&)
146 : {
147 :
148 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering CIB do_configure()";
149 :
150 : // this returns the structure of the CIBConf object
151 0 : auto conf = m_module->get_configuration();
152 : // this gets the CIBoardConf object
153 0 : auto board = m_module->get_board();
154 :
155 0 : auto trigger_conf = conf->get_cib_trigger();
156 :
157 : // identify the trigger bit that this receiver is assigned to
158 : // We need this to construct the HSI frame, right?
159 0 : if (!dunedaq::cibmodules::util::parse_hex(trigger_conf->get_trigger_bit(), m_trigger_bit)) {
160 0 : std::ostringstream msg("");
161 0 : msg << get_name() << ": Unable to parse trigger bit hex string : " << trigger_conf->get_trigger_bit();
162 0 : throw CIBModuleError(ERS_HERE, msg.str());
163 0 : } else {
164 0 : TLOG_DEBUG(TLVL_CIB_DEBUG) << get_name() << ": Parsed trigger bit hex string " << trigger_conf->get_trigger_bit()
165 0 : << " to 0x" << std::hex << m_trigger_bit << std::dec << "["
166 0 : << trigger_conf->get_trigger_id() << "]";
167 : }
168 :
169 : // init monitoring variables
170 0 : m_num_control_messages_sent = 0;
171 0 : m_num_control_responses_received = 0;
172 :
173 : // figure out the identifier of the CIB
174 : // this is set in the configuration
175 0 : auto geo_id = board->get_geo_id();
176 0 : m_det = geo_id->get_detector_id();
177 0 : m_crate = geo_id->get_crate_id();
178 0 : m_slot = geo_id->get_slot_id();
179 :
180 0 : auto session = m_cfg->get_session();
181 :
182 : // init trigger counters
183 0 : m_num_run_triggers_received.store(0);
184 :
185 0 : auto cib_host = conf->get_cib_host();
186 0 : auto cib_port = conf->get_cib_port();
187 :
188 : // network connection to the CIB module
189 0 : boost::asio::ip::tcp::resolver resolver(m_control_ios);
190 0 : boost::asio::ip::tcp::resolver::query query(cib_host, std::to_string(cib_port)); //"np04-iols-cib-02", 8992
191 0 : boost::asio::ip::tcp::resolver::iterator iter = resolver.resolve(query);
192 :
193 0 : m_control_endpoint = iter->endpoint();
194 : // attempt the connection.
195 0 : try {
196 :
197 0 : m_control_socket.connect(m_control_endpoint);
198 0 : m_control_socket.set_option(boost::asio::ip::tcp::no_delay(true));
199 0 : } catch (std::exception& e) {
200 0 : std::ostringstream msg("");
201 0 : msg << get_name() << "Exception caught while establishing connection to CIB : " << e.what();
202 : // do nothing more. Just exist
203 0 : m_is_configured.store(false);
204 0 : throw CIBCommunicationError(ERS_HERE, msg.str());
205 0 : }
206 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Successfully connected to CIB endpoint " << cib_host << ':' << cib_port
207 0 : << std::endl;
208 :
209 : // if necessary, set the calibration stream
210 0 : auto stream_conf = conf->get_calibration_stream();
211 0 : if (stream_conf) {
212 0 : TLOG() << "Calibration stream enabled";
213 0 : m_calibration_stream_enable = true;
214 0 : m_calibration_dir = stream_conf->get_output_directory();
215 0 : m_calibration_file_interval = std::chrono::duration_cast<decltype(m_calibration_file_interval)>(
216 0 : std::chrono::seconds(stream_conf->get_update_period_s()));
217 : }
218 :
219 : // these are for the local receiver operation
220 0 : m_receiver_port = board->get_receiver_port();
221 0 : m_receiver_timeout = std::chrono::milliseconds(conf->get_connection_timeout_ms());
222 0 : m_receiver_host = board->get_receiver_host();
223 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Default board receiver network location (PRELIMINARY) "
224 0 : << m_receiver_host << ':' << m_receiver_port << std::endl;
225 :
226 : // config fragment to be sent to the CIB
227 0 : nlohmann::json config;
228 : //
229 : // if the host is localhost, we need to resolve it to the actual hostname
230 : // and also check whether the port is available
231 : //
232 0 : if (m_receiver_host == "localhost") {
233 0 : boost::asio::ip::tcp::resolver resolver(m_receiver_ios);
234 : // Check if port is already in use, to try to avoid future conflicts
235 0 : unsigned short port = m_receiver_port;
236 0 : while (check_port_in_use(port)) {
237 : // the problem is that if the port is in use, the CIB will be sending the data to the wrong place
238 : // there is little point in continuing
239 0 : std::ostringstream msg("");
240 0 : msg << "Listener port [" << port << "] is in use by someone else. Trying another.";
241 0 : ers::warning(CIBMessage(ERS_HERE, msg.str()));
242 0 : port++;
243 0 : }
244 0 : if (port != m_receiver_port) {
245 0 : std::ostringstream msg("");
246 0 : msg << "Listener port [" << m_receiver_port << "] is in use. Relocating to port [" << port << "]";
247 0 : ers::warning(CIBMessage(ERS_HERE, msg.str()));
248 0 : m_receiver_port = port;
249 0 : } else {
250 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Listener port " << m_receiver_port << " is available." << std::endl;
251 : }
252 :
253 : // at this we have to find the hostname to tell the board where to send the data
254 0 : boost::asio::ip::tcp::resolver::query query_for_local(boost::asio::ip::host_name(), "");
255 0 : iter = resolver.resolve(query_for_local);
256 :
257 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Resolved localhost to " << iter->endpoint().address().to_string()
258 0 : << std::endl;
259 0 : m_receiver_host = iter->endpoint().address().to_string();
260 :
261 : // create the json string out of the config fragment
262 : // replacing the receiver address with the one that we just calculated
263 0 : try {
264 0 : config = board->get_cib_json(*session, m_receiver_host, m_receiver_port);
265 0 : } catch (nlohmann::json::exception& e) {
266 0 : std::ostringstream msg("");
267 0 : msg << get_name() << "Caught a JSON exception converting config fragment : " << e.what();
268 0 : m_is_configured.store(false);
269 0 : throw CIBModuleError(ERS_HERE, msg.str());
270 0 : } catch (std::exception& e) {
271 0 : std::ostringstream msg("");
272 0 : msg << get_name() << "Caught STD exception while converting config fragment : " << e.what();
273 : // do nothing more. Just exist
274 0 : m_is_configured.store(false);
275 0 : throw CIBModuleError(ERS_HERE, msg.str());
276 0 : }
277 0 : } else {
278 : /* just use the configuration information */
279 : // nlohmann::to_json(config, board->get_cib_json(*session));
280 0 : config = board->get_cib_json(*session);
281 : }
282 0 : auto json_dump = config.dump();
283 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << "Sending configuration: [" << json_dump << "] to CIB board";
284 0 : send_config(config.dump());
285 0 : m_is_configured.store(true);
286 0 : }
287 :
288 : void
289 0 : CIBModule::do_start(const CommandData_t& startobj)
290 : {
291 :
292 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_start() method";
293 : // actually, the first thing to check is whether the CIB has been configured
294 : // if not, this won't work
295 0 : if (!m_is_configured.load()) {
296 0 : throw CIBWrongState(ERS_HERE, "CIB has not been successfully configured.");
297 : }
298 :
299 : // Set this to false early so it doesn't interfere with the start
300 0 : m_stop_requested.store(false);
301 0 : m_run_number.store(startobj.at("run").get<daqdataformats::run_number_t>());
302 : // reset the metrics/counters
303 0 : m_num_run_triggers_received.store(0);
304 :
305 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Sending start of run command with run number " << m_run_number.load();
306 0 : m_thread_.start_working_thread();
307 :
308 : // NFB: There was a potential race condition here: the socket in the working thread
309 : // needs to be in place before the CIB receives order to send data, or we risk having a connection
310 : // failure, if for some reason the CIB attempts to connect before the working thread is ready to receive.
311 0 : if (m_calibration_stream_enable) {
312 0 : std::stringstream run;
313 0 : run << "run" << m_run_number.load();
314 0 : set_calibration_stream(run.str());
315 0 : }
316 : int cnt = 0;
317 0 : while (!m_receiver_ready.load()) {
318 0 : std::this_thread::sleep_for(std::chrono::milliseconds(10));
319 0 : cnt++;
320 0 : if (cnt > 50) {
321 : // the socket didn't get ready on time
322 0 : throw CIBModuleError(ERS_HERE, "Receiver socket timed out before becoming ready.");
323 : }
324 : }
325 0 : TLOG_DEBUG(TLVL_CIB_DEBUG) << get_name() << ": All ready to signal the CIB to start";
326 :
327 0 : nlohmann::json cmd;
328 0 : cmd["command"] = "start_run";
329 0 : cmd["run_number"] = m_run_number.load();
330 :
331 0 : if (send_message(cmd.dump())) {
332 0 : m_is_running.store(true);
333 0 : TLOG() << get_name() << ": CIB run started successfully";
334 : } else {
335 0 : throw CIBCommunicationError(ERS_HERE, "Unable to start CIB run");
336 : }
337 :
338 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_start() method";
339 0 : }
340 :
341 : void
342 0 : CIBModule::do_stop(const CommandData_t& /*stopobj*/)
343 : {
344 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_stop() method";
345 0 : TLOG_DEBUG(TLVL_CIB_DEBUG) << get_name() << ": Sending stop run command" << std::endl;
346 :
347 : // Set stop flag BEFORE sending command so receiver thread knows to expect EOF
348 0 : m_stop_requested.store(true);
349 :
350 0 : if (send_message("{\"command\":\"stop_run\"}")) {
351 : // Response arrival means CIB has closed its data socket (see Handler::stop_run())
352 : // Receiver thread will detect EOF and exit cleanly
353 0 : std::this_thread::sleep_for(std::chrono::milliseconds(2));
354 0 : TLOG() << get_name() << ": CIB run stopped successfully";
355 0 : m_is_running.store(false);
356 : } else {
357 : // failed to sent the message to stop the run.
358 : // stop the collecting thread and then throw, since that
359 : // attempts a cleaner exit
360 0 : m_thread_.stop_working_thread();
361 :
362 0 : throw CIBCommunicationError(ERS_HERE, "Unable to stop CIB");
363 : }
364 : //
365 0 : m_thread_.stop_working_thread();
366 :
367 : // -- print the counters for local info
368 0 : TLOG() << get_name() << ": CIB trigger counter summary after run [" << m_run_number << "]:\n\n"
369 0 : << "IOLS trigger counter in run : " << m_num_run_triggers_received << "\n"
370 0 : << "Global IOLS trigger count : " << m_num_total_triggers_received << std::endl;
371 :
372 : // reset counters
373 0 : m_num_run_triggers_received = 0;
374 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_stop() method";
375 0 : }
376 :
377 : // this is where most of the work is really done
378 : void
379 0 : CIBModule::do_hsi_work(std::atomic<bool>& running_flag)
380 : {
381 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_work() method";
382 :
383 0 : std::size_t n_bytes = 0;
384 0 : std::size_t n_words = 0;
385 0 : std::size_t prev_seq = 0;
386 0 : bool first = true;
387 :
388 : // connect to socket
389 : // should we keep everything local or under the class?
390 0 : boost::system::error_code ec;
391 :
392 : // check that this port is still available
393 0 : if (check_port_in_use(m_receiver_port)) {
394 : // the problem is that, at this stage, if the port is in use,
395 : // the CIB will be sending the data to the wrong place
396 : // there is little point in continuing
397 0 : std::ostringstream msg("");
398 0 : msg << "Listener port [" << m_receiver_port << "] is in use by someone else. Failing.";
399 0 : ers::error(CIBMessage(ERS_HERE, msg.str()));
400 0 : }
401 :
402 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Will set up the listener on port " << m_receiver_port << std::endl;
403 0 : boost::asio::ip::tcp::acceptor acceptor(m_receiver_ios,
404 0 : boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), m_receiver_port));
405 :
406 0 : acceptor.listen(boost::asio::ip::tcp::socket::max_connections, ec);
407 0 : if (ec) {
408 0 : std::ostringstream msg("");
409 0 : msg << get_name() << ": CIB got an error listening on socket: :" << m_receiver_port << " -- reason: '" << ec << "'";
410 0 : throw CIBCommunicationError(ERS_HERE, msg.str());
411 : return;
412 0 : } else {
413 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Waiting for an incoming connection on port " << m_receiver_port
414 0 : << std::endl;
415 : }
416 :
417 0 : std::future<void> accepting = async(std::launch::async, [&] { acceptor.accept(m_receiver_socket, ec); });
418 0 : if (ec) {
419 0 : std::stringstream msg;
420 0 : msg << "Socket opening failed:: " << ec.message();
421 0 : ers::error(CIBCommunicationError(ERS_HERE, msg.str()));
422 0 : return;
423 0 : }
424 : //
425 0 : m_receiver_ready.store(true);
426 :
427 0 : while (running_flag.load() && !m_stop_requested.load()) {
428 0 : if (accepting.wait_for(m_receiver_timeout) == std::future_status::ready) {
429 : break;
430 : }
431 : }
432 :
433 0 : TLOG() << get_name() << ": Connection received: start reading" << std::endl;
434 :
435 : // -- A couple of variables to help in the data parsing
436 : /**
437 : * The structure is a bit different than in the CTB
438 : * The TCP packet contains a single word (the trigger)
439 : * But other than that, everything are triggers
440 : */
441 :
442 0 : dunedaq::cib::daq::iols_tcp_packet_t tcp_packet;
443 0 : TLOG_DEBUG(TLVL_CIB_TRACE) << "Checking expected sizes: "
444 0 : << " sizeof(iols_tcp_packet_t)=" << sizeof(dunedaq::cib::daq::iols_tcp_packet_t)
445 0 : << " sizeof(iols_trigger_t)=" << sizeof(dunedaq::cib::daq::iols_trigger_t)
446 0 : << " sizeof(tcp_header_t)=" << sizeof(dunedaq::cib::daq::tcp_header_t) << std::endl;
447 :
448 : // boost::system::error_code receiving_error;
449 0 : bool connection_closed = false;
450 :
451 0 : while (running_flag.load() && !m_stop_requested.load()) {
452 0 : update_calibration_file();
453 :
454 0 : if (!read(m_receiver_socket, tcp_packet)) {
455 0 : connection_closed = true;
456 0 : break;
457 : }
458 :
459 0 : n_bytes = tcp_packet.header.packet_size;
460 0 : n_words = n_bytes / sizeof(dunedaq::cib::daq::iols_trigger_t);
461 :
462 0 : if (n_words != 1) {
463 0 : std::ostringstream msg("");
464 0 : msg << "Received more than one IoLS trigger word at once! This should never happen. Got " << n_bytes
465 0 : << " (expected " << sizeof(dunedaq::cib::daq::iols_trigger_t) << ")";
466 0 : ers::warning(CIBMessage(ERS_HERE, msg.str()));
467 0 : }
468 : // the CIB only ships one word per packet....so this error should be impossible
469 : // check continuity of the sequence numbers
470 0 : if (first) {
471 : // first word being fetched. The sequence number should be zero
472 0 : if (tcp_packet.header.sequence_id != 0) {
473 0 : std::ostringstream msg("");
474 0 : msg << "Missing sequence. First word should have sequence number 0. Got "
475 0 : << static_cast<int>(tcp_packet.header.sequence_id);
476 0 : ers::warning(CIBMessage(ERS_HERE, msg.str()));
477 0 : }
478 : first = false;
479 : } else {
480 0 : bool failed = false;
481 : // in case it rolled over, compare to 255
482 0 : if (tcp_packet.header.sequence_id == 0) {
483 0 : if (prev_seq != 255) {
484 : failed = true;
485 : }
486 : } else {
487 0 : if (tcp_packet.header.sequence_id != (prev_seq + 1)) {
488 : failed = true;
489 : }
490 : }
491 : if (failed) {
492 0 : std::ostringstream msg("");
493 0 : msg << "Skipped CIB word sequence. Prev word " << prev_seq << " current word "
494 0 : << static_cast<int>(tcp_packet.header.sequence_id);
495 0 : ers::warning(CIBMessage(ERS_HERE, msg.str()));
496 0 : }
497 : }
498 0 : prev_seq = tcp_packet.header.sequence_id;
499 :
500 0 : update_buffer_counts(n_words);
501 :
502 : // temporarily print the trigger
503 : // I leave this here for debugging purposes, but don´t even keep it in the logs, since it is too verbose. We can
504 : // always add it back if we need to debug something TLOG_DEBUG(TLVL_CIB_TRACE) << "TRIGGER : ts " <<
505 : // tcp_packet.word.timestamp
506 : // << " pos_m1 " << util::get_m1(tcp_packet.word)
507 : // << " pos_m2 " << util::get_m2(tcp_packet.word)
508 : // << " pos_m3 " << util::get_m3(tcp_packet.word);
509 :
510 0 : if (m_calibration_stream_enable) {
511 0 : m_calibration_file.write(reinterpret_cast<const char*>(&tcp_packet.word), sizeof(tcp_packet.word)); // NOLINT
512 0 : m_calibration_file.flush();
513 : } // word printing in calibration stream
514 :
515 0 : TLOG_DEBUG(TLVL_CIB_DEBUG) << get_name() << "Received IoLS trigger word!";
516 0 : ++m_num_total_triggers_received;
517 0 : ++m_num_run_triggers_received;
518 :
519 : // we do not need to know anything else
520 : // ideally, one could add other information such as the direction
521 : // this should be coming packed in the trigger word
522 : // note, however, that to reconstruct the trace direction we also would need the source position
523 : // and that we cannot afford to send, so we can just make it up out of the IoLS system
524 : //
525 : // Send HSI data to a DLH
526 0 : std::array<uint32_t, 7> hsi_struct;
527 0 : hsi_struct[0] = (0x1 << 26) | // some random bit that could indicate the type of frame
528 0 : (m_slot << 22) | // slot number
529 0 : (m_crate << 12) | // crate number
530 0 : (m_det << 6) | // detector number
531 : CIB_HSI_FRAME_VERSION;
532 : // timestamps - I like the explicit masking here to mistakes
533 0 : hsi_struct[1] = tcp_packet.word.timestamp & 0xFFFFFFFF; // ts low
534 0 : hsi_struct[2] = (tcp_packet.word.timestamp >> 32) & 0xFFFFFFFF; // ts high
535 :
536 : // we shall use these 2 sets of 32 bits to define the periscope position
537 : // pos_m3 == linear stage
538 0 : hsi_struct[3] = dunedaq::cibmodules::util::get_m3(tcp_packet.word); // lower 32b 0
539 : // pos_m3 == RNN600
540 0 : hsi_struct[4] = dunedaq::cibmodules::util::get_m2(tcp_packet.word); // upper 32b
541 : /**
542 : * A note about the 5th entry
543 : * The trigger bit is actually mapped into a single bit, that is then remapped back
544 : * into an index
545 : */
546 0 : hsi_struct[5] = m_trigger_bit; // trigger_map;
547 0 : hsi_struct[6] = m_num_run_triggers_received.load(); // m_generated_counter;
548 :
549 : // Same thing here. If something really bad happens, this code can be very useful, but we do not want to print this
550 : // in the logs, unless strictly necessary, since it is too verbose. TLOG_DEBUG(TLVL_CIB_TRACE) << "CIB HSI Frame: "
551 : // << "0x" << std::hex << hsi_struct[0]
552 : // << ", 0x" << hsi_struct[1]
553 : // << ", 0x" << hsi_struct[2]
554 : // << ", 0x" << hsi_struct[3]
555 : // << ", 0x" << hsi_struct[4]
556 : // << ", 0x" << hsi_struct[5]
557 : // << ", 0x" << hsi_struct[6]
558 : // << std::dec << std::endl;
559 :
560 0 : if (!m_cib_hsi_data_sender) {
561 0 : std::ostringstream msg("");
562 0 : msg << "HSI Data Sender not properly configured! This will go down in flames.";
563 0 : ers::error(CIBCommunicationError(ERS_HERE, msg.str()));
564 : // we cannot continue
565 0 : break;
566 0 : }
567 :
568 0 : send_raw_hsi_data(hsi_struct, m_cib_hsi_data_sender.get());
569 :
570 : // TODO Nuno Barros Apr-02-2024 : properly fill device id when someone explains me
571 : // how to get it
572 : // in fact, this may disappear in the future where the CIB is decoupled from
573 : // the HSI infrastructure
574 0 : dfmessages::HSIEvent event(
575 0 : m_det, m_trigger_bit, tcp_packet.word.timestamp, m_num_run_triggers_received.load(), m_run_number);
576 :
577 0 : send_hsi_event(event);
578 0 : if (connection_closed) {
579 : break;
580 : }
581 : }
582 :
583 : // Wait for stop signal (which only comes after CIB closed its sender socket)
584 0 : while (m_is_running.load()) {
585 0 : std::this_thread::sleep_for(std::chrono::microseconds(100));
586 : }
587 :
588 0 : boost::system::error_code closing_error;
589 : // if the system is already in an error state
590 : // we should call for a socket shutdown to force
591 : // the connection to close
592 0 : if (m_error_state.load()) {
593 :
594 0 : m_receiver_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_send, closing_error);
595 :
596 0 : if (closing_error) {
597 0 : std::stringstream msg;
598 0 : msg << "Error in shutdown " << closing_error.message();
599 0 : ers::error(CIBCommunicationError(ERS_HERE, msg.str()));
600 0 : }
601 : }
602 :
603 0 : m_receiver_socket.close(closing_error);
604 :
605 0 : if (closing_error) {
606 0 : std::stringstream msg;
607 0 : msg << "Socket closing failed:: " << closing_error.message();
608 0 : ers::error(CIBCommunicationError(ERS_HERE, msg.str()));
609 0 : }
610 :
611 0 : m_receiver_ready.store(false);
612 :
613 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": End of do_work loop: stop receiving data from the CIB";
614 :
615 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_work() method";
616 0 : }
617 :
618 : template<typename T>
619 : bool
620 0 : CIBModule::read(boost::asio::ip::tcp::socket& socket, T& obj)
621 : {
622 :
623 0 : boost::system::error_code receiving_error;
624 0 : boost::asio::read(socket, boost::asio::buffer(&obj, sizeof(T)), receiving_error);
625 :
626 0 : if (!receiving_error) {
627 : return true;
628 : }
629 :
630 0 : if (receiving_error == boost::asio::error::eof) {
631 0 : if (m_stop_requested.load()) {
632 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Socket closed after stop request." << std::endl;
633 : } else {
634 0 : std::string error_message = "Socket closed: " + receiving_error.message();
635 0 : ers::error(CIBCommunicationError(ERS_HERE, error_message));
636 0 : }
637 :
638 0 : return false;
639 : }
640 :
641 : if (receiving_error) {
642 0 : std::string error_message = "Read failure: " + receiving_error.message();
643 0 : ers::error(CIBCommunicationError(ERS_HERE, error_message));
644 : return false;
645 0 : }
646 :
647 : return true;
648 : }
649 :
650 : void
651 0 : CIBModule::init_calibration_file()
652 : {
653 0 : if (!m_calibration_stream_enable) {
654 0 : return;
655 : }
656 0 : char file_name[200] = "";
657 0 : time_t rawtime;
658 0 : time(&rawtime);
659 0 : struct tm local_tm;
660 0 : struct tm* timeinfo = localtime_r(&rawtime, &local_tm);
661 0 : strftime(file_name, sizeof(file_name), "%F_%H.%M.%S.iols.calib", timeinfo);
662 0 : std::string global_name = m_calibration_dir + m_calibration_prefix + file_name;
663 0 : m_calibration_file.open(global_name, std::ofstream::binary);
664 0 : if (!m_calibration_file.is_open()) {
665 0 : std::ostringstream msg;
666 0 : msg << get_name() << ": Unable to open calibration stream file: " << global_name;
667 0 : ers::warning(CIBMessage(ERS_HERE, msg.str()));
668 0 : m_calibration_stream_enable = false;
669 0 : return;
670 0 : }
671 0 : m_last_calibration_file_update = std::chrono::steady_clock::now();
672 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": New Calibration Stream file: " << global_name << std::endl;
673 0 : }
674 :
675 : void
676 0 : CIBModule::update_calibration_file()
677 : {
678 :
679 0 : if (!m_calibration_stream_enable) {
680 0 : return;
681 : }
682 :
683 0 : std::chrono::steady_clock::time_point check_point = std::chrono::steady_clock::now();
684 :
685 0 : if (check_point - m_last_calibration_file_update < m_calibration_file_interval) {
686 : return;
687 : }
688 :
689 0 : m_calibration_file.close();
690 0 : init_calibration_file();
691 : }
692 :
693 : bool
694 0 : CIBModule::set_calibration_stream(const std::string& prefix)
695 : {
696 :
697 0 : if (m_calibration_dir.back() != '/') {
698 0 : m_calibration_dir += '/';
699 : }
700 0 : m_calibration_prefix = prefix;
701 0 : if (prefix.size() > 0) {
702 0 : m_calibration_prefix += '_';
703 : }
704 : // possibly we could check here if the directory is valid and writable before assuming the calibration stream is
705 : // valid
706 0 : return true;
707 : }
708 :
709 : void
710 0 : CIBModule::send_config(const std::string& config)
711 : {
712 :
713 0 : TLOG_DEBUG(TLVL_CIB_INFO) << get_name() << ": Sending config" << std::endl;
714 :
715 : // structure the message to have a common management structure
716 0 : nlohmann::json conf;
717 0 : conf["command"] = "config";
718 0 : conf["config"] = nlohmann::json::parse(config);
719 :
720 0 : TLOG_DEBUG(TLVL_CIB_DEBUG) << get_name() << ": Shipped config : " << conf.dump() << std::endl;
721 :
722 0 : if (send_message(conf.dump())) {
723 0 : m_is_configured.store(true);
724 : } else {
725 0 : throw CIBCommunicationError(ERS_HERE, "Unable to configure CIB");
726 : }
727 0 : }
728 :
729 : bool
730 0 : CIBModule::send_message(const std::string& msg)
731 : {
732 :
733 : // add error options
734 0 : boost::system::error_code error;
735 0 : TLOG_DEBUG(1) << get_name() << ": Sending message: " << msg;
736 :
737 0 : m_num_control_messages_sent++;
738 :
739 0 : boost::asio::write(m_control_socket, boost::asio::buffer(msg), error);
740 0 : boost::array<char, 1024> reply_buf{ " " };
741 0 : m_control_socket.read_some(boost::asio::buffer(reply_buf), error);
742 0 : std::stringstream raw_answer(std::string(reply_buf.begin(), reply_buf.end()));
743 0 : TLOG_DEBUG(1) << get_name() << ": Unformatted answer: " << raw_answer.str();
744 :
745 0 : nlohmann::json answer;
746 0 : raw_answer >> answer;
747 0 : nlohmann::json& messages = answer["feedback"];
748 0 : TLOG_DEBUG(1) << get_name() << ": Received messages: " << messages.size();
749 :
750 0 : bool ret = true;
751 0 : for (nlohmann::json::size_type i = 0; i != messages.size(); ++i) {
752 :
753 0 : m_num_control_responses_received++;
754 :
755 0 : std::string type = messages[i]["type"].dump();
756 0 : if (type.find("error") != std::string::npos || type.find("Error") != std::string::npos ||
757 0 : type.find("ERROR") != std::string::npos) {
758 0 : ers::error(CIBMessage(ERS_HERE, messages[i]["message"].dump()));
759 0 : ret = false;
760 0 : } else if (type.find("warning") != std::string::npos || type.find("Warning") != std::string::npos ||
761 0 : type.find("WARNING") != std::string::npos) {
762 0 : ers::warning(CIBMessage(ERS_HERE, messages[i]["message"].dump()));
763 0 : } else if (type.find("info") != std::string::npos || type.find("Info") != std::string::npos ||
764 0 : type.find("INFO") != std::string::npos) {
765 0 : TLOG() << "Message from the CIB : " << messages[i]["message"].dump();
766 : } else {
767 0 : std::stringstream blob;
768 0 : blob << messages[i];
769 0 : TLOG() << get_name() << ": Unformatted feedback from the board: " << blob.str();
770 0 : }
771 0 : }
772 0 : return ret;
773 0 : }
774 :
775 : void
776 0 : CIBModule::update_buffer_counts(uint new_count) // NOLINT(build/unsigned)
777 : {
778 0 : std::unique_lock mon_data_lock(m_buffer_counts_mutex);
779 0 : if (m_buffer_counts.size() > 1000) {
780 0 : m_buffer_counts.pop_front();
781 : }
782 0 : m_buffer_counts.push_back(new_count);
783 0 : }
784 :
785 : double
786 0 : CIBModule::read_average_buffer_counts()
787 : {
788 0 : std::unique_lock mon_data_lock(m_buffer_counts_mutex);
789 :
790 0 : double total_counts;
791 0 : uint32_t number_of_counts; // NOLINT(build/unsigned)
792 :
793 0 : total_counts = 0;
794 0 : number_of_counts = m_buffer_counts.size();
795 :
796 0 : if (number_of_counts) {
797 0 : for (uint i = 0; i < number_of_counts; ++i) { // NOLINT(build/unsigned)
798 0 : total_counts = total_counts + m_buffer_counts.at(i);
799 : }
800 0 : return total_counts / number_of_counts;
801 : } else {
802 : return 0;
803 : }
804 0 : }
805 :
806 : void
807 0 : CIBModule::generate_opmon_data()
808 : {
809 0 : dunedaq::cibmodules::opmon::CIBModuleInfo module_info;
810 :
811 0 : module_info.set_num_control_messages_sent(m_num_control_messages_sent.load());
812 0 : module_info.set_num_control_responses_received(m_num_control_responses_received.load());
813 0 : module_info.set_hardware_running(m_is_running.load());
814 0 : module_info.set_hardware_configured(m_is_configured.load());
815 0 : module_info.set_num_total_triggers_received(m_num_total_triggers_received.load());
816 0 : module_info.set_num_run_triggers_received(m_num_run_triggers_received.load());
817 :
818 : // -- need to define these counters (and set the code to update them
819 0 : module_info.set_sent_hsi_events_counter(m_sent_counter.load());
820 0 : module_info.set_failed_to_send_hsi_events_counter(m_failed_to_send_counter.load());
821 : // module_info.set_last_sent_timestamp(m_last_sent_timestamp.load());
822 0 : module_info.set_average_buffer_occupancy(read_average_buffer_counts());
823 :
824 0 : publish(std::move(module_info));
825 :
826 : // should we also publish specific trigger info?
827 : // doesn't seem necessary at this time
828 0 : }
829 :
830 : bool
831 0 : CIBModule::check_port_in_use(unsigned short port)
832 : {
833 0 : using namespace boost::asio;
834 0 : using ip::tcp;
835 :
836 0 : io_service svc;
837 0 : tcp::acceptor a(svc);
838 :
839 0 : boost::system::error_code ec;
840 0 : a.open(tcp::v4(), ec) || a.bind({ tcp::v4(), port }, ec);
841 0 : a.close();
842 0 : return ec == error::address_in_use;
843 0 : }
844 : } // namespace dunedaq::cibmodules
845 :
846 0 : DEFINE_DUNE_DAQ_MODULE(dunedaq::cibmodules::CIBModule)
|
