dpdk: Add new DPDK stack to integrate with I/O services

docs: Update DPDK docs with new parameters:
Parameter names have had their hyphens changed to underscores, and
the I/O CPU argument is now named after the lcores and reflects
the naming used by DPDK.

transport: Add new udp_dpdk_link, based atop the new APIs:
This link is tightly coupled with the DPDK I/O service. The link class
carries all the address information to communicate with the other
host, and it can send packets directly through the DPDK NIC ports.

However, for receiving packets, the I/O service must pull the packets
from the DMA queue and attach them to the appropriate link object.
The link object merely formats the frame_buff object underneath, which
is embedded in the rte_mbuf container. For get_recv_buff, the link
will pull buffers only from its internal queue (the one filled by the
I/O service).

transport: Add DPDK-specific I/O service:
The I/O service is split into two parts, the user threads and the
I/O worker threads. The user threads submit requests through
various appropriate queues, and the I/O threads perform all the
I/O on their behalf. This includes routing UDP packets to the
correct receiver and getting the MAC address of a destination (by
performing the ARP request and handling the ARP replies).

The DPDK context stores I/O services. The context spawns all I/O
services on init(), and I/O services can be fetched from the dpdk_ctx
object by using a port ID.

I/O service clients:
The clients have two lockless ring buffers. One is to get a buffer
from the I/O service; the other is to release a buffer back to the
I/O service. Threads sleeping on buffer I/O are kept in a separate
list from the service queue and are processed in the course of doing
RX or TX.

The list nodes are embedded in the dpdk_io_if, and the head of the
list is on the dpdk_io_service. The I/O service will transfer the
embedded wait_req to the list if it cannot acquire the mutex to
complete the condition for waking.

Co-authored-by: Martin Braun <martin.braun@ettus.com>
Co-authored-by: Ciro Nishiguchi <ciro.nishiguchi@ni.com>
Co-authored-by: Brent Stapleton <brent.stapleton@ettus.com>

4 files changed, 317 insertions, 59 deletions

diff --git a/host/lib/include/uhdlib/transport/dpdk/arp.hpp b/host/lib/include/uhdlib/transport/dpdk/arp.hpp
new file mode 100644
index 000000000..e71119bb3
--- /dev/null
+++ b/host/lib/include/uhdlib/transport/dpdk/arp.hpp
@@ -0,0 +1,29 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+#ifndef _INCLUDED_UHDLIB_TRANSPORT_DPDK_ARP_HPP_
+#define _INCLUDED_UHDLIB_TRANSPORT_DPDK_ARP_HPP_
+
+#include <uhdlib/transport/dpdk/common.hpp>
+#include <uhdlib/transport/dpdk/service_queue.hpp>
+#include <rte_arp.h>
+
+namespace uhd { namespace transport { namespace dpdk {
+
+struct arp_request
+{
+    struct ether_addr tha;
+    port_id_t port;
+    ipv4_addr tpa;
+};
+
+struct arp_entry
+{
+    struct ether_addr mac_addr;
+    std::vector<wait_req*> reqs;
+};
+
+}}} /* namespace uhd::transport::dpdk */
+#endif /* _INCLUDED_UHDLIB_TRANSPORT_DPDK_ARP_HPP_ */
diff --git a/host/lib/include/uhdlib/transport/dpdk/common.hpp b/host/lib/include/uhdlib/transport/dpdk/common.hpp
index 1f4466a0f..ac3526e49 100644
--- a/host/lib/include/uhdlib/transport/dpdk/common.hpp
+++ b/host/lib/include/uhdlib/transport/dpdk/common.hpp
@@ -1,44 +1,130 @@
 //
-// Copyright 2019 Ettus Research, a National Instruments brand
+// Copyright 2019 Ettus Research, a National Instruments Brand
 //
 // SPDX-License-Identifier: GPL-3.0-or-later
 //
+
 #ifndef _INCLUDED_UHDLIB_TRANSPORT_DPDK_COMMON_HPP_
 #define _INCLUDED_UHDLIB_TRANSPORT_DPDK_COMMON_HPP_
 
 #include <uhd/config.hpp>
+#include <uhd/transport/frame_buff.hpp>
 #include <uhd/types/device_addr.hpp>
-#include <uhd/utils/noncopyable.hpp>
-#include <uhd/utils/static.hpp>
+#include <uhdlib/transport/adapter_info.hpp>
 #include <rte_ethdev.h>
 #include <rte_ether.h>
 #include <rte_flow.h>
 #include <rte_mbuf.h>
 #include <rte_mempool.h>
+#include <rte_spinlock.h>
 #include <rte_version.h>
 #include <unordered_map>
 #include <array>
 #include <atomic>
 #include <mutex>
+#include <set>
 #include <string>
 
-/* NOTE: There are changes to rte_eth_addr in 19.x */
+/* NOTE: There are changes to all the network standard fields in 19.x */
+
+
+namespace uhd { namespace transport {
+
+class dpdk_io_service;
+
+namespace dpdk {
 
-namespace uhd { namespace transport { namespace dpdk {
+struct arp_entry;
 
 using queue_id_t = uint16_t;
 using port_id_t  = uint16_t;
 using ipv4_addr  = uint32_t;
 
+class dpdk_adapter_info : public adapter_info
+{
+public:
+    dpdk_adapter_info(port_id_t port) : _port(port) {}
+    ~dpdk_adapter_info() {}
+
+    std::string to_string()
+    {
+        return std::string("DPDK:") + std::to_string(_port);
+    }
+
+    bool operator==(const dpdk_adapter_info& rhs) const
+    {
+        return (_port == rhs._port);
+    }
+
+private:
+    // Port ID
+    port_id_t _port;
+};
+
+
+/*!
+ * Packet/Frame buffer class for DPDK
+ *
+ * This class is intended to be placed in the private area of the rte_mbuf, and
+ * its memory is part of the rte_mbuf, so its life is tied to the underlying
+ * buffer (or more precisely, the encapsulating one).
+ */
+class dpdk_frame_buff : public frame_buff
+{
+public:
+    dpdk_frame_buff(struct rte_mbuf* mbuf) : _mbuf(mbuf)
+    {
+        _data        = rte_pktmbuf_mtod(mbuf, void*);
+        _packet_size = 0;
+    }
+
+    ~dpdk_frame_buff() = default;
+
+    /*!
+     * Simple getter for the underlying rte_mbuf.
+     * The rte_mbuf may need further modification before sending packets,
+     * like adjusting the IP and UDP lengths.
+     */
+    inline struct rte_mbuf* get_pktmbuf()
+    {
+        return _mbuf;
+    }
+
+    /*!
+     * Move the data pointer by the indicated size, to some desired
+     * encapsulated frame.
+     *
+     * \param hdr_size Size (in bytes) of the headers to skip. Can be negative
+     *                 to pull the header back.
+     */
+    inline void header_jump(ssize_t hdr_size)
+    {
+        _data = (void*)((uint8_t*)_data + hdr_size);
+    }
+
+    //! Embedded list node's next ptr
+    dpdk_frame_buff* next = nullptr;
+    //! Embedded list node's prev ptr
+    dpdk_frame_buff* prev = nullptr;
+
+private:
+    struct rte_mbuf* _mbuf;
+};
+
+
+/*!
+ * The size (in bytes) of the private area reserved within the rte_mbuf.
+ * This portion of the rte_mbuf is used for the embedded dpdk_frame_buff data
+ * structure.
+ */
+constexpr size_t DPDK_MBUF_PRIV_SIZE =
+    RTE_ALIGN(sizeof(struct dpdk_frame_buff), RTE_MBUF_PRIV_ALIGN);
+
 /*!
  * Class representing a DPDK NIC port
  *
  * The dpdk_port object possesses all the data needed to send and receive
- * packets between this port and a remote host. A logical link should specify
- * which packets are destined for it and allocate a DMA queue with the
- * dpdk_port::alloc_queue() function. A logical link should not, however,
- * specify ARP packets for its set of received packets. That functionality is
- * reserved for the special queue 0.
+ * packets between this port and a remote host.
  *
  * The logical link can then get the packet buffer pools associated with this
  * NIC port and use them to send and receive packets.
@@ -55,7 +141,7 @@ public:
      * \param port The port ID
      * \param mtu The intended MTU for the port
      * \param num_queues Number of DMA queues to reserve for this port
-     * \param num_mbufs The number of packet buffers per queue
+     * \param num_desc The number of descriptors per DMA queue
      * \param rx_pktbuf_pool A pointer to the port's RX packet buffer pool
      * \param tx_pktbuf_pool A pointer to the port's TX packet buffer pool
      * \param ipv4_address The IPv4 network address (w/ netmask)
@@ -64,7 +150,7 @@ public:
     static dpdk_port::uptr make(port_id_t port,
         size_t mtu,
         uint16_t num_queues,
-        size_t num_mbufs,
+        uint16_t num_desc,
         struct rte_mempool* rx_pktbuf_pool,
         struct rte_mempool* tx_pktbuf_pool,
         std::string ipv4_address);
@@ -72,11 +158,13 @@ public:
     dpdk_port(port_id_t port,
         size_t mtu,
         uint16_t num_queues,
-        size_t num_mbufs,
+        uint16_t num_desc,
         struct rte_mempool* rx_pktbuf_pool,
         struct rte_mempool* tx_pktbuf_pool,
         std::string ipv4_address);
 
+    ~dpdk_port();
+
     /*! Getter for this port's ID
      * \return this port's ID
      */
@@ -85,6 +173,19 @@ public:
         return _port;
     }
 
+    inline dpdk_adapter_info get_adapter_info() const
+    {
+        return dpdk_adapter_info(_port);
+    }
+
+    /*! Getter for this port's MAC address
+     * \return this port's MAC address
+     */
+    inline ether_addr get_mac_addr() const
+    {
+        return _mac_addr;
+    }
+
     /*! Getter for this port's MTU
      * \return this port's MTU
      */
@@ -141,68 +242,45 @@ public:
      * \param dst_ipv4_addr The destination IPv4 address (in network order)
      * \return whether the destination address matches this port's broadcast address
      */
-    inline bool dst_is_broadcast(const uint32_t dst_ipv4_addr) const
+    inline bool dst_is_broadcast(const ipv4_addr dst_ipv4_addr) const
     {
         uint32_t network = _netmask | ((~_netmask) & dst_ipv4_addr);
         return (network == 0xffffffff);
     }
 
-    /*! Allocate a DMA queue (TX/RX pair) and use the specified flow pattern
-     * to route packets to the RX queue.
-     *
-     * \pattern recv_pattern The flow pattern to use for directing traffic to
-     *                       the allocated RX queue.
-     * \return The queue ID for the allocated queue
-     * \throw uhd::runtime_error when there are no free queues
-     */
-    queue_id_t alloc_queue(struct rte_flow_pattern recv_pattern[]);
-
-    /*! Free a previously allocated queue and tear down the associated flow rule
-     * \param queue The ID of the queue to free
-     * \throw std::out_of_range when the queue ID is not currently allocated
-     */
-    void free_queue(queue_id_t queue);
-
     /*!
-     * Process ARP request/reply
+     * Allocate a UDP port and return it in network order
+     *
+     * \param udp_port UDP port to attempt to allocate. Use 0 for no preference.
+     * \return 0 for failure, else the allocated UDP port in network order.
      */
-    // int process_arp(struct rte_mempool *tx_pktbuf_pool, struct arp_hdr *arp_frame);
+    uint16_t alloc_udp_port(uint16_t udp_port);
 
 private:
+    friend uhd::transport::dpdk_io_service;
+
     /*!
      * Construct and transmit an ARP reply (for the given ARP request)
      */
-    int _arp_reply(struct rte_mempool* tx_pktbuf_pool, struct arp_hdr* arp_req);
+    int _arp_reply(queue_id_t queue_id, struct arp_hdr* arp_req);
 
     port_id_t _port;
     size_t _mtu;
+    size_t _num_queues;
     struct rte_mempool* _rx_pktbuf_pool;
     struct rte_mempool* _tx_pktbuf_pool;
     struct ether_addr _mac_addr;
     ipv4_addr _ipv4;
     ipv4_addr _netmask;
-    size_t _num_queues;
-    std::vector<queue_id_t> _free_queues;
-    std::unordered_map<queue_id_t, struct rte_flow*> _flow_rules;
-    /* Need ARP table
-     * To implement ARP service, maybe create ARP xport
-     * Then need dpdk_udp_link and dpdk_raw_link
-     *
-     * ...Or just do it inline with dpdk_ctx
-     *
-     * And link can just save the result (do it during constructor)
-     *
-     *
-     * But what about the service that _responds_ to ARP requests?!
-     *
-     * Maybe have to connect a DPDK link in stages:
-     * First, create the ARP service and attach it to the dpdk_ctx
-     *  dpdk_ctx must own the links...?
-     *  Or! Always burn a DMA engine for ARP
-     *
-     * Maybe have a shared_ptr to an ARP service here?
-     */
+
+    // Structures protected by mutex
     std::mutex _mutex;
+    std::set<uint16_t> _udp_ports;
+    uint16_t _next_udp_port = 0xffff;
+
+    // Structures protected by spin lock
+    rte_spinlock_t _spinlock = RTE_SPINLOCK_INITIALIZER;
+    std::unordered_map<ipv4_addr, struct arp_entry*> _arp_table;
 };
 
 
@@ -267,17 +345,21 @@ public:
     int get_port_link_status(port_id_t portid) const;
 
     /*!
-     * Get port ID for routing packet destined for given address
+     * Get port for routing packet destined for given address
      * \param addr Destination address
-     * \return port ID from routing table
+     * \return pointer to the port from routing table
      */
-    int get_route(const std::string& addr) const;
+    dpdk_port* get_route(const std::string& addr) const;
 
     /*!
      * \return whether init() has been called
      */
     bool is_init_done(void) const;
 
+    /*! Return a reference to an IO service given a port ID
+     */
+    std::shared_ptr<uhd::transport::dpdk_io_service> get_io_service(const size_t port_id);
+
 private:
     /*! Convert the args to DPDK's EAL args and Initialize the EAL
      *
@@ -312,8 +394,12 @@ private:
     std::unordered_map<port_id_t, dpdk_port::uptr> _ports;
     std::vector<struct rte_mempool*> _rx_pktbuf_pools;
     std::vector<struct rte_mempool*> _tx_pktbuf_pools;
+    // Store all the I/O services, and also store the corresponding port ID
+    std::map<std::shared_ptr<uhd::transport::dpdk_io_service>, std::vector<size_t>>
+        _io_srv_portid_map;
 };
 
-}}} // namespace uhd::transport::dpdk
+} // namespace dpdk
+}} // namespace uhd::transport
 
 #endif /* _INCLUDED_UHDLIB_TRANSPORT_DPDK_COMMON_HPP_ */
diff --git a/host/lib/include/uhdlib/transport/dpdk/service_queue.hpp b/host/lib/include/uhdlib/transport/dpdk/service_queue.hpp
index 7c9917079..c95786864 100644
--- a/host/lib/include/uhdlib/transport/dpdk/service_queue.hpp
+++ b/host/lib/include/uhdlib/transport/dpdk/service_queue.hpp
@@ -28,6 +28,14 @@ enum wait_type {
     WAIT_FLOW_OPEN,
     //! Wake once the flow/socket is destroyed
     WAIT_FLOW_CLOSE,
+    //! Wake once the new transport is connected
+    WAIT_XPORT_CONNECT,
+    //! Wake once the transport is disconnected
+    WAIT_XPORT_DISCONNECT,
+    //! Wake when MAC address found for IP address
+    WAIT_ARP,
+    //! Wake once the I/O worker terminates
+    WAIT_LCORE_TERM,
     //! Number of possible reasons for waiting
     WAIT_TYPE_COUNT
 };
@@ -50,6 +58,8 @@ struct wait_req
     std::mutex mutex;
     //! Whether the request was completed
     bool complete;
+    //! The status or error code associated with the request
+    int retval;
     //! An atomic reference counter for managing this request object's memory
     rte_atomic32_t refcnt;
 };
@@ -110,7 +120,7 @@ class service_queue
 public:
     /*!
      * Create a service queue
-     * \param depth Must be a power of 2
+     * \param depth Must be a power of 2. Actual size is less.
      * \param lcore_id The DPDK lcore_id associated with this service queue
      */
     service_queue(size_t depth, unsigned int lcore_id)
@@ -227,6 +237,24 @@ public:
         return stat;
     }
 
+    /*!
+     * Get the size of the service queue
+     * \return size of the service queue
+     */
+    inline size_t get_size()
+    {
+        return rte_ring_get_size(_waiter_ring);
+    }
+
+    /*!
+     * Get the capacity of the service queue
+     * \return capacity of the service queue
+     */
+    inline size_t get_capacity()
+    {
+        return rte_ring_get_capacity(_waiter_ring);
+    }
+
 private:
     //! Multi-producer, single-consumer ring for requests
     struct rte_ring* _waiter_ring;
diff --git a/host/lib/include/uhdlib/transport/dpdk/udp.hpp b/host/lib/include/uhdlib/transport/dpdk/udp.hpp
new file mode 100644
index 000000000..65e561315
--- /dev/null
+++ b/host/lib/include/uhdlib/transport/dpdk/udp.hpp
@@ -0,0 +1,115 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+#ifndef _INCLUDED_UHDLIB_TRANSPORT_DPDK_UDP_HPP_
+#define _INCLUDED_UHDLIB_TRANSPORT_DPDK_UDP_HPP_
+
+#include <uhdlib/transport/dpdk/common.hpp>
+#include <arpa/inet.h>
+#include <netinet/udp.h>
+#include <rte_ip.h>
+#include <rte_udp.h>
+#include <boost/format.hpp>
+
+namespace uhd { namespace transport { namespace dpdk {
+
+constexpr size_t HDR_SIZE_UDP_IPV4 =
+    (sizeof(struct ether_hdr) + sizeof(struct ipv4_hdr) + sizeof(struct udp_hdr));
+
+/*!
+ * An enumerated type representing the type of flow for an IPv4 client
+ * Currently, only UDP is supported (FLOW_TYPE_UDP)
+ */
+enum flow_type {
+    FLOW_TYPE_UDP,
+    FLOW_TYPE_COUNT,
+};
+
+/*!
+ * A tuple for IPv4 flows that can be used for hashing
+ */
+struct ipv4_5tuple
+{
+    enum flow_type flow_type;
+    ipv4_addr src_ip;
+    ipv4_addr dst_ip;
+    uint16_t src_port;
+    uint16_t dst_port;
+};
+
+inline void fill_ipv4_hdr(struct rte_mbuf* mbuf,
+    const dpdk_port* port,
+    uint32_t dst_ipv4_addr,
+    uint8_t proto_id,
+    uint32_t payload_len)
+{
+    struct ether_hdr* eth_hdr = rte_pktmbuf_mtod(mbuf, struct ether_hdr*);
+    struct ipv4_hdr* ip_hdr   = (struct ipv4_hdr*)&eth_hdr[1];
+
+    ip_hdr->version_ihl     = 0x40 | 5;
+    ip_hdr->type_of_service = 0;
+    ip_hdr->total_length    = rte_cpu_to_be_16(20 + payload_len);
+    ip_hdr->packet_id       = 0;
+    ip_hdr->fragment_offset = rte_cpu_to_be_16(IPV4_HDR_DF_FLAG);
+    ip_hdr->time_to_live    = 64;
+    ip_hdr->next_proto_id   = proto_id;
+    ip_hdr->hdr_checksum    = 0; // Require HW offload
+    ip_hdr->src_addr        = port->get_ipv4();
+    ip_hdr->dst_addr        = dst_ipv4_addr;
+
+    mbuf->ol_flags = PKT_TX_IP_CKSUM | PKT_TX_IPV4;
+    mbuf->l2_len   = sizeof(struct ether_hdr);
+    mbuf->l3_len   = sizeof(struct ipv4_hdr);
+    mbuf->pkt_len  = sizeof(struct ether_hdr) + sizeof(struct ipv4_hdr) + payload_len;
+    mbuf->data_len = sizeof(struct ether_hdr) + sizeof(struct ipv4_hdr) + payload_len;
+}
+
+/* All values except payload length must be in network order */
+inline void fill_udp_hdr(struct rte_mbuf* mbuf,
+    const dpdk_port* port,
+    uint32_t dst_ipv4_addr,
+    uint16_t src_port,
+    uint16_t dst_port,
+    uint32_t payload_len)
+{
+    struct ether_hdr* eth_hdr;
+    struct ipv4_hdr* ip_hdr;
+    struct udp_hdr* tx_hdr;
+
+    fill_ipv4_hdr(
+        mbuf, port, dst_ipv4_addr, IPPROTO_UDP, sizeof(struct udp_hdr) + payload_len);
+
+    eth_hdr = rte_pktmbuf_mtod(mbuf, struct ether_hdr*);
+    ip_hdr  = (struct ipv4_hdr*)&eth_hdr[1];
+    tx_hdr  = (struct udp_hdr*)&ip_hdr[1];
+
+    tx_hdr->src_port    = src_port;
+    tx_hdr->dst_port    = dst_port;
+    tx_hdr->dgram_len   = rte_cpu_to_be_16(8 + payload_len);
+    tx_hdr->dgram_cksum = 0;
+    mbuf->l4_len        = sizeof(struct udp_hdr);
+}
+
+//! Return an IPv4 address (numeric, in network order) into a string
+inline std::string ipv4_num_to_str(const uint32_t ip_addr)
+{
+    char addr_str[INET_ADDRSTRLEN];
+    struct in_addr ipv4_addr;
+    ipv4_addr.s_addr = ip_addr;
+    inet_ntop(AF_INET, &ipv4_addr, addr_str, sizeof(addr_str));
+    return std::string(addr_str);
+}
+
+inline std::string eth_addr_to_string(const struct ether_addr mac_addr)
+{
+    auto mac_stream = boost::format("%02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx");
+    mac_stream % (uint32_t)mac_addr.addr_bytes[0] % (uint32_t)mac_addr.addr_bytes[1]
+        % (uint32_t)mac_addr.addr_bytes[2] % (uint32_t)mac_addr.addr_bytes[3]
+        % (uint32_t)mac_addr.addr_bytes[4] % (uint32_t)mac_addr.addr_bytes[5];
+    return mac_stream.str();
+}
+
+}}} /* namespace uhd::transport::dpdk */
+#endif /* _INCLUDED_UHDLIB_TRANSPORT_DPDK_UDP_HPP_ */