prov/efa: Fix the rx flags procedure.

This patch fixed two issues in efa provider that will cause
trouble when sharing rx to a peer provider.

1. rxr_msg_recv (trecv, recvv, trecvv) should pass flags as
rxr_rx_flags(ep) to rxr_msg_generic_recv, which means it should
enable FI_COMPLETION in the flags as long as the prov_info->rx_attr.flags
support FI_COMPLETION. For rxr_msg_recvmsg (trecvmsg), it should
pass application flags |= util_ep.rx_msg_flags, which will have NO FI_COMPLETION
when application binds rx cq with FI_SELECTIVE_COMPLETION, and does not have
FI_COMPLETION in the flags of fi_recvmsg.

2. when calling ofi_need_completion in rxr_rx_entry_report_completion, the rxr_rx_flags(ep)
should not be passed in as the cq_flags. Instead, cq_flags should be either 0 or FI_SELECTIVE_COMPLETION
which can be derived from util_ep.rx_msg_flags.

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

Author: shijin-aws

prov/efa/src/rdm/rxr_msg.c

@@ -1145,20 +1145,13 @@ ssize_t rxr_msg_generic_recv(struct fid_ep *ep, const struct fi_msg *msg,
 	struct rxr_ep *rxr_ep;
 	struct dlist_entry *unexp_list;
 	struct rxr_op_entry *rx_entry;
-	uint64_t rx_op_flags;
 
 	rxr_ep = container_of(ep, struct rxr_ep, base_ep.util_ep.ep_fid.fid);
 
 	assert(msg->iov_count <= rxr_ep->rx_iov_limit);
 
 	efa_perfset_start(rxr_ep, perf_efa_recv);
 
-	assert(rxr_ep->base_ep.util_ep.rx_msg_flags == 0 || rxr_ep->base_ep.util_ep.rx_msg_flags == FI_COMPLETION);
-	rx_op_flags = rxr_ep->base_ep.util_ep.rx_op_flags;
-	if (rxr_ep->base_ep.util_ep.rx_msg_flags == 0)
-		rx_op_flags &= ~FI_COMPLETION;
-	flags = flags | rx_op_flags;
-
 	ofi_mutex_lock(&rxr_ep->base_ep.util_ep.lock);
 	if (OFI_UNLIKELY(is_rx_res_full(rxr_ep))) {
 		ret = -FI_EAGAIN;
@@ -1355,32 +1348,42 @@ static
 ssize_t rxr_msg_recvmsg(struct fid_ep *ep_fid, const struct fi_msg *msg,
 			uint64_t flags)
 {
-	return rxr_msg_generic_recv(ep_fid, msg, 0, 0, ofi_op_msg, flags);
+	struct rxr_ep *ep;
+
+	ep = container_of(ep_fid, struct rxr_ep, base_ep.util_ep.ep_fid.fid);
+
+	return rxr_msg_generic_recv(ep_fid, msg, 0, 0, ofi_op_msg, flags | ep->base_ep.util_ep.rx_msg_flags);
 }
 
 static
-ssize_t rxr_msg_recv(struct fid_ep *ep, void *buf, size_t len,
+ssize_t rxr_msg_recv(struct fid_ep *ep_fid, void *buf, size_t len,
 		     void *desc, fi_addr_t src_addr, void *context)
 {
 	struct fi_msg msg = {0};
 	struct iovec iov;
+	struct rxr_ep *ep;
+
+	ep = container_of(ep_fid, struct rxr_ep, base_ep.util_ep.ep_fid.fid);
 
 	iov.iov_base = buf;
 	iov.iov_len = len;
 
 	rxr_msg_construct(&msg, &iov, &desc, 1, src_addr, context, 0);
-	return rxr_msg_recvmsg(ep, &msg, 0);
+	return rxr_msg_recvmsg(ep_fid, &msg, rxr_rx_flags(ep));
 }
 
 static
-ssize_t rxr_msg_recvv(struct fid_ep *ep, const struct iovec *iov,
+ssize_t rxr_msg_recvv(struct fid_ep *ep_fid, const struct iovec *iov,
 		      void **desc, size_t count, fi_addr_t src_addr,
 		      void *context)
 {
 	struct fi_msg msg = {0};
+	struct rxr_ep *ep;
+
+	ep = container_of(ep_fid, struct rxr_ep, base_ep.util_ep.ep_fid.fid);
 
 	rxr_msg_construct(&msg, iov, desc, count, src_addr, context, 0);
-	return rxr_msg_recvmsg(ep, &msg, 0);
+	return rxr_msg_recvmsg(ep_fid, &msg, rxr_rx_flags(ep));
 }
 
 /**
@@ -1393,12 +1396,15 @@ ssize_t rxr_msg_trecv(struct fid_ep *ep_fid, void *buf, size_t len, void *desc,
 {
 	struct fi_msg msg = {0};
 	struct iovec iov;
+	struct rxr_ep *ep;
+
+	ep = container_of(ep_fid, struct rxr_ep, base_ep.util_ep.ep_fid.fid);
 
 	iov.iov_base = (void *)buf;
 	iov.iov_len = len;
 
 	rxr_msg_construct(&msg, &iov, &desc, 1, src_addr, context, 0);
-	return rxr_msg_generic_recv(ep_fid, &msg, tag, ignore, ofi_op_tagged, 0);
+	return rxr_msg_generic_recv(ep_fid, &msg, tag, ignore, ofi_op_tagged, rxr_rx_flags(ep));
 }
 
 static
@@ -1407,9 +1413,12 @@ ssize_t rxr_msg_trecvv(struct fid_ep *ep_fid, const struct iovec *iov,
 		       uint64_t tag, uint64_t ignore, void *context)
 {
 	struct fi_msg msg = {0};
+	struct rxr_ep *ep;
+
+	ep = container_of(ep_fid, struct rxr_ep, base_ep.util_ep.ep_fid.fid);
 
 	rxr_msg_construct(&msg, iov, desc, count, src_addr, context, 0);
-	return rxr_msg_generic_recv(ep_fid, &msg, tag, ignore, ofi_op_tagged, 0);
+	return rxr_msg_generic_recv(ep_fid, &msg, tag, ignore, ofi_op_tagged, rxr_rx_flags(ep));
 }
 
 static
@@ -1418,6 +1427,9 @@ ssize_t rxr_msg_trecvmsg(struct fid_ep *ep_fid, const struct fi_msg_tagged *tmsg
 {
 	ssize_t ret;
 	struct fi_msg msg = {0};
+	struct rxr_ep *ep;
+
+	ep = container_of(ep_fid, struct rxr_ep, base_ep.util_ep.ep_fid.fid);
 
 	if (flags & FI_PEEK) {
 		ret = rxr_msg_peek_trecv(ep_fid, tmsg, flags);
@@ -1429,7 +1441,7 @@ ssize_t rxr_msg_trecvmsg(struct fid_ep *ep_fid, const struct fi_msg_tagged *tmsg
 
 	rxr_msg_construct(&msg, tmsg->msg_iov, tmsg->desc, tmsg->iov_count, tmsg->addr, tmsg->context, tmsg->data);
 	ret = rxr_msg_generic_recv(ep_fid, &msg, tmsg->tag, tmsg->ignore,
-				   ofi_op_tagged, flags);
+				   ofi_op_tagged, flags | ep->base_ep.util_ep.rx_msg_flags);
 
 out:
 	return ret;

prov/efa/src/rdm/rxr_op_entry.c

@@ -716,7 +716,9 @@ void rxr_rx_entry_report_completion(struct rxr_op_entry *rx_entry)
 	struct rxr_ep *ep = rx_entry->ep;
 	struct util_cq *rx_cq = ep->base_ep.util_ep.rx_cq;
 	int ret = 0;
+	uint64_t cq_flags;
 
+	cq_flags = (ep->base_ep.util_ep.rx_msg_flags == FI_COMPLETION) ? 0 : FI_SELECTIVE_COMPLETION;
 	if (OFI_UNLIKELY(rx_entry->cq_entry.len < rx_entry->total_len)) {
 		EFA_WARN(FI_LOG_CQ,
 			"Message truncated! tag: %"PRIu64" incoming message size: %"PRIu64" receiving buffer size: %zu\n",
@@ -748,7 +750,7 @@ void rxr_rx_entry_report_completion(struct rxr_op_entry *rx_entry)
 	}
 
 	if (!(rx_entry->rxr_flags & RXR_RX_ENTRY_RECV_CANCEL) &&
-	    (ofi_need_completion(rxr_rx_flags(ep), rx_entry->fi_flags) ||
+	    (ofi_need_completion(cq_flags, rx_entry->fi_flags) ||
 	     (rx_entry->cq_entry.flags & FI_MULTI_RECV))) {
 		EFA_DBG(FI_LOG_CQ,
 		       "Writing recv completion for rx_entry from peer: %"