prov/efa: Bypass rdma-core in data path.

This patch allows libfabric to bypass rdma-core API in
data path to reduce layering overhead and cache misses.
It starts from making libfabric call internal ops that
is the equivalent of the rdma-core API. Further optimization
is needed to get rid of the rdma-core API workflow restriction
and implement the logic directly for the transmission and cq
poll calls.

Co-authored-by: Luke Robison <[email protected]>

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

Author: shijin-aws

include/windows/config.h

@@ -43,6 +43,9 @@
 /* Indicates if ibv_reg_dmabuf_mr verbs is available */
 #define HAVE_EFA_DMABUF_MR 0
 
+/* Indicates if efa data path direct is available */
+#define HAVE_EFA_DATA_PATH_DIRECT 0
+
 /* Define to 1 if host_clock_get_service is available. */
 /* #undef HAVE_HOST_GET_CLOCK_SERVICE */
 

man/fi_efa.7.md

@@ -522,6 +522,11 @@ Setting this environment variable to 0 can disable this feature.
 When the number of internal rx pkts to post is lower than this threshold,
 the refill will be skipped.
 
+*FI_EFA_USE_DATA_PATH_DIRECT*
+
+: Use the direct data path implementation that bypasses rdma-core on data path, including the CQ polling and TX/RX submissions, when it's available.
+Setting this variable as 1 will enable this feature (Default: false).
+
 # SEE ALSO
 
 [`fabric`(7)](fabric.7.html),

prov/efa/Makefile.include

@@ -51,6 +51,7 @@ _efa_files = \
 	prov/efa/src/efa_rma.c \
 	prov/efa/src/efa_cq.c \
 	prov/efa/src/efa_ep.c \
+	prov/efa/src/efa_data_path_direct.c \
 	prov/efa/src/rdm/efa_rdm_peer.c	\
 	prov/efa/src/rdm/efa_rdm_cq.c \
 	prov/efa/src/rdm/efa_rdm_ep_utils.c \
@@ -75,6 +76,10 @@ _efa_files = \
 	prov/efa/src/rdm/efa_rdm_srx.c \
 	prov/efa/src/rdm/efa_rdm_util.c
 
+if ENABLE_EFA_UNIT_TEST
+_efa_files += prov/efa/test/efa_unit_test_data_path_ops.c
+endif ENABLE_EFA_UNIT_TEST
+
 _efa_headers = \
 	prov/efa/src/efa.h \
 	prov/efa/src/efa_av.h \
@@ -95,6 +100,13 @@ _efa_headers = \
 	prov/efa/src/efa_prov.h \
 	prov/efa/src/efa_env.h \
 	prov/efa/src/fi_ext_efa.h \
+	prov/efa/src/efa_data_path_ops.h \
+	prov/efa/src/efa_io_defs.h \
+	prov/efa/src/efa_data_path_direct_structs.h \
+	prov/efa/src/efa_data_path_direct_entry.h \
+	prov/efa/src/efa_data_path_direct.h \
+	prov/efa/src/efa_data_path_direct_internal.h \
+	prov/efa/src/efa_mmio.h \
 	prov/efa/src/rdm/efa_rdm_peer.h	\
 	prov/efa/src/rdm/efa_rdm_cq.h \
 	prov/efa/src/rdm/efa_rdm_ep.h \
@@ -166,7 +178,31 @@ prov_efa_test_efa_unit_test_LDFLAGS = $(cmocka_rpath) $(efa_LDFLAGS) $(cmocka_LD
 					-Wl,--wrap=efa_rdm_pke_read \
 					-Wl,--wrap=efa_rdm_pke_proc_matched_rtm \
 					-Wl,--wrap=efa_rdm_ope_post_send \
-					-Wl,--wrap=efa_device_support_unsolicited_write_recv
+					-Wl,--wrap=efa_device_support_unsolicited_write_recv \
+					-Wl,--wrap=efa_qp_post_recv \
+					-Wl,--wrap=efa_qp_wr_complete \
+					-Wl,--wrap=efa_qp_wr_send \
+					-Wl,--wrap=efa_qp_wr_start \
+					-Wl,--wrap=efa_qp_wr_rdma_read \
+					-Wl,--wrap=efa_qp_wr_rdma_write \
+					-Wl,--wrap=efa_qp_wr_rdma_write_imm \
+					-Wl,--wrap=efa_qp_wr_send_imm \
+					-Wl,--wrap=efa_qp_wr_set_inline_data_list \
+					-Wl,--wrap=efa_qp_wr_set_sge_list \
+					-Wl,--wrap=efa_qp_wr_set_ud_addr \
+					-Wl,--wrap=efa_ibv_cq_start_poll \
+					-Wl,--wrap=efa_ibv_cq_next_poll \
+					-Wl,--wrap=efa_ibv_cq_end_poll \
+					-Wl,--wrap=efa_ibv_cq_wc_read_opcode \
+					-Wl,--wrap=efa_ibv_cq_wc_read_qp_num \
+					-Wl,--wrap=efa_ibv_cq_wc_read_vendor_err \
+					-Wl,--wrap=efa_ibv_cq_wc_read_src_qp \
+					-Wl,--wrap=efa_ibv_cq_wc_read_slid \
+					-Wl,--wrap=efa_ibv_cq_wc_read_byte_len \
+					-Wl,--wrap=efa_ibv_cq_wc_read_wc_flags \
+					-Wl,--wrap=efa_ibv_cq_wc_read_imm_data \
+					-Wl,--wrap=efa_ibv_cq_wc_is_unsolicited \
+					-Wl,--wrap=efa_ibv_cq_wc_read_sgid
 
 if HAVE_EFADV_CQ_EX
 prov_efa_test_efa_unit_test_LDFLAGS += -Wl,--wrap=efadv_create_cq

prov/efa/configure.m4

@@ -226,6 +226,12 @@ AC_DEFUN([FI_EFA_CONFIGURE],[
 	AC_DEFINE_UNQUOTED([HAVE_EFADV_QUERY_CQ],
 		[$have_efadv_query_cq],
 		[Indicates if efadv_query_cq is available])
+	AS_IF([test "$have_efadv_query_qp_wqs" = "1" -a "$have_efadv_query_cq" = "1"],
+		[have_efa_data_path_direct=1],
+		[have_efa_data_path_direct=0])
+	AC_DEFINE_UNQUOTED([HAVE_EFA_DATA_PATH_DIRECT],
+		[$have_efa_data_path_direct],
+		[Indicates if data path direct is available (requires both QUERY_QP_WQS and QUERY_CQ)])
 
 
 	CPPFLAGS=$save_CPPFLAGS

prov/efa/src/efa_base_ep.c

@@ -7,6 +7,7 @@
 #include "efa_cq.h"
 #include "efa_cntr.h"
 #include "rdm/efa_rdm_protocol.h"
+#include "efa_data_path_direct.h"
 
 int efa_base_ep_bind_av(struct efa_base_ep *base_ep, struct efa_av *av)
 {
@@ -251,9 +252,12 @@ int efa_qp_create(struct efa_qp **qp, struct ibv_qp_init_attr_ex *init_attr_ex,
 	}
 
 	(*qp)->ibv_qp_ex = ibv_qp_to_qp_ex((*qp)->ibv_qp);
+	/* Initialize it explicitly for safety */
+	(*qp)->data_path_direct_enabled = false;
 	return FI_SUCCESS;
 }
 
+static
 int efa_base_ep_create_qp(struct efa_base_ep *base_ep,
 			  struct ibv_qp_init_attr_ex *init_attr_ex)
 {
@@ -312,6 +316,10 @@ void efa_qp_destruct(struct efa_qp *qp)
 	err = -ibv_destroy_qp(qp->ibv_qp);
 	if (err)
 		EFA_INFO(FI_LOG_CORE, "destroy qp[%u] failed, err: %s\n", qp->qp_num, fi_strerror(-err));
+#if HAVE_EFA_DATA_PATH_DIRECT
+	if (qp->data_path_direct_enabled)
+		efa_data_path_direct_qp_finalize(qp);
+#endif
 	free(qp);
 }
 
@@ -725,40 +733,51 @@ 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)
 {
 	struct ibv_qp_init_attr_ex attr_ex = { 0 };
-	struct efa_cq *scq, *rcq;
-	struct ibv_cq_ex *tx_ibv_cq, *rx_ibv_cq;
+	struct efa_cq *scq, *rcq, *txcq, *rxcq;
 	int err;
 
-	scq = efa_base_ep_get_tx_cq(ep);
-	rcq = efa_base_ep_get_rx_cq(ep);
+	txcq = efa_base_ep_get_tx_cq(ep);
+	rxcq = efa_base_ep_get_rx_cq(ep);
 
-	if (!scq && !rcq) {
+	if (!txcq && !rxcq) {
 		EFA_WARN(FI_LOG_EP_CTRL,
 			"Endpoint is not bound to a send or receive completion queue\n");
 		return -FI_ENOCQ;
 	}
 
-	if (!scq && ofi_needs_tx(ep->info->caps)) {
+	if (!txcq && ofi_needs_tx(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 (!rcq && ofi_needs_rx(ep->info->caps)) {
+	if (!rxcq && ofi_needs_rx(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;
 	}
 
-	tx_ibv_cq = scq ? scq->ibv_cq.ibv_cq_ex : rcq->ibv_cq.ibv_cq_ex;
-	rx_ibv_cq = rcq ? rcq->ibv_cq.ibv_cq_ex : scq->ibv_cq.ibv_cq_ex;
+	scq = txcq ? txcq : rxcq;
+	rcq = rxcq ? rxcq : txcq;
 
-	efa_base_ep_construct_ibv_qp_init_attr_ex(ep, &attr_ex, tx_ibv_cq, rx_ibv_cq);
+	efa_base_ep_construct_ibv_qp_init_attr_ex(ep, &attr_ex, scq->ibv_cq.ibv_cq_ex, rcq->ibv_cq.ibv_cq_ex);
 
 	err = efa_base_ep_create_qp(ep, &attr_ex);
 	if (err)
 		return err;
 
+#if HAVE_EFA_DATA_PATH_DIRECT
+	/* Only enable direct QP when direct CQ is enabled */
+	assert(scq->ibv_cq.data_path_direct_enabled == rcq->ibv_cq.data_path_direct_enabled);
+	if (scq->ibv_cq.data_path_direct_enabled) {
+		err = efa_data_path_direct_qp_initialize(ep->qp);
+		if (err) {
+			efa_base_ep_destruct_qp(ep);
+			return err;
+		}
+	}
+#endif
+
 	if (create_user_recv_qp) {
 		err = efa_qp_create(&ep->user_recv_qp, &attr_ex, ep->info->tx_attr->tclass);
 		if (err) {

prov/efa/src/efa_base_ep.h

@@ -12,6 +12,7 @@
 #include "ofi_util.h"
 #include "efa_av.h"
 #include "rdm/efa_rdm_protocol.h"
+#include "efa_data_path_direct_structs.h"
 
 #define EFA_QP_DEFAULT_SERVICE_LEVEL 0
 #define EFA_QP_LOW_LATENCY_SERVICE_LEVEL 8
@@ -46,9 +47,12 @@ struct efa_qp {
 	struct efa_base_ep *base_ep;
 	uint32_t qp_num;
 	uint32_t qkey;
+	bool data_path_direct_enabled;
+#if HAVE_EFA_DATA_PATH_DIRECT
+	struct efa_data_path_direct_qp data_path_direct_qp;
+#endif
 };
 
-
 struct efa_av;
 
 struct efa_recv_wr {
@@ -111,9 +115,6 @@ int efa_qp_create(struct efa_qp **qp, struct ibv_qp_init_attr_ex *init_attr_ex,
 
 void efa_qp_destruct(struct efa_qp *qp);
 
-int efa_base_ep_create_qp(struct efa_base_ep *base_ep,
-			  struct ibv_qp_init_attr_ex *init_attr_ex);
-
 void efa_base_ep_close_util_ep(struct efa_base_ep *base_ep);
 
 int efa_base_ep_destruct_qp(struct efa_base_ep *base_ep);