mpmd,transport,prefs: Add xport_mgr for dpdk_zero_copy

Add configuration sections to the UHD config file for NIC entries. Keys
are based on MAC addresses, and the entries beneath the section describe
which CPU and I/O thread to use for the NIC and its IPv4 address.

Make ring sizes configurable for uhd-dpdk. Ring size is now an argument
for packet buffers. Note that the maximum number of available buffers
is still determined at init!

Add ability to receive broadcasts to uhd-dpdk. This is controllable by
a boolean in the sockarg during socket creation. dpdk_zero_copy will
filter broadcast packets out.

Add dpdk_simple transport (to mirror udp_simple). This transport allows
receiving from broadcast addresses, but it only permits one outstanding
buffer at a time.

Fix IP checksum handling in UHD-DPDK.
TX checksums were not being calculated in the NIC, and in RX, the check
for IP checksums allowed values of zero (reported as none). Now packets
with bad IP checksums will be dropped.

2 files changed, 43 insertions, 41 deletions

diff --git a/host/tests/CMakeLists.txt b/host/tests/CMakeLists.txt
index 9eb48eedb..01076540d 100644
--- a/host/tests/CMakeLists.txt
+++ b/host/tests/CMakeLists.txt
@@ -113,6 +113,10 @@ if(ENABLE_DPDK)
     include_directories(${DPDK_INCLUDE_DIR})
     add_executable(dpdk_test
         dpdk_test.cpp
+        ${CMAKE_SOURCE_DIR}/lib/utils/config_parser.cpp
+        ${CMAKE_SOURCE_DIR}/lib/utils/paths.cpp
+        ${CMAKE_SOURCE_DIR}/lib/utils/pathslib.cpp
+        ${CMAKE_SOURCE_DIR}/lib/utils/prefs.cpp
         ${CMAKE_SOURCE_DIR}/lib/transport/dpdk_zero_copy.cpp
     )
     target_link_libraries(dpdk_test uhd ${Boost_LIBRARIES} ${DPDK_LIBRARIES})
diff --git a/host/tests/dpdk_test.cpp b/host/tests/dpdk_test.cpp
index 43ef3d388..b506c7b30 100644
--- a/host/tests/dpdk_test.cpp
+++ b/host/tests/dpdk_test.cpp
@@ -43,7 +43,7 @@ struct dpdk_test_args
     std::string src_port;
     std::string dst_ip;
     std::string dst_port;
-    pthread_cond_t* cond;
+    pthread_cond_t *cond;
     pthread_mutex_t mutex;
     bool started;
     int cpu;
@@ -61,7 +61,7 @@ struct dpdk_test_stats
 };
 
 
-static void process_udp(int id, uint32_t* udp_data, struct dpdk_test_stats* stats)
+static void process_udp(int id, uint32_t *udp_data, struct dpdk_test_stats *stats)
 {
     if (udp_data[0] != stats[id].last_seqno + 1) {
         stats[id].lasts[stats[id].dropped_packets & 0xf] = stats[id].last_seqno;
@@ -76,21 +76,21 @@ static void process_udp(int id, uint32_t* udp_data, struct dpdk_test_stats* stat
 static void send_udp(uhd::transport::dpdk_zero_copy::sptr& stream,
     int id,
     bool fc_only,
-    struct dpdk_test_stats* stats)
+    struct dpdk_test_stats *stats)
 {
     uhd::transport::managed_send_buffer::sptr mbuf = stream->get_send_buff(0);
     if (mbuf.get() == nullptr) {
         printf("Could not get TX buffer!\n");
         return;
     }
-    auto* tx_data = mbuf->cast<uint32_t*>();
+    auto *tx_data = mbuf->cast<uint32_t *>();
     tx_data[0]    = fc_only ? stats[id].tx_seqno - 1 : stats[id].tx_seqno;
     tx_data[1]    = stats[id].last_seqno;
     if (!fc_only) {
-        memset(&tx_data[2], stats[id].last_seqno, 8 * BENCH_SPP);
-        stats[id].tx_xfer += 8 * BENCH_SPP;
+        memset(&tx_data[2], stats[id].last_seqno, 8*BENCH_SPP);
+        stats[id].tx_xfer += 8*BENCH_SPP;
     }
-    size_t num_bytes = 8 + (fc_only ? 0 : 8 * BENCH_SPP);
+    size_t num_bytes = 8 + (fc_only ? 0 : 8*BENCH_SPP);
     mbuf->commit(num_bytes);
     mbuf.reset();
 
@@ -100,12 +100,11 @@ static void send_udp(uhd::transport::dpdk_zero_copy::sptr& stream,
 }
 
 static void bench(
-    uhd::transport::dpdk_zero_copy::sptr* stream, uint32_t nb_ports, double timeout)
+    uhd::transport::dpdk_zero_copy::sptr *stream, uint32_t nb_ports, double timeout)
 {
     uint64_t total_xfer[NUM_PORTS];
     uint32_t id;
-    struct dpdk_test_stats* stats =
-        (struct dpdk_test_stats*)malloc(sizeof(*stats) * nb_ports);
+    struct dpdk_test_stats *stats = (struct dpdk_test_stats *) malloc(sizeof(*stats)*nb_ports);
     for (id = 0; id < nb_ports; id++) {
         stats[id].tx_seqno        = 1;
         stats[id].tx_xfer         = 0;
@@ -137,7 +136,7 @@ static void bench(
             if (nb_rx <= 0) {
                 consec_no_rx++;
                 if (consec_no_rx >= 100000) {
-                    uint32_t skt_drops = stream[id]->get_drop_count();
+                    // uint32_t skt_drops = stream[id]->get_drop_count();
                     // printf("TX seq %d, TX ack %d, RX seq %d, %d drops!\n",
                     // stats[id].tx_seqno, stats[id].last_ackno, stats[id].last_seqno,
                     // skt_drops);
@@ -151,7 +150,7 @@ static void bench(
 
             for (unsigned int buf = 0; buf < nb_rx; buf++) {
                 total_xfer[id] += bufs[buf]->size();
-                auto data = bufs[buf]->cast<uint32_t*>();
+                auto data = bufs[buf]->cast<uint32_t *>();
                 process_udp(id, data, stats);
             }
 
@@ -185,15 +184,15 @@ static void bench(
         printf("Bytes received = %ld\n", total_xfer[id]);
         printf("Bytes sent = %ld\n", stats[id].tx_xfer);
         printf("Time taken = %ld us\n",
-            (bench_end.tv_sec - bench_start.tv_sec) * 1000000
+            (bench_end.tv_sec - bench_start.tv_sec)*1000000
                 + (bench_end.tv_usec - bench_start.tv_usec));
-        double elapsed_time = (bench_end.tv_sec - bench_start.tv_sec) * 1000000
-                              + (bench_end.tv_usec - bench_start.tv_usec);
+        double elapsed_time = (bench_end.tv_sec - bench_start.tv_sec)*1000000
+            + (bench_end.tv_usec - bench_start.tv_usec);
         elapsed_time *= 1.0e-6;
         double elapsed_bytes = total_xfer[id];
-        printf("RX Performance = %e Gbps\n", elapsed_bytes * 8.0 / 1.0e9 / elapsed_time);
+        printf("RX Performance = %e Gbps\n", elapsed_bytes*8.0/1.0e9/elapsed_time);
         elapsed_bytes = stats[id].tx_xfer;
-        printf("TX Performance = %e Gbps\n", elapsed_bytes * 8.0 / 1.0e9 / elapsed_time);
+        printf("TX Performance = %e Gbps\n", elapsed_bytes*8.0/1.0e9/elapsed_time);
         uint32_t skt_drops = stream[id]->get_drop_count();
         printf("Dropped %d packets\n", stats[id].dropped_packets);
         printf("Socket reports dropped %d packets\n", skt_drops);
@@ -221,9 +220,9 @@ static inline void set_cpu(pthread_t t, int cpu)
     }
 }
 
-void* prepare_and_bench_blocking(void* arg)
+void *prepare_and_bench_blocking(void *arg)
 {
-    struct dpdk_test_args* args = (struct dpdk_test_args*)arg;
+    struct dpdk_test_args *args = (struct dpdk_test_args *) arg;
     pthread_mutex_lock(&args->mutex);
     pthread_t t = pthread_self();
     set_cpu(t, args->cpu);
@@ -236,14 +235,16 @@ void* prepare_and_bench_blocking(void* arg)
     buff_args.send_frame_size = 8000;
     buff_args.num_send_frames = 8;
     buff_args.num_recv_frames = 8;
-    auto dev_addr             = uhd::device_addr_t();
-    eth_data[0]               = uhd::transport::dpdk_zero_copy::make(ctx,
+    auto dev_addr = uhd::device_addr_t();
+    eth_data[0] = uhd::transport::dpdk_zero_copy::make(
+        ctx,
         args->portid,
         args->dst_ip,
         args->src_port,
         args->dst_port,
         buff_args,
-        dev_addr);
+        dev_addr
+    );
 
     bench(eth_data, 1, 0.1);
     return 0;
@@ -274,29 +275,31 @@ void prepare_and_bench_polling(void)
     buff_args.num_recv_frames = 8;
     auto dev_addr             = uhd::device_addr_t();
     for (unsigned int i = 0; i < NUM_PORTS; i++) {
-        eth_data[i] = uhd::transport::dpdk_zero_copy::make(ctx,
+        eth_data[i] = uhd::transport::dpdk_zero_copy::make(
+            ctx,
             bench_args[i].portid,
             bench_args[i].dst_ip,
             bench_args[i].src_port,
             bench_args[i].dst_port,
             buff_args,
-            dev_addr);
+            dev_addr
+        );
     }
 
     bench(eth_data, NUM_PORTS, 0.0);
 }
 
-int main(int argc, char** argv)
+int main(int argc, char **argv)
 {
-    int retval, io0_cpu = 1, io1_cpu = 1, user0_cpu = 0, user1_cpu = 2;
+    int retval, user0_cpu = 0, user1_cpu = 2;
     std::string args, cpusets;
     po::options_description desc("Allowed options");
-    desc.add_options()("help", "help message")(
-        "args", po::value<std::string>(&args)->default_value(""), "UHD-DPDK args")(
-        "polling-mode", "Use polling mode (single thread on own core)")("cpusets",
-        po::value<std::string>(&cpusets)->default_value(""),
-        "which core(s) to use for a given thread (specify something like "
-        "\"io0=1,io1=1,user0=0,user1=2\")");
+    desc.add_options()
+        ("help", "help message")
+        ("args", po::value<std::string>(&args)->default_value(""), "UHD-DPDK args")
+        ("polling-mode", "Use polling mode (single thread on own core)")
+        ("cpusets", po::value<std::string>(&cpusets)->default_value(""), "which core(s) to use for a given thread (specify something like \"user0=0,user1=2\")")
+    ;
     po::variables_map vm;
     po::store(po::parse_command_line(argc, argv, desc), vm);
     po::notify(vm);
@@ -317,20 +320,15 @@ int main(int argc, char** argv)
 
     auto cpuset_map = uhd::device_addr_t(cpusets);
     for (std::string& key : cpuset_map.keys()) {
-        if (key == "io0") {
-            io0_cpu = std::stoi(cpuset_map[key], NULL, 0);
-        } else if (key == "io1") {
-            io1_cpu = std::stoi(cpuset_map[key], NULL, 0);
-        } else if (key == "user0") {
+        if (key == "user0") {
             user0_cpu = std::stoi(cpuset_map[key], NULL, 0);
         } else if (key == "user1") {
             user1_cpu = std::stoi(cpuset_map[key], NULL, 0);
         }
     }
 
-    int port_thread_mapping[2] = {io0_cpu, io1_cpu};
     auto& ctx                  = uhd::transport::uhd_dpdk_ctx::get();
-    ctx.init(dpdk_args, 2, &port_thread_mapping[0], NUM_MBUFS, MBUF_CACHE_SIZE, 9000);
+    ctx.init(dpdk_args);
 
     uint32_t eth_ip   = htonl(0xc0a80003);
     uint32_t eth_mask = htonl(0xffffff00);
@@ -391,12 +389,12 @@ int main(int argc, char** argv)
 
         pthread_cond_broadcast(&cond);
 
-        status = pthread_join(threads[0], (void**)&retval);
+        status = pthread_join(threads[0], (void **) &retval);
         if (status) {
             perror("Error while joining thread");
             return status;
         }
-        status = pthread_join(threads[1], (void**)&retval);
+        status = pthread_join(threads[1], (void **) &retval);
         if (status) {
             perror("Error while joining thread");
             return status;