nvmf/rdma: Add WR batch rdma new transport option

With x86 and low queue depths with multiple QPs/initiators
there is a benefit from disable batch when we have randread IO pattern.

Testing environment: x86, Intel(R) Xeon(R) Silver 4116 CPU @ 2.10GH
Several results:
init_cores | QD | BS | Avg IOPs/BW without batch | Avg IOPS/BW with batch
  8 cores  | 4  | 4K |     1870087 / 7305        |     1594014 / 6226
  8 cores  | 8  | 4K |     1853573 / 7240        |     1576400 / 6157
  8 cores  | 16 | 4K |     1819643 / 7108        |     1569487 / 6130
  8 cores  | 32 | 4K |     1815467 / 7092        |     1569909 / 6132
 16 cores  | 4  | 4K |     1908018 / 7453        |     1566843 / 6120
 16 cores  | 8  | 4K |     1906081 / 7446        |     1562110 / 6102
 16 cores  | 16 | 4K |     1880706 / 7346        |     1555060 / 6074
 16 cores  | 32 | 4K |     1835878 / 7171        |     1548156 / 6046

Signed-off-by: Ivan Betsis <c_ivanb@mellanox.com>
Signed-off-by: Evgeniy Kochetov <evgeniik@mellanox.com>
Signed-off-by: Alexey Marchuk <alexeymar@mellanox.com>
Signed-off-by: Sasha Kotchubievsky <sashakot@mellanox.com>

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## v20.04: (Upcoming Release)
 
+### nvmf
+
+Add 'no_wr_batching' parameter in 'spdk_nvmf_transport_opts' for the ability disable WR batching RDMA. 
+
 ### vmd
 A new function, `spdk_vmd_fini`, has been added. It releases all resources acquired by the VMD
 library through the `spdk_vmd_init` call.

include/spdk/nvmf.h

@@ -83,6 +83,7 @@ struct spdk_nvmf_transport_opts {
 	bool		no_srq;
 	bool		c2h_success;
 	bool		dif_insert_or_strip;
+	bool            no_wr_batching;
 	uint32_t	sock_priority;
 };
 

lib/nvmf/rdma.c

@@ -522,6 +522,14 @@ struct spdk_nvmf_rdma_transport {
 static inline void
 spdk_nvmf_rdma_start_disconnect(struct spdk_nvmf_rdma_qpair *rqpair);
 
+static void
+_poller_submit_sends(struct spdk_nvmf_rdma_transport *rtransport,
+		     struct spdk_nvmf_rdma_poller *rpoller);
+
+static void
+_poller_submit_recvs(struct spdk_nvmf_rdma_transport *rtransport,
+		     struct spdk_nvmf_rdma_poller *rpoller);
+
 static inline int
 spdk_nvmf_rdma_check_ibv_state(enum ibv_qp_state state)
 {
@@ -1105,6 +1113,7 @@ static void
 nvmf_rdma_qpair_queue_recv_wrs(struct spdk_nvmf_rdma_qpair *rqpair, struct ibv_recv_wr *first)
 {
 	struct ibv_recv_wr *last;
+	struct spdk_nvmf_rdma_transport *rtransport = SPDK_CONTAINEROF(rqpair->qpair.transport, struct spdk_nvmf_rdma_transport, transport);
 
 	last = first;
 	while (last->next != NULL) {
@@ -1121,6 +1130,10 @@ nvmf_rdma_qpair_queue_recv_wrs(struct spdk_nvmf_rdma_qpair *rqpair, struct ibv_r
 		rqpair->resources->recvs_to_post.last->next = first;
 		rqpair->resources->recvs_to_post.last = last;
 	}
+
+	if (rtransport->transport.opts.no_wr_batching) {
+		_poller_submit_recvs(rtransport, rqpair->poller);
+	}
 }
 
 /* Append the given send wr structure to the qpair's outstanding sends list. */
@@ -1129,6 +1142,7 @@ static void
 nvmf_rdma_qpair_queue_send_wrs(struct spdk_nvmf_rdma_qpair *rqpair, struct ibv_send_wr *first)
 {
 	struct ibv_send_wr *last;
+	struct spdk_nvmf_rdma_transport *rtransport = SPDK_CONTAINEROF(rqpair->qpair.transport, struct spdk_nvmf_rdma_transport, transport);
 
 	last = first;
 	while (last->next != NULL) {
@@ -1143,6 +1157,10 @@ nvmf_rdma_qpair_queue_send_wrs(struct spdk_nvmf_rdma_qpair *rqpair, struct ibv_s
 		rqpair->sends_to_post.last->next = first;
 		rqpair->sends_to_post.last = last;
 	}
+
+	if (rtransport->transport.opts.no_wr_batching) {
+		_poller_submit_sends(rtransport, rqpair->poller);
+	}
 }
 
 static int
@@ -2295,6 +2313,7 @@ spdk_nvmf_rdma_request_process(struct spdk_nvmf_rdma_transport *rtransport,
 #define SPDK_NVMF_RDMA_DEFAULT_BUFFER_CACHE_SIZE 32
 #define SPDK_NVMF_RDMA_DEFAULT_NO_SRQ false
 #define SPDK_NVMF_RDMA_DIF_INSERT_OR_STRIP false
+#define SPDK_NVMF_RDMA_DEFAULT_NO_WR_BATCHING false
 
 static void
 spdk_nvmf_rdma_opts_init(struct spdk_nvmf_transport_opts *opts)
@@ -2310,6 +2329,7 @@ spdk_nvmf_rdma_opts_init(struct spdk_nvmf_transport_opts *opts)
 	opts->max_srq_depth =		SPDK_NVMF_RDMA_DEFAULT_SRQ_DEPTH;
 	opts->no_srq =			SPDK_NVMF_RDMA_DEFAULT_NO_SRQ;
 	opts->dif_insert_or_strip =	SPDK_NVMF_RDMA_DIF_INSERT_OR_STRIP;
+	opts->no_wr_batching =          SPDK_NVMF_RDMA_DEFAULT_NO_WR_BATCHING;
 }
 
 const struct spdk_mem_map_ops g_nvmf_rdma_map_ops = {
@@ -2370,7 +2390,8 @@ spdk_nvmf_rdma_create(struct spdk_nvmf_transport_opts *opts)
 		     "  Transport opts:  max_ioq_depth=%d, max_io_size=%d,\n"
 		     "  max_qpairs_per_ctrlr=%d, io_unit_size=%d,\n"
 		     "  in_capsule_data_size=%d, max_aq_depth=%d,\n"
-		     "  num_shared_buffers=%d, max_srq_depth=%d, no_srq=%d\n",
+		     "  num_shared_buffers=%d, max_srq_depth=%d, no_srq=%d,\n"
+		     "  no_wr_batching=%d\n",
 		     opts->max_queue_depth,
 		     opts->max_io_size,
 		     opts->max_qpairs_per_ctrlr,
@@ -2379,7 +2400,8 @@ spdk_nvmf_rdma_create(struct spdk_nvmf_transport_opts *opts)
 		     opts->max_aq_depth,
 		     opts->num_shared_buffers,
 		     opts->max_srq_depth,
-		     opts->no_srq);
+		     opts->no_srq,
+		     opts->no_wr_batching);
 
 	/* I/O unit size cannot be larger than max I/O size */
 	if (opts->io_unit_size > opts->max_io_size) {
@@ -3725,7 +3747,7 @@ _qp_reset_failed_recvs(struct spdk_nvmf_rdma_qpair *rqpair, struct ibv_recv_wr *
 	spdk_nvmf_rdma_start_disconnect(rqpair);
 }
 
-static void
+void
 _poller_submit_recvs(struct spdk_nvmf_rdma_transport *rtransport,
 		     struct spdk_nvmf_rdma_poller *rpoller)
 {
@@ -3818,7 +3840,7 @@ _qp_reset_failed_sends(struct spdk_nvmf_rdma_transport *rtransport,
 
 }
 
-static void
+void
 _poller_submit_sends(struct spdk_nvmf_rdma_transport *rtransport,
 		     struct spdk_nvmf_rdma_poller *rpoller)
 {