Add dpdk per-interface startup config parameter to specify worker threads

New parameter allows specifying which worker threads will process rx
queues. Parameter arguments is list of cores and number of worker specified
must be equal to the number of rx queues configured (num-rx-queues). If
num-rx-queues is not specified, it will be automatically set to
number of workers.

Sample config:

dpdk {
  dev 0000:86:00.0 {
    workers 2,3
    num-rx-queues 2
  }
}

Change-Id: I88bc381e0e542eb02def09a726c6f04de3e1ae17
Signed-off-by: Damjan Marion <damarion@cisco.com>

3 files changed, 56 insertions, 23 deletions

diff --git a/vnet/vnet/devices/dpdk/dpdk.h b/vnet/vnet/devices/dpdk/dpdk.h
index ded9d2db709..eb0a11f91a3 100644
--- a/vnet/vnet/devices/dpdk/dpdk.h
+++ b/vnet/vnet/devices/dpdk/dpdk.h
@@ -323,6 +323,7 @@ typedef struct {
 #define _(x) uword x;
     foreach_dpdk_device_config_item
 #undef _
+    clib_bitmap_t * workers;
 } dpdk_device_config_t;
 
 typedef struct {
diff --git a/vnet/vnet/devices/dpdk/init.c b/vnet/vnet/devices/dpdk/init.c
index 398bd19e193..c685fb4170e 100644
--- a/vnet/vnet/devices/dpdk/init.c
+++ b/vnet/vnet/devices/dpdk/init.c
@@ -571,29 +571,44 @@ dpdk_lib_init (dpdk_main_t * dm)
       dpdk_device_and_queue_t * dq;
       int q;
 
-      for (q = 0; q < xd->rx_q_used; q++)
-        {
-          int cpu = dm->input_cpu_first_index + next_cpu;
-          unsigned lcore = vlib_worker_threads[cpu].dpdk_lcore_id;
-
-          /*
-           * numa node for worker thread handling this queue
-           * needed for taking buffers from the right mempool
-           */
-          vec_validate(xd->cpu_socket_id_by_queue, q);
-          xd->cpu_socket_id_by_queue[q] = rte_lcore_to_socket_id(lcore);
-
-          /*
-           * construct vector of (device,queue) pairs for each worker thread
-           */
-          vec_add2(dm->devices_by_cpu[cpu], dq, 1);
-          dq->device = xd->device_index;
-          dq->queue_id = q;
-
-          next_cpu++;
-          if (next_cpu == dm->input_cpu_count)
-            next_cpu = 0;
-        }
+      if (devconf->workers)
+	{
+	  int i;
+	  q = 0;
+	  clib_bitmap_foreach (i, devconf->workers, ({
+	    int cpu = dm->input_cpu_first_index + i;
+	    unsigned lcore = vlib_worker_threads[cpu].dpdk_lcore_id;
+	    vec_validate(xd->cpu_socket_id_by_queue, q);
+	    xd->cpu_socket_id_by_queue[q] = rte_lcore_to_socket_id(lcore);
+	    vec_add2(dm->devices_by_cpu[cpu], dq, 1);
+	    dq->device = xd->device_index;
+	    dq->queue_id = q++;
+	  }));
+	}
+      else
+	for (q = 0; q < xd->rx_q_used; q++)
+	  {
+	    int cpu = dm->input_cpu_first_index + next_cpu;
+	    unsigned lcore = vlib_worker_threads[cpu].dpdk_lcore_id;
+
+	    /*
+	     * numa node for worker thread handling this queue
+	     * needed for taking buffers from the right mempool
+	     */
+	    vec_validate(xd->cpu_socket_id_by_queue, q);
+	    xd->cpu_socket_id_by_queue[q] = rte_lcore_to_socket_id(lcore);
+
+	    /*
+	     * construct vector of (device,queue) pairs for each worker thread
+	     */
+	    vec_add2(dm->devices_by_cpu[cpu], dq, 1);
+	    dq->device = xd->device_index;
+	    dq->queue_id = q;
+
+	    next_cpu++;
+	    if (next_cpu == dm->input_cpu_count)
+	      next_cpu = 0;
+	  }
 
       vec_validate_aligned (xd->tx_vectors, tm->n_vlib_mains,
                             CLIB_CACHE_LINE_BYTES);
@@ -873,6 +888,9 @@ dpdk_device_config (dpdk_config_main_t * conf, vlib_pci_addr_t pci_addr, unforma
 	;
       else if (unformat (input, "num-tx-desc %u", &devconf->num_tx_desc))
 	;
+      else if (unformat (input, "workers %U", unformat_bitmap_list,
+			 &devconf->workers))
+	;
       else
 	{
 	  error = clib_error_return (0, "unknown input `%U'",
@@ -880,6 +898,18 @@ dpdk_device_config (dpdk_config_main_t * conf, vlib_pci_addr_t pci_addr, unforma
 	  break;
 	}
     }
+
+  if (error)
+    return error;
+
+  if (devconf->workers && devconf->num_rx_queues == 0)
+    devconf->num_rx_queues = clib_bitmap_count_set_bits(devconf->workers);
+  else if (devconf->workers &&
+	   clib_bitmap_count_set_bits(devconf->workers) != devconf->num_rx_queues)
+    error = clib_error_return (0, "%U: number of worker threadds must be "
+			       "equal to number of rx queues",
+			       format_vlib_pci_addr, &pci_addr);
+
   return error;
 }
 
diff --git a/vppinfra/vppinfra/bitmap.h b/vppinfra/vppinfra/bitmap.h
index 986c322e86c..a89aa399e2f 100644
--- a/vppinfra/vppinfra/bitmap.h
+++ b/vppinfra/vppinfra/bitmap.h
@@ -45,6 +45,8 @@
 #include <vppinfra/error.h>
 #include <vppinfra/bitops.h>	/* for count_set_bits */
 
+typedef uword clib_bitmap_t;
+
 /* Returns 1 if the entire bitmap is zero, 0 otherwise */
 always_inline uword
 clib_bitmap_is_zero (uword * ai)