rdma: introduce direct verb for Cx4/5 tx

Direct Verb allows for direct access to NIC HW rx/tx rings. This patch
introduce TX direct verb support for Mellanox ConnectX-4/5 adapters.
'dv' mode must be explicitely selected at interface creation to benefit
from this.

Type: feature

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

4 files changed, 549 insertions, 73 deletions

diff --git a/src/plugins/rdma/device.c b/src/plugins/rdma/device.c
index f33d55c85a0..eb13f855b1a 100644
--- a/src/plugins/rdma/device.c
+++ b/src/plugins/rdma/device.c
@@ -535,7 +535,8 @@ rdma_txq_init (vlib_main_t * vm, rdma_device_t * rd, u16 qid, u32 n_desc)
 
   vec_validate_aligned (rd->txqs, qid, CLIB_CACHE_LINE_BYTES);
   txq = vec_elt_at_index (rd->txqs, qid);
-  txq->size = n_desc;
+  ASSERT (is_pow2 (n_desc));
+  txq->bufs_log2sz = min_log2 (n_desc);
   vec_validate_aligned (txq->bufs, n_desc - 1, CLIB_CACHE_LINE_BYTES);
 
   if ((txq->cq = ibv_create_cq (rd->ctx, n_desc, NULL, NULL, 0)) == 0)
@@ -569,6 +570,57 @@ rdma_txq_init (vlib_main_t * vm, rdma_device_t * rd, u16 qid, u32 n_desc)
   qpa.qp_state = IBV_QPS_RTS;
   if (ibv_modify_qp (txq->qp, &qpa, qp_flags) != 0)
     return clib_error_return_unix (0, "Modify QP (send) Failed");
+
+  txq->ibv_cq = txq->cq;
+  txq->ibv_qp = txq->qp;
+
+  if (rd->flags & RDMA_DEVICE_F_MLX5DV)
+    {
+      rdma_mlx5_wqe_t *tmpl = (void *) txq->dv_wqe_tmpl;
+      struct mlx5dv_cq dv_cq;
+      struct mlx5dv_qp dv_qp;
+      struct mlx5dv_obj obj = { };
+
+      obj.cq.in = txq->cq;
+      obj.cq.out = &dv_cq;
+      obj.qp.in = txq->qp;
+      obj.qp.out = &dv_qp;
+
+      if (mlx5dv_init_obj (&obj, MLX5DV_OBJ_CQ | MLX5DV_OBJ_QP))
+	return clib_error_return_unix (0, "DV init obj failed");
+
+      if (RDMA_TXQ_BUF_SZ (txq) > dv_qp.sq.wqe_cnt
+	  || !is_pow2 (dv_qp.sq.wqe_cnt)
+	  || sizeof (rdma_mlx5_wqe_t) != dv_qp.sq.stride
+	  || (uword) dv_qp.sq.buf % sizeof (rdma_mlx5_wqe_t))
+	return clib_error_return (0, "Unsupported DV SQ parameters");
+
+      if (RDMA_TXQ_BUF_SZ (txq) > dv_cq.cqe_cnt
+	  || !is_pow2 (dv_cq.cqe_cnt)
+	  || sizeof (struct mlx5_cqe64) != dv_cq.cqe_size
+	  || (uword) dv_cq.buf % sizeof (struct mlx5_cqe64))
+	return clib_error_return (0, "Unsupported DV CQ parameters");
+
+      /* get SQ and doorbell addresses */
+      txq->dv_sq_wqes = dv_qp.sq.buf;
+      txq->dv_sq_dbrec = dv_qp.dbrec;
+      txq->dv_sq_db = dv_qp.bf.reg;
+      txq->dv_sq_log2sz = min_log2 (dv_qp.sq.wqe_cnt);
+
+      /* get CQ and doorbell addresses */
+      txq->dv_cq_cqes = dv_cq.buf;
+      txq->dv_cq_dbrec = dv_cq.dbrec;
+      txq->dv_cq_log2sz = min_log2 (dv_cq.cqe_cnt);
+
+      /* init tx desc template */
+      STATIC_ASSERT_SIZEOF (txq->dv_wqe_tmpl, sizeof (*tmpl));
+      mlx5dv_set_ctrl_seg (&tmpl->ctrl, 0, MLX5_OPCODE_SEND, 0,
+			   txq->qp->qp_num, 0, RDMA_MLX5_WQE_DS, 0,
+			   RDMA_TXQ_DV_INVALID_ID);
+      /* FIXME: mlx5dv_set_eth_seg(&tmpl->eseg, MLX5_ETH_WQE_L3_CSUM | MLX5_ETH_WQE_L4_CSUM, 0, 0, 0); */
+      mlx5dv_set_data_seg (&tmpl->dseg, 0, rd->lkey, 0);
+    }
+
   return 0;
 }
 
@@ -587,6 +639,13 @@ rdma_dev_init (vlib_main_t * vm, rdma_device_t * rd, u32 rxq_size,
   if ((rd->pd = ibv_alloc_pd (rd->ctx)) == 0)
     return clib_error_return_unix (0, "PD Alloc Failed");
 
+  if ((rd->mr = ibv_reg_mr (rd->pd, (void *) bm->buffer_mem_start,
+			    bm->buffer_mem_size,
+			    IBV_ACCESS_LOCAL_WRITE)) == 0)
+    return clib_error_return_unix (0, "Register MR Failed");
+
+  rd->lkey = rd->mr->lkey;	/* avoid indirection in datapath */
+
   ethernet_mac_address_generate (rd->hwaddr.bytes);
 
   if ((rd->mr = ibv_reg_mr (rd->pd, (void *) bm->buffer_mem_start,
@@ -657,28 +716,16 @@ rdma_create_if (vlib_main_t * vm, rdma_create_if_args_t * args)
     }
 
   if (args->rxq_size < VLIB_FRAME_SIZE || args->txq_size < VLIB_FRAME_SIZE ||
+      args->rxq_size > 65535 || args->txq_size > 65535 ||
       !is_pow2 (args->rxq_size) || !is_pow2 (args->txq_size))
     {
       args->rv = VNET_API_ERROR_INVALID_VALUE;
-      args->error =
-	clib_error_return (0, "queue size must be a power of two >= %i",
-			   VLIB_FRAME_SIZE);
+      args->error = clib_error_return (0, "queue size must be a power of two "
+				       "between %i and 65535",
+				       VLIB_FRAME_SIZE);
       goto err0;
     }
 
-  switch (args->mode)
-    {
-    case RDMA_MODE_AUTO:
-      break;
-    case RDMA_MODE_IBV:
-      break;
-    case RDMA_MODE_DV:
-      args->rv = VNET_API_ERROR_INVALID_VALUE;
-      args->error = clib_error_return (0, "unsupported mode");
-      goto err0;
-      break;
-    }
-
   dev_list = ibv_get_device_list (&n_devs);
   if (n_devs == 0)
     {
@@ -762,8 +809,8 @@ rdma_create_if (vlib_main_t * vm, rdma_create_if_args_t * args)
 	}
     }
 
-  if ((args->error =
-       rdma_dev_init (vm, rd, args->rxq_size, args->txq_size, args->rxq_num)))
+  if ((args->error = rdma_dev_init (vm, rd, args->rxq_size, args->txq_size,
+				    args->rxq_num)))
     goto err2;
 
   if ((args->error = rdma_register_interface (vnm, rd)))
diff --git a/src/plugins/rdma/output.c b/src/plugins/rdma/output.c
index a93c7c8c644..015208c023e 100644
--- a/src/plugins/rdma/output.c
+++ b/src/plugins/rdma/output.c
@@ -21,47 +21,359 @@
 #include <vppinfra/ring.h>
 #include <vnet/ethernet/ethernet.h>
 #include <vnet/devices/devices.h>
-
 #include <rdma/rdma.h>
 
+#ifndef MLX5_ETH_L2_INLINE_HEADER_SIZE
+#define MLX5_ETH_L2_INLINE_HEADER_SIZE  18
+#endif
+
+#define RDMA_TX_RETRIES 5
+
+#define RDMA_TXQ_DV_DSEG_SZ(txq)        (RDMA_MLX5_WQE_DS * RDMA_TXQ_DV_SQ_SZ(txq))
+#define RDMA_TXQ_DV_DSEG2WQE(d)         (((d) + RDMA_MLX5_WQE_DS - 1) / RDMA_MLX5_WQE_DS)
+
+/*
+ * MLX5 direct verbs tx/free functions
+ */
+
+static_always_inline void
+rdma_device_output_free_mlx5 (vlib_main_t * vm,
+			      const vlib_node_runtime_t * node,
+			      rdma_txq_t * txq)
+{
+  u16 idx = txq->dv_cq_idx;
+  u32 cq_mask = pow2_mask (txq->dv_cq_log2sz);
+  u32 sq_mask = pow2_mask (txq->dv_sq_log2sz);
+  u32 mask = pow2_mask (txq->bufs_log2sz);
+  u32 buf_sz = RDMA_TXQ_BUF_SZ (txq);
+  u32 log2_cq_sz = txq->dv_cq_log2sz;
+  struct mlx5_cqe64 *cqes = txq->dv_cq_cqes, *cur = cqes + (idx & cq_mask);
+  u8 op_own, saved;
+  const rdma_mlx5_wqe_t *wqe;
+
+  for (;;)
+    {
+      op_own = *(volatile u8 *) &cur->op_own;
+      if (((idx >> log2_cq_sz) & MLX5_CQE_OWNER_MASK) !=
+	  (op_own & MLX5_CQE_OWNER_MASK) || (op_own >> 4) == MLX5_CQE_INVALID)
+	break;
+      if (PREDICT_FALSE ((op_own >> 4)) != MLX5_CQE_REQ)
+	vlib_error_count (vm, node->node_index, RDMA_TX_ERROR_COMPLETION, 1);
+      idx++;
+      cur = cqes + (idx & cq_mask);
+    }
+
+  if (idx == txq->dv_cq_idx)
+    return;			/* nothing to do */
+
+  cur = cqes + ((idx - 1) & cq_mask);
+  saved = cur->op_own;
+  (void) saved;
+  cur->op_own = 0xf0;
+  txq->dv_cq_idx = idx;
+
+  /* retrieve original WQE and get new tail counter */
+  wqe = txq->dv_sq_wqes + (be16toh (cur->wqe_counter) & sq_mask);
+  if (PREDICT_FALSE (wqe->ctrl.imm == RDMA_TXQ_DV_INVALID_ID))
+    return;			/* can happen if CQE reports error for an intermediate WQE */
+
+  ASSERT (RDMA_TXQ_USED_SZ (txq->head, wqe->ctrl.imm) <= buf_sz &&
+	  RDMA_TXQ_USED_SZ (wqe->ctrl.imm, txq->tail) < buf_sz);
+
+  /* free sent buffers and update txq head */
+  vlib_buffer_free_from_ring (vm, txq->bufs, txq->head & mask, buf_sz,
+			      RDMA_TXQ_USED_SZ (txq->head, wqe->ctrl.imm));
+  txq->head = wqe->ctrl.imm;
+
+  /* ring doorbell */
+  CLIB_MEMORY_STORE_BARRIER ();
+  txq->dv_cq_dbrec[0] = htobe32 (idx);
+}
+
+static_always_inline void
+rdma_device_output_tx_mlx5_doorbell (rdma_txq_t * txq, rdma_mlx5_wqe_t * last,
+				     const u16 tail, u32 sq_mask)
+{
+  last->ctrl.imm = tail;	/* register item to free */
+  last->ctrl.fm_ce_se = MLX5_WQE_CTRL_CQ_UPDATE;	/* generate a CQE so we can free buffers */
+
+  ASSERT (tail != txq->tail &&
+	  RDMA_TXQ_AVAIL_SZ (txq, txq->head, txq->tail) >=
+	  RDMA_TXQ_USED_SZ (txq->tail, tail));
+
+  CLIB_MEMORY_STORE_BARRIER ();
+  txq->dv_sq_dbrec[MLX5_SND_DBR] = htobe32 (tail);
+  CLIB_COMPILER_BARRIER ();
+  txq->dv_sq_db[0] = *(u64 *) (txq->dv_sq_wqes + (txq->tail & sq_mask));
+  txq->tail = tail;
+}
+
+static_always_inline void
+rdma_mlx5_wqe_init (rdma_mlx5_wqe_t * wqe, const void *tmpl,
+		    vlib_buffer_t * b, const u16 tail)
+{
+  u16 sz = b->current_length;
+  u16 inline_sz = clib_min (sz, MLX5_ETH_L2_INLINE_HEADER_SIZE);
+
+  clib_memcpy_fast (wqe, tmpl, RDMA_MLX5_WQE_SZ);
+
+  wqe->ctrl.opmod_idx_opcode |= ((u32) htobe16 (tail)) << 8;
+  /* speculatively copy at least MLX5_ETH_L2_INLINE_HEADER_SIZE (18-bytes) */
+  const void *cur = vlib_buffer_get_current (b);
+  clib_memcpy_fast (wqe->eseg.inline_hdr_start,
+		    cur, MLX5_ETH_L2_INLINE_HEADER_SIZE);
+  wqe->eseg.inline_hdr_sz = htobe16 (inline_sz);
+  wqe->dseg.byte_count = htobe32 (sz - inline_sz);
+  wqe->dseg.addr = htobe64 (pointer_to_uword (cur) + inline_sz);
+}
+
+/*
+ * specific data path for chained buffers, supporting ring wrap-around
+ * contrary to the normal path - otherwise we may fail to enqueue chained
+ * buffers because we are close to the end of the ring while we still have
+ * plenty of descriptors available
+ */
+static_always_inline u32
+rdma_device_output_tx_mlx5_chained (vlib_main_t * vm,
+				    const vlib_node_runtime_t * node,
+				    const rdma_device_t * rd,
+				    rdma_txq_t * txq, u32 n_left_from, u32 n,
+				    u32 * bi, vlib_buffer_t ** b,
+				    rdma_mlx5_wqe_t * wqe, u16 tail)
+{
+  rdma_mlx5_wqe_t *last = wqe;
+  u32 wqe_n = RDMA_TXQ_AVAIL_SZ (txq, txq->head, tail);
+  u32 sq_mask = pow2_mask (txq->dv_sq_log2sz);
+  u32 mask = pow2_mask (txq->bufs_log2sz);
+  u32 dseg_mask = RDMA_TXQ_DV_DSEG_SZ (txq) - 1;
+  const u32 lkey = wqe[0].dseg.lkey;
+
+  vlib_buffer_copy_indices (txq->bufs + (txq->tail & mask), bi,
+			    n_left_from - n);
+
+  while (n >= 1 && wqe_n >= 1)
+    {
+      u32 *bufs = txq->bufs + (tail & mask);
+      rdma_mlx5_wqe_t *wqe = txq->dv_sq_wqes + (tail & sq_mask);
+
+      /* setup the head WQE */
+      rdma_mlx5_wqe_init (wqe, txq->dv_wqe_tmpl, b[0], tail);
+
+      bufs[0] = bi[0];
+
+      if (b[0]->flags & VLIB_BUFFER_NEXT_PRESENT)
+	{
+	  /*
+	   * max number of available dseg:
+	   *  - 4 dseg per WQEBB available
+	   *  - max 32 dseg per WQE (5-bits length field in WQE ctrl)
+	   */
+#define RDMA_MLX5_WQE_DS_MAX    (1 << 5)
+	  const u32 dseg_max =
+	    clib_min (RDMA_MLX5_WQE_DS * (wqe_n - 1), RDMA_MLX5_WQE_DS_MAX);
+	  vlib_buffer_t *chained_b = b[0];
+	  u32 chained_n = 0;
+
+	  /* there are exactly 4 dseg per WQEBB and we rely on that */
+	  STATIC_ASSERT (RDMA_MLX5_WQE_DS *
+			 sizeof (struct mlx5_wqe_data_seg) ==
+			 MLX5_SEND_WQE_BB, "wrong size");
+
+	  /*
+	   * iterate over fragments, supporting ring wrap-around contrary to
+	   * the normal path - otherwise we may fail to enqueue chained
+	   * buffers because we are close to the end of the ring while we
+	   * still have plenty of descriptors available
+	   */
+	  while (chained_n < dseg_max
+		 && chained_b->flags & VLIB_BUFFER_NEXT_PRESENT)
+	    {
+	      struct mlx5_wqe_data_seg *dseg = (void *) txq->dv_sq_wqes;
+	      dseg += ((tail + 1) * RDMA_MLX5_WQE_DS + chained_n) & dseg_mask;
+	      if (((clib_address_t) dseg & (MLX5_SEND_WQE_BB - 1)) == 0)
+		{
+		  /*
+		   * start of new WQEBB
+		   * head/tail are shared between buffers and descriptor
+		   * In order to maintain 1:1 correspondance between
+		   * buffer index and descriptor index, we build
+		   * 4-fragments chains and save the head
+		   */
+		  chained_b->flags &= ~(VLIB_BUFFER_NEXT_PRESENT |
+					VLIB_BUFFER_TOTAL_LENGTH_VALID);
+		  u32 idx = tail + 1 + RDMA_TXQ_DV_DSEG2WQE (chained_n);
+		  idx &= mask;
+		  txq->bufs[idx] = chained_b->next_buffer;
+		}
+
+	      chained_b = vlib_get_buffer (vm, chained_b->next_buffer);
+	      dseg->byte_count = htobe32 (chained_b->current_length);
+	      dseg->lkey = lkey;
+	      dseg->addr = htobe64 (vlib_buffer_get_current_va (chained_b));
+
+	      chained_n += 1;
+	    }
+
+	  if (chained_b->flags & VLIB_BUFFER_NEXT_PRESENT)
+	    {
+	      /*
+	       * no descriptors left: drop the chain including 1st WQE
+	       * skip the problematic packet and continue
+	       */
+	      vlib_buffer_free_from_ring (vm, txq->bufs, tail & mask,
+					  RDMA_TXQ_BUF_SZ (txq), 1 +
+					  RDMA_TXQ_DV_DSEG2WQE (chained_n));
+	      vlib_error_count (vm, node->node_index,
+				dseg_max == chained_n ?
+				RDMA_TX_ERROR_SEGMENT_SIZE_EXCEEDED :
+				RDMA_TX_ERROR_NO_FREE_SLOTS, 1);
+
+	      /* fixup tail to overwrite wqe head with next packet */
+	      tail -= 1;
+	    }
+	  else
+	    {
+	      /* update WQE descriptor with new dseg number */
+	      ((u8 *) & wqe[0].ctrl.qpn_ds)[3] = RDMA_MLX5_WQE_DS + chained_n;
+
+	      tail += RDMA_TXQ_DV_DSEG2WQE (chained_n);
+	      wqe_n -= RDMA_TXQ_DV_DSEG2WQE (chained_n);
+
+	      last = wqe;
+	    }
+	}
+      else
+	{
+	  /* not chained */
+	  last = wqe;
+	}
+
+      tail += 1;
+      bi += 1;
+      b += 1;
+      wqe_n -= 1;
+      n -= 1;
+    }
+
+  if (n == n_left_from)
+    return 0;			/* we fail to enqueue even a single packet */
+
+  rdma_device_output_tx_mlx5_doorbell (txq, last, tail, sq_mask);
+  return n_left_from - n;
+}
+
+static_always_inline u32
+rdma_device_output_tx_mlx5 (vlib_main_t * vm,
+			    const vlib_node_runtime_t * node,
+			    const rdma_device_t * rd, rdma_txq_t * txq,
+			    const u32 n_left_from, u32 * bi,
+			    vlib_buffer_t ** b)
+{
+  u32 sq_mask = pow2_mask (txq->dv_sq_log2sz);
+  u32 mask = pow2_mask (txq->bufs_log2sz);
+  rdma_mlx5_wqe_t *wqe = txq->dv_sq_wqes + (txq->tail & sq_mask);
+  u32 n = n_left_from;
+  u16 tail = txq->tail;
+
+  ASSERT (RDMA_TXQ_BUF_SZ (txq) <= RDMA_TXQ_DV_SQ_SZ (txq));
+
+  while (n >= 4)
+    {
+      u32 flags = b[0]->flags | b[1]->flags | b[2]->flags | b[3]->flags;
+      if (PREDICT_FALSE (flags & VLIB_BUFFER_NEXT_PRESENT))
+	return rdma_device_output_tx_mlx5_chained (vm, node, rd, txq,
+						   n_left_from, n, bi, b, wqe,
+						   tail);
+
+      if (PREDICT_TRUE (n >= 8))
+	{
+	  vlib_prefetch_buffer_header (b + 4, LOAD);
+	  vlib_prefetch_buffer_header (b + 5, LOAD);
+	  vlib_prefetch_buffer_header (b + 6, LOAD);
+	  vlib_prefetch_buffer_header (b + 7, LOAD);
+	  clib_prefetch_load (wqe + 4);
+	}
+
+      rdma_mlx5_wqe_init (wqe + 0, txq->dv_wqe_tmpl, b[0], tail + 0);
+      rdma_mlx5_wqe_init (wqe + 1, txq->dv_wqe_tmpl, b[1], tail + 1);
+      rdma_mlx5_wqe_init (wqe + 2, txq->dv_wqe_tmpl, b[2], tail + 2);
+      rdma_mlx5_wqe_init (wqe + 3, txq->dv_wqe_tmpl, b[3], tail + 3);
+
+      b += 4;
+      tail += 4;
+      wqe += 4;
+      n -= 4;
+    }
+
+  while (n >= 1)
+    {
+      if (PREDICT_FALSE (b[0]->flags & VLIB_BUFFER_NEXT_PRESENT))
+	return rdma_device_output_tx_mlx5_chained (vm, node, rd, txq,
+						   n_left_from, n, bi, b, wqe,
+						   tail);
+
+      rdma_mlx5_wqe_init (wqe, txq->dv_wqe_tmpl, b[0], tail);
+
+      b += 1;
+      tail += 1;
+      wqe += 1;
+      n -= 1;
+    }
+
+  vlib_buffer_copy_indices (txq->bufs + (txq->tail & mask), bi, n_left_from);
+
+  rdma_device_output_tx_mlx5_doorbell (txq, &wqe[-1], tail, sq_mask);
+  return n_left_from;
+}
+
+/*
+ * standard ibverb tx/free functions
+ */
+
 static_always_inline void
-rdma_device_output_free (vlib_main_t * vm, rdma_txq_t * txq)
+rdma_device_output_free_ibverb (vlib_main_t * vm,
+				const vlib_node_runtime_t * node,
+				rdma_txq_t * txq)
 {
   struct ibv_wc wc[VLIB_FRAME_SIZE];
-  u32 tail, slot;
+  u32 mask = pow2_mask (txq->bufs_log2sz);
+  u16 tail;
   int n;
 
-  n = ibv_poll_cq (txq->cq, VLIB_FRAME_SIZE, wc);
+  n = ibv_poll_cq (txq->ibv_cq, VLIB_FRAME_SIZE, wc);
   if (n <= 0)
-    return;
+    {
+      if (PREDICT_FALSE (n < 0))
+	vlib_error_count (vm, node->node_index, RDMA_TX_ERROR_COMPLETION, 1);
+      return;
+    }
+
+  while (PREDICT_FALSE (IBV_WC_SUCCESS != wc[n - 1].status))
+    {
+      vlib_error_count (vm, node->node_index, RDMA_TX_ERROR_COMPLETION, 1);
+      n--;
+      if (0 == n)
+	return;
+    }
 
   tail = wc[n - 1].wr_id;
-  slot = txq->head & (txq->size - 1);
-  vlib_buffer_free_from_ring (vm, txq->bufs, slot, txq->size,
-			      tail - txq->head);
+  vlib_buffer_free_from_ring (vm, txq->bufs, txq->head & mask,
+			      RDMA_TXQ_BUF_SZ (txq),
+			      RDMA_TXQ_USED_SZ (txq->head, tail));
   txq->head = tail;
 }
 
 static_always_inline u32
-rmda_device_output_tx (vlib_main_t * vm, const rdma_device_t * rd,
-		       rdma_txq_t * txq, u32 n_left_from, u32 * bi)
+rdma_device_output_tx_ibverb (vlib_main_t * vm,
+			      const vlib_node_runtime_t * node,
+			      const rdma_device_t * rd, rdma_txq_t * txq,
+			      u32 n_left_from, u32 * bi, vlib_buffer_t ** b)
 {
-  vlib_buffer_t *bufs[VLIB_FRAME_SIZE], **b = bufs;
   struct ibv_send_wr wr[VLIB_FRAME_SIZE], *w = wr;
   struct ibv_sge sge[VLIB_FRAME_SIZE], *s = sge;
-  u32 n, slot = txq->tail & (txq->size - 1);
-  u32 *tx = &txq->bufs[slot];
-
-  /* do not enqueue more packet than ring space */
-  n_left_from = clib_min (n_left_from, txq->size - (txq->tail - txq->head));
-  /* avoid wrap-around logic in core loop */
-  n = n_left_from = clib_min (n_left_from, txq->size - slot);
+  u32 mask = txq->bufs_log2sz;
+  u32 n = n_left_from;
 
-  /* if ring is full, do nothing */
-  if (PREDICT_FALSE (0 == n_left_from))
-    return 0;
-
-  vlib_get_buffers (vm, bi, bufs, n_left_from);
   memset (w, 0, n_left_from * sizeof (w[0]));
 
   while (n >= 4)
@@ -81,8 +393,6 @@ rmda_device_output_tx (vlib_main_t * vm, const rdma_device_t * rd,
 	  CLIB_PREFETCH (&w[4 + 3], CLIB_CACHE_LINE_BYTES, STORE);
 	}
 
-      vlib_buffer_copy_indices (tx, bi, 4);
-
       s[0].addr = vlib_buffer_get_current_va (b[0]);
       s[0].length = b[0]->current_length;
       s[0].lkey = rd->lkey;
@@ -122,15 +432,11 @@ rmda_device_output_tx (vlib_main_t * vm, const rdma_device_t * rd,
       s += 4;
       w += 4;
       b += 4;
-      bi += 4;
-      tx += 4;
       n -= 4;
     }
 
   while (n >= 1)
     {
-      vlib_buffer_copy_indices (tx, bi, 1);
-
       s[0].addr = vlib_buffer_get_current_va (b[0]);
       s[0].length = b[0]->current_length;
       s[0].lkey = rd->lkey;
@@ -143,30 +449,71 @@ rmda_device_output_tx (vlib_main_t * vm, const rdma_device_t * rd,
       s += 1;
       w += 1;
       b += 1;
-      bi += 1;
-      tx += 1;
       n -= 1;
     }
 
-  w[-1].wr_id = txq->tail + n_left_from;	/* register item to free */
+  w[-1].wr_id = txq->tail;	/* register item to free */
   w[-1].next = 0;		/* fix next pointer in WR linked-list */
   w[-1].send_flags = IBV_SEND_SIGNALED;	/* generate a CQE so we can free buffers */
 
   w = wr;
-  if (PREDICT_FALSE (0 != ibv_post_send (txq->qp, w, &w)))
-    n_left_from = w - wr;
+  if (PREDICT_FALSE (0 != ibv_post_send (txq->ibv_qp, w, &w)))
+    {
+      vlib_error_count (vm, node->node_index, RDMA_TX_ERROR_SUBMISSION,
+			n_left_from - (w - wr));
+      n_left_from = w - wr;
+    }
 
+  vlib_buffer_copy_indices (txq->bufs + (txq->tail & mask), bi, n_left_from);
   txq->tail += n_left_from;
   return n_left_from;
 }
 
-VNET_DEVICE_CLASS_TX_FN (rdma_device_class) (vlib_main_t * vm,
-					     vlib_node_runtime_t * node,
-					     vlib_frame_t * frame)
+/*
+ * common tx/free functions
+ */
+
+static_always_inline void
+rdma_device_output_free (vlib_main_t * vm, const vlib_node_runtime_t * node,
+			 rdma_txq_t * txq, int is_mlx5dv)
+{
+  if (is_mlx5dv)
+    rdma_device_output_free_mlx5 (vm, node, txq);
+  else
+    rdma_device_output_free_ibverb (vm, node, txq);
+}
+
+static_always_inline u32
+rdma_device_output_tx_try (vlib_main_t * vm, const vlib_node_runtime_t * node,
+			   const rdma_device_t * rd, rdma_txq_t * txq,
+			   u32 n_left_from, u32 * bi, int is_mlx5dv)
+{
+  vlib_buffer_t *b[VLIB_FRAME_SIZE];
+  u32 mask = pow2_mask (txq->bufs_log2sz);
+
+  /* do not enqueue more packet than ring space */
+  n_left_from = clib_min (n_left_from, RDMA_TXQ_AVAIL_SZ (txq, txq->head,
+							  txq->tail));
+  /* avoid wrap-around logic in core loop */
+  n_left_from = clib_min (n_left_from, RDMA_TXQ_BUF_SZ (txq) -
+			  (txq->tail & mask));
+
+  /* if ring is full, do nothing */
+  if (PREDICT_FALSE (n_left_from == 0))
+    return 0;
+
+  vlib_get_buffers (vm, bi, b, n_left_from);
+
+  return is_mlx5dv ?
+    rdma_device_output_tx_mlx5 (vm, node, rd, txq, n_left_from, bi, b) :
+    rdma_device_output_tx_ibverb (vm, node, rd, txq, n_left_from, bi, b);
+}
+
+static_always_inline uword
+rdma_device_output_tx (vlib_main_t * vm, vlib_node_runtime_t * node,
+		       vlib_frame_t * frame, rdma_device_t * rd,
+		       int is_mlx5dv)
 {
-  rdma_main_t *rm = &rdma_main;
-  vnet_interface_output_runtime_t *ord = (void *) node->runtime_data;
-  rdma_device_t *rd = pool_elt_at_index (rm->devices, ord->dev_instance);
   u32 thread_index = vm->thread_index;
   rdma_txq_t *txq =
     vec_elt_at_index (rd->txqs, thread_index % vec_len (rd->txqs));
@@ -174,19 +521,20 @@ VNET_DEVICE_CLASS_TX_FN (rdma_device_class) (vlib_main_t * vm,
   u32 n_left_from;
   int i;
 
-  ASSERT (txq->size >= VLIB_FRAME_SIZE && is_pow2 (txq->size));
-  ASSERT (txq->tail - txq->head <= txq->size);
+  ASSERT (RDMA_TXQ_BUF_SZ (txq) >= VLIB_FRAME_SIZE);
 
   from = vlib_frame_vector_args (frame);
   n_left_from = frame->n_vectors;
 
   clib_spinlock_lock_if_init (&txq->lock);
 
-  for (i = 0; i < 5 && n_left_from > 0; i++)
+  for (i = 0; i < RDMA_TX_RETRIES && n_left_from > 0; i++)
     {
       u32 n_enq;
-      rdma_device_output_free (vm, txq);
-      n_enq = rmda_device_output_tx (vm, rd, txq, n_left_from, from);
+      rdma_device_output_free (vm, node, txq, is_mlx5dv);
+      n_enq = rdma_device_output_tx_try (vm, node, rd, txq, n_left_from, from,
+					 is_mlx5dv);
+
       n_left_from -= n_enq;
       from += n_enq;
     }
@@ -203,6 +551,20 @@ VNET_DEVICE_CLASS_TX_FN (rdma_device_class) (vlib_main_t * vm,
   return frame->n_vectors - n_left_from;
 }
 
+VNET_DEVICE_CLASS_TX_FN (rdma_device_class) (vlib_main_t * vm,
+					     vlib_node_runtime_t * node,
+					     vlib_frame_t * frame)
+{
+  rdma_main_t *rm = &rdma_main;
+  vnet_interface_output_runtime_t *ord = (void *) node->runtime_data;
+  rdma_device_t *rd = pool_elt_at_index (rm->devices, ord->dev_instance);
+
+  if (PREDICT_TRUE (rd->flags & RDMA_DEVICE_F_MLX5DV))
+    return rdma_device_output_tx (vm, node, frame, rd, 1 /* is_mlx5dv */ );
+
+  return rdma_device_output_tx (vm, node, frame, rd, 0 /* is_mlx5dv */ );
+}
+
 /*
  * fd.io coding-style-patch-verification: ON
  *
diff --git a/src/plugins/rdma/rdma.h b/src/plugins/rdma/rdma.h
index 1e2f3d9ea0e..82f32ec9d01 100644
--- a/src/plugins/rdma/rdma.h
+++ b/src/plugins/rdma/rdma.h
@@ -41,6 +41,19 @@ enum
 
 typedef struct
 {
+  CLIB_ALIGN_MARK (align0, MLX5_SEND_WQE_BB);
+  struct mlx5_wqe_ctrl_seg ctrl;
+  struct mlx5_wqe_eth_seg eseg;
+  struct mlx5_wqe_data_seg dseg;
+} rdma_mlx5_wqe_t;
+#define RDMA_MLX5_WQE_SZ        sizeof(rdma_mlx5_wqe_t)
+#define RDMA_MLX5_WQE_DS        (RDMA_MLX5_WQE_SZ/sizeof(struct mlx5_wqe_data_seg))
+STATIC_ASSERT (RDMA_MLX5_WQE_SZ == MLX5_SEND_WQE_BB &&
+	       RDMA_MLX5_WQE_SZ % sizeof (struct mlx5_wqe_data_seg) == 0,
+	       "bad size");
+
+typedef struct
+{
   CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
   struct ibv_cq *cq;
   struct ibv_wq *wq;
@@ -65,14 +78,60 @@ typedef struct
 typedef struct
 {
   CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
+
+  /* following fields are accessed in datapath */
   clib_spinlock_t lock;
+
+  union
+  {
+    struct
+    {
+      /* ibverb datapath. Cache of cq, sq below */
+      struct ibv_cq *ibv_cq;
+      struct ibv_qp *ibv_qp;
+    };
+    struct
+    {
+      /* direct verbs datapath */
+      rdma_mlx5_wqe_t *dv_sq_wqes;
+      volatile u32 *dv_sq_dbrec;
+      volatile u64 *dv_sq_db;
+      struct mlx5_cqe64 *dv_cq_cqes;
+      volatile u32 *dv_cq_dbrec;
+    };
+  };
+
+  u32 *bufs;			/* vlib_buffer ring buffer */
+  u16 head;
+  u16 tail;
+  u16 dv_cq_idx;		/* monotonic CQE index (valid only for direct verbs) */
+  u8 bufs_log2sz;		/* log2 vlib_buffer entries */
+  u8 dv_sq_log2sz:4;		/* log2 SQ WQE entries (valid only for direct verbs) */
+  u8 dv_cq_log2sz:4;		/* log2 CQ CQE entries (valid only for direct verbs) */
+    STRUCT_MARK (cacheline1);
+
+  /* WQE template (valid only for direct verbs) */
+  u8 dv_wqe_tmpl[64];
+
+  /* end of 2nd 64-bytes cacheline (or 1st 128-bytes cacheline) */
+    STRUCT_MARK (cacheline2);
+
+  /* fields below are not accessed in datapath */
   struct ibv_cq *cq;
   struct ibv_qp *qp;
-  u32 *bufs;
-  u32 size;
-  u32 head;
-  u32 tail;
+
 } rdma_txq_t;
+STATIC_ASSERT_OFFSET_OF (rdma_txq_t, cacheline1, 64);
+STATIC_ASSERT_OFFSET_OF (rdma_txq_t, cacheline2, 128);
+
+#define RDMA_TXQ_DV_INVALID_ID  0xffffffff
+
+#define RDMA_TXQ_BUF_SZ(txq)    (1U << (txq)->bufs_log2sz)
+#define RDMA_TXQ_DV_SQ_SZ(txq)  (1U << (txq)->dv_sq_log2sz)
+#define RDMA_TXQ_DV_CQ_SZ(txq)  (1U << (txq)->dv_cq_log2sz)
+
+#define RDMA_TXQ_USED_SZ(head, tail)            ((u16)((u16)(tail) - (u16)(head)))
+#define RDMA_TXQ_AVAIL_SZ(txq, head, tail)      ((u16)(RDMA_TXQ_BUF_SZ (txq) - RDMA_TXQ_USED_SZ (head, tail)))
 
 typedef struct
 {
@@ -170,8 +229,11 @@ typedef struct
   u16 cqe_flags;
 } rdma_input_trace_t;
 
-#define foreach_rdma_tx_func_error	       \
-_(NO_FREE_SLOTS, "no free tx slots")
+#define foreach_rdma_tx_func_error \
+_(SEGMENT_SIZE_EXCEEDED, "segment size exceeded") \
+_(NO_FREE_SLOTS, "no free tx slots") \
+_(SUBMISSION, "tx submission errors") \
+_(COMPLETION, "tx completion errors")
 
 typedef enum
 {
@@ -181,7 +243,7 @@ typedef enum
     RDMA_TX_N_ERROR,
 } rdma_tx_func_error_t;
 
-#endif /* AVF_H */
+#endif /* _RDMA_H_ */
 
 /*
  * fd.io coding-style-patch-verification: ON
diff --git a/src/plugins/rdma/rdma_doc.md b/src/plugins/rdma/rdma_doc.md
index 3c79f9aefd3..3fed5b6fc49 100644
--- a/src/plugins/rdma/rdma_doc.md
+++ b/src/plugins/rdma/rdma_doc.md
@@ -44,13 +44,13 @@ vpp# set int st rdma-0 up
 vpp# ping 1.1.1.100`
 ```
 
-### Containers support
+## Containers support
 It should work in containers as long as:
  - the `ib_uverbs` module is loaded
  - the device nodes `/dev/infiniband/uverbs[0-9]+` are usable from the
    container (but see [security considerations](#Security considerations))
 
-### SR-IOV VFs support
+## SR-IOV VFs support
 It should work on SR-IOV VFs the same way it does with PFs. Because of VFs
 security containment features, make sure the MAC address of the rdma VPP
 interface matches the MAC address assigned to the underlying VF.
@@ -68,3 +68,8 @@ aware of the [security considerations](#Security considerations)):
 ```
 host# ip l set dev enp94s0f0 vf 0 spoof off trust on
 ```
+
+## Direct Verb mode
+Direct Verb allows the driver to access the NIC HW RX/TX rings directly
+instead of having to go through libibverb and suffering associated overhead.
+It will be automatically selected if the adapter supports it.