prov/efa: Add packet entry flag tracking for double linked list management

shijin-aws · shijin-aws · commit ef67891db9eb · 2025-10-09T15:26:38.000-07:00
Add flags EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS and EFA_RDM_PKE_IN_OPE_QUEUED_PKTS
to track packet entry insertion status in outstanding_tx_pkts and queued_pkts lists.
Update insertion/removal operations to set/clear these flags and modify
efa_rdm_pke_release_tx() to safely remove packets from lists based on flag state.
Add comprehensive unit test to verify flag tracking behavior.

Signed-off-by: Shi Jin &lt;sjina@amazon.com&gt;
diff --git a/prov/efa/src/rdm/efa_rdm_ep_utils.c b/prov/efa/src/rdm/efa_rdm_ep_utils.c
@@ -409,9 +409,11 @@ void efa_rdm_ep_record_tx_op_submitted(struct efa_rdm_ep *ep, struct efa_rdm_pke
 	 * and the RMA is a local read toward the endpoint itself
 	 */
 	peer = ope->peer;
-	if (peer)
+	if (peer) {
 		dlist_insert_tail(&pkt_entry->entry,
 				  &peer->outstanding_tx_pkts);
+		pkt_entry->flags |= EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS;
+	}
 
 	assert(pkt_entry->alloc_type == EFA_RDM_PKE_FROM_EFA_TX_POOL);
 	ep->efa_outstanding_tx_ops++;
@@ -471,8 +473,10 @@ void efa_rdm_ep_record_tx_op_completed(struct efa_rdm_ep *ep, struct efa_rdm_pke
 	 *    a new peer has the same GID+QPN was inserted to address, or because
 	 *    application removed the peer from address vector.
 	 */
-	if (pkt_entry->peer)
+	if (pkt_entry->peer) {
 		dlist_remove(&pkt_entry->entry);
+		pkt_entry->flags &= ~EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS;
+	}
 
 	assert(pkt_entry->alloc_type == EFA_RDM_PKE_FROM_EFA_TX_POOL);
 	ep->efa_outstanding_tx_ops--;
@@ -543,6 +547,7 @@ void efa_rdm_ep_queue_rnr_pkt(struct efa_rdm_ep *ep, struct efa_rdm_pke *pkt_ent
 
 	assert(ope);
 	dlist_insert_tail(&pkt_entry->entry, &ope->queued_pkts);
+	pkt_entry->flags |= EFA_RDM_PKE_IN_OPE_QUEUED_PKTS;
 	ep->efa_rnr_queued_pkt_cnt += 1;
 	assert(peer);
 	if (!(ope->internal_flags & EFA_RDM_OPE_QUEUED_RNR)) {
@@ -783,6 +788,7 @@ ssize_t efa_rdm_ep_post_queued_pkts(struct efa_rdm_ep *ep,
 		 * be removed from the list before send.
 		 */
 		dlist_remove(&pkt_entry->entry);
+		pkt_entry->flags &= ~EFA_RDM_PKE_IN_OPE_QUEUED_PKTS;
 
 		switch (efa_rdm_pkt_type_of(pkt_entry)) {
 		case EFA_RDM_RMA_CONTEXT_PKT:
@@ -798,6 +804,7 @@ ssize_t efa_rdm_ep_post_queued_pkts(struct efa_rdm_ep *ep,
 			if (ret == -FI_EAGAIN) {
 				/* add the pkt back to pkts, so it can be resent again */
 				dlist_insert_tail(&pkt_entry->entry, pkts);
+				pkt_entry->flags |= EFA_RDM_PKE_IN_OPE_QUEUED_PKTS;
 			}
 
 			return ret;
diff --git a/prov/efa/src/rdm/efa_rdm_pke.c b/prov/efa/src/rdm/efa_rdm_pke.c
@@ -118,6 +118,14 @@ void efa_rdm_pke_release_tx(struct efa_rdm_pke *pkt_entry)
 #if ENABLE_DEBUG
 	dlist_remove(&pkt_entry->dbg_entry);
 #endif
+	/*
+	 * Remove packet from double linked lists if it was inserted
+	 */
+	if (pkt_entry->flags & (EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS | EFA_RDM_PKE_IN_OPE_QUEUED_PKTS)) {
+		dlist_remove(&pkt_entry->entry);
+		pkt_entry->flags &= ~(EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS | EFA_RDM_PKE_IN_OPE_QUEUED_PKTS);
+	}
+
 	/*
 	 * Decrement rnr_queued_pkts counter and reset backoff for this peer if
 	 * we get a send completion for a retransmitted packet.
diff --git a/prov/efa/src/rdm/efa_rdm_pke.h b/prov/efa/src/rdm/efa_rdm_pke.h
@@ -16,6 +16,8 @@
 #define EFA_RDM_PKE_LOCAL_WRITE		BIT_ULL(4) /**< this packet entry is used as context of an RDMA Write to self */
 #define EFA_RDM_PKE_SEND_TO_USER_RECV_QP	BIT_ULL(5) /**< this packet entry is used for posting send to a dedicated QP that doesn't expect any pkt hdrs */
 #define EFA_RDM_PKE_HAS_NO_BASE_HDR	BIT_ULL(6)	/**< This packet entry's wiredata contains no base header */
+#define EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS	BIT_ULL(7) /**< this packet entry is in peer->outstanding_tx_pkts list */
+#define EFA_RDM_PKE_IN_OPE_QUEUED_PKTS	BIT_ULL(8) /**< this packet entry is in ope->queued_pkts list */
 
 #define EFA_RDM_PKE_ALIGNMENT		128
 
@@ -137,6 +139,8 @@ struct efa_rdm_pke {
 	 * @see #EFA_RDM_PKE_LOCAL_WRITE
 	 * @see #EFA_RDM_PKE_SEND_TO_USER_RECV_QP
 	 * @see #EFA_RDM_PKE_HAS_NO_BASE_HDR
+	 * @see #EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS
+	 * @see #EFA_RDM_PKE_IN_OPE_QUEUED_PKTS
 	 */
 	uint32_t flags;
 
diff --git a/prov/efa/test/efa_unit_test_pke.c b/prov/efa/test/efa_unit_test_pke.c
@@ -275,3 +275,96 @@ void test_efa_rdm_pke_get_unexp(struct efa_resource **state)
 	assert_int_equal(efa_unit_test_get_dlist_length(&efa_rdm_ep->rx_pkt_list), 1);
 #endif
 }
+
+/**
+ * @brief Test packet entry flag tracking for double linked list insertion/removal
+ *
+ * This test verifies that packet entries correctly track their insertion status
+ * in outstanding_tx_pkts and queued_pkts lists using internal flags, and that
+ * the release function properly removes them from the appropriate lists.
+ *
+ * @param state
+ */
+void test_efa_rdm_pke_flag_tracking(struct efa_resource **state)
+{
+	struct efa_resource *resource = *state;
+	struct efa_rdm_ep *efa_rdm_ep;
+	struct efa_rdm_pke *pkt_entry;
+	struct efa_rdm_peer *peer;
+	struct efa_rdm_ope *txe;
+	struct fi_msg msg = {0};
+	char buf[16];
+	struct iovec iov = {
+		.iov_base = buf,
+		.iov_len = sizeof(buf)
+	};
+	struct efa_ep_addr raw_addr = {0};
+	size_t raw_addr_len = sizeof(struct efa_ep_addr);
+	fi_addr_t peer_addr;
+	int err, numaddr;
+
+	efa_unit_test_resource_construct(resource, FI_EP_RDM, EFA_FABRIC_NAME);
+
+	efa_rdm_ep = container_of(resource->ep, struct efa_rdm_ep, base_ep.util_ep.ep_fid);
+
+	/* Create a fake peer */
+	err = fi_getname(&resource->ep->fid, &raw_addr, &raw_addr_len);
+	assert_int_equal(err, 0);
+	raw_addr.qpn = 1;
+	raw_addr.qkey = 0x1234;
+	numaddr = fi_av_insert(resource->av, &raw_addr, 1, &peer_addr, 0, NULL);
+	assert_int_equal(numaddr, 1);
+	peer = efa_rdm_ep_get_peer(efa_rdm_ep, peer_addr);
+	assert_non_null(peer);
+
+	/* Create a txe */
+	msg.addr = peer_addr;
+	msg.iov_count = 1;
+	msg.msg_iov = &iov;
+	msg.desc = NULL;
+	txe = efa_rdm_ep_alloc_txe(efa_rdm_ep, peer, &msg, ofi_op_msg, 0, 0);
+	assert_non_null(txe);
+
+	/* Allocate a packet entry */
+	pkt_entry = efa_rdm_pke_alloc(efa_rdm_ep, efa_rdm_ep->efa_tx_pkt_pool, EFA_RDM_PKE_FROM_EFA_TX_POOL);
+	assert_non_null(pkt_entry);
+	pkt_entry->ope = txe;
+	pkt_entry->peer = peer;
+
+	/* Initially, packet should not be in any list */
+	assert_int_equal(pkt_entry->flags & EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS, 0);
+	assert_int_equal(pkt_entry->flags & EFA_RDM_PKE_IN_OPE_QUEUED_PKTS, 0);
+	assert_true(dlist_empty(&peer->outstanding_tx_pkts));
+	assert_true(dlist_empty(&txe->queued_pkts));
+
+	/* Test insertion into outstanding_tx_pkts */
+	efa_rdm_ep_record_tx_op_submitted(efa_rdm_ep, pkt_entry);
+	assert_int_not_equal(pkt_entry->flags & EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS, 0);
+	assert_int_equal(pkt_entry->flags & EFA_RDM_PKE_IN_OPE_QUEUED_PKTS, 0);
+	assert_false(dlist_empty(&peer->outstanding_tx_pkts));
+	assert_true(dlist_empty(&txe->queued_pkts));
+
+	/* Test removal from outstanding_tx_pkts */
+	efa_rdm_ep_record_tx_op_completed(efa_rdm_ep, pkt_entry);
+	assert_int_equal(pkt_entry->flags & EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS, 0);
+	assert_int_equal(pkt_entry->flags & EFA_RDM_PKE_IN_OPE_QUEUED_PKTS, 0);
+	assert_true(dlist_empty(&peer->outstanding_tx_pkts));
+	assert_true(dlist_empty(&txe->queued_pkts));
+
+	/* Test insertion into queued_pkts */
+	efa_rdm_ep_queue_rnr_pkt(efa_rdm_ep, pkt_entry);
+	assert_int_equal(pkt_entry->flags & EFA_RDM_PKE_IN_PEER_OUTSTANDING_TX_PKTS, 0);
+	assert_int_not_equal(pkt_entry->flags & EFA_RDM_PKE_IN_OPE_QUEUED_PKTS, 0);
+	assert_true(dlist_empty(&peer->outstanding_tx_pkts));
+	assert_false(dlist_empty(&txe->queued_pkts));
+
+	/* Test that release_tx removes from queued_pkts */
+	efa_rdm_pke_release_tx(pkt_entry);
+	assert_true(dlist_empty(&peer->outstanding_tx_pkts));
+	assert_true(dlist_empty(&txe->queued_pkts));
+
+	/* Clean up */
+	efa_rdm_txe_release(txe);
+}
+
+
diff --git a/prov/efa/test/efa_unit_tests.c b/prov/efa/test/efa_unit_tests.c
@@ -322,6 +322,8 @@ int main(void)
 		cmocka_unit_test_setup_teardown(test_efa_rdm_pke_alloc_rtw_rxe, efa_unit_test_mocks_setup, efa_unit_test_mocks_teardown),
 		cmocka_unit_test_setup_teardown(test_efa_rdm_pke_alloc_rtr_rxe, efa_unit_test_mocks_setup, efa_unit_test_mocks_teardown),
 		cmocka_unit_test_setup_teardown(test_efa_rdm_pke_get_unexp, efa_unit_test_mocks_setup, efa_unit_test_mocks_teardown),
+		cmocka_unit_test_setup_teardown(test_efa_rdm_pke_flag_tracking, efa_unit_test_mocks_setup, efa_unit_test_mocks_teardown),
+		cmocka_unit_test_setup_teardown(test_efa_rdm_pke_flag_tracking, efa_unit_test_mocks_setup, efa_unit_test_mocks_teardown),
 		/* end of efa_unit_test_pke.c */
 
 		/* begin efa_unit_test_domain.c */
diff --git a/prov/efa/test/efa_unit_tests.h b/prov/efa/test/efa_unit_tests.h
@@ -320,6 +320,7 @@ void test_efa_rdm_pke_alloc_rta_rxe();
 void test_efa_rdm_pke_alloc_rtw_rxe();
 void test_efa_rdm_pke_alloc_rtr_rxe();
 void test_efa_rdm_pke_get_unexp();
+void test_efa_rdm_pke_flag_tracking();
 /* end of efa_unit_test_pke.c */
 
 void test_efa_msg_fi_recv();
