//
// Copyright 2019 Ettus Research, a National Instruments Brand
//
// SPDX-License-Identifier: GPL-3.0-or-later
//
#include <uhd/utils/log.hpp>
#include <uhd/utils/thread.hpp>
#include <uhdlib/transport/dpdk/arp.hpp>
#include <uhdlib/transport/dpdk/udp.hpp>
#include <uhdlib/transport/dpdk_io_service_client.hpp>
#include <uhdlib/utils/narrow.hpp>
#include <cmath>
/*
* Memory management
*
* Every object that allocates and frees DPDK memory has a reference to the
* dpdk_ctx.
*
* Ownership hierarchy:
*
* dpdk_io_service_mgr (1) =>
* dpdk_ctx::sptr
* dpdk_io_service::sptr
*
* xport (1) =>
* dpdk_send_io::sptr
* dpdk_recv_io::sptr
*
* usrp link_mgr (1) =>
* udp_dpdk_link::sptr
*
* dpdk_send_io (2) =>
* dpdk_ctx::sptr
* dpdk_io_service::sptr
*
* dpdk_recv_io (2) =>
* dpdk_ctx::sptr
* dpdk_io_service::sptr
*
* dpdk_io_service (3) =>
* dpdk_ctx::wptr (weak_ptr)
* udp_dpdk_link::sptr
*
* udp_dpdk_link (4) =>
* dpdk_ctx::sptr
*/
using namespace uhd::transport;
dpdk_io_service::dpdk_io_service(
unsigned int lcore_id, std::vector<dpdk::dpdk_port*> ports, size_t servq_depth)
: _ctx(dpdk::dpdk_ctx::get())
, _lcore_id(lcore_id)
, _ports(ports)
, _servq(servq_depth, lcore_id)
{
UHD_LOG_TRACE("DPDK::IO_SERVICE", "Launching I/O service for lcore " << lcore_id);
for (auto port : _ports) {
UHD_LOG_TRACE("DPDK::IO_SERVICE",
"lcore_id " << lcore_id << ": Adding port index " << port->get_port_id());
_tx_queues[port->get_port_id()] = std::list<dpdk_send_io*>();
_recv_xport_map[port->get_port_id()] = std::list<dpdk_recv_io*>();
}
int status = rte_eal_remote_launch(_io_worker, this, lcore_id);
if (status) {
throw uhd::runtime_error("DPDK: I/O service cannot launch on busy lcore");
}
}
dpdk_io_service::sptr dpdk_io_service::make(
unsigned int lcore_id, std::vector<dpdk::dpdk_port*> ports, size_t servq_depth)
{
return dpdk_io_service::sptr(new dpdk_io_service(lcore_id, ports, servq_depth));
}
dpdk_io_service::~dpdk_io_service()
{
UHD_LOG_TRACE(
"DPDK::IO_SERVICE", "Shutting down I/O service for lcore " << _lcore_id);
dpdk::wait_req* req = dpdk::wait_req_alloc(dpdk::wait_type::WAIT_LCORE_TERM, NULL);
if (!req) {
UHD_LOG_ERROR("DPDK::IO_SERVICE",
"Could not allocate request for lcore termination for lcore " << _lcore_id);
return;
}
dpdk::wait_req_get(req);
_servq.submit(req, std::chrono::microseconds(-1));
dpdk::wait_req_put(req);
}
void dpdk_io_service::attach_recv_link(recv_link_if::sptr link)
{
struct dpdk_flow_data data;
data.link = dynamic_cast<udp_dpdk_link*>(link.get());
data.is_recv = true;
assert(data.link);
auto req = wait_req_alloc(dpdk::wait_type::WAIT_FLOW_OPEN, (void*)&data);
if (!req) {
UHD_LOG_ERROR(
"DPDK::IO_SERVICE", "Could not allocate wait_req to attach recv_link");
throw uhd::runtime_error("DPDK: Could not allocate wait_req to attach recv_link");
}
_servq.submit(req, std::chrono::microseconds(-1));
dpdk::wait_req_put(req);
{
std::lock_guard<std::mutex> lock(_mutex);
_recv_links.push_back(link);
}
}
void dpdk_io_service::attach_send_link(send_link_if::sptr link)
{
udp_dpdk_link* dpdk_link = dynamic_cast<udp_dpdk_link*>(link.get());
assert(dpdk_link);
// First, fill in destination MAC address
struct dpdk::arp_request arp_data;
arp_data.tpa = dpdk_link->get_remote_ipv4();
arp_data.port = dpdk_link->get_port()->get_port_id();
if (dpdk_link->get_port()->dst_is_broadcast(arp_data.tpa)) {
// If a broadcast IP, skip the ARP and fill with broadcast MAC addr
memset(arp_data.tha.addr_bytes, 0xFF, 6);
} else {
auto arp_req = wait_req_alloc(dpdk::wait_type::WAIT_ARP, (void*)&arp_data);
if (!arp_req) {
UHD_LOG_ERROR(
"DPDK::IO_SERVICE", "Could not allocate wait_req for ARP request");
throw uhd::runtime_error("DPDK: Could not allocate wait_req for ARP request");
}
if (_servq.submit(arp_req, std::chrono::microseconds(3000000))) {
// Try one more time...
auto arp_req2 = wait_req_alloc(dpdk::wait_type::WAIT_ARP, (void*)&arp_data);
if (_servq.submit(arp_req2, std::chrono::microseconds(30000000))) {
wait_req_put(arp_req);
wait_req_put(arp_req2);
throw uhd::io_error("DPDK: Could not reach host");
}
wait_req_put(arp_req2);
}
wait_req_put(arp_req);
}
dpdk_link->set_remote_mac(arp_data.tha);
// Then, submit the link to the I/O service thread
struct dpdk_flow_data data;
data.link = dpdk_link;
data.is_recv = false;
auto req = wait_req_alloc(dpdk::wait_type::WAIT_FLOW_OPEN, (void*)&data);
if (!req) {
UHD_LOG_ERROR(
"DPDK::IO_SERVICE", "Could not allocate wait_req to attach send_link");
throw uhd::runtime_error("DPDK: Could not allocate wait_req to attach send_link");
}
_servq.submit(req, std::chrono::microseconds(-1));
wait_req_put(req);
{
std::lock_guard<std::mutex> lock(_mutex);
_send_links.push_back(link);
}
}
void dpdk_io_service::detach_recv_link(recv_link_if::sptr link)
{
auto link_ptr = link.get();
struct dpdk_flow_data data;
data.link = dynamic_cast<udp_dpdk_link*>(link_ptr);
data.is_recv = true;
auto req = wait_req_alloc(dpdk::wait_type::WAIT_FLOW_CLOSE, (void*)&data);
if (!req) {
UHD_LOG_ERROR(
"DPDK::IO_SERVICE", "Could not allocate wait_req to detach recv_link");
throw uhd::runtime_error("DPDK: Could not allocate wait_req to detach recv_link");
}
_servq.submit(req, std::chrono::microseconds(-1));
wait_req_put(req);
{
std::lock_guard<std::mutex> lock(_mutex);
_recv_links.remove_if(
[link_ptr](recv_link_if::sptr& item) { return item.get() == link_ptr; });
}
}
void dpdk_io_service::detach_send_link(send_link_if::sptr link)
{
auto link_ptr = link.get();
struct dpdk_flow_data data;
data.link = dynamic_cast<udp_dpdk_link*>(link_ptr);
data.is_recv = false;
auto req = wait_req_alloc(dpdk::wait_type::WAIT_FLOW_CLOSE, (void*)&data);
if (!req) {
UHD_LOG_ERROR(
"DPDK::IO_SERVICE", "Could not allocate wait_req to detach send_link");
throw uhd::runtime_error("DPDK: Could not allocate wait_req to detach send_link");
}
_servq.submit(req, std::chrono::microseconds(-1));
wait_req_put(req);
{
std::lock_guard<std::mutex> lock(_mutex);
_send_links.remove_if(
[link_ptr](send_link_if::sptr& item) { return item.get() == link_ptr; });
}
}
recv_io_if::sptr dpdk_io_service::make_recv_client(recv_link_if::sptr data_link,
size_t num_recv_frames,
recv_callback_t cb,
send_link_if::sptr /*fc_link*/,
size_t num_send_frames,
recv_io_if::fc_callback_t fc_cb)
{
auto link = dynamic_cast<udp_dpdk_link*>(data_link.get());
auto recv_io = std::make_shared<dpdk_recv_io>(
shared_from_this(), link, num_recv_frames, cb, num_send_frames, fc_cb);
// Register with I/O service
recv_io->_dpdk_io_if.io_client = static_cast<void*>(recv_io.get());
auto xport_req = dpdk::wait_req_alloc(
dpdk::wait_type::WAIT_XPORT_CONNECT, (void*)&recv_io->_dpdk_io_if);
_servq.submit(xport_req, std::chrono::microseconds(-1));
wait_req_put(xport_req);
return recv_io;
}
send_io_if::sptr dpdk_io_service::make_send_client(send_link_if::sptr send_link,
size_t num_send_frames,
send_io_if::send_callback_t send_cb,
recv_link_if::sptr /*recv_link*/,
size_t num_recv_frames,
recv_callback_t recv_cb,
send_io_if::fc_callback_t fc_cb)
{
auto link = dynamic_cast<udp_dpdk_link*>(send_link.get());
auto send_io = std::make_shared<dpdk_send_io>(shared_from_this(),
link,
num_send_frames,
send_cb,
num_recv_frames,
recv_cb,
fc_cb);
// Register with I/O service
send_io->_dpdk_io_if.io_client = static_cast<void*>(send_io.get());
auto xport_req = dpdk::wait_req_alloc(
dpdk::wait_type::WAIT_XPORT_CONNECT, (void*)&send_io->_dpdk_io_if);
_servq.submit(xport_req, std::chrono::microseconds(-1));
wait_req_put(xport_req);
return send_io;
}
int dpdk_io_service::_io_worker(void* arg)
{
if (!arg)
return -EINVAL;
dpdk_io_service* srv = (dpdk_io_service*)arg;
/* Check that this is a valid lcore */
unsigned int lcore_id = rte_lcore_id();
if (lcore_id == LCORE_ID_ANY)
return -ENODEV;
/* Check that this lcore has ports */
if (srv->_ports.size() == 0)
return -ENODEV;
char name[16];
snprintf(name, sizeof(name), "dpdk-io_%hu", (uint16_t)lcore_id);
rte_thread_setname(pthread_self(), name);
UHD_LOG_TRACE("DPDK::IO_SERVICE",
"I/O service thread '" << name << "' started on lcore " << lcore_id);
uhd::set_thread_priority_safe();
snprintf(name, sizeof(name), "rx-tbl_%hu", (uint16_t)lcore_id);
struct rte_hash_parameters hash_params = {.name = name,
.entries = MAX_FLOWS,
.reserved = 0,
.key_len = sizeof(struct dpdk::ipv4_5tuple),
.hash_func = NULL,
.hash_func_init_val = 0,
.socket_id = uhd::narrow_cast<int>(rte_socket_id()),
.extra_flag = 0};
srv->_rx_table = rte_hash_create(&hash_params);
if (srv->_rx_table == NULL) {
return rte_errno;
}
int status = 0;
while (!status) {
/* For each port, attempt to receive packets and process */
for (auto port : srv->_ports) {
srv->_rx_burst(port, 0);
}
/* For each port's TX queues, do TX */
for (auto port : srv->_ports) {
srv->_tx_burst(port);
}
/* For each port's RX release queues, release buffers */
for (auto port : srv->_ports) {
srv->_rx_release(port);
}
/* Retry waking clients */
if (srv->_retry_head) {
dpdk_io_if* node = srv->_retry_head;
dpdk_io_if* end = srv->_retry_head->prev;
while (true) {
dpdk_io_if* next = node->next;
srv->_wake_client(node);
if (node == end) {
break;
} else {
node = next;
next = node->next;
}
}
}
/* Check for open()/close()/term() requests and service 1 at a time
* Leave this last so we immediately terminate if requested
*/
status = srv->_service_requests();
}
return status;
}
int dpdk_io_service::_service_requests()
{
for (int i = 0; i < MAX_PENDING_SERVICE_REQS; i++) {
/* Dequeue */
dpdk::wait_req* req = _servq.pop();
if (!req) {
break;
}
switch (req->reason) {
case dpdk::wait_type::WAIT_SIMPLE:
while (_servq.complete(req) == -ENOBUFS)
;
break;
case dpdk::wait_type::WAIT_RX:
case dpdk::wait_type::WAIT_TX_BUF:
throw uhd::not_implemented_error(
"DPDK: _service_requests(): DPDK is still a WIP");
case dpdk::wait_type::WAIT_FLOW_OPEN:
_service_flow_open(req);
break;
case dpdk::wait_type::WAIT_FLOW_CLOSE:
_service_flow_close(req);
break;
case dpdk::wait_type::WAIT_XPORT_CONNECT:
_service_xport_connect(req);
break;
case dpdk::wait_type::WAIT_XPORT_DISCONNECT:
_service_xport_disconnect(req);
break;
case dpdk::wait_type::WAIT_ARP: {
assert(req->data != NULL);
int arp_status = _service_arp_request(req);
assert(arp_status != -ENOMEM);
if (arp_status == 0) {
while (_servq.complete(req) == -ENOBUFS)
;
}
break;
}
case dpdk::wait_type::WAIT_LCORE_TERM:
rte_free(_rx_table);
while (_servq.complete(req) == -ENOBUFS)
;
// Return a positive value to indicate we should terminate
return 1;
default:
UHD_LOG_ERROR(
"DPDK::IO_SERVICE", "Invalid reason associated with wait request");
while (_servq.complete(req) == -ENOBUFS)
;
break;
}
}
return 0;
}
void dpdk_io_service::_service_flow_open(dpdk::wait_req* req)
{
auto flow_req_data = (struct dpdk_flow_data*)req->data;
assert(flow_req_data);
if (flow_req_data->is_recv) {
// If RX, add to RX table. Currently, nothing to do for TX.
struct dpdk::ipv4_5tuple ht_key = {.flow_type = dpdk::flow_type::FLOW_TYPE_UDP,
.src_ip = 0,
.dst_ip = flow_req_data->link->get_port()->get_ipv4(),
.src_port = 0,
.dst_port = flow_req_data->link->get_local_port()};
// Check the UDP port isn't in use
if (rte_hash_lookup(_rx_table, &ht_key) > 0) {
req->retval = -EADDRINUSE;
UHD_LOG_ERROR("DPDK::IO_SERVICE", "Cannot add to RX table");
while (_servq.complete(req) == -ENOBUFS)
;
return;
}
// Add xport list for this UDP port
auto rx_entry = new std::list<dpdk_io_if*>();
if (rte_hash_add_key_data(_rx_table, &ht_key, rx_entry)) {
UHD_LOG_ERROR("DPDK::IO_SERVICE", "Could not add new RX list to table");
delete rx_entry;
req->retval = -ENOMEM;
while (_servq.complete(req) == -ENOBUFS)
;
return;
}
}
while (_servq.complete(req) == -ENOBUFS)
;
}
void dpdk_io_service::_service_flow_close(dpdk::wait_req* req)
{
auto flow_req_data = (struct dpdk_flow_data*)req->data;
assert(flow_req_data);
if (flow_req_data->is_recv) {
// If RX, remove from RX table. Currently, nothing to do for TX.
struct dpdk::ipv4_5tuple ht_key = {.flow_type = dpdk::flow_type::FLOW_TYPE_UDP,
.src_ip = 0,
.dst_ip = flow_req_data->link->get_port()->get_ipv4(),
.src_port = 0,
.dst_port = flow_req_data->link->get_local_port()};
std::list<dpdk_io_if*>* xport_list;
if (rte_hash_lookup_data(_rx_table, &ht_key, (void**)&xport_list) > 0) {
UHD_ASSERT_THROW(xport_list->empty());
delete xport_list;
rte_hash_del_key(_rx_table, &ht_key);
while (_servq.complete(req) == -ENOBUFS)
;
return;
}
}
while (_servq.complete(req) == -ENOBUFS)
;
}
void dpdk_io_service::_service_xport_connect(dpdk::wait_req* req)
{
auto dpdk_io = static_cast<dpdk_io_if*>(req->data);
UHD_ASSERT_THROW(dpdk_io);
auto port = dpdk_io->link->get_port();
if (dpdk_io->recv_cb) {
// Add to RX table only if have a callback.
struct dpdk::ipv4_5tuple ht_key = {.flow_type = dpdk::flow_type::FLOW_TYPE_UDP,
.src_ip = 0,
.dst_ip = port->get_ipv4(),
.src_port = 0,
.dst_port = dpdk_io->link->get_local_port()};
void* hash_data;
if (rte_hash_lookup_data(_rx_table, &ht_key, &hash_data) < 0) {
req->retval = -ENOENT;
UHD_LOG_ERROR("DPDK::IO_SERVICE", "Cannot add xport to RX table");
while (_servq.complete(req) == -ENOBUFS)
;
return;
}
// Add to xport list for this UDP port
auto rx_entry = (std::list<dpdk_io_if*>*)(hash_data);
rx_entry->push_back(dpdk_io);
}
if (dpdk_io->is_recv) {
UHD_LOG_TRACE("DPDK::IO_SERVICE", "Servicing RX connect request...");
// Add to xport list for this NIC port
auto& xport_list = _recv_xport_map.at(port->get_port_id());
xport_list.push_back((dpdk_recv_io*)dpdk_io->io_client);
} else {
UHD_LOG_TRACE("DPDK::IO_SERVICE", "Servicing TX connect request...");
dpdk_send_io* send_io = static_cast<dpdk_send_io*>(dpdk_io->io_client);
// Add to xport list for this NIC port
auto& xport_list = _tx_queues.at(port->get_port_id());
xport_list.push_back(send_io);
for (size_t i = 0; i < send_io->_num_send_frames; i++) {
auto buff_ptr =
(dpdk::dpdk_frame_buff*)dpdk_io->link->get_send_buff(0).release();
if (!buff_ptr) {
UHD_LOG_ERROR("DPDK::IO_SERVICE",
"TX mempool out of memory. Please increase dpdk_num_mbufs.");
break;
}
if (rte_ring_enqueue(send_io->_buffer_queue, buff_ptr)) {
rte_pktmbuf_free(buff_ptr->get_pktmbuf());
break;
}
send_io->_num_frames_in_use++;
}
}
while (_servq.complete(req) == -ENOBUFS)
;
}
void dpdk_io_service::_service_xport_disconnect(dpdk::wait_req* req)
{
auto dpdk_io = (struct dpdk_io_if*)req->data;
assert(dpdk_io);
auto port = dpdk_io->link->get_port();
if (dpdk_io->recv_cb) {
// Remove from RX table only if have a callback.
struct dpdk::ipv4_5tuple ht_key = {.flow_type = dpdk::flow_type::FLOW_TYPE_UDP,
.src_ip = 0,
.dst_ip = port->get_ipv4(),
.src_port = 0,
.dst_port = dpdk_io->link->get_local_port()};
void* hash_data;
if (rte_hash_lookup_data(_rx_table, &ht_key, &hash_data) >= 0) {
// Remove from xport list for this UDP port
auto rx_entry = (std::list<dpdk_io_if*>*)(hash_data);
rx_entry->remove(dpdk_io);
} else {
req->retval = -EINVAL;
UHD_LOG_ERROR("DPDK::IO_SERVICE", "Cannot remove xport from RX table");
}
}
if (dpdk_io->is_recv) {
UHD_LOG_TRACE("DPDK::IO_SERVICE", "Servicing RX disconnect request...");
dpdk_recv_io* recv_client = static_cast<dpdk_recv_io*>(dpdk_io->io_client);
// Remove from xport list for this NIC port
auto& xport_list = _recv_xport_map.at(port->get_port_id());
xport_list.remove(recv_client);
while (!rte_ring_empty(recv_client->_recv_queue)) {
frame_buff* buff_ptr;
rte_ring_dequeue(recv_client->_recv_queue, (void**)&buff_ptr);
dpdk_io->link->release_recv_buff(frame_buff::uptr(buff_ptr));
}
while (!rte_ring_empty(recv_client->_release_queue)) {
frame_buff* buff_ptr;
rte_ring_dequeue(recv_client->_release_queue, (void**)&buff_ptr);
dpdk_io->link->release_recv_buff(frame_buff::uptr(buff_ptr));
}
} else {
UHD_LOG_TRACE("DPDK::IO_SERVICE", "Servicing TX disconnect request...");
dpdk_send_io* send_client = static_cast<dpdk_send_io*>(dpdk_io->io_client);
// Remove from xport list for this NIC port
auto& xport_list = _tx_queues.at(port->get_port_id());
xport_list.remove(send_client);
while (!rte_ring_empty(send_client->_send_queue)) {
frame_buff* buff_ptr;
rte_ring_dequeue(send_client->_send_queue, (void**)&buff_ptr);
dpdk_io->link->release_send_buff(frame_buff::uptr(buff_ptr));
}
while (!rte_ring_empty(send_client->_buffer_queue)) {
frame_buff* buff_ptr;
rte_ring_dequeue(send_client->_buffer_queue, (void**)&buff_ptr);
dpdk_io->link->release_send_buff(frame_buff::uptr(buff_ptr));
}
}
// Now remove the node if it's on the retry list
if ((_retry_head == dpdk_io) && (dpdk_io->next == dpdk_io)) {
_retry_head = NULL;
} else if (_retry_head) {
dpdk_io_if* node = _retry_head->next;
while (node != _retry_head) {
if (node == dpdk_io) {
dpdk_io->prev->next = dpdk_io->next;
dpdk_io->next->prev = dpdk_io->prev;
break;
}
node = node->next;
}
}
while (_servq.complete(req) == -ENOBUFS)
;
}
int dpdk_io_service::_service_arp_request(dpdk::wait_req* req)
{
int status = 0;
auto arp_req_data = (struct dpdk::arp_request*)req->data;
dpdk::ipv4_addr dst_addr = arp_req_data->tpa;
auto ctx_sptr = _ctx.lock();
UHD_ASSERT_THROW(ctx_sptr);
dpdk::dpdk_port* port = ctx_sptr->get_port(arp_req_data->port);
UHD_LOG_TRACE("DPDK::IO_SERVICE",
"ARP: Requesting address for " << dpdk::ipv4_num_to_str(dst_addr));
rte_spinlock_lock(&port->_spinlock);
struct dpdk::arp_entry* entry = NULL;
if (port->_arp_table.count(dst_addr) == 0) {
entry = (struct dpdk::arp_entry*)rte_zmalloc(NULL, sizeof(*entry), 0);
if (!entry) {
status = -ENOMEM;
goto arp_end;
}
entry = new (entry) dpdk::arp_entry();
entry->reqs.push_back(req);
port->_arp_table[dst_addr] = entry;
status = -EAGAIN;
UHD_LOG_TRACE("DPDK::IO_SERVICE", "Address not in table. Sending ARP request.");
_send_arp_request(port, 0, arp_req_data->tpa);
} else {
entry = port->_arp_table.at(dst_addr);
if (is_zero_ether_addr(&entry->mac_addr)) {
UHD_LOG_TRACE("DPDK::IO_SERVICE",
"ARP: Address in table, but not populated yet. Resending ARP request.");
port->_arp_table.at(dst_addr)->reqs.push_back(req);
status = -EAGAIN;
_send_arp_request(port, 0, arp_req_data->tpa);
} else {
UHD_LOG_TRACE("DPDK::IO_SERVICE", "ARP: Address in table.");
ether_addr_copy(&entry->mac_addr, &arp_req_data->tha);
status = 0;
}
}
arp_end:
rte_spinlock_unlock(&port->_spinlock);
return status;
}
int dpdk_io_service::_send_arp_request(
dpdk::dpdk_port* port, dpdk::queue_id_t queue, dpdk::ipv4_addr ip)
{
struct rte_mbuf* mbuf;
struct ether_hdr* hdr;
struct arp_hdr* arp_frame;
mbuf = rte_pktmbuf_alloc(port->get_tx_pktbuf_pool());
if (unlikely(mbuf == NULL)) {
UHD_LOG_WARNING(
"DPDK::IO_SERVICE", "Could not allocate packet buffer for ARP request");
return -ENOMEM;
}
hdr = rte_pktmbuf_mtod(mbuf, struct ether_hdr*);
arp_frame = (struct arp_hdr*)&hdr[1];
memset(hdr->d_addr.addr_bytes, 0xFF, ETHER_ADDR_LEN);
hdr->s_addr = port->get_mac_addr();
hdr->ether_type = rte_cpu_to_be_16(ETHER_TYPE_ARP);
arp_frame->arp_hrd = rte_cpu_to_be_16(ARP_HRD_ETHER);
arp_frame->arp_pro = rte_cpu_to_be_16(ETHER_TYPE_IPv4);
arp_frame->arp_hln = 6;
arp_frame->arp_pln = 4;
arp_frame->arp_op = rte_cpu_to_be_16(ARP_OP_REQUEST);
arp_frame->arp_data.arp_sha = port->get_mac_addr();
arp_frame->arp_data.arp_sip = port->get_ipv4();
memset(arp_frame->arp_data.arp_tha.addr_bytes, 0x00, ETHER_ADDR_LEN);
arp_frame->arp_data.arp_tip = ip;
mbuf->pkt_len = 42;
mbuf->data_len = 42;
if (rte_eth_tx_burst(port->get_port_id(), queue, &mbuf, 1) != 1) {
UHD_LOG_WARNING("DPDK::IO_SERVICE", "ARP request not sent: Descriptor ring full");
rte_pktmbuf_free(mbuf);
return -EAGAIN;
}
return 0;
}
/* Do a burst of RX on port */
int dpdk_io_service::_rx_burst(dpdk::dpdk_port* port, dpdk::queue_id_t queue)
{
struct ether_hdr* hdr;
char* l2_data;
struct rte_mbuf* bufs[RX_BURST_SIZE];
const uint16_t num_rx =
rte_eth_rx_burst(port->get_port_id(), queue, bufs, RX_BURST_SIZE);
if (unlikely(num_rx == 0)) {
return 0;
}
for (int buf = 0; buf < num_rx; buf++) {
uint64_t ol_flags = bufs[buf]->ol_flags;
hdr = rte_pktmbuf_mtod(bufs[buf], struct ether_hdr*);
l2_data = (char*)&hdr[1];
switch (rte_be_to_cpu_16(hdr->ether_type)) {
case ETHER_TYPE_ARP:
_process_arp(port, queue, (struct arp_hdr*)l2_data);
rte_pktmbuf_free(bufs[buf]);
break;
case ETHER_TYPE_IPv4:
if ((ol_flags & PKT_RX_IP_CKSUM_MASK) == PKT_RX_IP_CKSUM_BAD) {
UHD_LOG_WARNING("DPDK::IO_SERVICE", "RX packet has bad IP cksum");
} else if ((ol_flags & PKT_RX_IP_CKSUM_MASK) == PKT_RX_IP_CKSUM_NONE) {
UHD_LOG_WARNING("DPDK::IO_SERVICE", "RX packet missing IP cksum");
} else {
_process_ipv4(port, bufs[buf], (struct ipv4_hdr*)l2_data);
}
break;
default:
rte_pktmbuf_free(bufs[buf]);
break;
}
}
return num_rx;
}
int dpdk_io_service::_process_arp(
dpdk::dpdk_port* port, dpdk::queue_id_t queue_id, struct arp_hdr* arp_frame)
{
uint32_t dest_ip = arp_frame->arp_data.arp_sip;
struct ether_addr dest_addr = arp_frame->arp_data.arp_sha;
UHD_LOG_TRACE("DPDK::IO_SERVICE",
"Processing ARP packet: " << dpdk::ipv4_num_to_str(dest_ip) << " -> "
<< dpdk::eth_addr_to_string(dest_addr));
/* Add entry to ARP table */
rte_spinlock_lock(&port->_spinlock);
struct dpdk::arp_entry* entry = NULL;
if (port->_arp_table.count(dest_ip) == 0) {
entry = (struct dpdk::arp_entry*)rte_zmalloc(NULL, sizeof(*entry), 0);
if (!entry) {
return -ENOMEM;
}
entry = new (entry) dpdk::arp_entry();
ether_addr_copy(&dest_addr, &entry->mac_addr);
port->_arp_table[dest_ip] = entry;
} else {
entry = port->_arp_table.at(dest_ip);
ether_addr_copy(&dest_addr, &entry->mac_addr);
for (auto req : entry->reqs) {
auto arp_data = (struct dpdk::arp_request*)req->data;
ether_addr_copy(&dest_addr, &arp_data->tha);
while (_servq.complete(req) == -ENOBUFS)
;
}
entry->reqs.clear();
}
rte_spinlock_unlock(&port->_spinlock);
/* Respond if this was an ARP request */
if (arp_frame->arp_op == rte_cpu_to_be_16(ARP_OP_REQUEST)
&& arp_frame->arp_data.arp_tip == port->get_ipv4()) {
UHD_LOG_TRACE("DPDK::IO_SERVICE", "Sending ARP reply.");
port->_arp_reply(queue_id, arp_frame);
}
return 0;
}
int dpdk_io_service::_process_ipv4(
dpdk::dpdk_port* port, struct rte_mbuf* mbuf, struct ipv4_hdr* pkt)
{
bool bcast = port->dst_is_broadcast(pkt->dst_addr);
if (pkt->dst_addr != port->get_ipv4() && !bcast) {
rte_pktmbuf_free(mbuf);
return -ENODEV;
}
if (pkt->next_proto_id == IPPROTO_UDP) {
return _process_udp(port, mbuf, (struct udp_hdr*)&pkt[1], bcast);
}
rte_pktmbuf_free(mbuf);
return -EINVAL;
}
int dpdk_io_service::_process_udp(
dpdk::dpdk_port* port, struct rte_mbuf* mbuf, struct udp_hdr* pkt, bool /*bcast*/)
{
// Get the link
struct dpdk::ipv4_5tuple ht_key = {.flow_type = dpdk::flow_type::FLOW_TYPE_UDP,
.src_ip = 0,
.dst_ip = port->get_ipv4(),
.src_port = 0,
.dst_port = pkt->dst_port};
void* hash_data;
if (rte_hash_lookup_data(_rx_table, &ht_key, &hash_data) < 0) {
UHD_LOG_WARNING("DPDK::IO_SERVICE", "Dropping packet: No link entry in rx table");
rte_pktmbuf_free(mbuf);
return -ENOENT;
}
// Get xport list for this UDP port
auto rx_entry = (std::list<dpdk_io_if*>*)(hash_data);
if (rx_entry->empty()) {
UHD_LOG_WARNING("DPDK::IO_SERVICE", "Dropping packet: No xports for link");
rte_pktmbuf_free(mbuf);
return -ENOENT;
}
// Turn rte_mbuf -> dpdk_frame_buff
auto link = rx_entry->front()->link;
link->enqueue_recv_mbuf(mbuf);
auto buff = link->get_recv_buff(0);
bool rcvr_found = false;
for (auto client_if : *rx_entry) {
// Check all the muxed receivers...
if (client_if->recv_cb(buff, link, link)) {
rcvr_found = true;
if (buff) {
assert(client_if->is_recv);
auto recv_io = (dpdk_recv_io*)client_if->io_client;
auto buff_ptr = (dpdk::dpdk_frame_buff*)buff.release();
if (rte_ring_enqueue(recv_io->_recv_queue, buff_ptr)) {
rte_pktmbuf_free(buff_ptr->get_pktmbuf());
UHD_LOG_WARNING(
"DPDK::IO_SERVICE", "Dropping packet: No space in recv queue");
} else {
recv_io->_num_frames_in_use++;
assert(recv_io->_num_frames_in_use <= recv_io->_num_recv_frames);
_wake_client(client_if);
}
}
break;
}
}
if (!rcvr_found) {
UHD_LOG_WARNING("DPDK::IO_SERVICE", "Dropping packet: No receiver xport found");
// Release the buffer if no receiver found
link->release_recv_buff(std::move(buff));
return -ENOENT;
}
return 0;
}
/* Do a burst of TX on port's tx queues */
int dpdk_io_service::_tx_burst(dpdk::dpdk_port* port)
{
unsigned int total_tx = 0;
auto& queues = _tx_queues.at(port->get_port_id());
for (auto& send_io : queues) {
unsigned int num_tx = rte_ring_count(send_io->_send_queue);
num_tx = (num_tx < TX_BURST_SIZE) ? num_tx : TX_BURST_SIZE;
bool replaced_buffers = false;
for (unsigned int i = 0; i < num_tx; i++) {
size_t frame_size = send_io->_dpdk_io_if.link->get_send_frame_size();
if (send_io->_fc_cb && !send_io->_fc_cb(frame_size)) {
break;
}
dpdk::dpdk_frame_buff* buff_ptr;
int status = rte_ring_dequeue(send_io->_send_queue, (void**)&buff_ptr);
if (status) {
UHD_LOG_ERROR("DPDK::IO_SERVICE", "TX Q Count doesn't match actual");
break;
}
send_io->_send_cb(frame_buff::uptr(buff_ptr), send_io->_dpdk_io_if.link);
// Attempt to replace buffer
buff_ptr = (dpdk::dpdk_frame_buff*)send_io->_dpdk_io_if.link->get_send_buff(0)
.release();
if (!buff_ptr) {
UHD_LOG_ERROR("DPDK::IO_SERVICE",
"TX mempool out of memory. Please increase dpdk_num_mbufs.");
send_io->_num_frames_in_use--;
} else if (rte_ring_enqueue(send_io->_buffer_queue, buff_ptr)) {
rte_pktmbuf_free(buff_ptr->get_pktmbuf());
send_io->_num_frames_in_use--;
} else {
replaced_buffers = true;
}
}
if (replaced_buffers) {
_wake_client(&send_io->_dpdk_io_if);
}
total_tx += num_tx;
}
return total_tx;
}
int dpdk_io_service::_rx_release(dpdk::dpdk_port* port)
{
unsigned int total_bufs = 0;
auto& queues = _recv_xport_map.at(port->get_port_id());
for (auto& recv_io : queues) {
unsigned int num_buf = rte_ring_count(recv_io->_release_queue);
num_buf = (num_buf < RX_BURST_SIZE) ? num_buf : RX_BURST_SIZE;
for (unsigned int i = 0; i < num_buf; i++) {
dpdk::dpdk_frame_buff* buff_ptr;
int status = rte_ring_dequeue(recv_io->_release_queue, (void**)&buff_ptr);
if (status) {
UHD_LOG_ERROR("DPDK::IO_SERVICE", "RX Q Count doesn't match actual");
break;
}
recv_io->_fc_cb(frame_buff::uptr(buff_ptr),
recv_io->_dpdk_io_if.link,
recv_io->_dpdk_io_if.link);
recv_io->_num_frames_in_use--;
}
total_bufs += num_buf;
}
return total_bufs;
}
uint16_t dpdk_io_service::_get_unique_client_id()
{
std::lock_guard<std::mutex> lock(_mutex);
if (_client_id_set.size() >= MAX_CLIENTS) {
UHD_LOG_ERROR("DPDK::IO_SERVICE", "Exceeded maximum number of clients");
throw uhd::runtime_error("DPDK::IO_SERVICE: Exceeded maximum number of clients");
}
uint16_t id = _next_client_id++;
while (_client_id_set.count(id)) {
id = _next_client_id++;
}
_client_id_set.insert(id);
return id;
}
void dpdk_io_service::_wake_client(dpdk_io_if* dpdk_io)
{
dpdk::wait_req* req;
if (dpdk_io->is_recv) {
auto recv_io = static_cast<dpdk_recv_io*>(dpdk_io->io_client);
req = recv_io->_waiter;
} else {
auto send_io = static_cast<dpdk_send_io*>(dpdk_io->io_client);
req = send_io->_waiter;
}
bool stat = req->mutex.try_lock();
if (stat) {
bool active_req = !req->complete;
if (dpdk_io->next) {
// On the list: Take it off
if (dpdk_io->next == dpdk_io) {
// Only node on the list
_retry_head = NULL;
} else {
// Other nodes are on the list
if (_retry_head == dpdk_io) {
// Move list head to next
_retry_head = dpdk_io->next;
}
dpdk_io->next->prev = dpdk_io->prev;
dpdk_io->prev->next = dpdk_io->next;
}
dpdk_io->next = NULL;
dpdk_io->prev = NULL;
}
if (active_req) {
req->complete = true;
req->cond.notify_one();
}
req->mutex.unlock();
if (active_req) {
wait_req_put(req);
}
} else {
// Put on the retry list, if it isn't already
if (!dpdk_io->next) {
if (_retry_head) {
dpdk_io->next = _retry_head;
dpdk_io->prev = _retry_head->prev;
_retry_head->prev->next = dpdk_io;
_retry_head->prev = dpdk_io;
} else {
_retry_head = dpdk_io;
dpdk_io->next = dpdk_io;
dpdk_io->prev = dpdk_io;
}
}
}
}