Line data Source code
1 : /**
2 : * @file IfaceWrapper.cpp DPDK based Interface wrapper
3 : *
4 : * This is part of the DUNE DAQ , copyright 2020.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 : #include "datahandlinglibs/DataHandlingIssues.hpp"
9 : #include "logging/Logging.hpp"
10 :
11 : #include "opmonlib/Utils.hpp"
12 :
13 : #include "dpdklibs/Issues.hpp"
14 :
15 : #include "dpdklibs/nicreader/Structs.hpp"
16 :
17 : #include "IfaceWrapper.hpp"
18 : #include "dpdklibs/EALSetup.hpp"
19 : #include "dpdklibs/FlowControl.hpp"
20 : #include "dpdklibs/arp/ARP.hpp"
21 : #include "dpdklibs/ipv4_addr.hpp"
22 : #include "dpdklibs/udp/PacketCtor.hpp"
23 : #include "dpdklibs/udp/Utils.hpp"
24 :
25 : #include "appfwk/ConfigurationManager.hpp"
26 : // #include "confmodel/DROStreamConf.hpp"
27 : // #include "confmodel/StreamParameters.hpp"
28 : #include "appmodel/DPDKPortConfiguration.hpp"
29 : #include "confmodel/DetectorStream.hpp"
30 : #include "confmodel/GeoId.hpp"
31 : #include "confmodel/HostCores.hpp"
32 : // #include "confmodel/NetworkDevice.hpp"
33 : // #include "appmodel/NICInterfaceConfiguration.hpp"
34 : // #include "appmodel/NICStatsConf.hpp"
35 : // #include "appmodel/EthStreamParameters.hpp"
36 :
37 : #include "dpdklibs/opmon/IfaceWrapper.pb.h"
38 :
39 : #include <chrono>
40 : #include <format>
41 : #include <memory>
42 : #include <regex>
43 : #include <stdexcept>
44 : #include <string>
45 :
46 : /**
47 : * @brief TRACE debug levels used in this source file
48 : */
49 : enum
50 : {
51 : TLVL_ENTER_EXIT_METHODS = 5,
52 : TLVL_WORK_STEPS = 10,
53 : TLVL_BOOKKEEPING = 15
54 : };
55 :
56 : namespace dunedaq {
57 : namespace dpdklibs {
58 :
59 : //-----------------------------------------------------------------------------
60 : // TODO: the constructor signature shall be reviewd.
61 : // The current constructor takes a set of largely correlated arguments
62 : // - a receiver object
63 : // - the list of active senders
64 : // - the list of active streams
65 : //
66 : // These arguments are created by applying the enable mask to detector2daq connection object.
67 : // They are preferred to the d2d object not to expose the IfaceWrapper code to the System class
68 0 : IfaceWrapper::IfaceWrapper(uint iface_id,
69 : const appmodel::DPDKReceiver* receiver,
70 : const std::vector<const appmodel::NWDetDataSender*>& nw_senders,
71 : const std::vector<const confmodel::DetectorStream*>& active_streams,
72 : sid_to_source_map_t& sources,
73 0 : std::atomic<bool>& run_marker)
74 0 : : m_sources(sources)
75 0 : , m_run_marker(run_marker)
76 : {
77 :
78 : // Arguments consistency check: collect source ids in senders
79 0 : std::set<int> src_in_d2d;
80 :
81 0 : for (auto& det_stream : active_streams) {
82 0 : src_in_d2d.insert(det_stream->get_source_id());
83 : }
84 :
85 : // Arguments consistency check: collect source ids in source model map
86 0 : std::set<int> src_models;
87 0 : for (const auto& [src_id, _] : m_sources) {
88 0 : src_models.insert(src_id);
89 : }
90 :
91 : // check that the 2 sets are identical.
92 0 : if (!std::includes(src_models.begin(), src_models.end(), src_in_d2d.begin(), src_in_d2d.end())) {
93 :
94 : // TODO: remove, possibly
95 0 : for (auto src : src_models)
96 0 : TLOG_DEBUG(TLVL_BOOKKEEPING) << "model srcid " << src;
97 0 : for (auto src : src_in_d2d)
98 0 : TLOG_DEBUG(TLVL_BOOKKEEPING) << "d2d srcid " << src;
99 :
100 : // D2D sources are not included in the source model list
101 : // Extract the differences: src_in_d2d - src_models
102 :
103 0 : std::vector<int> src_missing;
104 0 : std::set_difference(
105 : src_models.begin(), src_models.end(), src_in_d2d.begin(), src_in_d2d.end(), std::back_inserter(src_missing));
106 :
107 0 : std::stringstream ss;
108 0 : for (int src : src_missing) {
109 0 : ss << src << " ";
110 : }
111 :
112 : // TLOG() << std::format("WARNING : these source ids are present in the d2d connection but no corresponding source
113 : // objects are found {}", ss.str());
114 0 : throw MissingSourceIDOutputs(ERS_HERE, m_iface_id, ss.str());
115 0 : }
116 :
117 0 : auto net_device = receiver->get_uses();
118 :
119 0 : m_iface_id = iface_id;
120 0 : m_mac_addr = net_device->get_mac_address();
121 0 : m_ip_addr = net_device->get_ip_address();
122 :
123 0 : TLOG() << "Building IfaceWrapper " << m_iface_id;
124 0 : std::stringstream s;
125 0 : s << 'IfaceWrapper (port ' << m_iface_id << ") responding to : ";
126 0 : for (const std::string& ip_addr : m_ip_addr) {
127 0 : s << ip_addr << " ";
128 : }
129 :
130 0 : TLOG() << s.str();
131 :
132 0 : for (const std::string& ip_addr : m_ip_addr) {
133 0 : IpAddr ip_addr_struct(ip_addr);
134 0 : m_ip_addr_bin.push_back(udp::ip_address_dotdecimal_to_binary(ip_addr_struct.addr_bytes[0],
135 0 : ip_addr_struct.addr_bytes[1],
136 0 : ip_addr_struct.addr_bytes[2],
137 0 : ip_addr_struct.addr_bytes[3]));
138 : }
139 :
140 0 : auto iface_cfg = receiver->get_configuration();
141 :
142 0 : m_with_flow = iface_cfg->get_flow_control();
143 0 : m_prom_mode = iface_cfg->get_promiscuous_mode();
144 0 : ;
145 0 : m_mtu = iface_cfg->get_mtu();
146 0 : m_max_block_words = unsigned(m_mtu) / sizeof(uint64_t);
147 0 : m_rx_ring_size = iface_cfg->get_rx_ring_size();
148 0 : m_tx_ring_size = iface_cfg->get_tx_ring_size();
149 0 : m_num_mbufs = iface_cfg->get_num_bufs();
150 0 : m_burst_size = iface_cfg->get_burst_size();
151 0 : m_mbuf_cache_size = iface_cfg->get_mbuf_cache_size();
152 :
153 0 : m_lcore_sleep_ns = iface_cfg->get_lcore_sleep_us() * 1000;
154 0 : m_socket_id = rte_eth_dev_socket_id(m_iface_id);
155 :
156 0 : m_iface_id_str = iface_cfg->UID();
157 :
158 : // Here is my list of cores
159 0 : for (const auto* proc_res : iface_cfg->get_used_lcores()) {
160 0 : m_rte_cores.insert(m_rte_cores.end(), proc_res->get_cpu_cores().begin(), proc_res->get_cpu_cores().end());
161 : }
162 0 : if (std::find(m_rte_cores.begin(), m_rte_cores.end(), rte_get_main_lcore()) != m_rte_cores.end()) {
163 0 : TLOG() << "ERROR! Throw ERS error here that LCore=0 should not be used, as it's a control RTE core!";
164 0 : throw std::runtime_error(
165 0 : std::string("ERROR! Throw ERS here that LCore=0 should not be used, as it's a control RTE core!"));
166 : }
167 :
168 : // iterate through active streams
169 :
170 : // Create a map of sender ni (ip) to streams from the d2d connection object
171 0 : std::map<std::string, std::map<uint, uint>> ip_to_stream_src_groups;
172 :
173 0 : for (auto nw_sender : nw_senders) {
174 0 : auto sender_ni = nw_sender->get_uses();
175 :
176 0 : std::string tx_ip = sender_ni->get_ip_address().at(0);
177 :
178 : // Loop over streams
179 0 : for (auto det_stream : nw_sender->get_streams()) {
180 :
181 : // Only include active streams
182 0 : if (std::find(active_streams.begin(), active_streams.end(), det_stream) == active_streams.end())
183 0 : continue;
184 :
185 0 : uint32_t tx_geo_stream_id = det_stream->get_geo_id()->get_stream_id();
186 : // (tx, geo_stream) -> source_id
187 0 : ip_to_stream_src_groups[tx_ip][tx_geo_stream_id] = det_stream->get_source_id();
188 : }
189 0 : }
190 :
191 : // RS FIXME: Is this RX_Q bump is enough??? I don't remember how the RX_Qs are assigned...
192 0 : uint32_t core_idx(0), rx_q(0); // RS FIXME: Ensure that no RX_Q=0 is used for UDP RX, ever.
193 :
194 0 : m_rx_qs.insert(rx_q);
195 0 : m_arp_rx_queue = rx_q;
196 0 : ++rx_q;
197 :
198 : // Build additional helper maps
199 0 : for (const auto& [tx_ip, strm_src] : ip_to_stream_src_groups) {
200 0 : m_ips.insert(tx_ip);
201 0 : m_rx_qs.insert(rx_q);
202 0 : m_num_frames_rxq[rx_q] = { 0 };
203 0 : m_num_bytes_rxq[rx_q] = { 0 };
204 :
205 0 : m_rx_core_map[m_rte_cores[core_idx]][rx_q] = tx_ip;
206 0 : m_stream_id_to_source_id[rx_q] = strm_src;
207 : // TLOG() << "+++ ip, rx_q : (" << tx_ip << ", " << rx_q << ") -> " << strm_src;
208 :
209 0 : ++rx_q;
210 0 : if (++core_idx == m_rte_cores.size()) {
211 0 : core_idx = 0;
212 : }
213 : }
214 :
215 : // Log mapping
216 0 : for (auto const& [lcore, rx_qs] : m_rx_core_map) {
217 0 : TLOG() << "Lcore=" << lcore << " handles: ";
218 0 : for (auto const& [rx_q, src_ip] : rx_qs) {
219 0 : TLOG() << " rx_q=" << rx_q << " src_ip=" << src_ip;
220 : }
221 : }
222 :
223 : // Adding single TX queue for ARP responses
224 0 : TLOG() << "Append TX_Q=0 for ARP responses.";
225 0 : m_tx_qs.insert(0);
226 :
227 : // Strict parsing (DAQ protocol) or pass through of UDP payloads to SourceModels
228 0 : for (auto const& [sid, src_concept] : m_sources) {
229 0 : if (!src_concept->m_daq_protocol_ensured) {
230 0 : m_strict_parsing = false;
231 : }
232 : }
233 0 : }
234 :
235 : //-----------------------------------------------------------------------------
236 0 : IfaceWrapper::~IfaceWrapper()
237 : {
238 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "IfaceWrapper destructor called. First stop check, then closing iface.";
239 :
240 0 : struct rte_flow_error error;
241 0 : rte_flow_flush(m_iface_id, &error);
242 : // graceful_stop();
243 : // close_iface();
244 0 : TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "IfaceWrapper destroyed.";
245 0 : }
246 :
247 : //-----------------------------------------------------------------------------
248 : void
249 0 : IfaceWrapper::allocate_mbufs()
250 : {
251 0 : TLOG() << "Allocating pools and mbufs for UDP, GARP, and ARP.";
252 :
253 : // Pools for UDP RX messages
254 0 : for (size_t i = 0; i < m_rx_qs.size(); ++i) {
255 0 : std::stringstream bufss;
256 0 : bufss << "MBP-" << m_iface_id << '-' << i;
257 0 : TLOG() << "Acquire pool with name=" << bufss.str() << " for iface_id=" << m_iface_id << " rxq=" << i;
258 0 : m_mbuf_pools[i] = ealutils::get_mempool(bufss.str(), m_num_mbufs, m_mbuf_cache_size, 16384, m_socket_id);
259 0 : m_bufs[i] = (rte_mbuf**)malloc(sizeof(struct rte_mbuf*) * m_burst_size);
260 : // No need to alloc?
261 : // rte_pktmbuf_alloc_bulk(m_mbuf_pools[i].get(), m_bufs[i], m_burst_size);
262 0 : }
263 :
264 : // Pools for GARP messages
265 0 : std::stringstream garpss;
266 0 : garpss << "GARPMBP-" << m_iface_id;
267 0 : TLOG() << "Acquire GARP pool with name=" << garpss.str() << " for iface_id=" << m_iface_id;
268 0 : m_garp_mbuf_pool = ealutils::get_mempool(garpss.str());
269 0 : m_garp_bufs[0] = (rte_mbuf**)malloc(sizeof(struct rte_mbuf*) * m_burst_size);
270 0 : rte_pktmbuf_alloc_bulk(m_garp_mbuf_pool.get(), m_garp_bufs[0], m_burst_size);
271 :
272 : // Pools for ARP request/responses
273 0 : std::stringstream arpss;
274 0 : arpss << "ARPMBP-" << m_iface_id;
275 0 : TLOG() << "Acquire ARP pool with name=" << arpss.str() << " for iface_id=" << m_iface_id;
276 0 : m_arp_mbuf_pool = ealutils::get_mempool(arpss.str());
277 0 : m_arp_bufs[0] = (rte_mbuf**)malloc(sizeof(struct rte_mbuf*) * m_burst_size);
278 0 : rte_pktmbuf_alloc_bulk(m_arp_mbuf_pool.get(), m_arp_bufs[0], m_burst_size);
279 0 : }
280 :
281 : //-----------------------------------------------------------------------------
282 : void
283 0 : IfaceWrapper::setup_interface()
284 : {
285 0 : TLOG() << "Initialize interface " << m_iface_id;
286 0 : bool with_reset = false, with_mq_mode = true; // go to config
287 0 : bool check_link_status = false;
288 :
289 0 : int retval = ealutils::iface_init(m_iface_id,
290 0 : m_rx_qs.size(),
291 0 : m_tx_qs.size(),
292 0 : m_rx_ring_size,
293 0 : m_tx_ring_size,
294 0 : m_mbuf_pools,
295 : with_reset,
296 : with_mq_mode,
297 : check_link_status);
298 0 : if (retval != 0) {
299 0 : throw FailedToSetupInterface(ERS_HERE, m_iface_id, retval);
300 : }
301 : // Promiscuous mode
302 0 : ealutils::iface_promiscuous_mode(m_iface_id, m_prom_mode); // should come from config
303 0 : }
304 :
305 : //-----------------------------------------------------------------------------
306 : void
307 0 : IfaceWrapper::setup_flow_steering()
308 : {
309 : // Flow steering setup
310 0 : TLOG() << "Configuring Flow steering rules for iface=" << m_iface_id;
311 0 : struct rte_flow_error error;
312 0 : struct rte_flow* flow;
313 0 : TLOG() << "Attempt to flush previous flow rules...";
314 0 : rte_flow_flush(m_iface_id, &error);
315 : #warning RS: FIXME -> Check for flow flush return!
316 :
317 0 : TLOG() << "Create control flow rules (ARP) assinged to rxq=" << m_arp_rx_queue;
318 0 : flow = generate_arp_flow(m_iface_id, m_arp_rx_queue, &error);
319 0 : if (not flow) { // ers::fatal
320 0 : TLOG() << "ARP flow can't be created for " << m_arp_rx_queue << " Error type: " << (unsigned)error.type
321 0 : << " Message: '" << error.message << "'";
322 0 : ers::fatal(dunedaq::datahandlinglibs::InitializationError(ERS_HERE, "Couldn't create ARP flow API rules!"));
323 0 : rte_exit(EXIT_FAILURE, "error in creating ARP flow");
324 : }
325 :
326 0 : TLOG() << "Create flow rules for UDP RX.";
327 0 : for (auto const& [lcoreid, rxqs] : m_rx_core_map) {
328 0 : for (auto const& [rxqid, srcip] : rxqs) {
329 : // Put the IP numbers temporarily in a vector, so they can be converted easily to uint32_t
330 0 : TLOG() << "Creating flow rule for src_ip=" << srcip << " assigned to rxq=" << rxqid;
331 0 : size_t ind = 0, current_ind = 0;
332 0 : std::vector<uint8_t> v;
333 0 : for (int i = 0; i < 4; ++i) {
334 0 : v.push_back(std::stoi(srcip.substr(current_ind, srcip.size() - current_ind), &ind));
335 0 : current_ind += ind + 1;
336 : }
337 :
338 0 : flow = generate_ipv4_flow(m_iface_id, rxqid, RTE_IPV4(v[0], v[1], v[2], v[3]), 0xffffffff, 0, 0, &error);
339 :
340 0 : if (not flow) { // ers::fatal
341 0 : TLOG() << "Flow can't be created for " << rxqid << " Error type: " << (unsigned)error.type << " Message: '"
342 0 : << error.message << "'";
343 0 : ers::fatal(dunedaq::datahandlinglibs::InitializationError(ERS_HERE, "Couldn't create Flow API rules!"));
344 0 : rte_exit(EXIT_FAILURE, "error in creating flow");
345 : }
346 0 : }
347 : }
348 :
349 0 : return;
350 : }
351 :
352 : //-----------------------------------------------------------------------------
353 : void
354 0 : IfaceWrapper::setup_xstats()
355 : {
356 : // Stats setup
357 0 : m_iface_xstats.setup(m_iface_id);
358 0 : m_iface_xstats.reset_counters();
359 0 : }
360 :
361 : //-----------------------------------------------------------------------------
362 : void
363 0 : IfaceWrapper::stop_xstats()
364 : {
365 : // Stopping stats
366 0 : m_iface_xstats.reset_counters();
367 0 : m_iface_xstats.stop();
368 0 : }
369 :
370 : //-----------------------------------------------------------------------------
371 : void
372 0 : IfaceWrapper::start()
373 : {
374 : // Reset counters for RX queues
375 0 : for (auto const& [rx_q, _] : m_num_frames_rxq) {
376 0 : m_num_frames_rxq[rx_q] = { 0 };
377 0 : m_num_bytes_rxq[rx_q] = { 0 };
378 0 : m_num_full_bursts[rx_q] = { 0 };
379 0 : m_max_burst_size[rx_q] = { 0 };
380 : }
381 :
382 : // Reset counters for rte_workers
383 0 : for (auto const& [lcore, _] : m_rx_core_map) {
384 0 : m_num_unhandled_non_ipv4[lcore] = { 0 };
385 0 : m_num_unhandled_non_udp[lcore] = { 0 };
386 0 : m_num_unhandled_non_jumbo_udp[lcore] = { 0 };
387 : }
388 :
389 0 : m_lcore_enable_flow.store(false);
390 0 : m_lcore_quit_signal.store(false);
391 0 : TLOG() << "Interface id=" << m_iface_id << " Launching GARP thread with garp_func...";
392 0 : m_garp_thread = std::thread(&IfaceWrapper::garp_func, this);
393 :
394 0 : TLOG() << "Interface id=" << m_iface_id << " starting ARP LCore processor:";
395 0 : m_arp_thread = std::thread(&IfaceWrapper::IfaceWrapper::arp_response_runner, this, nullptr);
396 :
397 0 : TLOG() << "Interface id=" << m_iface_id << " starting LCore processors:";
398 0 : for (auto const& [lcoreid, _] : m_rx_core_map) {
399 0 : int ret = rte_eal_remote_launch((int (*)(void*))(&IfaceWrapper::rx_runner), this, lcoreid);
400 0 : TLOG() << " -> LCore[" << lcoreid << "] launched with return code=" << ret << " "
401 0 : << (ret < 0 ? rte_strerror(-ret) : "");
402 : }
403 0 : }
404 :
405 : //-----------------------------------------------------------------------------
406 : void
407 0 : IfaceWrapper::stop()
408 : {
409 0 : m_lcore_enable_flow.store(false);
410 0 : m_lcore_quit_signal.store(true);
411 : // Stop GARP sender thread
412 0 : if (m_garp_thread.joinable()) {
413 0 : m_garp_thread.join();
414 : } else {
415 0 : TLOG() << "GARP thread is not joinable!";
416 : }
417 :
418 0 : if (m_arp_thread.joinable()) {
419 0 : m_arp_thread.join();
420 : } else {
421 0 : TLOG() << "ARP thread is not joinable!";
422 : }
423 0 : }
424 : /*
425 : void
426 : IfaceWrapper::scrap()
427 : {
428 : struct rte_flow_error error;
429 : rte_flow_flush(m_iface_id, &error);
430 : }
431 : */
432 :
433 : //-----------------------------------------------------------------------------
434 : void
435 0 : IfaceWrapper::generate_opmon_data()
436 : {
437 :
438 0 : if (m_iface_xstats.m_enabled) {
439 : // Poll stats from HW
440 0 : m_iface_xstats.poll();
441 :
442 0 : opmon::EthStats s;
443 0 : s.set_ipackets(m_iface_xstats.m_eth_stats.ipackets);
444 0 : s.set_opackets(m_iface_xstats.m_eth_stats.opackets);
445 0 : s.set_ibytes(m_iface_xstats.m_eth_stats.ibytes);
446 0 : s.set_obytes(m_iface_xstats.m_eth_stats.obytes);
447 0 : s.set_imissed(m_iface_xstats.m_eth_stats.imissed);
448 0 : s.set_ierrors(m_iface_xstats.m_eth_stats.ierrors);
449 0 : s.set_oerrors(m_iface_xstats.m_eth_stats.oerrors);
450 0 : s.set_rx_nombuf(m_iface_xstats.m_eth_stats.rx_nombuf);
451 0 : publish(std::move(s));
452 :
453 0 : if (m_iface_xstats.m_eth_stats.imissed > 0) {
454 0 : ers::warning(PacketErrors(ERS_HERE, m_iface_id_str, "missed", m_iface_xstats.m_eth_stats.imissed));
455 : }
456 0 : if (m_iface_xstats.m_eth_stats.ierrors > 0) {
457 0 : ers::warning(PacketErrors(ERS_HERE, m_iface_id_str, "dropped", m_iface_xstats.m_eth_stats.ierrors));
458 : }
459 :
460 : // loop over all the xstats information
461 0 : opmon::EthXStatsInfo xinfos;
462 0 : opmon::EthXStatsErrors xerrs;
463 0 : std::map<std::string, opmon::QueueEthXStats> xq;
464 :
465 0 : for (int i = 0; i < m_iface_xstats.m_len; ++i) {
466 :
467 0 : std::string name(m_iface_xstats.m_xstats_names[i].name);
468 :
469 : // first we select the info from the queue
470 0 : static std::regex queue_regex(R"((rx|tx)_q(\d+)_([^_]+))");
471 0 : std::smatch match;
472 :
473 0 : if (std::regex_match(name, match, queue_regex)) {
474 0 : auto queue_name = match[1].str() + '-' + match[2].str();
475 0 : auto& entry = xq[queue_name];
476 0 : try {
477 0 : opmonlib::set_value(entry, match[3], m_iface_xstats.m_xstats_values[i]);
478 0 : } catch (const ers::Issue& e) {
479 0 : ers::warning(MetricPublishFailed(ERS_HERE, name, e));
480 0 : }
481 0 : continue;
482 0 : }
483 :
484 0 : google::protobuf::Message* metric_p = nullptr;
485 0 : static std::regex err_regex(R"(.+error.*)");
486 0 : if (std::regex_match(name, err_regex))
487 : metric_p = &xerrs;
488 : else
489 0 : metric_p = &xinfos;
490 :
491 0 : try {
492 0 : opmonlib::set_value(*metric_p, name, m_iface_xstats.m_xstats_values[i]);
493 0 : } catch (const ers::Issue& e) {
494 0 : ers::warning(MetricPublishFailed(ERS_HERE, name, e));
495 0 : }
496 :
497 0 : } // loop over xstats
498 :
499 : // Reset HW counters
500 0 : m_iface_xstats.reset_counters();
501 : // finally we publish the information
502 0 : publish(std::move(xinfos));
503 0 : publish(std::move(xerrs));
504 0 : for (auto [id, stat] : xq) {
505 0 : publish(std::move(stat), { { "queue", id } });
506 0 : }
507 0 : }
508 :
509 0 : for (const auto& [src_rx_q, _] : m_num_frames_rxq) {
510 0 : opmon::QueueInfo i;
511 0 : i.set_packets_received(m_num_frames_rxq[src_rx_q].load());
512 0 : i.set_bytes_received(m_num_bytes_rxq[src_rx_q].load());
513 0 : i.set_full_rx_burst(m_num_full_bursts[src_rx_q].load());
514 0 : i.set_max_burst_size(m_max_burst_size[src_rx_q].exchange(0));
515 :
516 0 : publish(std::move(i), { { "queue", std::to_string(src_rx_q) } });
517 0 : }
518 :
519 : // RTE Workers
520 0 : for (auto const& [lcore, _] : m_rx_core_map) {
521 0 : opmon::RTEWorkerInfo info;
522 0 : info.set_num_unhandled_non_ipv4(m_num_unhandled_non_ipv4[lcore].exchange(0));
523 0 : info.set_num_unhandled_non_udp(m_num_unhandled_non_udp[lcore].exchange(0));
524 0 : info.set_num_unhandled_non_jumbo_udp(m_num_unhandled_non_jumbo_udp[lcore].exchange(0));
525 0 : publish(std::move(info), { { "rte_worker_id", std::to_string(lcore) } });
526 0 : }
527 :
528 0 : for (auto& [id, counter] : m_num_unexid_frames) {
529 0 : auto val = counter.exchange(0);
530 0 : if (val > 0) {
531 0 : ers::warning(UnexpectedStreamID(ERS_HERE, id, val));
532 : }
533 : }
534 0 : }
535 :
536 : //-----------------------------------------------------------------------------
537 : void
538 0 : IfaceWrapper::garp_func()
539 : {
540 0 : TLOG() << "Launching GARP sender...";
541 0 : while (m_run_marker.load()) {
542 0 : for (const auto& ip_addr_bin : m_ip_addr_bin) {
543 0 : arp::pktgen_send_garp(m_garp_bufs[0][0], m_iface_id, ip_addr_bin);
544 : }
545 0 : ++m_garps_sent;
546 0 : std::this_thread::sleep_for(std::chrono::seconds(1));
547 : }
548 0 : TLOG() << "GARP function joins.";
549 0 : }
550 :
551 : //-----------------------------------------------------------------------------
552 : void
553 0 : IfaceWrapper::parse_udp_payload(int src_rx_q, char* payload, std::size_t size)
554 : {
555 : // Pointers for parsing and to the end of the UDP payload.
556 0 : char* plptr = payload;
557 0 : const char* plendptr = payload + size;
558 :
559 : // Process every DAQEth frame within UDP payload
560 0 : while (plptr + sizeof(dunedaq::detdataformats::DAQEthHeader) < plendptr) { // Scatter loop start
561 :
562 : // Reinterpret directly to DAQEthHeader
563 0 : auto daqhdrptr = reinterpret_cast<dunedaq::detdataformats::DAQEthHeader*>(plptr);
564 :
565 : // Check number of DAQEth block_words
566 0 : unsigned block_words = unsigned(daqhdrptr->block_length) - 1; // removing timestamp word from the block length.
567 :
568 : // Check for corrupted DAQEth frame length
569 0 : if (block_words == 0 || block_words > m_max_block_words) {
570 : // RS FIXME: corrupted length -> stop, add opmon counter or warning
571 0 : return;
572 : }
573 :
574 : // Calculate data bytes after DAQEthHeader based on block_words
575 0 : std::size_t data_bytes = std::size_t(block_words) * sizeof(dunedaq::detdataformats::DAQEthHeader::word_t);
576 :
577 : // Grab end pointer of DAQEth frame
578 0 : char* daqframe_endptr = plptr + sizeof(dunedaq::detdataformats::DAQEthHeader) + data_bytes;
579 :
580 : // Check if full DAQEth frame fits
581 0 : if (daqframe_endptr > plendptr) {
582 : // RS FIXME: truncated payload -> stop, add opmon counter or warning
583 : return;
584 : }
585 :
586 : // Calculate DAQEth frame size (used both for handling and advancing)
587 0 : std::size_t daq_frame_size = sizeof(dunedaq::detdataformats::DAQEthHeader) + data_bytes;
588 :
589 : // Sadly, cannot take a reference to a bitfield
590 0 : uint strm_id = daqhdrptr->stream_id;
591 :
592 : // Check that stream id is corresponds to a registered source
593 0 : auto& strm_to_src = m_stream_id_to_source_id[src_rx_q];
594 :
595 0 : if (auto strm_it = strm_to_src.find(strm_id); strm_it != strm_to_src.end()) {
596 :
597 0 : m_sources[strm_it->second]->handle_daq_frame((char*)daqhdrptr, daq_frame_size);
598 :
599 : } else {
600 : // Really bad -> unexpeced StreamID in UDP Payload.
601 : // This check is needed in order to avoid dynamically add thousands
602 : // of Sources on the fly, in case the data corruption is extremely severe.
603 0 : if (m_num_unexid_frames.count(strm_id) == 0) {
604 0 : m_num_unexid_frames[strm_id] = 0;
605 : }
606 0 : m_num_unexid_frames[strm_id]++;
607 : }
608 :
609 : // Advance to next payload
610 0 : plptr += daq_frame_size;
611 :
612 : } // Scatter loop end
613 : }
614 :
615 : //-----------------------------------------------------------------------------
616 : void
617 0 : IfaceWrapper::passthrough_udp_payload(int src_rx_q, char* payload, std::size_t size)
618 : {
619 : // Get DAQ Header and its StreamID
620 0 : auto* daqhdrptr = reinterpret_cast<dunedaq::detdataformats::DAQEthHeader*>(payload);
621 :
622 : // Sadly, cannot take a reference to a bitfield
623 0 : uint strm_id = daqhdrptr->stream_id;
624 :
625 : // Check that stream id is corresponds to a registered source
626 0 : auto& strm_to_src = m_stream_id_to_source_id[src_rx_q];
627 :
628 0 : if (auto strm_it = strm_to_src.find(strm_id); strm_it != strm_to_src.end()) {
629 0 : m_sources[strm_it->second]->handle_daq_frame(payload, size);
630 : } else {
631 : // Really bad -> unexpeced StreamID in UDP Payload.
632 : // This check is needed in order to avoid dynamically add thousands
633 : // of Sources on the fly, in case the data corruption is extremely severe.
634 0 : if (m_num_unexid_frames.count(strm_id) == 0) {
635 0 : m_num_unexid_frames[strm_id] = 0;
636 : }
637 0 : m_num_unexid_frames[strm_id]++;
638 : }
639 0 : }
640 :
641 : } // namespace dpdklibs
642 : } // namespace dunedaq
643 :
644 : //
645 : #include "detail/IfaceWrapper.hxx"
|
