Line data Source code
1 : /**
2 : * @file test_frame_transmitter.cxx Construct UDP packets to be sent over the wire using DPDK
3 : *
4 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
5 : * Licensing/copyright details are in the COPYING file that you should have
6 : * received with this code.
7 : */
8 :
9 : #include "dpdklibs/EALSetup.hpp"
10 : #include "dpdklibs/udp/PacketCtor.hpp"
11 : #include "dpdklibs/udp/Utils.hpp"
12 :
13 : #include "detdataformats/DAQEthHeader.hpp"
14 : #include "logging/Logging.hpp"
15 :
16 : #include "rte_cycles.h"
17 : #include "rte_dev.h"
18 : #include "rte_eal.h"
19 : #include "rte_ethdev.h"
20 : #include "rte_lcore.h"
21 : #include "rte_mbuf.h"
22 :
23 : #include "CLI/App.hpp"
24 : #include "CLI/Config.hpp"
25 : #include "CLI/Formatter.hpp"
26 :
27 : #include <chrono>
28 : #include <iostream>
29 : #include <thread>
30 :
31 : namespace {
32 :
33 : // Only 8 and above works
34 : constexpr int burst_size = 256;
35 :
36 : constexpr int buffer_size = 9800; // Same number as in EALSetup.hpp
37 :
38 : std::string dst_mac_addr = "";
39 : int payload_bytes = 0; // payload past the DAQEthHeader
40 :
41 : std::atomic<long> num_bursts = 0;
42 : std::atomic<long> num_packets = 0;
43 : std::atomic<long> num_bytes = 0;
44 : }
45 :
46 : using namespace dunedaq;
47 : using namespace dpdklibs;
48 : using namespace udp;
49 :
50 : void
51 0 : lcore_main(void* arg)
52 : {
53 :
54 0 : uint16_t iface = *(static_cast<uint16_t*>(arg));
55 :
56 0 : uint16_t lid = rte_lcore_id();
57 :
58 0 : if (lid != 1) {
59 0 : return;
60 : }
61 :
62 0 : if (rte_eth_dev_socket_id(iface) >= 0 && rte_eth_dev_socket_id(iface) != static_cast<int>(rte_socket_id())) {
63 0 : TLOG(TLVL_WARNING) << "WARNING, dpdk interface " << iface
64 0 : << " is on remote NUMA node to polling thread.\nrte_eth_dev_socket_id(" << iface
65 0 : << ") == " << rte_eth_dev_socket_id(iface)
66 0 : << ", rte_socket_id() == " << static_cast<int>(rte_socket_id())
67 0 : << "\nPerformance will not be optimal.\n";
68 : }
69 :
70 0 : struct rte_mempool* mbuf_pool = rte_pktmbuf_pool_create((std::string("MBUF_POOL") + std::to_string(lid)).c_str(),
71 0 : NUM_MBUFS * rte_eth_dev_count_avail(),
72 : MBUF_CACHE_SIZE,
73 : 0,
74 : buffer_size,
75 0 : rte_socket_id());
76 :
77 0 : if (mbuf_pool == NULL) {
78 0 : rte_exit(EXIT_FAILURE, "ERROR: call to rte_pktmbuf_pool_create failed, info: %s\n", rte_strerror(rte_errno));
79 : }
80 :
81 0 : struct rte_mbuf** bufs = (rte_mbuf**)malloc(sizeof(struct rte_mbuf*) * burst_size);
82 0 : if (bufs == NULL) {
83 0 : TLOG(TLVL_ERROR) << "Failure trying to acquire memory for transmission buffers on thread #" << lid
84 0 : << "; exiting...";
85 0 : std::exit(1);
86 : }
87 :
88 0 : int retval = rte_pktmbuf_alloc_bulk(mbuf_pool, bufs, burst_size);
89 :
90 0 : if (retval != 0) {
91 0 : rte_exit(EXIT_FAILURE, "ERROR: call to rte_pktmbuf_alloc_bulk failed, info: %s\n", strerror(abs(retval)));
92 : }
93 :
94 0 : TLOG() << "In thread, we have iface == " << iface;
95 0 : construct_packets_for_burst(iface, dst_mac_addr, payload_bytes, burst_size, bufs);
96 :
97 0 : TLOG() << "Dump of the first packet header: ";
98 0 : TLOG() << get_udp_header_str(bufs[0]);
99 :
100 0 : TLOG() << "Dump of the first packet rte_mbuf object: ";
101 0 : TLOG() << get_rte_mbuf_str(bufs[0]);
102 :
103 0 : rte_mbuf_sanity_check(bufs[0], 1);
104 :
105 0 : TLOG() << "\n\nCore " << rte_lcore_id() << " transmitting packets. [Ctrl+C to quit]\n\n";
106 :
107 0 : uint16_t nb_tx = 0;
108 0 : int cntr = 0;
109 0 : while (cntr < std::numeric_limits<int>::max()) {
110 0 : cntr++;
111 :
112 0 : nb_tx = rte_eth_tx_burst(iface, lid - 1, bufs, burst_size);
113 0 : num_packets += nb_tx;
114 0 : num_bytes += nb_tx * bufs[0]->pkt_len; // n.b. Assumption in this line is that each packet is the same size
115 0 : num_bursts++;
116 :
117 : // Free any unsent packets
118 0 : if (unlikely(nb_tx < burst_size)) {
119 0 : rte_pktmbuf_free_bulk(bufs, burst_size - nb_tx);
120 : }
121 : }
122 : }
123 :
124 : int
125 0 : main(int argc, char* argv[])
126 : {
127 :
128 0 : CLI::App app{ "test frame transmitter" };
129 :
130 0 : const std::string default_mac_address = "6c:fe:54:47:98:20";
131 :
132 0 : app.add_option("--dst-mac", dst_mac_addr, "Destination MAC address (default " + default_mac_address + ")");
133 0 : app.add_option("--payload", payload_bytes, "Bytes of payload past the DAQEthHeader header");
134 :
135 0 : if (dst_mac_addr == "") {
136 0 : dst_mac_addr = default_mac_address;
137 : }
138 :
139 0 : CLI11_PARSE(app, argc, argv);
140 :
141 0 : argc = 1; // Set to 1 so rte_eal_init ignores all the CLI-parsed arguments
142 :
143 0 : int retval = rte_eal_init(argc, argv);
144 0 : if (retval < 0) {
145 0 : rte_exit(EXIT_FAILURE, "ERROR: EAL initialization failed, info: %s\n", strerror(abs(retval)));
146 : }
147 :
148 : // Check that there is an even number of ports to send/receive on
149 0 : auto nb_ports = rte_eth_dev_count_avail();
150 0 : TLOG() << "There are " << nb_ports << " ethernet ports available out of a total of " << rte_eth_dev_count_total();
151 0 : if (nb_ports == 0) {
152 0 : rte_exit(EXIT_FAILURE,
153 : "ERROR: 0 ethernet ports are available. This can be caused either by someone else currently\nusing "
154 : "dpdk-based code or the necessary drivers not being bound to the NICs\n(see "
155 : "https://github.com/DUNE-DAQ/dpdklibs#readme for more)\n");
156 : }
157 :
158 0 : const uint16_t n_tx_qs = 1;
159 0 : const uint16_t n_rx_qs = 0;
160 0 : const uint16_t rx_ring_size = 1024;
161 0 : const uint16_t tx_ring_size = 1024;
162 :
163 0 : uint16_t portid = 0;
164 0 : std::map<int, std::unique_ptr<rte_mempool>> dummyarg;
165 0 : retval = ealutils::iface_init(portid, n_rx_qs, n_tx_qs, rx_ring_size, tx_ring_size, dummyarg);
166 :
167 0 : if (retval != 0) {
168 0 : TLOG(TLVL_ERROR) << "A failure occurred initializing ethernet interface #" << portid << "; exiting...";
169 0 : rte_eal_cleanup();
170 0 : std::exit(2);
171 : }
172 :
173 0 : struct rte_eth_dev_info dev_info;
174 0 : retval = rte_eth_dev_info_get(portid, &dev_info);
175 :
176 0 : if (retval != 0) {
177 0 : TLOG(TLVL_ERROR) << "A failure occured trying to get info on ethernet interface #" << portid << "; exiting...";
178 0 : rte_eal_cleanup();
179 0 : std::exit(4);
180 : }
181 :
182 0 : TLOG() << "Name of the interface is " << rte_dev_name(dev_info.device);
183 :
184 0 : auto stats = std::thread([&]() {
185 0 : long sleep_time = 1;
186 0 : while (true) {
187 0 : TLOG() << "Rates: " << num_packets / static_cast<double>(sleep_time) << " packets/s, "
188 0 : << num_bytes / static_cast<double>(sleep_time) << " bytes/s, "
189 0 : << num_bursts / static_cast<double>(sleep_time) << " bursts/s" << "\n";
190 0 : num_packets.exchange(0);
191 0 : num_bytes.exchange(0);
192 0 : num_bursts.exchange(0);
193 0 : std::this_thread::sleep_for(std::chrono::seconds(sleep_time));
194 0 : }
195 0 : });
196 :
197 0 : rte_eal_mp_remote_launch((lcore_function_t*)lcore_main, &portid, SKIP_MAIN);
198 :
199 0 : rte_eal_mp_wait_lcore();
200 0 : rte_eal_cleanup();
201 :
202 0 : return 0;
203 0 : }
|
