Line data Source code
1 : /**
2 : *
3 : * @file test_transmit_and_receive.cxx
4 : *
5 : * This is a standalone program which will transmit ethernet packets
6 : * out on one dpdk-enabled network interface port and receive them on
7 : * another. Obviously you actually need two such ports for this to
8 : * work.
9 : *
10 : *
11 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
12 : * Licensing/copyright details are in the COPYING file that you should have
13 : * received with this code.
14 : */
15 :
16 : #include "dpdklibs/EALSetup.hpp"
17 : #include "dpdklibs/udp/PacketCtor.hpp"
18 : #include "dpdklibs/udp/Utils.hpp"
19 :
20 : #include "detdataformats/DAQEthHeader.hpp"
21 : #include "logging/Logging.hpp"
22 :
23 : #include "rte_common.h"
24 : #include "rte_eal.h"
25 : #include "rte_ethdev.h"
26 :
27 : #include "CLI/App.hpp"
28 : #include "CLI/Config.hpp"
29 : #include "CLI/Formatter.hpp"
30 :
31 : #include <limits>
32 : #include <mutex>
33 :
34 : using namespace dunedaq::dpdklibs;
35 : using dunedaq::detdataformats::DAQEthHeader;
36 :
37 : namespace {
38 :
39 : constexpr int burst_size = 256;
40 :
41 : std::mutex lcore_thread_mutex;
42 :
43 : int payload_bytes = 9000;
44 :
45 : std::atomic<long> num_sent_bursts = 0;
46 : std::atomic<long> num_sent_packets = 0;
47 : std::atomic<long> num_received_packets = 0;
48 : std::atomic<long> num_sent_bytes = 0;
49 : std::atomic<long> num_received_bytes = 0;
50 : std::atomic<long> num_bad_packets = 0;
51 :
52 : }; // namespace
53 :
54 : struct lcore_thread_arg
55 : {
56 : rte_mempool* pool;
57 : int iface_id;
58 : bool is_receiver;
59 : };
60 :
61 : inline bool
62 0 : operator!=(const DAQEthHeader& header1, const DAQEthHeader& header2)
63 : {
64 0 : return (header1.version != header2.version || header1.det_id != header2.det_id ||
65 0 : header1.crate_id != header2.crate_id || header1.slot_id != header2.slot_id ||
66 0 : header1.stream_id != header2.stream_id || header1.seq_id != header2.seq_id ||
67 0 : header1.block_length != header2.block_length || header1.timestamp != header2.timestamp);
68 : }
69 :
70 : int
71 0 : lcore_main(void* arg)
72 : {
73 :
74 0 : uint16_t lid = rte_lcore_id();
75 :
76 : // 0: main thread (doesn't use this function), 1: receiver thread, 2: transmitter thread
77 0 : if (lid == 0 || lid >= 3) {
78 : return 0;
79 : }
80 :
81 0 : TLOG() << "From lcore_main, lid == " << lid << ", rte_socket_id() == " << rte_socket_id();
82 :
83 0 : lcore_thread_arg info;
84 0 : const auto ptr_to_vector_of_thread_configs = static_cast<std::vector<lcore_thread_arg>*>(arg);
85 :
86 0 : {
87 0 : std::lock_guard<std::mutex> guard(lcore_thread_mutex);
88 0 : info = ptr_to_vector_of_thread_configs->at(lid - 1);
89 0 : }
90 0 : TLOG() << "In lcore_main, thread #" << lid << ", iface_id == " << info.iface_id
91 0 : << ", is_receiver == " << info.is_receiver;
92 :
93 0 : if (rte_eth_dev_socket_id(info.iface_id) >= 0 &&
94 0 : rte_eth_dev_socket_id(info.iface_id) != static_cast<int>(rte_socket_id())) {
95 0 : TLOG(TLVL_WARNING) << "WARNING, dpdk interface " << info.iface_id
96 0 : << " is on remote NUMA node to polling thread.\nrte_eth_dev_socket_id(" << info.iface_id
97 0 : << ") == " << rte_eth_dev_socket_id(info.iface_id)
98 0 : << ", rte_socket_id() == " << static_cast<int>(rte_socket_id())
99 0 : << "\nPerformance will not be optimal.\n";
100 : }
101 :
102 0 : auto bufs = static_cast<rte_mbuf**>(malloc(sizeof(struct rte_mbuf*) * burst_size));
103 :
104 0 : if (bufs == NULL) {
105 0 : TLOG(TLVL_ERROR) << "Unable to allocate memory for buffers; returning from thread...";
106 0 : return 1;
107 : }
108 :
109 0 : int retval = 0;
110 0 : retval = rte_pktmbuf_alloc_bulk(info.pool, bufs, burst_size);
111 :
112 0 : if (retval != 0) {
113 0 : std::stringstream errstr;
114 0 : errstr << "A failure occurred calling rte_pktmbuf_alloc_bulk (" << strerror(abs(retval))
115 0 : << "); returning from thread...";
116 0 : return 2;
117 0 : }
118 :
119 0 : if (info.is_receiver) {
120 :
121 0 : uint16_t nb_rx = 0;
122 0 : int bytes_in_burst = 0;
123 0 : int bad_packets_in_burst = 0;
124 :
125 : // Here, I'm taking advantage of the fact that the transmitting
126 : // thread calls construct_packets_for_burst, which itself fills
127 : // the DAQEthHeader structure inside the packet it sends using
128 : // "set_daqethheader_test_values". I'll compare the received
129 : // packet's DAQEthHeader to this one and make sure they're equal
130 :
131 0 : DAQEthHeader daqethheader_reference;
132 0 : udp::set_daqethheader_test_values(daqethheader_reference);
133 0 : DAQEthHeader* daqethheader_ptr = nullptr;
134 :
135 0 : while (true) {
136 :
137 0 : bytes_in_burst = 0; // Quicker to add up the bytes in the packets in a local variables, then attach the total to
138 : // the atomic variable used in stats reporting
139 0 : bad_packets_in_burst = 0; // Same thinking
140 :
141 0 : nb_rx = rte_eth_rx_burst(info.iface_id, 0, bufs, burst_size);
142 0 : num_received_packets += nb_rx;
143 :
144 0 : for (int i_p = 0; i_p < nb_rx; ++i_p) {
145 0 : bytes_in_burst += bufs[i_p]->pkt_len;
146 :
147 : // DAQEthHeader (should be) the first thing after the ethernet + IPv4 + UDP headers
148 0 : daqethheader_ptr = rte_pktmbuf_mtod_offset(bufs[0], DAQEthHeader*, sizeof(udp::ipv4_udp_packet_hdr));
149 0 : if (*daqethheader_ptr != daqethheader_reference) {
150 0 : bad_packets_in_burst++;
151 : }
152 : }
153 :
154 0 : num_received_bytes += bytes_in_burst;
155 0 : num_bad_packets += bad_packets_in_burst;
156 :
157 : // Free any unsent packets
158 0 : if (unlikely(nb_rx < burst_size)) {
159 0 : rte_pktmbuf_free_bulk(bufs, burst_size - nb_rx);
160 : }
161 : }
162 :
163 : } else {
164 :
165 0 : rte_ether_addr dst_mac_addr_struct;
166 :
167 : // Notice that in the snippet of code below, the assumption is that
168 : // the first entry in the vector of thread configurations
169 : // corresponds to the receiving thread
170 0 : uint16_t receiver_iface = std::numeric_limits<uint16_t>::max();
171 0 : {
172 0 : std::lock_guard<std::mutex> guard(lcore_thread_mutex);
173 0 : receiver_iface = ptr_to_vector_of_thread_configs->at(0).iface_id;
174 0 : }
175 :
176 0 : retval = rte_eth_macaddr_get(receiver_iface, &dst_mac_addr_struct);
177 :
178 0 : if (retval == 0) {
179 0 : TLOG() << "Will send to ethernet interface with MAC address " << ealutils::get_mac_addr_str(dst_mac_addr_struct);
180 : } else {
181 0 : TLOG() << "Problem trying to obtain the MAC address of the destination port; returning...";
182 0 : return 3;
183 : }
184 :
185 0 : const std::string dst_mac_addr = ealutils::get_mac_addr_str(dst_mac_addr_struct);
186 0 : udp::construct_packets_for_burst(info.iface_id, dst_mac_addr, payload_bytes, burst_size, bufs);
187 :
188 0 : TLOG() << "Dump of the first packet header: ";
189 0 : TLOG() << udp::get_udp_header_str(bufs[0]);
190 :
191 0 : TLOG() << "Dump of the first packet rte_mbuf object: ";
192 0 : TLOG() << udp::get_rte_mbuf_str(bufs[0]);
193 :
194 0 : rte_mbuf_sanity_check(bufs[0], 1);
195 :
196 0 : uint16_t nb_tx = 0;
197 :
198 0 : while (true) {
199 :
200 0 : int queue_id = 0;
201 : // TLOG() << "Just about to call rte_eth_tx_burst with arguments info.iface_id == " << info.iface_id << ", second
202 : // argument " << queue_id << ", bufs at " << bufs << " and burst size " << burst_size;
203 0 : nb_tx = rte_eth_tx_burst(info.iface_id, queue_id, bufs, burst_size);
204 0 : num_sent_packets += nb_tx;
205 0 : num_sent_bytes += nb_tx * bufs[0]->pkt_len;
206 0 : num_sent_bursts++;
207 : // TLOG() << "Sent " << nb_tx;
208 :
209 : // Free any unsent packets
210 0 : if (unlikely(nb_tx < burst_size)) {
211 0 : rte_pktmbuf_free_bulk(bufs, burst_size - nb_tx);
212 : }
213 : }
214 0 : }
215 : return 0;
216 : }
217 :
218 : int
219 0 : main(int argc, char** argv)
220 : {
221 :
222 0 : CLI::App app{ "test transmit and receive" };
223 :
224 0 : std::stringstream payload_desc;
225 0 : payload_desc << "Bytes of payload past the DAQEthHeader header (default " << payload_bytes << " bytes)";
226 0 : app.add_option("--payload", payload_bytes, payload_desc.str());
227 :
228 0 : CLI11_PARSE(app, argc, argv);
229 :
230 0 : argc = 1; // So rte_eal_init doesn't interpret what gets passed to the command line
231 :
232 0 : int retval = rte_eal_init(argc, argv);
233 :
234 0 : if (retval < 0) {
235 0 : rte_exit(EXIT_FAILURE, "ERROR: EAL initialization failed, info: %s\n", strerror(abs(retval)));
236 : }
237 :
238 0 : int nb_ports = rte_eth_dev_count_avail();
239 0 : TLOG() << "There are " << nb_ports << " ethernet ports available out of a total of " << rte_eth_dev_count_total();
240 :
241 0 : if (nb_ports < 2) {
242 0 : rte_exit(EXIT_FAILURE,
243 : "ERROR: fewer than 2 ethernet ports are available.\nYou need one port for transmitting packets and one "
244 : "port for receiving them\n");
245 : }
246 :
247 0 : auto stats = std::thread([&]() {
248 0 : long sleep_time = 1;
249 0 : while (true) {
250 0 : TLOG() << "\nRates: \n"
251 0 : << num_sent_bursts / static_cast<double>(sleep_time) << " sent bursts/s\n"
252 0 : << num_sent_packets / static_cast<double>(sleep_time) << " sent packets/s\n"
253 0 : << num_received_packets / static_cast<double>(sleep_time) << " received packets/s\n"
254 0 : << num_bad_packets / static_cast<double>(sleep_time) << " bad packets/s\n"
255 0 : << num_sent_bytes / static_cast<double>(sleep_time) << " sent bytes/s\n"
256 0 : << num_received_bytes / static_cast<double>(sleep_time) << " received bytes/s\n";
257 0 : num_sent_packets.exchange(0);
258 0 : num_sent_bytes.exchange(0);
259 0 : num_sent_bursts.exchange(0);
260 0 : num_received_packets.exchange(0);
261 0 : num_received_bytes.exchange(0);
262 0 : num_bad_packets.exchange(0);
263 0 : std::this_thread::sleep_for(std::chrono::seconds(sleep_time));
264 0 : }
265 0 : });
266 :
267 0 : uint16_t portid = std::numeric_limits<uint16_t>::max();
268 0 : int n_rx_qs = 0;
269 0 : int n_tx_qs = 0;
270 0 : uint16_t rx_ring_size = 1024;
271 0 : uint16_t tx_ring_size = 1024;
272 0 : int port_cntr = -1;
273 0 : std::map<int, std::unique_ptr<rte_mempool>> mbuf_pools;
274 0 : std::vector<lcore_thread_arg> lcore_thread_args;
275 :
276 0 : RTE_ETH_FOREACH_DEV(portid)
277 : { // RTE_ETH_FOREACH_DEV since no guarantee dpdk counts port IDs as 0, 1, 2 ...
278 :
279 0 : port_cntr++; // Initialize at -1, start at 0
280 :
281 : // Right now (May-2-2023) iface_init's implementation explicitly
282 : // assumes that the pools corresponding to <N> receiver queues
283 : // correspond to keys 0...N-1 in a map whose values are
284 : // unique_ptrs to pools
285 :
286 0 : std::stringstream poolname;
287 0 : poolname << "MBP-" << portid;
288 0 : mbuf_pools[port_cntr] = ealutils::get_mempool(poolname.str());
289 :
290 0 : bool is_receiver = false;
291 :
292 0 : if (port_cntr == 0) {
293 : n_tx_qs = 0;
294 : n_rx_qs = 1;
295 : is_receiver = true;
296 0 : } else if (port_cntr == 1) {
297 0 : n_tx_qs = 1;
298 0 : n_rx_qs = 0;
299 0 : is_receiver = false;
300 : }
301 :
302 0 : retval = ealutils::iface_init(portid, n_rx_qs, n_tx_qs, rx_ring_size, tx_ring_size, mbuf_pools);
303 :
304 0 : if (retval != 0) {
305 0 : rte_eal_cleanup();
306 :
307 0 : std::stringstream errstr;
308 0 : errstr << "A failure occurred initializing ethernet interface #" << portid << " (" << strerror(abs(retval))
309 0 : << "); exiting...";
310 0 : rte_exit(EXIT_FAILURE, errstr.str().c_str());
311 0 : }
312 :
313 0 : auto pool = mbuf_pools[port_cntr].release();
314 0 : mbuf_pools.erase(mbuf_pools.find(port_cntr));
315 :
316 0 : lcore_thread_args.emplace_back(lcore_thread_arg{ pool, portid, is_receiver });
317 0 : }
318 :
319 0 : rte_eal_mp_remote_launch(lcore_main, &lcore_thread_args, SKIP_MAIN);
320 :
321 0 : rte_eal_mp_wait_lcore();
322 0 : rte_eal_cleanup();
323 :
324 0 : int mynum = 55;
325 0 : TLOG() << "Things are still OK at this line, and here's a number: " << mynum;
326 :
327 0 : return 0;
328 0 : }
|
