dpdk: add txq struct and fix dpdk tx lock

This introduces a txq structure mirroring the rxq structure.
This fixes the case when #txq > #rxq, because lock must be per txq.

Type: fix
Fixes: dfb19cabe20ccf1cbd1aa714f493ccd322839b91

Change-Id: Ic1bce64d2b08b9a98c8242a1ba1bfcdbda322bec
Signed-off-by: Benoît Ganne <bganne@cisco.com>

4 files changed, 23 insertions, 19 deletions

diff --git a/src/plugins/dpdk/device/common.c b/src/plugins/dpdk/device/common.c
index 18d4555315a..0c43bfe02bb 100644
--- a/src/plugins/dpdk/device/common.c
+++ b/src/plugins/dpdk/device/common.c
@@ -41,6 +41,7 @@ dpdk_device_setup (dpdk_device_t * xd)
   dpdk_main_t *dm = &dpdk_main;
   vlib_main_t *vm = vlib_get_main ();
   vnet_main_t *vnm = vnet_get_main ();
+  vlib_thread_main_t *tm = vlib_get_thread_main ();
   vnet_sw_interface_t *sw = vnet_get_sw_interface (vnm, xd->sw_if_index);
   vnet_hw_interface_t *hi = vnet_get_hw_interface (vnm, xd->hw_if_index);
   struct rte_eth_dev_info dev_info;
@@ -95,7 +96,8 @@ dpdk_device_setup (dpdk_device_t * xd)
       goto error;
     }
 
-  /* Set up one TX-queue per worker thread */
+  vec_validate_aligned (xd->tx_queues, xd->tx_q_used - 1,
+			CLIB_CACHE_LINE_BYTES);
   for (j = 0; j < xd->tx_q_used; j++)
     {
       rv =
@@ -110,6 +112,9 @@ dpdk_device_setup (dpdk_device_t * xd)
 				  &xd->tx_conf);
       if (rv < 0)
 	dpdk_device_error (xd, "rte_eth_tx_queue_setup", rv);
+
+      if (xd->tx_q_used < tm->n_vlib_mains)
+	clib_spinlock_init (&vec_elt (xd->tx_queues, j).lock);
     }
 
   vec_validate_aligned (xd->rx_queues, xd->rx_q_used - 1,
diff --git a/src/plugins/dpdk/device/device.c b/src/plugins/dpdk/device/device.c
index 2467b7d70e5..eb2dbbcf5df 100644
--- a/src/plugins/dpdk/device/device.c
+++ b/src/plugins/dpdk/device/device.c
@@ -158,18 +158,18 @@ static_always_inline
 				struct rte_mbuf **mb, u32 n_left)
 {
   dpdk_main_t *dm = &dpdk_main;
-  dpdk_rx_queue_t *rxq;
+  dpdk_tx_queue_t *txq;
   u32 n_retry;
   int n_sent = 0;
   int queue_id;
 
   n_retry = 16;
   queue_id = vm->thread_index % xd->tx_q_used;
-  rxq = vec_elt_at_index (xd->rx_queues, queue_id);
+  txq = vec_elt_at_index (xd->tx_queues, queue_id);
 
   do
     {
-      clib_spinlock_lock_if_init (&rxq->lock);
+      clib_spinlock_lock_if_init (&txq->lock);
 
       if (PREDICT_TRUE (xd->flags & DPDK_DEVICE_FLAG_PMD))
 	{
@@ -183,7 +183,7 @@ static_always_inline
 	  n_sent = 0;
 	}
 
-      clib_spinlock_unlock_if_init (&rxq->lock);
+      clib_spinlock_unlock_if_init (&txq->lock);
 
       if (PREDICT_FALSE (n_sent < 0))
 	{
diff --git a/src/plugins/dpdk/device/dpdk.h b/src/plugins/dpdk/device/dpdk.h
index 0ace4f3a0ed..963bc811ee1 100644
--- a/src/plugins/dpdk/device/dpdk.h
+++ b/src/plugins/dpdk/device/dpdk.h
@@ -156,13 +156,21 @@ typedef struct
 typedef struct
 {
   CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
-  clib_spinlock_t lock;
   u8 buffer_pool_index;
 } dpdk_rx_queue_t;
 
 typedef struct
 {
   CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
+  clib_spinlock_t lock;
+} dpdk_tx_queue_t;
+
+typedef struct
+{
+  CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
+
+  dpdk_rx_queue_t *rx_queues;
+  dpdk_tx_queue_t *tx_queues;
 
   /* Instance ID to access internal device array. */
   u32 device_index;
@@ -173,18 +181,16 @@ typedef struct
   /* next node index if we decide to steal the rx graph arc */
   u32 per_interface_next_index;
 
+  u16 rx_q_used;
+  u16 tx_q_used;
+  u16 flags;
+
   /* DPDK device port number */
   dpdk_portid_t port_id;
   dpdk_pmd_t pmd:8;
   i8 cpu_socket;
 
-  u16 flags;
-
-  dpdk_rx_queue_t *rx_queues;
-  u16 tx_q_used;
-
     CLIB_CACHE_LINE_ALIGN_MARK (cacheline1);
-  u16 rx_q_used;
   u16 nb_tx_desc;
   u16 nb_rx_desc;
 
diff --git a/src/plugins/dpdk/device/init.c b/src/plugins/dpdk/device/init.c
index b69e2b1b7c7..f1d1b008279 100644
--- a/src/plugins/dpdk/device/init.c
+++ b/src/plugins/dpdk/device/init.c
@@ -609,13 +609,6 @@ dpdk_lib_init (dpdk_main_t * dm)
       else
 	rte_eth_macaddr_get (i, (void *) addr);
 
-      if (xd->tx_q_used < tm->n_vlib_mains)
-        for (int q = 0; q < xd->tx_q_used; q++)
-	  {
-	    dpdk_rx_queue_t *rxq = vec_elt_at_index (xd->rx_queues, q);
-	    clib_spinlock_init (&rxq->lock);
-	  }
-
       xd->port_id = i;
       xd->device_index = xd - dm->devices;
       xd->per_interface_next_index = ~0;