Line data Source code
1 :
2 : /* Application will run until quit or killed. */
3 :
4 : #include <inttypes.h>
5 : #include <rte_cycles.h>
6 : #include <rte_eal.h>
7 : #include <rte_ethdev.h>
8 : #include <rte_lcore.h>
9 : #include <rte_mbuf.h>
10 : #include <stdint.h>
11 : #include <tuple>
12 :
13 : #include <csignal>
14 : #include <fstream>
15 : #include <iomanip>
16 : #include <limits>
17 : #include <sstream>
18 :
19 : #include "CLI/App.hpp"
20 : #include "CLI/Config.hpp"
21 : #include "CLI/Formatter.hpp"
22 :
23 : #include "detdataformats/DAQEthHeader.hpp"
24 : #include "dpdklibs/EALSetup.hpp"
25 : #include "dpdklibs/udp/PacketCtor.hpp"
26 : #include "dpdklibs/udp/Utils.hpp"
27 : #include "logging/Logging.hpp"
28 :
29 : #include "dpdklibs/arp/ARP.hpp"
30 : #include "dpdklibs/ipv4_addr.hpp"
31 : #include "dpdklibs/udp/PacketCtor.hpp"
32 : #include "dpdklibs/udp/Utils.hpp"
33 :
34 : #include <fmt/core.h>
35 : #include <fmt/ranges.h>
36 :
37 : #include <regex>
38 :
39 : #define RX_RING_SIZE 1024
40 : #define TX_RING_SIZE 1024
41 :
42 : #define NUM_MBUFS 8191
43 : #define MBUF_CACHE_SIZE 250
44 :
45 : #define NSTREAM 128
46 :
47 : #define PG_JUMBO_FRAME_LEN (9600 + RTE_ETHER_CRC_LEN + RTE_ETHER_HDR_LEN)
48 : #ifndef RTE_JUMBO_ETHER_MTU
49 : #define RTE_JUMBO_ETHER_MTU (PG_JUMBO_FRAME_LEN - RTE_ETHER_HDR_LEN - RTE_ETHER_CRC_LEN) /*< Ethernet MTU. */
50 : #endif
51 :
52 : using namespace dunedaq;
53 : using namespace dpdklibs;
54 : using namespace udp;
55 :
56 : namespace {
57 :
58 : std::ostream&
59 : operator<<(std::ostream& out, const StreamUID& obj)
60 : {
61 : return out << static_cast<std::string>(obj);
62 : }
63 :
64 : struct StreamStats
65 : {
66 : std::atomic<uint64_t> total_packets = 0;
67 : std::atomic<uint64_t> num_packets = 0;
68 : std::atomic<uint64_t> num_bytes = 0;
69 : std::atomic<uint64_t> num_bad_timestamp = 0;
70 : std::atomic<uint64_t> max_timestamp_skip = 0;
71 : std::atomic<uint64_t> num_bad_seq_id = 0;
72 : std::atomic<uint64_t> max_seq_id_skip = 0;
73 : std::atomic<uint64_t> num_bad_payload_size = 0;
74 : std::atomic<uint64_t> min_payload_size = 0;
75 : std::atomic<uint64_t> max_payload_size = 0;
76 : std::atomic<uint64_t> prev_seq_id = 0;
77 : std::atomic<uint64_t> prev_timestamp = 0;
78 :
79 0 : void reset()
80 : {
81 0 : num_packets.exchange(0);
82 0 : num_bytes.exchange(0);
83 0 : num_bad_timestamp.exchange(0);
84 0 : max_timestamp_skip.exchange(0);
85 0 : num_bad_seq_id.exchange(0);
86 0 : max_seq_id_skip.exchange(0);
87 0 : num_bad_payload_size.exchange(0);
88 0 : min_payload_size.exchange(0);
89 0 : max_payload_size.exchange(0);
90 0 : }
91 : };
92 :
93 : std::map<StreamUID, StreamStats> stream_stats;
94 :
95 : // Apparently only 8 and above works for "burst_size"
96 :
97 : // From the dpdk documentation, describing the rte_eth_rx_burst
98 : // function (and keeping in mind that their "nb_pkts" variable is the
99 : // same as our "burst size" variable below):
100 : // "Some drivers using vector instructions require that nb_pkts is
101 : // divisible by 4 or 8, depending on the driver implementation."
102 :
103 : bool check_timestamp = false;
104 : bool per_stream_reports = false;
105 :
106 : constexpr int burst_size = 256;
107 :
108 : constexpr uint32_t expected_packet_type = 0x291;
109 :
110 : constexpr int default_mbuf_size = 9000; // As opposed to RTE_MBUF_DEFAULT_BUF_SIZE
111 :
112 : constexpr int max_packets_to_dump = 10;
113 : int dumped_packet_count = 0;
114 :
115 : uint16_t expected_packet_size = 0; // 7243; // i.e., every packet that isn't the initial one
116 :
117 : std::atomic<uint64_t> num_packets = 0;
118 : std::atomic<uint64_t> num_bytes = 0;
119 : std::atomic<uint64_t> total_packets = 0;
120 : std::atomic<uint64_t> non_ipv4_packets = 0;
121 : std::atomic<uint64_t> total_bad_timestamp = 0;
122 : std::atomic<uint64_t> num_bad_timestamp = 0;
123 : std::atomic<uint64_t> max_timestamp_skip = 0;
124 : std::atomic<uint64_t> total_bad_seq_id = 0;
125 : std::atomic<uint64_t> num_bad_seq_id = 0;
126 : std::atomic<uint64_t> max_seq_id_skip = 0;
127 : std::atomic<uint64_t> total_bad_payload_size = 0;
128 : std::atomic<uint64_t> num_bad_payload_size = 0;
129 : std::atomic<uint64_t> min_payload_size = 0;
130 : std::atomic<uint64_t> max_payload_size = 0;
131 : std::atomic<uint64_t> udp_pkt_counter = 0;
132 :
133 : std::atomic<int64_t> garps_sent = 0;
134 :
135 : // std::ofstream datafile;
136 : // const std::string output_data_filename = "dpdklibs_test_frame_receiver.dat";
137 :
138 : } // namespace
139 :
140 : std::vector<char*>
141 0 : construct_argv(std::vector<std::string>& std_argv)
142 : {
143 0 : std::vector<char*> vec_argv;
144 0 : for (int i = 0; i < std_argv.size(); i++) {
145 0 : vec_argv.insert(vec_argv.end(), std_argv[i].data());
146 : }
147 0 : return vec_argv;
148 0 : }
149 :
150 : static inline int
151 0 : check_against_previous_stream(const detdataformats::DAQEthHeader* daq_hdr, uint64_t exp_ts_diff)
152 : {
153 : // uint64_t unique_str_id = (daq_hdr->det_id<<22) + (daq_hdr->crate_id<<12) + (daq_hdr->slot_id<<8) +
154 : // daq_hdr->stream_id;
155 0 : StreamUID unique_str_id(*daq_hdr);
156 0 : uint64_t stream_ts = daq_hdr->timestamp;
157 0 : uint64_t seq_id = daq_hdr->seq_id;
158 0 : int ret_val = 0;
159 :
160 0 : if (check_timestamp) {
161 0 : if (stream_stats[unique_str_id].prev_timestamp == 0) {
162 0 : stream_stats[unique_str_id].prev_timestamp = stream_ts;
163 : } else {
164 0 : uint64_t expected_ts = stream_stats[unique_str_id].prev_timestamp + exp_ts_diff;
165 0 : if (stream_ts != expected_ts) {
166 0 : uint64_t ts_difference = stream_ts - stream_stats[unique_str_id].prev_timestamp;
167 0 : ret_val = 1;
168 0 : ++num_bad_timestamp;
169 0 : ++stream_stats[unique_str_id].num_bad_timestamp;
170 0 : ++total_bad_timestamp;
171 0 : if (ts_difference > max_timestamp_skip) {
172 0 : max_timestamp_skip = ts_difference;
173 : }
174 0 : if (ts_difference > stream_stats[unique_str_id].max_timestamp_skip) {
175 0 : stream_stats[unique_str_id].max_timestamp_skip = ts_difference;
176 : }
177 : }
178 0 : stream_stats[unique_str_id].prev_timestamp = stream_ts;
179 : }
180 : }
181 :
182 0 : uint64_t expected_seq_id =
183 0 : (stream_stats[unique_str_id].prev_seq_id == 4095) ? 0 : stream_stats[unique_str_id].prev_seq_id + 1;
184 0 : if (seq_id != expected_seq_id) {
185 0 : uint64_t adj_expected_seq_id = (expected_seq_id == 0) ? 4096 : expected_seq_id;
186 0 : uint64_t adj_seq_id = (seq_id < adj_expected_seq_id) ? (4096 + seq_id) : seq_id;
187 0 : uint64_t seq_id_difference = adj_seq_id - adj_expected_seq_id;
188 0 : ret_val += 2;
189 0 : ++num_bad_seq_id;
190 0 : ++total_bad_seq_id;
191 0 : ++stream_stats[unique_str_id].num_bad_seq_id;
192 0 : if (seq_id_difference > max_seq_id_skip) {
193 0 : max_seq_id_skip = seq_id_difference;
194 : }
195 0 : if (seq_id_difference > stream_stats[unique_str_id].max_seq_id_skip) {
196 0 : stream_stats[unique_str_id].max_seq_id_skip = seq_id_difference;
197 : }
198 : }
199 :
200 0 : return ret_val;
201 : }
202 :
203 : static inline int
204 0 : check_packet_size(struct rte_mbuf* mbuf, StreamUID unique_str_id)
205 : {
206 0 : std::size_t packet_size = mbuf->data_len;
207 :
208 0 : if (packet_size > max_payload_size) {
209 0 : max_payload_size = packet_size;
210 : }
211 0 : if (packet_size < min_payload_size) {
212 0 : min_payload_size = packet_size;
213 : }
214 0 : if (packet_size > stream_stats[unique_str_id].max_payload_size) {
215 0 : stream_stats[unique_str_id].max_payload_size = packet_size;
216 : }
217 0 : if (packet_size < stream_stats[unique_str_id].min_payload_size) {
218 0 : stream_stats[unique_str_id].min_payload_size = packet_size;
219 : }
220 :
221 0 : if (expected_packet_size and (packet_size != expected_packet_size)) {
222 0 : ++num_bad_payload_size;
223 0 : ++total_bad_payload_size;
224 0 : ++stream_stats[unique_str_id].num_bad_payload_size;
225 0 : return 1;
226 : }
227 : return 0;
228 : }
229 :
230 : static int
231 0 : lcore_main(struct rte_mempool* mbuf_pool,
232 : uint16_t iface,
233 : uint64_t time_per_report,
234 : const std::vector<std::string>& garp_ip_addr_strs)
235 : {
236 : /*
237 : * Check that the iface is on the same NUMA node as the polling thread
238 : * for best performance.
239 : */
240 :
241 0 : if (rte_eth_dev_socket_id(iface) >= 0 && rte_eth_dev_socket_id(iface) != static_cast<int>(rte_socket_id())) {
242 0 : fmt::print("WARNING, iface {} is on remote NUMA node to polling thread.\n\tPerformance will not be optimal.\n",
243 : iface);
244 : }
245 :
246 0 : auto stats = std::thread([&]() {
247 0 : while (true) {
248 0 : uint64_t packets_per_second = num_packets / time_per_report;
249 0 : uint64_t bytes_per_second = num_bytes / time_per_report;
250 0 : fmt::print("Since the last report {} seconds ago:\n"
251 : "Packets/s: {} Bytes/s: {} Total packets: {} Non-IPV4 packets: {} Total UDP packets: {}\n"
252 : "Packets with wrong sequence id: {}, Max wrong seq_id jump {}, Total Packets with Wrong seq_id {}\n"
253 : "Max udp payload: {}, Min udp payload: {}\n",
254 : time_per_report,
255 : packets_per_second,
256 : bytes_per_second,
257 : total_packets,
258 : non_ipv4_packets,
259 : udp_pkt_counter,
260 : num_bad_seq_id,
261 : max_seq_id_skip,
262 : total_bad_seq_id,
263 : max_payload_size,
264 : min_payload_size);
265 :
266 0 : if (expected_packet_size) {
267 0 : fmt::print("Packets with wrong payload size: {}, Total Packets with Wrong size {}\n",
268 : num_bad_payload_size,
269 : total_bad_payload_size);
270 : }
271 :
272 0 : if (check_timestamp) {
273 0 : fmt::print("Wrong Timestamp difference Packets: {}, Max wrong Timestamp difference {}, Total Packets with "
274 : "Wrong Timestamp {}\n",
275 : num_bad_timestamp,
276 : max_timestamp_skip,
277 : total_bad_timestamp);
278 : }
279 :
280 0 : fmt::print("\n");
281 : // for (auto stream = stream_stats.begin(); stream != stream_stats.end(); stream++) {
282 0 : for (auto& [suid, stats] : stream_stats) {
283 0 : fmt::print("Stream {:15}: n.pkts {} (tot. {})", (std::string)suid, stats.num_packets, stats.total_packets);
284 0 : if (per_stream_reports) {
285 0 : float stream_bytes_per_second = (float)stats.num_bytes / (1024. * 1024.) / (float)time_per_report;
286 0 : fmt::print(" {:8.3f} MB/s, seq. jumps: {}", stream_bytes_per_second, stats.num_bad_seq_id);
287 : }
288 0 : stats.reset();
289 0 : fmt::print("\n");
290 : }
291 :
292 0 : fmt::print("\n");
293 0 : num_packets.exchange(0);
294 0 : num_bytes.exchange(0);
295 0 : num_bad_timestamp.exchange(0);
296 0 : num_bad_seq_id.exchange(0);
297 0 : num_bad_payload_size.exchange(0);
298 0 : max_payload_size.exchange(0);
299 0 : min_payload_size.exchange(0);
300 0 : max_seq_id_skip.exchange(0);
301 0 : max_timestamp_skip.exchange(0);
302 0 : std::this_thread::sleep_for(std::chrono::seconds(time_per_report));
303 0 : }
304 0 : });
305 :
306 : //
307 : // Prepare and start GARP thread
308 : //
309 0 : std::vector<rte_be32_t> ip_addr_bin_vector;
310 0 : for (const auto& ip_addr_str : garp_ip_addr_strs) {
311 0 : TLOG() << "IP address for ARP responses: " << ip_addr_str;
312 0 : IpAddr ip_addr(ip_addr_str);
313 0 : rte_be32_t ip_addr_bin = ip_address_dotdecimal_to_binary(
314 0 : ip_addr.addr_bytes[0], ip_addr.addr_bytes[1], ip_addr.addr_bytes[2], ip_addr.addr_bytes[3]);
315 0 : ip_addr_bin_vector.push_back(ip_addr_bin);
316 : }
317 :
318 0 : struct rte_mbuf** tx_bufs = (rte_mbuf**)malloc(sizeof(struct rte_mbuf*) * burst_size);
319 0 : rte_pktmbuf_alloc_bulk(mbuf_pool, tx_bufs, burst_size);
320 :
321 0 : auto garp = std::thread([&]() {
322 0 : while (true) {
323 : // TLOG() << "Packets/s: " << num_packets << " Bytes/s: " << num_bytes << " Total packets: " << total_packets << "
324 : // Failed packets: " << failed_packets; num_packets.exchange(0); num_bytes.exchange(0);
325 :
326 0 : for (const auto& ip_addr_bin : ip_addr_bin_vector) {
327 0 : arp::pktgen_send_garp(tx_bufs[0], iface, ip_addr_bin);
328 0 : ++garps_sent;
329 : }
330 :
331 0 : std::this_thread::sleep_for(
332 0 : std::chrono::seconds(1)); // If we sample for anything other than 1s, the rate calculation will need to change
333 0 : }
334 0 : });
335 :
336 0 : struct rte_mbuf** rx_bufs = (rte_mbuf**)malloc(sizeof(struct rte_mbuf*) * burst_size);
337 0 : rte_pktmbuf_alloc_bulk(mbuf_pool, rx_bufs, burst_size);
338 :
339 : // datafile.open(output_data_filename, std::ios::out | std::ios::binary);
340 : // if ( (datafile.rdstate() & std::ofstream::failbit ) != 0 ) {
341 : // fmt::print("WARNING: Unable to open output file \"{}\"\n", output_data_filename);
342 : // }
343 :
344 0 : while (true) {
345 : /* Get burst of RX packets, from first iface of pair. */
346 0 : const uint16_t nb_rx = rte_eth_rx_burst(iface, 0, rx_bufs, burst_size);
347 :
348 0 : num_packets += nb_rx;
349 0 : total_packets += nb_rx;
350 :
351 0 : for (int i_b = 0; i_b < nb_rx; ++i_b) {
352 0 : num_bytes += rx_bufs[i_b]->pkt_len;
353 :
354 0 : bool dump_packet = false;
355 0 : if (not RTE_ETH_IS_IPV4_HDR(rx_bufs[i_b]->packet_type)) {
356 0 : non_ipv4_packets++;
357 0 : dump_packet = true;
358 0 : continue;
359 : }
360 0 : ++udp_pkt_counter;
361 :
362 0 : char* udp_payload = udp::get_udp_payload(rx_bufs[i_b]);
363 0 : const detdataformats::DAQEthHeader* daq_hdr = reinterpret_cast<const detdataformats::DAQEthHeader*>(udp_payload);
364 :
365 : // uint64_t unique_str_id = (daq_hdr->det_id<<22) + (daq_hdr->crate_id<<12) + (daq_hdr->slot_id<<8) +
366 : // daq_hdr->stream_id;
367 0 : StreamUID unique_str_id(*daq_hdr);
368 :
369 : // if ((udp_pkt_counter % 1000000) == 0 ) {
370 : // std::cout << "\nDAQ HEADER:\n" << *daq_hdr<< "\n";
371 : // }
372 0 : if (stream_stats.find(unique_str_id) == stream_stats.end()) {
373 0 : stream_stats[unique_str_id];
374 0 : stream_stats[unique_str_id].prev_seq_id = daq_hdr->seq_id - 1;
375 : }
376 0 : stream_stats[unique_str_id].num_bytes += rx_bufs[i_b]->pkt_len;
377 :
378 0 : if (check_against_previous_stream(daq_hdr, 2048) != 0) {
379 0 : dump_packet = true;
380 : }
381 0 : if (check_packet_size(rx_bufs[i_b], unique_str_id) != 0) {
382 0 : dump_packet = true;
383 : }
384 :
385 0 : ++stream_stats[unique_str_id].total_packets;
386 0 : ++stream_stats[unique_str_id].num_packets;
387 0 : stream_stats[unique_str_id].prev_seq_id = daq_hdr->seq_id;
388 :
389 0 : if (dump_packet && dumped_packet_count < max_packets_to_dump) {
390 0 : dumped_packet_count++;
391 :
392 0 : rte_pktmbuf_dump(stdout, rx_bufs[i_b], rx_bufs[i_b]->pkt_len);
393 : }
394 : }
395 :
396 0 : rte_pktmbuf_free_bulk(rx_bufs, nb_rx);
397 : }
398 :
399 : return 0;
400 0 : }
401 :
402 : // Define the function to be called when ctrl-c (SIGINT) is sent to process
403 : void
404 0 : signal_callback_handler(int signum)
405 : {
406 0 : fmt::print("Caught signal {}\n", signum);
407 0 : std::exit(signum);
408 : }
409 :
410 : int
411 0 : main(int argc, char** argv)
412 : {
413 0 : uint64_t time_per_report = 1;
414 0 : uint16_t iface = 0;
415 0 : std::vector<std::string> garp_ip_addresses;
416 0 : std::vector<std::string> pcie_addresses;
417 :
418 0 : CLI::App app{ "test frame receiver" };
419 0 : app.add_option("-g,--garp-ip-address", garp_ip_addresses, "IP Addresses");
420 0 : app.add_option("-m,--pcie-mask", pcie_addresses, "PCIE Addresses device mask");
421 0 : app.add_option("-s,--exp-frame-size", expected_packet_size, "Expected frame size");
422 0 : app.add_option("-i,--iface", iface, "Interface to init");
423 0 : app.add_option("-t,--report-interval-time", time_per_report, "Time Per Report");
424 0 : app.add_flag("--check-time", check_timestamp, "Report back differences in timestamp");
425 0 : app.add_flag("-p,--per-stream-reports", per_stream_reports, "Detailed per stream reports");
426 :
427 0 : CLI11_PARSE(app, argc, argv);
428 :
429 : // Validate arguments
430 0 : fmt::print("ips : {}\n", fmt::join(garp_ip_addresses, " | "));
431 0 : fmt::print("pcies : {}\n", fmt::join(pcie_addresses, " | "));
432 :
433 0 : std::regex re_ipv4("[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}");
434 0 : std::regex re_pcie("^0{0,4}:[a-fA-F0-9]{2}:[a-fA-F0-9]{2}.[0-9]$");
435 :
436 0 : fmt::print("IP addresses\n");
437 0 : bool all_ip_ok = true;
438 0 : for (const auto& ip : garp_ip_addresses) {
439 0 : bool ip_ok = std::regex_match(ip, re_ipv4);
440 0 : fmt::print("- {} {}\n", ip, ip_ok);
441 0 : all_ip_ok &= ip_ok;
442 : }
443 :
444 0 : fmt::print("PCIE addresses\n");
445 0 : bool all_pcie_ok = true;
446 0 : for (const auto& pcie : pcie_addresses) {
447 0 : bool pcie_ok = std::regex_match(pcie, re_pcie);
448 0 : fmt::print("- {} {}\n", pcie, pcie_ok);
449 0 : all_pcie_ok &= pcie_ok;
450 : }
451 :
452 0 : if (!all_ip_ok or !all_pcie_ok) {
453 : return -1;
454 : }
455 :
456 : // define function to be called when ctrl+c is called.
457 0 : std::signal(SIGINT, signal_callback_handler);
458 :
459 0 : std::vector<std::string> eal_args;
460 0 : eal_args.push_back("dpdklibds_test_garp");
461 0 : for (const auto& pcie : pcie_addresses) {
462 0 : eal_args.push_back("-a");
463 0 : eal_args.push_back(pcie);
464 : }
465 :
466 0 : dunedaq::dpdklibs::ealutils::init_eal(eal_args);
467 :
468 0 : auto n_ifaces = rte_eth_dev_count_avail();
469 0 : fmt::print("# of available ifaces: {}\n", n_ifaces);
470 0 : if (n_ifaces == 0) {
471 0 : std::cout << "WARNING: no available ifaces. exiting...\n";
472 0 : rte_eal_cleanup();
473 0 : return 1;
474 : }
475 :
476 : // Allocate pools and mbufs per queue
477 0 : std::map<int, std::unique_ptr<rte_mempool>> mbuf_pools;
478 0 : std::map<int, struct rte_mbuf**> bufs;
479 0 : const uint16_t n_rx_qs = 1;
480 0 : const uint16_t n_tx_qs = 1;
481 0 : const uint16_t rx_ring_size = 2048;
482 0 : const uint16_t tx_ring_size = 2048;
483 :
484 0 : std::cout << "Allocating pools and mbufs.\n";
485 0 : for (size_t i = 0; i < n_rx_qs; ++i) {
486 0 : std::stringstream ss;
487 0 : ss << "MBP-" << i;
488 0 : fmt::print("Pool acquire: {}\n", ss.str());
489 0 : mbuf_pools[i] = ealutils::get_mempool(ss.str());
490 0 : bufs[i] = (rte_mbuf**)malloc(sizeof(struct rte_mbuf*) * burst_size);
491 0 : rte_pktmbuf_alloc_bulk(mbuf_pools[i].get(), bufs[i], burst_size);
492 0 : }
493 :
494 : // Setting up only one iface
495 0 : fmt::print("Initialize only iface {}!\n", iface);
496 0 : ealutils::iface_init(
497 : iface, n_rx_qs, n_tx_qs, rx_ring_size, tx_ring_size, mbuf_pools); // just init iface, no TX queues
498 :
499 0 : lcore_main(mbuf_pools[0].get(), iface, time_per_report, garp_ip_addresses);
500 :
501 : rte_eal_cleanup();
502 :
503 : return 0;
504 0 : }
|
