rdma: Add functions to queue and flush recv WRs

Change-Id: I77e286efd48330fcb6c1c85d1515656e49587983
Signed-off-by: Alexey Marchuk <alexeymar@mellanox.com>
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/6290
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Community-CI: Mellanox Build Bot

include/spdk_internal/rdma.h

@@ -64,6 +64,7 @@ struct spdk_rdma_qp {
 	struct ibv_qp *qp;
 	struct rdma_cm_id *cm_id;
 	struct spdk_rdma_send_wr_list send_wrs;
+	struct spdk_rdma_recv_wr_list recv_wrs;
 };
 
 struct spdk_rdma_mem_map;
@@ -189,6 +190,24 @@ bool spdk_rdma_qp_queue_send_wrs(struct spdk_rdma_qp *spdk_rdma_qp, struct ibv_s
  */
 int spdk_rdma_qp_flush_send_wrs(struct spdk_rdma_qp *spdk_rdma_qp, struct ibv_send_wr **bad_wr);
 
+/**
+ * Append the given recv wr structure to the qpair's outstanding recv list.
+ * This function accepts either a single Work Request or the first WR in a linked list.
+ *
+ * \param spdk_rdma_qp Pointer to SPDK RDMA qpair
+ * \param first Pointer to the first Work Request
+ * \return true if there were no outstanding WRs before, false otherwise
+ */
+bool spdk_rdma_qp_queue_recv_wrs(struct spdk_rdma_qp *spdk_rdma_qp, struct ibv_recv_wr *first);
+
+/**
+ * Submit all queued recv Work Request
+ * \param spdk_rdma_qp Pointer to SPDK RDMA qpair
+ * \param bad_wr Stores a pointer to the first failed WR if this function return nonzero value
+ * \return 0 on succes, errno on failure
+ */
+int spdk_rdma_qp_flush_recv_wrs(struct spdk_rdma_qp *spdk_rdma_qp, struct ibv_recv_wr **bad_wr);
+
 /**
  * Create a memory map which is used to register Memory Regions and perform address -> memory
  * key translations

lib/rdma/common.c

@@ -269,8 +269,8 @@ spdk_rdma_srq_destroy(struct spdk_rdma_srq *rdma_srq)
 	return rc;
 }
 
-bool
-spdk_rdma_srq_queue_recv_wrs(struct spdk_rdma_srq *rdma_srq, struct ibv_recv_wr *first)
+static inline bool
+rdma_queue_recv_wrs(struct spdk_rdma_recv_wr_list *recv_wrs, struct ibv_recv_wr *first)
 {
 	struct ibv_recv_wr *last;
 
@@ -279,17 +279,26 @@ spdk_rdma_srq_queue_recv_wrs(struct spdk_rdma_srq *rdma_srq, struct ibv_recv_wr
 		last = last->next;
 	}
 
-	if (rdma_srq->recv_wrs.first == NULL) {
-		rdma_srq->recv_wrs.first = first;
-		rdma_srq->recv_wrs.last = last;
+	if (recv_wrs->first == NULL) {
+		recv_wrs->first = first;
+		recv_wrs->last = last;
 		return true;
 	} else {
-		rdma_srq->recv_wrs.last->next = first;
-		rdma_srq->recv_wrs.last = last;
+		recv_wrs->last->next = first;
+		recv_wrs->last = last;
 		return false;
 	}
 }
 
+bool
+spdk_rdma_srq_queue_recv_wrs(struct spdk_rdma_srq *rdma_srq, struct ibv_recv_wr *first)
+{
+	assert(rdma_srq);
+	assert(first);
+
+	return rdma_queue_recv_wrs(&rdma_srq->recv_wrs, first);
+}
+
 int
 spdk_rdma_srq_flush_recv_wrs(struct spdk_rdma_srq *rdma_srq, struct ibv_recv_wr **bad_wr)
 {
@@ -305,3 +314,28 @@ spdk_rdma_srq_flush_recv_wrs(struct spdk_rdma_srq *rdma_srq, struct ibv_recv_wr
 
 	return rc;
 }
+
+bool
+spdk_rdma_qp_queue_recv_wrs(struct spdk_rdma_qp *spdk_rdma_qp, struct ibv_recv_wr *first)
+{
+	assert(spdk_rdma_qp);
+	assert(first);
+
+	return rdma_queue_recv_wrs(&spdk_rdma_qp->recv_wrs, first);
+}
+
+int
+spdk_rdma_qp_flush_recv_wrs(struct spdk_rdma_qp *spdk_rdma_qp, struct ibv_recv_wr **bad_wr)
+{
+	int rc;
+
+	if (spdk_unlikely(spdk_rdma_qp->recv_wrs.first == NULL)) {
+		return 0;
+	}
+
+	rc = ibv_post_recv(spdk_rdma_qp->qp, spdk_rdma_qp->recv_wrs.first, bad_wr);
+
+	spdk_rdma_qp->recv_wrs.first = NULL;
+
+	return rc;
+}

lib/rdma/spdk_rdma.map

@@ -16,6 +16,8 @@
 	spdk_rdma_create_mem_map;
 	spdk_rdma_free_mem_map;
 	spdk_rdma_get_translation;
+	spdk_rdma_qp_queue_recv_wrs;
+	spdk_rdma_qp_flush_recv_wrs;
 
 	local: *;
 };