rpma: make gpspm server use seperate RCQ

Signed-off-by: Xiao Yang <[email protected]>

Author: yangx-jy

engines/librpma_gpspm.c

@@ -368,11 +368,8 @@ struct server_data {
 	/* resources for messaging buffer from DRAM allocated by fio */
 	struct rpma_mr_local *msg_mr;
 
-	uint32_t msg_sqe_available; /* # of free SQ slots */
-
-	/* in-memory queues */
-	struct ibv_wc *msgs_queued;
-	uint32_t msg_queued_nr;
+	/* receive CQ */
+	struct rpma_cq *rcq;
 };
 
 static int server_init(struct thread_data *td)
@@ -393,13 +390,6 @@ static int server_init(struct thread_data *td)
 		goto err_server_cleanup;
 	}
 
-	/* allocate in-memory queue */
-	sd->msgs_queued = calloc(td->o.iodepth, sizeof(*sd->msgs_queued));
-	if (sd->msgs_queued == NULL) {
-		td_verror(td, errno, "calloc");
-		goto err_free_sd;
-	}
-
 	/*
 	 * Assure a single io_u buffer can store both SEND and RECV messages and
 	 * an io_us buffer allocation is page-size-aligned which is required
@@ -412,9 +402,6 @@ static int server_init(struct thread_data *td)
 
 	return 0;
 
-err_free_sd:
-	free(sd);
-
 err_server_cleanup:
 	librpma_fio_server_cleanup(td);
 
@@ -485,7 +472,6 @@ static void server_cleanup(struct thread_data *td)
 		if ((ret = rpma_mr_dereg(&sd->msg_mr)))
 			librpma_td_verror(td, ret, "rpma_mr_dereg");
 
-		free(sd->msgs_queued);
 		free(sd);
 	}
 
@@ -501,7 +487,6 @@ static int prepare_connection(struct thread_data *td,
 	int i;
 
 	/* prepare buffers for a flush requests */
-	sd->msg_sqe_available = td->o.iodepth;
 	for (i = 0; i < td->o.iodepth; i++) {
 		size_t offset_recv_msg = IO_U_BUFF_OFF_SERVER(i) + RECV_OFFSET;
 		if ((ret = rpma_conn_req_recv(conn_req, sd->msg_mr,
@@ -518,6 +503,7 @@ static int prepare_connection(struct thread_data *td,
 static int server_open_file(struct thread_data *td, struct fio_file *f)
 {
 	struct librpma_fio_server_data *csd = td->io_ops_data;
+	struct server_data *sd = csd->server_data;
 	struct rpma_conn_cfg *cfg = NULL;
 	uint16_t max_msg_num = td->o.iodepth;
 	int ret;
@@ -536,8 +522,10 @@ static int server_open_file(struct thread_data *td, struct fio_file *f)
 	 *   all possible flush requests (SENDs)
 	 * - the receive queue (RQ) has to be big enough to accommodate
 	 *   all flush responses (RECVs)
-	 * - the completion queue (CQ) has to be big enough to accommodate
-	 *   all success and error completions (sq_size + rq_size)
+	 * - the main completion queue (CQ) has to be big enough to
+	 *   accommodate all success and error completions (sq_size)
+	 * - the receive completion queue (RCQ) has to be big enough to
+	 *   accommodate all success and error completions (rq_size)
 	 */
 	if ((ret = rpma_conn_cfg_set_sq_size(cfg, max_msg_num))) {
 		librpma_td_verror(td, ret, "rpma_conn_cfg_set_sq_size");
@@ -547,12 +535,24 @@ static int server_open_file(struct thread_data *td, struct fio_file *f)
 		librpma_td_verror(td, ret, "rpma_conn_cfg_set_rq_size");
 		goto err_cfg_delete;
 	}
-	if ((ret = rpma_conn_cfg_set_cq_size(cfg, max_msg_num * 2))) {
+	if ((ret = rpma_conn_cfg_set_cq_size(cfg, max_msg_num))) {
 		librpma_td_verror(td, ret, "rpma_conn_cfg_set_cq_size");
 		goto err_cfg_delete;
 	}
 
-	ret = librpma_fio_server_open_file(td, f, cfg);
+	if ((ret = rpma_conn_cfg_set_rcq_size(cfg, max_msg_num))) {
+		librpma_td_verror(td, ret, "rpma_conn_cfg_set_rcq_size");
+		goto err_cfg_delete;
+	}
+
+	if ((ret = librpma_fio_server_open_file(td, f, cfg))) {
+		librpma_td_verror(td, ret, "librpma_fio_server_open_file");
+		goto err_cfg_delete;
+	}
+
+	/* get the connection's receive CQ */
+	if ((ret = rpma_conn_get_rcq(csd->conn, &sd->rcq)))
+		librpma_td_verror(td, ret, "rpma_conn_get_rcq");
 
 err_cfg_delete:
 	(void) rpma_conn_cfg_delete(&cfg);
@@ -630,7 +630,6 @@ static int server_qe_process(struct thread_data *td, struct ibv_wc *wc)
 		librpma_td_verror(td, ret, "rpma_send");
 		goto err_free_unpacked;
 	}
-	--sd->msg_sqe_available;
 
 	gpspm_flush_request__free_unpacked(flush_req, NULL);
 
@@ -645,48 +644,17 @@ static int server_qe_process(struct thread_data *td, struct ibv_wc *wc)
 	return -1;
 }
 
-static inline int server_queue_process(struct thread_data *td)
-{
-	struct librpma_fio_server_data *csd = td->io_ops_data;
-	struct server_data *sd = csd->server_data;
-	int ret;
-	int i;
-
-	/* min(# of queue entries, # of SQ entries available) */
-	uint32_t qes_to_process = min(sd->msg_queued_nr, sd->msg_sqe_available);
-	if (qes_to_process == 0)
-		return 0;
-
-	/* process queued completions */
-	for (i = 0; i < qes_to_process; ++i) {
-		if ((ret = server_qe_process(td, &sd->msgs_queued[i])))
-			return ret;
-	}
-
-	/* progress the queue */
-	for (i = 0; i < sd->msg_queued_nr - qes_to_process; ++i) {
-		memcpy(&sd->msgs_queued[i],
-			&sd->msgs_queued[qes_to_process + i],
-			sizeof(sd->msgs_queued[i]));
-	}
-
-	sd->msg_queued_nr -= qes_to_process;
-
-	return 0;
-}
-
-static int server_cmpl_process(struct thread_data *td)
+static int server_cmpl_process(struct thread_data *td, struct rpma_cq *cq, struct ibv_wc *wc)
 {
-	struct librpma_fio_server_data *csd = td->io_ops_data;
-	struct server_data *sd = csd->server_data;
-	struct ibv_wc *wc = &sd->msgs_queued[sd->msg_queued_nr];
 	struct librpma_fio_options_values *o = td->eo;
 	int ret;
 
-	ret = rpma_cq_get_wc(csd->cq, 1, wc, NULL);
+	memset(wc, 0, sizeof(struct ibv_wc));
+
+	ret = rpma_cq_get_wc(cq, 1, wc, NULL);
 	if (ret == RPMA_E_NO_COMPLETION) {
 		if (o->busy_wait_polling == 0) {
-			ret = rpma_cq_wait(csd->cq);
+			ret = rpma_cq_wait(cq);
 			if (ret == RPMA_E_NO_COMPLETION) {
 				/* lack of completion is not an error */
 				return 0;
@@ -695,7 +663,7 @@ static int server_cmpl_process(struct thread_data *td)
 				goto err_terminate;
 			}
 
-			ret = rpma_cq_get_wc(csd->cq, 1, wc, NULL);
+			ret = rpma_cq_get_wc(cq, 1, wc, NULL);
 			if (ret == RPMA_E_NO_COMPLETION) {
 				/* lack of completion is not an error */
 				return 0;
@@ -716,10 +684,8 @@ static int server_cmpl_process(struct thread_data *td)
 	if (wc->status != IBV_WC_SUCCESS)
 		goto err_terminate;
 
-	if (wc->opcode == IBV_WC_RECV)
-		++sd->msg_queued_nr;
-	else if (wc->opcode == IBV_WC_SEND)
-		++sd->msg_sqe_available;
+	if (wc->opcode != IBV_WC_RECV && wc->opcode != IBV_WC_SEND)
+		goto err_terminate;
 
 	return 0;
 
@@ -729,12 +695,35 @@ static int server_cmpl_process(struct thread_data *td)
 	return -1;
 }
 
+static inline int server_queue_process(struct thread_data *td)
+{
+	struct librpma_fio_server_data *csd = td->io_ops_data;
+	struct server_data *sd = csd->server_data;
+	struct ibv_wc wc = {0};
+	int ret;
+
+	/* process the receive completion */
+	ret = server_cmpl_process(td, sd->rcq, &wc);
+	if (ret)
+		return ret;
+
+	if (wc.opcode == IBV_WC_RECV) {
+		ret = server_qe_process(td, &wc);
+		if (ret)
+			return ret;
+	}
+
+	/* process the send completion */
+	ret = server_cmpl_process(td, csd->cq, &wc);
+	if (ret)
+		return ret;
+
+	return 0;
+}
+
 static enum fio_q_status server_queue(struct thread_data *td, struct io_u *io_u)
 {
 	do {
-		if (server_cmpl_process(td))
-			return FIO_Q_BUSY;
-
 		if (server_queue_process(td))
 			return FIO_Q_BUSY;