prov/efa: Introduce new protocol interface and implement it for the TX path

prov/efa/Makefile.include

@@ -78,7 +78,13 @@ _efa_files = \
 	prov/efa/src/rdm/efa_rdm_tracepoint_def.c \
 	prov/efa/src/rdm/efa_rdm_srx.c \
 	prov/efa/src/rdm/efa_rdm_util.c \
-	prov/efa/src/rdm/efa_rdm_mr.c
+	prov/efa/src/rdm/efa_rdm_mr.c \
+	prov/efa/src/rdm/efa_rdm_proto.c \
+	prov/efa/src/rdm/efa_rdm_proto_eager.c \
+	prov/efa/src/rdm/efa_rdm_proto_medium.c \
+	prov/efa/src/rdm/efa_rdm_proto_longcts.c \
+	prov/efa/src/rdm/efa_rdm_proto_longread.c \
+	prov/efa/src/rdm/efa_rdm_proto_runtread.c
 
 if ENABLE_EFA_UNIT_TEST
 _efa_files += prov/efa/test/efa_unit_test_data_path_ops.c
@@ -140,7 +146,13 @@ _efa_headers = \
 	prov/efa/src/rdm/efa_rdm_tracepoint.h \
 	prov/efa/src/rdm/efa_rdm_srx.h \
 	prov/efa/src/rdm/efa_rdm_util.h \
-	prov/efa/src/rdm/efa_rdm_mr.h
+	prov/efa/src/rdm/efa_rdm_mr.h \
+	prov/efa/src/rdm/efa_rdm_proto.h \
+	prov/efa/src/rdm/efa_rdm_proto_eager.h \
+	prov/efa/src/rdm/efa_rdm_proto_medium.h \
+	prov/efa/src/rdm/efa_rdm_proto_longcts.h \
+	prov/efa/src/rdm/efa_rdm_proto_longread.h \
+	prov/efa/src/rdm/efa_rdm_proto_runtread.h
 
 if HAVE_LTTNG
 efa_LDFLAGS += -llttng-ust
@@ -174,7 +186,8 @@ nodist_prov_efa_test_efa_unit_test_SOURCES = \
 	prov/efa/test/efa_unit_test_msg.c \
 	prov/efa/test/efa_unit_test_rma.c \
 	prov/efa/test/efa_unit_test_rdm_rma.c \
-	prov/efa/test/efa_unit_test_data_path_direct.c
+	prov/efa/test/efa_unit_test_data_path_direct.c \
+	prov/efa/test/efa_unit_test_proto.c
 
 
 efa_CPPFLAGS += -I$(top_srcdir)/include -I$(top_srcdir)/prov/efa/test $(cmocka_CPPFLAGS)

prov/efa/src/efa.h

@@ -93,21 +93,6 @@
 #define EFA_DEFAULT_INTER_MIN_READ_WRITE_SIZE (65536)
 #define EFA_DEFAULT_INTRA_MAX_GDRCOPY_FROM_DEV_SIZE (3072)
 
-/*
- * The default memory alignment
- */
-#define EFA_RDM_DEFAULT_MEMORY_ALIGNMENT (8)
-
-/*
- * The CUDA memory alignment
- */
-#define EFA_RDM_CUDA_MEMORY_ALIGNMENT (64)
-
-/*
- * The alignment to support in-order aligned ops.
- */
-#define EFA_RDM_IN_ORDER_ALIGNMENT (128)
-
 /*
  * Set alignment to x86 cache line size.
  */

prov/efa/src/rdm/efa_rdm_atomic.c

@@ -58,10 +58,6 @@ efa_rdm_atomic_alloc_txe(struct efa_rdm_ep *efa_rdm_ep,
 		return NULL;
 	}
 
-	efa_domain_ope_list_lock(efa_rdm_ep_domain(efa_rdm_ep));
-	dlist_insert_tail(&txe->ep_entry, &efa_rdm_ep->txe_list);
-	efa_domain_ope_list_unlock(efa_rdm_ep_domain(efa_rdm_ep));
-
 	ofi_ioc_to_iov(msg_atomic->msg_iov, iov, msg_atomic->iov_count, datatype_size);
 	msg.addr = msg_atomic->addr;
 	msg.msg_iov = iov;

prov/efa/src/rdm/efa_rdm_cq.c

@@ -842,8 +842,36 @@ enum ibv_wc_status efa_rdm_cq_process_wc(struct efa_ibv_cq *cq, struct efa_rdm_e
 #if ENABLE_DEBUG
 			ep->send_comps++;
 #endif
-			efa_rdm_pke_handle_send_completion(pkt_entry);
-			efa_rdm_cq_increment_pkt_entry_gen(pkt_entry);
+			if (pkt_entry->callback) {
+				efa_rdm_ep_record_tx_op_completed(pkt_entry->ep,
+								  pkt_entry);
+				/*
+				 * For a send completion, pkt_entry->peer can be
+				 * NULL in 3 situations:
+				 * 1. the pkt_entry is used for a local read
+				 * operation
+				 * 2. a new peer with same gid+qpn was inserted
+				 * to av, thus the peer was removed from AV.
+				 * 3. application removed the peer's address
+				 * from av. In 1, we should proceed. For 2 and
+				 * 3, the send completion should be ignored.
+				 */
+				if (!pkt_entry->peer &&
+				    !(pkt_entry->flags &
+				      EFA_RDM_PKE_LOCAL_READ)) {
+					EFA_WARN(
+						FI_LOG_CQ,
+						"ignoring send completion of a "
+						"packet to a removed peer.\n");
+					efa_rdm_pke_release_tx(pkt_entry);
+				} else {
+					pkt_entry->callback(pkt_entry);
+				}
+				efa_rdm_cq_increment_pkt_entry_gen(pkt_entry);
+			} else {
+				efa_rdm_pke_handle_send_completion(pkt_entry);
+				efa_rdm_cq_increment_pkt_entry_gen(pkt_entry);
+			}
 			break;
 		case IBV_WC_RECV:
 			/* efa_rdm_cq_handle_recv_completion does additional work to determine the source

prov/efa/src/rdm/efa_rdm_ep.h

@@ -29,6 +29,21 @@ struct efa_rdm_ep_queued_copy {
 
 #define EFA_RDM_MAX_QUEUED_COPY (8)
 
+/*
+ * The default memory alignment
+ */
+#define EFA_RDM_DEFAULT_MEMORY_ALIGNMENT (8)
+
+/*
+ * The CUDA memory alignment
+ */
+#define EFA_RDM_CUDA_MEMORY_ALIGNMENT (64)
+
+/*
+ * The alignment to support in-order aligned ops.
+ */
+#define EFA_RDM_IN_ORDER_ALIGNMENT (128)
+
 /**
  * Max number of opes queued before handshake is made
  * with their peers. This cnt is per EP.
@@ -195,7 +210,8 @@ struct efa_rdm_ep {
 	struct efa_rdm_pke **pke_vec;
 	/* Work arrays for efa_rdm_ope_post_send to avoid stack allocation */
 	struct efa_rdm_pke **send_pkt_entry_vec;
-	int *send_pkt_entry_size_vec;
+	size_t *send_pkt_entry_data_sizes;
+	size_t send_pkt_entry_vec_size;
 	struct dlist_entry entry;
 	/* the count of opes queued before handshake is made with their peers */
 	size_t ope_queued_before_handshake_cnt;
@@ -220,13 +236,6 @@ struct efa_rdm_peer *efa_rdm_ep_get_peer_explicit(struct efa_rdm_ep *ep, fi_addr
 int32_t efa_rdm_ep_get_peer_ahn(struct efa_rdm_ep *ep, fi_addr_t addr);
 struct efa_rdm_peer *efa_rdm_ep_get_peer_implicit(struct efa_rdm_ep *ep, fi_addr_t addr);
 
-struct efa_rdm_ope *efa_rdm_ep_alloc_txe(struct efa_rdm_ep *efa_rdm_ep,
-					 struct efa_rdm_peer *peer,
-					 const struct fi_msg *msg,
-					 uint32_t op,
-					 uint64_t tag,
-					 uint64_t flags);
-
 struct efa_rdm_ope *efa_rdm_ep_alloc_rxe(struct efa_rdm_ep *ep,
 					   struct efa_rdm_peer *peer, uint32_t op);
 
@@ -253,7 +262,26 @@ void efa_rdm_ep_queue_rnr_pkt(struct efa_rdm_ep *ep, struct efa_rdm_pke *pkt_ent
 ssize_t efa_rdm_ep_post_queued_pkts(struct efa_rdm_ep *ep,
 				    struct dlist_entry *pkts);
 
-size_t efa_rdm_ep_get_memory_alignment(struct efa_rdm_ep *ep, enum fi_hmem_iface iface);
+/**
+ * @brief Get memory alignment for given ep and hmem iface
+ *
+ * @param ep efa rdm ep
+ * @param iface hmem iface
+ * @return size_t the memory alignment
+ */
+static inline size_t efa_rdm_ep_get_memory_alignment(struct efa_rdm_ep *ep,
+						     enum fi_hmem_iface iface)
+{
+	size_t memory_alignment = EFA_RDM_DEFAULT_MEMORY_ALIGNMENT;
+
+	if (ep->sendrecv_in_order_aligned_128_bytes) {
+		memory_alignment = EFA_RDM_IN_ORDER_ALIGNMENT;
+	} else if (iface == FI_HMEM_CUDA) {
+		memory_alignment = EFA_RDM_CUDA_MEMORY_ALIGNMENT;
+	}
+
+	return memory_alignment;
+}
 
 static inline
 struct efa_domain *efa_rdm_ep_domain(struct efa_rdm_ep *ep)

prov/efa/src/rdm/efa_rdm_ep_fiops.c

@@ -680,9 +680,13 @@ int efa_rdm_ep_open(struct fid_domain *domain, struct fi_info *info,
 		goto err_free_pke_vec;
 	}
 
-	efa_rdm_ep->send_pkt_entry_size_vec = calloc(sizeof(int), efa_base_ep_get_tx_pool_size(&efa_rdm_ep->base_ep));
-	if (!efa_rdm_ep->send_pkt_entry_size_vec) {
-		EFA_WARN(FI_LOG_EP_CTRL, "cannot alloc memory for efa_rdm_ep->send_pkt_entry_size_vec!\n");
+	efa_rdm_ep->send_pkt_entry_data_sizes =
+		calloc(sizeof(size_t),
+		       efa_base_ep_get_tx_pool_size(&efa_rdm_ep->base_ep));
+	if (!efa_rdm_ep->send_pkt_entry_data_sizes) {
+		EFA_WARN(FI_LOG_EP_CTRL,
+			 "cannot alloc memory for "
+			 "efa_rdm_ep->send_pkt_entry_data_sizes!\n");
 		ret = -FI_ENOMEM;
 		goto err_free_send_pkt_entry_vec;
 	}
@@ -1190,8 +1194,8 @@ static int efa_rdm_ep_close(struct fid *fid)
 		free(efa_rdm_ep->pke_vec);
 	if (efa_rdm_ep->send_pkt_entry_vec)
 		free(efa_rdm_ep->send_pkt_entry_vec);
-	if (efa_rdm_ep->send_pkt_entry_size_vec)
-		free(efa_rdm_ep->send_pkt_entry_size_vec);
+	if (efa_rdm_ep->send_pkt_entry_data_sizes)
+		free(efa_rdm_ep->send_pkt_entry_data_sizes);
 
 	ofi_genlock_unlock(&domain->srx_lock);
 

prov/efa/src/rdm/efa_rdm_ep_utils.c

@@ -317,36 +317,6 @@ int efa_rdm_ep_post_user_recv_buf(struct efa_rdm_ep *ep, struct efa_rdm_ope *rxe
 	return err;
 }
 
-
-
-/* create a new txe */
-struct efa_rdm_ope *efa_rdm_ep_alloc_txe(struct efa_rdm_ep *efa_rdm_ep,
-					 struct efa_rdm_peer *peer,
-					 const struct fi_msg *msg,
-					 uint32_t op,
-					 uint64_t tag,
-					 uint64_t flags)
-{
-	struct efa_rdm_ope *txe;
-
-	txe = ofi_buf_alloc(efa_rdm_ep->ope_pool);
-	if (OFI_UNLIKELY(!txe)) {
-		EFA_DBG(FI_LOG_EP_CTRL, "TX entries exhausted.\n");
-		return NULL;
-	}
-
-	efa_rdm_txe_construct(txe, efa_rdm_ep, peer, msg, op, flags);
-	if (op == ofi_op_tagged) {
-		txe->cq_entry.tag = tag;
-		txe->tag = tag;
-	}
-
-	efa_domain_ope_list_lock(efa_rdm_ep_domain(efa_rdm_ep));
-	dlist_insert_tail(&txe->ep_entry, &efa_rdm_ep->txe_list);
-	efa_domain_ope_list_unlock(efa_rdm_ep_domain(efa_rdm_ep));
-	return txe;
-}
-
 /**
  * @brief record the event that a TX op has been submitted
  *
@@ -674,15 +644,18 @@ static ssize_t efa_rdm_ep_handshake_common(struct efa_rdm_ep *ep, struct efa_rdm
 
 	msg.addr = peer->conn->fi_addr;
 
-	txe = efa_rdm_ep_alloc_txe(ep, peer, &msg, ofi_op_write, 0, 0);
-
+	txe = ofi_buf_alloc(ep->ope_pool);
 	if (OFI_UNLIKELY(!txe)) {
 		EFA_WARN(FI_LOG_EP_CTRL, "TX entries exhausted.\n");
 		return -FI_EAGAIN;
 	}
 
-	/* efa_rdm_ep_alloc_txe() joins ep->base_ep.util_ep.tx_op_flags and passed in flags,
-	 * reset to desired flags (remove things like FI_DELIVERY_COMPLETE, and FI_COMPLETION)
+	efa_rdm_txe_construct(txe, ep, peer, &msg, ofi_op_write, 0);
+
+	/*
+	 * efa_rdm_txe_construct() joins ep->base_ep.util_ep.tx_op_flags and
+	 * passed in flags, reset to desired flags (remove things like
+	 * FI_DELIVERY_COMPLETE, and FI_COMPLETION)
 	 */
 	txe->fi_flags = EFA_RDM_TXE_NO_COMPLETION | EFA_RDM_TXE_NO_COUNTER;
 	txe->internal_flags |= EFA_RDM_OPE_INTERNAL;
@@ -1068,26 +1041,6 @@ void efa_rdm_ep_post_internal_rx_pkts(struct efa_rdm_ep *ep)
 	efa_base_ep_write_eq_error(&ep->base_ep, err, FI_EFA_ERR_INTERNAL_RX_BUF_POST);
 }
 
-/**
- * @brief Get memory alignment for given ep and hmem iface
- *
- * @param ep efa rdm ep
- * @param iface hmem iface
- * @return size_t the memory alignment
- */
-size_t efa_rdm_ep_get_memory_alignment(struct efa_rdm_ep *ep, enum fi_hmem_iface iface)
-{
-	size_t memory_alignment = EFA_RDM_DEFAULT_MEMORY_ALIGNMENT;
-
-	if (ep->sendrecv_in_order_aligned_128_bytes) {
-		memory_alignment = EFA_RDM_IN_ORDER_ALIGNMENT;
-	} else if (iface == FI_HMEM_CUDA) {
-		memory_alignment = EFA_RDM_CUDA_MEMORY_ALIGNMENT;
-	}
-
-	return memory_alignment;
-}
-
 /**
  * @brief Enforce a handshake to made for given txe.
  * It will trigger a handshake with peer and choose to