prov/efa: acquire the same lock for qp lifecycle

In efa ep close, qp destroy + qp table update + the cq
flush must happen in the same locking block, and the same
lock must be acquired by the qp creation + qp table update
in the ep enable.

This ensure that the cq flush will exactly flush out the
stale cqe belongs to the old qp before the new qp that likely
has the same qpn is ingested by rdma-core's qp table, which
will make rdma-core regard the stale cqe as a valid cqe for the
new qp and cause unexpected behaviors.

Signed-off-by: Shi Jin <[email protected]>

Author: shijin-aws

prov/efa/src/efa_base_ep.c

@@ -58,6 +58,8 @@ int efa_base_ep_destruct_qp(struct efa_base_ep *base_ep)
 	struct efa_qp *qp = base_ep->qp;
 	struct efa_qp *user_recv_qp = base_ep->user_recv_qp;
 	uint32_t qp_num;
+	struct efa_cq *tx_cq, *rx_cq;
+	int err;
 
 	/*
 	 * Acquire the lock to prevent race conditions when CQ polling accesses the qp_table
@@ -79,6 +81,25 @@ int efa_base_ep_destruct_qp(struct efa_base_ep *base_ep)
 		domain->qp_table[qp_num & domain->qp_table_sz_m1] = NULL;
 		base_ep->user_recv_qp = NULL;
 	}
+
+	/* Flush the cq to poll out all stale cqes for the destroyed qp */
+	tx_cq = efa_base_ep_get_tx_cq(base_ep);
+	rx_cq = efa_base_ep_get_rx_cq(base_ep);
+
+	if (tx_cq) {
+		err = 0;
+		while (err != ENOENT) {
+			err = tx_cq->poll_ibv_cq(-1, &tx_cq->ibv_cq);
+		}
+	}
+
+	if (rx_cq && rx_cq != tx_cq) {
+		err = 0;
+		while (err != ENOENT) {
+			err = rx_cq->poll_ibv_cq(-1, &rx_cq->ibv_cq);
+		}
+	}
+
 	efa_base_ep_unlock_cq(base_ep);
 
 	return 0;
@@ -364,10 +385,7 @@ int efa_base_ep_enable_qp(struct efa_base_ep *base_ep, struct efa_qp *qp)
 
 	qp->qp_num = qp->ibv_qp->qp_num;
 
-	/* Acquire the lock to prevent race conditions when CQ polling accesses the qp_table */
-	efa_base_ep_lock_cq(base_ep);
 	base_ep->domain->qp_table[qp->qp_num & base_ep->domain->qp_table_sz_m1] = qp;
-	efa_base_ep_unlock_cq(base_ep);
 
 	EFA_INFO(FI_LOG_EP_CTRL, "QP enabled! qp_n: %d qkey: %d\n", qp->qp_num, qp->qkey);
 
@@ -822,37 +840,19 @@ int efa_base_ep_create_and_enable_qp(struct efa_base_ep *ep, bool create_user_re
 	scq = txcq ? txcq : rxcq;
 	rcq = rxcq ? rxcq : txcq;
 
+	/* Acquire the lock to prevent race conditions when CQ polling accesses the qp_table */
+	efa_base_ep_lock_cq(ep);
 	err = efa_base_ep_create_qp(ep, &scq->ibv_cq, &rcq->ibv_cq, create_user_recv_qp);
 	if (err)
-		return err;
-
-	return efa_base_ep_enable(ep);
-}
-
-void efa_base_ep_flush_cq(struct efa_base_ep *base_ep)
-{
-	struct efa_cq *tx_cq, *rx_cq;
-	int err;
-
-	efa_base_ep_lock_cq(base_ep);
-	tx_cq = efa_base_ep_get_tx_cq(base_ep);
-	rx_cq = efa_base_ep_get_rx_cq(base_ep);
+		goto out;
 
-	if (tx_cq) {
-		err = 0;
-		while (err != ENOENT) {
-			err = tx_cq->poll_ibv_cq(-1, &tx_cq->ibv_cq);
-		}
-	}
+	err = efa_base_ep_enable(ep);
 
-	if (rx_cq && rx_cq != tx_cq) {
-		err = 0;
-		while (err != ENOENT) {
-			err = rx_cq->poll_ibv_cq(-1, &rx_cq->ibv_cq);
-		}
-	}
-	efa_base_ep_unlock_cq(base_ep);
+out:
+	efa_base_ep_unlock_cq(ep);
+	return err;
 }
+
 #if ENABLE_DEBUG
 void efa_ep_addr_print(char *prefix, struct efa_ep_addr *addr) {
 	char raw_gid_str[INET6_ADDRSTRLEN];

prov/efa/src/efa_base_ep.h

@@ -151,8 +151,6 @@ void efa_base_ep_remove_cntr_ibv_cq_poll_list(struct efa_base_ep *ep);
 
 int efa_base_ep_create_and_enable_qp(struct efa_base_ep *ep, bool create_user_recv_qp);
 
-void efa_base_ep_flush_cq(struct efa_base_ep *base_ep);
-
 #if ENABLE_DEBUG
 void efa_ep_addr_print(char *prefix, struct efa_ep_addr *addr);
 #endif

prov/efa/src/efa_ep.c

@@ -208,8 +208,6 @@ static int efa_ep_close(fid_t fid)
 		EFA_WARN(FI_LOG_EP_CTRL, "Unable to close base endpoint\n");
 	}
 
-	efa_base_ep_flush_cq(ep);
-
 	free(ep);
 
 	return 0;

prov/efa/src/rdm/efa_rdm_ep_fiops.c

@@ -1033,8 +1033,6 @@ static int efa_rdm_ep_close(struct fid *fid)
 		retv = ret;
 	}
 
-	efa_base_ep_flush_cq(&efa_rdm_ep->base_ep);
-
 	ret = efa_rdm_ep_close_shm_ep_resources(efa_rdm_ep);
 	if (ret)
 		retv = ret;