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