Line data Source code
1 : #include "dpdklibs/udp/IPV4UDPPacket.hpp"
2 : #include <rte_arp.h>
3 : #include <rte_ethdev.h>
4 : #include <time.h>
5 :
6 : namespace dunedaq {
7 : namespace dpdklibs {
8 :
9 : int
10 0 : IfaceWrapper::rx_runner(void* arg __rte_unused)
11 : {
12 :
13 : // Timespec for opportunistic sleep. Nanoseconds configured in conf.
14 0 : struct timespec sleep_request = { 0, (long)m_lcore_sleep_ns };
15 :
16 : // bool once = true; // One shot action variable.
17 0 : uint16_t iface = m_iface_id;
18 :
19 0 : const uint16_t lid = rte_lcore_id();
20 0 : auto queues = m_rx_core_map[lid];
21 :
22 0 : if (rte_eth_dev_socket_id(iface) >= 0 && rte_eth_dev_socket_id(iface) != (int)rte_socket_id()) {
23 0 : TLOG() << "WARNING, iface " << iface << " is on remote NUMA node to polling thread! "
24 0 : << "Performance will not be optimal.";
25 : }
26 :
27 0 : TLOG() << "LCore RX runner on CPU[" << lid << "]: Main loop starts for iface " << iface << " !";
28 :
29 0 : std::map<int, int> nb_rx_map;
30 : // While loop of quit atomic member in IfaceWrapper
31 0 : while (!this->m_lcore_quit_signal.load()) {
32 :
33 : // Loop over assigned queues to process
34 0 : uint8_t fb_count(0);
35 0 : for (const auto& q : queues) {
36 0 : auto src_rx_q = q.first;
37 0 : auto* q_bufs = m_bufs[src_rx_q];
38 :
39 : // Get burst from queue
40 0 : const uint16_t nb_rx = rte_eth_rx_burst(iface, src_rx_q, q_bufs, m_burst_size);
41 0 : nb_rx_map[src_rx_q] = nb_rx;
42 : }
43 :
44 0 : for (const auto& q : queues) {
45 :
46 0 : auto src_rx_q = q.first;
47 0 : auto* q_bufs = m_bufs[src_rx_q];
48 0 : const uint16_t nb_rx = nb_rx_map[src_rx_q];
49 :
50 : // We got packets from burst on this queue
51 0 : if (nb_rx != 0) [[likely]] {
52 :
53 : // Update max burst size counter of this queue
54 0 : m_max_burst_size[src_rx_q] = std::max(nb_rx, m_max_burst_size[src_rx_q].load());
55 :
56 : // -------
57 : // Iterate on burst packets
58 0 : for (int i_b = 0; i_b < nb_rx; ++i_b) {
59 :
60 : // Check if packet is segmented. Implement support for it if needed.
61 : // if (q_bufs[i_b]->nb_segs > 1) [[unlikely]] {
62 : // TLOG_DEBUG(10) << "It appears a packet is spread across more than one receiving buffer;"
63 : // << " there's currently no logic in this program to handle this";
64 : //}
65 :
66 : // Check packet type, decide their fate: ignore unexpected ones, FIXME: monitor occurrences
67 0 : auto pkt_type = q_bufs[i_b]->packet_type;
68 : // Handle non IPV4 frames.
69 0 : if (not RTE_ETH_IS_IPV4_HDR(pkt_type)) [[unlikely]] {
70 : // TLOG_DEBUG(10) << "Non-Ethernet packet type: " << (unsigned)pkt_type << " original: " << pkt_type;
71 0 : if (pkt_type == RTE_PTYPE_L2_ETHER_ARP) {
72 : // TLOG() << "Unexpected: Should handle an ARP request from lcore=" << lid << " rx_q=" << src_rx_q << "!
73 : // Flow should be steered to dedicated RX Queue.";
74 0 : } else if (pkt_type == RTE_PTYPE_L2_ETHER_LLDP) {
75 : // TLOG_DEBUG(10) << "TODO: Handle LLDP packet!";
76 : } else {
77 : // TLOG_DEBUG(10) << "Unidentified! Dumping...";
78 : // rte_pktmbuf_dump(stdout, q_bufs[i_b], m_bufs[src_rx_q][i_b]->pkt_len);
79 : }
80 0 : ++m_num_unhandled_non_ipv4[lid];
81 0 : continue;
82 : }
83 :
84 : // Check if frame is non UDP: in that case, ignore it.
85 0 : if ((pkt_type & RTE_PTYPE_L4_MASK) != RTE_PTYPE_L4_UDP) [[unlikely]] {
86 0 : ++m_num_unhandled_non_udp[lid];
87 0 : continue; // ommit it
88 : }
89 :
90 : // Check for JUMBO frames (bigger than 1500 Bytes)
91 0 : if (q_bufs[i_b]->pkt_len > 1500) [[likely]] { // RS FIXME: do proper check on data length later
92 :
93 : // If flow enabled, handle the payload.
94 0 : if (m_lcore_enable_flow.load()) [[likely]] {
95 : // Get length of user payload. (Ethernet headers excluded.)
96 0 : struct udp::ipv4_udp_packet_hdr* udp_packet =
97 0 : rte_pktmbuf_mtod(q_bufs[i_b], struct udp::ipv4_udp_packet_hdr*);
98 0 : char* message = udp::get_udp_payload(q_bufs[i_b]);
99 0 : std::size_t udp_payload_len = udp::get_payload_size_udp_hdr(&udp_packet->udp_hdr);
100 :
101 0 : if (m_strict_parsing) { // all sources maintain DAQ protocol
102 0 : parse_udp_payload(src_rx_q, message, udp_payload_len);
103 : } else { // avoid size checks and scattering
104 0 : passthrough_udp_payload(src_rx_q, message, udp_payload_len);
105 : }
106 : }
107 :
108 : // Update metrics of queue: frame and Byte counters
109 0 : ++m_num_frames_rxq[src_rx_q];
110 0 : std::size_t data_len = q_bufs[i_b]->data_len;
111 0 : m_num_bytes_rxq[src_rx_q] += data_len;
112 : } else {
113 0 : ++m_num_unhandled_non_jumbo_udp[lid];
114 : }
115 : }
116 :
117 : // Bulk free of mbufs
118 0 : rte_pktmbuf_free_bulk(q_bufs, nb_rx);
119 :
120 : } // per burst
121 : // -----------
122 :
123 : // Full burst counter
124 0 : if (nb_rx == m_burst_size) {
125 0 : ++fb_count;
126 0 : ++m_num_full_bursts[src_rx_q];
127 : }
128 : } // per queue
129 :
130 : // If no full buffers in burst...
131 0 : if (!fb_count) {
132 0 : if (m_lcore_sleep_ns) {
133 : // Sleep n nanoseconds... (value from config, timespec initialized in lcore first lines)
134 0 : /*int response =*/nanosleep(&sleep_request, nullptr);
135 : }
136 : }
137 :
138 : } // main while(quit) loop
139 :
140 0 : TLOG() << "LCore RX runner on CPU[" << lid << "] returned.";
141 0 : return 0;
142 0 : }
143 :
144 : int
145 0 : IfaceWrapper::arp_response_runner(void* arg __rte_unused)
146 : {
147 :
148 : // Timespec for opportunistic sleep. Nanoseconds configured in conf.
149 0 : struct timespec sleep_request = { 0, (long)900000 };
150 :
151 : // bool once = true; // One shot action variable.
152 0 : uint16_t iface = m_iface_id;
153 :
154 0 : const uint16_t lid = rte_lcore_id();
155 0 : unsigned arp_rx_queue = m_arp_rx_queue;
156 :
157 0 : TLOG() << "LCore ARP responder on CPU[" << lid << "]: Main loop starts for iface " << iface
158 0 : << " rx queue: " << arp_rx_queue;
159 :
160 : // While loop of quit atomic member in IfaceWrapper
161 0 : while (!this->m_lcore_quit_signal.load()) {
162 :
163 0 : const uint16_t nb_rx = rte_eth_rx_burst(iface, arp_rx_queue, m_arp_bufs[arp_rx_queue], m_burst_size);
164 :
165 : // We got packets from burst on this queue
166 0 : if (nb_rx != 0) {
167 : // Iterate on burst packets
168 0 : for (int i_b = 0; i_b < nb_rx; ++i_b) {
169 :
170 : // Check packet type, ommit/drop unexpected ones.
171 0 : auto pkt_type = m_arp_bufs[arp_rx_queue][i_b]->packet_type;
172 : //// Handle non IPV4 packets
173 0 : if (not RTE_ETH_IS_IPV4_HDR(pkt_type)) {
174 : // TLOG_DEBUG(10) << "Non-Ethernet packet type: " << (unsigned)pkt_type << " original: " << pkt_type;
175 0 : if (pkt_type == RTE_PTYPE_L2_ETHER_ARP) {
176 0 : TLOG_DEBUG(10) << "Handling ARP request";
177 0 : struct rte_ether_hdr* eth_hdr = rte_pktmbuf_mtod(m_arp_bufs[arp_rx_queue][i_b], struct rte_ether_hdr*);
178 0 : struct rte_arp_hdr* arp_hdr = (struct rte_arp_hdr*)(eth_hdr + 1);
179 :
180 0 : std::string srcaddr = dunedaq::dpdklibs::udp::get_ipv4_decimal_addr_str(
181 0 : dunedaq::dpdklibs::udp::ip_address_binary_to_dotdecimal(rte_be_to_cpu_32(arp_hdr->arp_data.arp_sip)));
182 0 : TLOG_DEBUG(10) << "SRC IP: " << srcaddr;
183 0 : std::string dstaddr = dunedaq::dpdklibs::udp::get_ipv4_decimal_addr_str(
184 0 : dunedaq::dpdklibs::udp::ip_address_binary_to_dotdecimal(rte_be_to_cpu_32(arp_hdr->arp_data.arp_tip)));
185 0 : TLOG_DEBUG(10) << "DEST IP: " << dstaddr;
186 :
187 0 : for (const auto& ip_addr_bin : m_ip_addr_bin) {
188 0 : std::string localaddr = dunedaq::dpdklibs::udp::get_ipv4_decimal_addr_str(
189 0 : dunedaq::dpdklibs::udp::ip_address_binary_to_dotdecimal(rte_be_to_cpu_32(ip_addr_bin)));
190 0 : TLOG_DEBUG(10) << "LOCAL IP: " << localaddr;
191 0 : }
192 :
193 0 : if (std::find(m_ip_addr_bin.begin(), m_ip_addr_bin.end(), arp_hdr->arp_data.arp_tip) !=
194 0 : m_ip_addr_bin.end()) {
195 0 : arp::pktgen_process_arp(m_arp_bufs[arp_rx_queue][i_b], m_iface_id, arp_hdr->arp_data.arp_tip);
196 : } else {
197 0 : TLOG_DEBUG(10) << "I'm not the ARP target";
198 : }
199 0 : } else if (pkt_type == RTE_PTYPE_L2_ETHER_LLDP) {
200 : // TLOG_DEBUG(10) << "TODO: Handle LLDP packet!";
201 : } else {
202 : // TLOG_DEBUG(10) << "Unidentified! Dumping...";
203 : // rte_pktmbuf_dump(stdout, m_arp_bufs[arp_rx_queue][i_b], m_bufs[src_rx_q][i_b]->pkt_len);
204 : }
205 0 : continue;
206 0 : }
207 : }
208 :
209 : // Bulk free of mbufs
210 0 : rte_pktmbuf_free_bulk(m_arp_bufs[arp_rx_queue], nb_rx);
211 :
212 : } // per burst
213 :
214 : // If no full buffers in burst...
215 0 : if (m_lcore_sleep_ns) {
216 : // Sleep n nanoseconds... (value from config, timespec initialized in lcore first lines)
217 0 : /*int response =*/nanosleep(&sleep_request, nullptr);
218 : }
219 :
220 : } // main while(quit) loop
221 :
222 0 : TLOG() << "LCore ARP responder on CPU[" << lid << "] returned.";
223 0 : return 0;
224 : }
225 :
226 : } // namespace dpdklibs
227 : } // namespace dunedaq
|
