prov/efa: Implement the cq progress

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

Author: jiaxiyan

prov/efa/src/dgram/efa_dgram_cq.c

@@ -4,25 +4,117 @@
 #ifndef EFA_DGRAM_CQ_H
 #define EFA_DGRAM_CQ_H
 
-typedef void (*efa_dgram_cq_read_entry)(struct ibv_cq_ex *ibv_cqx, int index, void *buf);
+#include "efa_av.h"
+#include "efa_cntr.h"
 
-struct efa_dgram_cq {
+#define EFA_ERROR_MSG_BUFFER_LENGTH 1024
+
+typedef void (*efa_cq_read_entry)(struct ibv_cq_ex *ibv_cqx, void *buf);
+
+struct efa_cq {
 	struct util_cq		util_cq;
 	struct efa_domain	*domain;
-	size_t			entry_size;
-	efa_dgram_cq_read_entry	read_entry;
-	ofi_spin_t		lock;
-	struct ofi_bufpool	*wce_pool;
-	uint32_t	flags; /* User defined capability mask */
-
+	efa_cq_read_entry	read_entry;
 	struct ibv_cq_ex	*ibv_cq_ex;
 };
 
-int efa_dgram_cq_open(struct fid_domain *domain_fid, struct fi_cq_attr *attr,
-		      struct fid_cq **cq_fid, void *context);
+int efa_cq_open(struct fid_domain *domain_fid, struct fi_cq_attr *attr,
+		struct fid_cq **cq_fid, void *context);
+
+static inline
+bool efa_use_unsolicited_write_recv()
+{
+	return efa_env.use_unsolicited_write_recv && efa_device_support_unsolicited_write_recv();
+}
+
+#if HAVE_CAPS_UNSOLICITED_WRITE_RECV
+/**
+ * @brief Check whether a completion consumes recv buffer
+ *
+ * @param ibv_cq_ex extended ibv cq
+ * @return true the wc consumes a recv buffer
+ * @return false the wc doesn't consume a recv buffer
+ */
+static inline
+bool efa_cq_wc_is_unsolicited(struct ibv_cq_ex *ibv_cq_ex)
+{
+	return efa_use_unsolicited_write_recv() && efadv_wc_is_unsolicited(efadv_cq_from_ibv_cq_ex(ibv_cq_ex));
+}
+
+#else
+
+static inline
+bool efa_cq_wc_is_unsolicited(struct ibv_cq_ex *ibv_cq_ex)
+{
+	return false;
+}
+
+#endif
+
+
+static inline const char *efa_ep_raw_addr_str(struct efa_base_ep *base_ep, char *buf, size_t *buflen)
+{
+	return ofi_straddr(buf, buflen, FI_ADDR_EFA, &base_ep->src_addr);
+}
+
+static inline const char *efa_ep_get_peer_raw_addr_str(struct efa_base_ep *base_ep, fi_addr_t addr, char *buf, size_t *buflen)
+{
+	struct efa_conn *efa_conn;
+	efa_conn = efa_av_addr_to_conn(base_ep->av, addr);
+
+	return ofi_straddr(buf, buflen, FI_ADDR_EFA, efa_conn ? efa_conn->ep_addr : NULL);
+}
+
+/**
+ * @brief Write the error message and return its byte length
+ * @param[in]    ep          EFA base endpoint
+ * @param[in]    addr        Remote peer fi_addr_t
+ * @param[in]    prov_errno  EFA provider * error code(must be positive)
+ * @param[out]   buf         Pointer to the address of error data written by this function
+ * @param[out]   buflen      Pointer to the returned error data size
+ * @return       A status code. 0 if the error data was written successfully, otherwise a negative FI error code.
+ */
+static inline int efa_write_error_msg(struct efa_base_ep *ep, fi_addr_t addr, int prov_errno, void **buf, size_t *buflen)
+{
+    char ep_addr_str[OFI_ADDRSTRLEN] = {0}, peer_addr_str[OFI_ADDRSTRLEN] = {0};
+    char peer_host_id_str[EFA_HOST_ID_STRING_LENGTH + 1] = {0};
+    char local_host_id_str[EFA_HOST_ID_STRING_LENGTH + 1] = {0};
+    const char *base_msg = efa_strerror(prov_errno);
+	char err_msg[EFA_ERROR_MSG_BUFFER_LENGTH];
+    size_t len = 0;
+	uint64_t local_host_id;
+
+    *buf = NULL;
+    *buflen = 0;
+
+    len = sizeof(ep_addr_str);
+    efa_ep_raw_addr_str(ep, ep_addr_str, &len);
+    len = sizeof(peer_addr_str);
+    efa_ep_get_peer_raw_addr_str(ep, addr, peer_addr_str, &len);
+
+	local_host_id = efa_get_host_id(efa_env.host_id_file);
+    if (!local_host_id || EFA_HOST_ID_STRING_LENGTH != snprintf(local_host_id_str, EFA_HOST_ID_STRING_LENGTH + 1, "i-%017lx", local_host_id)) {
+        strcpy(local_host_id_str, "N/A");
+    }
+
+	/* efa-raw cannot get peer host id without a handshake */
+	strcpy(peer_host_id_str, "N/A");
+
+    int ret = snprintf(err_msg, EFA_ERROR_MSG_BUFFER_LENGTH, "%s My EFA addr: %s My host id: %s Peer EFA addr: %s Peer host id: %s",
+                       base_msg, ep_addr_str, local_host_id_str, peer_addr_str, peer_host_id_str);
+
+    if (ret < 0 || ret > EFA_ERROR_MSG_BUFFER_LENGTH - 1) {
+        return -FI_EINVAL;
+    }
+
+    if (strlen(err_msg) >= EFA_ERROR_MSG_BUFFER_LENGTH) {
+        return -FI_ENOBUFS;
+    }
 
-ssize_t efa_dgram_cq_readfrom(struct fid_cq *cq_fid, void *buf, size_t count, fi_addr_t *src_addr);
+    *buf = err_msg;
+    *buflen = EFA_ERROR_MSG_BUFFER_LENGTH;
 
-ssize_t efa_dgram_cq_readerr(struct fid_cq *cq_fid, struct fi_cq_err_entry *entry, uint64_t flags);
+    return 0;
+}
 
 #endif

prov/efa/src/dgram/efa_dgram_cq.h

@@ -71,7 +71,7 @@ static int efa_dgram_ep_close(fid_t fid)
 static int efa_dgram_ep_bind(struct fid *fid, struct fid *bfid, uint64_t flags)
 {
 	struct efa_dgram_ep *ep;
-	struct efa_dgram_cq *cq;
+	struct efa_cq *cq;
 	struct efa_av *av;
 	struct util_eq *eq;
 	struct util_cntr *cntr;
@@ -94,24 +94,14 @@ static int efa_dgram_ep_bind(struct fid *fid, struct fid *bfid, uint64_t flags)
 		if (!(flags & (FI_RECV | FI_TRANSMIT)))
 			return -FI_EBADFLAGS;
 
-		cq = container_of(bfid, struct efa_dgram_cq, util_cq.cq_fid);
+		cq = container_of(bfid, struct efa_cq, util_cq.cq_fid);
 		if (ep->base_ep.domain != cq->domain)
 			return -FI_EINVAL;
 
 		ret = ofi_ep_bind_cq(&ep->base_ep.util_ep, &cq->util_cq, flags);
 		if (ret)
 			return ret;
 
-		if (flags & FI_RECV) {
-			if (ep->rcq)
-				return -EINVAL;
-			ep->rcq = cq;
-		}
-		if (flags & FI_TRANSMIT) {
-			if (ep->scq)
-				return -EINVAL;
-			ep->scq = cq;
-		}
 		break;
 	case FI_CLASS_AV:
 		av = container_of(bfid, struct efa_av, util_av.av_fid.fid);
@@ -186,46 +176,47 @@ static int efa_dgram_ep_setflags(struct fid_ep *ep_fid, uint64_t flags)
 static int efa_dgram_ep_enable(struct fid_ep *ep_fid)
 {
 	struct ibv_qp_init_attr_ex attr_ex = { 0 };
-	struct ibv_pd *ibv_pd;
 	struct efa_dgram_ep *ep;
+	struct efa_cq *scq, *rcq;
 	int err;
 
 	ep = container_of(ep_fid, struct efa_dgram_ep, base_ep.util_ep.ep_fid);
 
-	if (!ep->scq && !ep->rcq) {
+	scq = ep->base_ep.util_ep.tx_cq ? container_of(ep->base_ep.util_ep.tx_cq, struct efa_cq, util_cq) : NULL;
+	rcq = ep->base_ep.util_ep.rx_cq ? container_of(ep->base_ep.util_ep.rx_cq, struct efa_cq, util_cq) : NULL;
+
+	if (!scq && !rcq) {
 		EFA_WARN(FI_LOG_EP_CTRL,
 			"Endpoint is not bound to a send or receive completion queue\n");
 		return -FI_ENOCQ;
 	}
 
-	if (!ep->scq && ofi_send_allowed(ep->base_ep.info->caps)) {
+	if (!scq && ofi_needs_tx(ep->base_ep.info->caps)) {
 		EFA_WARN(FI_LOG_EP_CTRL,
 			"Endpoint is not bound to a send completion queue when it has transmit capabilities enabled (FI_SEND).\n");
 		return -FI_ENOCQ;
 	}
 
-	if (!ep->rcq && ofi_recv_allowed(ep->base_ep.info->caps)) {
+	if (!rcq && ofi_needs_rx(ep->base_ep.info->caps)) {
 		EFA_WARN(FI_LOG_EP_CTRL,
 			"Endpoint is not bound to a receive completion queue when it has receive capabilities enabled. (FI_RECV)\n");
 		return -FI_ENOCQ;
 	}
 
-	if (ep->scq) {
+	if (scq) {
 		attr_ex.cap.max_send_wr = ep->base_ep.info->tx_attr->size;
 		attr_ex.cap.max_send_sge = ep->base_ep.info->tx_attr->iov_limit;
-		attr_ex.send_cq = ibv_cq_ex_to_cq(ep->scq->ibv_cq_ex);
-		ibv_pd = ep->scq->domain->ibv_pd;
+		attr_ex.send_cq = ibv_cq_ex_to_cq(scq->ibv_cq_ex);
 	} else {
-		attr_ex.send_cq = ibv_cq_ex_to_cq(ep->rcq->ibv_cq_ex);
-		ibv_pd = ep->rcq->domain->ibv_pd;
+		attr_ex.send_cq = ibv_cq_ex_to_cq(rcq->ibv_cq_ex);
 	}
 
-	if (ep->rcq) {
+	if (rcq) {
 		attr_ex.cap.max_recv_wr = ep->base_ep.info->rx_attr->size;
 		attr_ex.cap.max_recv_sge = ep->base_ep.info->rx_attr->iov_limit;
-		attr_ex.recv_cq = ibv_cq_ex_to_cq(ep->rcq->ibv_cq_ex);
+		attr_ex.recv_cq = ibv_cq_ex_to_cq(rcq->ibv_cq_ex);
 	} else {
-		attr_ex.recv_cq = ibv_cq_ex_to_cq(ep->scq->ibv_cq_ex);
+		attr_ex.recv_cq = ibv_cq_ex_to_cq(scq->ibv_cq_ex);
 	}
 
 	attr_ex.cap.max_inline_data =
@@ -234,7 +225,7 @@ static int efa_dgram_ep_enable(struct fid_ep *ep_fid)
 	assert(EFA_EP_TYPE_IS_DGRAM(ep->base_ep.domain->info));
 	attr_ex.qp_type = IBV_QPT_UD;
 	attr_ex.comp_mask = IBV_QP_INIT_ATTR_PD;
-	attr_ex.pd = ibv_pd;
+	attr_ex.pd = container_of(ep->base_ep.util_ep.domain, struct efa_domain, util_domain)->ibv_pd;
 
 	attr_ex.qp_context = ep;
 	attr_ex.sq_sig_all = 1;
@@ -277,89 +268,19 @@ static struct fi_ops efa_dgram_ep_ops = {
 	.ops_open = fi_no_ops_open,
 };
 
-static void efa_dgram_ep_progress_internal(struct efa_dgram_ep *ep, struct efa_dgram_cq *efa_dgram_cq)
+/**
+ * @brief progress engine for the EFA dgram endpoint
+ *
+ * This function now a no-op.
+ *
+ * @param[in] util_ep The endpoint FID to progress
+ */
+static
+void efa_ep_progress_no_op(struct util_ep *util_ep)
 {
-	struct util_cq *cq;
-	struct fi_cq_tagged_entry cq_entry[efa_dgram_cq_PROGRESS_ENTRIES] = {0};
-	struct fi_cq_tagged_entry *temp_cq_entry;
-	struct fi_cq_err_entry cq_err_entry = {0};
-	fi_addr_t src_addr[efa_dgram_cq_PROGRESS_ENTRIES];
-	uint64_t flags;
-	int i;
-	ssize_t ret, err;
-
-	cq = &efa_dgram_cq->util_cq;
-	flags = ep->base_ep.util_ep.caps;
-
-	VALGRIND_MAKE_MEM_DEFINED(&cq_entry, sizeof(cq_entry));
-
-	ret = efa_dgram_cq_readfrom(&cq->cq_fid, cq_entry, efa_dgram_cq_PROGRESS_ENTRIES,
-			      (flags & FI_SOURCE) ? src_addr : NULL);
-	if (ret == -FI_EAGAIN)
-		return;
-
-	if (OFI_UNLIKELY(ret < 0)) {
-		if (OFI_UNLIKELY(ret != -FI_EAVAIL)) {
-			EFA_WARN(FI_LOG_CQ, "no error available errno: %ld\n", ret);
-			efa_base_ep_write_eq_error(&ep->base_ep, -ret, FI_EFA_ERR_DGRAM_CQ_READ);
-			return;
-		}
-
-		err = efa_dgram_cq_readerr(&cq->cq_fid, &cq_err_entry, flags);
-		if (OFI_UNLIKELY(err < 0)) {
-			EFA_WARN(FI_LOG_CQ, "unable to read error entry errno: %ld\n", err);
-			efa_base_ep_write_eq_error(&ep->base_ep, FI_EIO, cq_err_entry.prov_errno);
-			return;
-		}
-
-		ofi_cq_write_error(cq, &cq_err_entry);
-		return;
-	}
-
-	temp_cq_entry = (struct fi_cq_tagged_entry *)cq_entry;
-	for (i = 0; i < ret; i++) {
-		(flags & FI_SOURCE) ?
-			ofi_cq_write_src(cq, temp_cq_entry->op_context,
-					 temp_cq_entry->flags,
-					 temp_cq_entry->len,
-					 temp_cq_entry->buf,
-					 temp_cq_entry->data,
-					 temp_cq_entry->tag,
-					 src_addr[i]) :
-			ofi_cq_write(cq, temp_cq_entry->op_context,
-				     temp_cq_entry->flags,
-				     temp_cq_entry->len,
-				     temp_cq_entry->buf,
-				     temp_cq_entry->data,
-				     temp_cq_entry->tag);
-
-		temp_cq_entry = (struct fi_cq_tagged_entry *)
-				((uint8_t *)temp_cq_entry + efa_dgram_cq->entry_size);
-	}
 	return;
 }
 
-void efa_dgram_ep_progress(struct util_ep *ep)
-{
-	struct efa_dgram_ep *efa_dgram_ep;
-	struct efa_dgram_cq *rcq;
-	struct efa_dgram_cq *scq;
-
-	efa_dgram_ep = container_of(ep, struct efa_dgram_ep, base_ep.util_ep);
-	rcq = efa_dgram_ep->rcq;
-	scq = efa_dgram_ep->scq;
-
-	ofi_genlock_lock(&ep->lock);
-
-	if (rcq)
-		efa_dgram_ep_progress_internal(efa_dgram_ep, rcq);
-
-	if (scq && scq != rcq)
-		efa_dgram_ep_progress_internal(efa_dgram_ep, scq);
-
-	ofi_genlock_unlock(&ep->lock);
-}
-
 static struct fi_ops_atomic efa_dgram_ep_atomic_ops = {
 	.size = sizeof(struct fi_ops_atomic),
 	.write = fi_no_atomic_write,
@@ -433,7 +354,7 @@ int efa_dgram_ep_open(struct fid_domain *domain_fid, struct fi_info *user_info,
 	if (!ep)
 		return -FI_ENOMEM;
 
-	ret = efa_base_ep_construct(&ep->base_ep, domain_fid, user_info, efa_dgram_ep_progress, context);
+	ret = efa_base_ep_construct(&ep->base_ep, domain_fid, user_info, efa_ep_progress_no_op, context);
 	if (ret)
 		goto err_ep_destroy;
 

prov/efa/src/dgram/efa_dgram_ep.c

@@ -8,9 +8,6 @@
 
 struct efa_dgram_ep {
 	struct efa_base_ep base_ep;
-
-	struct efa_dgram_cq	*rcq;
-	struct efa_dgram_cq	*scq;
 };
 
 int efa_dgram_ep_open(struct fid_domain *domain_fid, struct fi_info *info,

prov/efa/src/dgram/efa_dgram_ep.h

@@ -70,15 +70,15 @@ struct efa_conn *efa_av_addr_to_conn(struct efa_av *av, fi_addr_t fi_addr)
 }
 
 /**
- * @brief find fi_addr for dgram endpoint
+ * @brief find fi_addr for efa endpoint
  *
  * @param[in]	av	address vector
  * @param[in]	ahn	address handle number
  * @param[in]	qpn	QP number
  * @return	On success, return fi_addr to the peer who send the packet
  * 		If no such peer exist, return FI_ADDR_NOTAVAIL
  */
-fi_addr_t efa_av_reverse_lookup_dgram(struct efa_av *av, uint16_t ahn, uint16_t qpn)
+fi_addr_t efa_av_reverse_lookup(struct efa_av *av, uint16_t ahn, uint16_t qpn)
 {
 	struct efa_cur_reverse_av *cur_entry;
 	struct efa_cur_reverse_av_key cur_key;

prov/efa/src/efa_av.c

@@ -86,6 +86,6 @@ struct efa_conn *efa_av_addr_to_conn(struct efa_av *av, fi_addr_t fi_addr);
 
 fi_addr_t efa_av_reverse_lookup_rdm(struct efa_av *av, uint16_t ahn, uint16_t qpn, struct efa_rdm_pke *pkt_entry);
 
-fi_addr_t efa_av_reverse_lookup_dgram(struct efa_av *av, uint16_t ahn, uint16_t qpn);
+fi_addr_t efa_av_reverse_lookup(struct efa_av *av, uint16_t ahn, uint16_t qpn);
 
 #endif