From b7609fe9165b720f71d456ac9b23f1d7e8dcb08c Mon Sep 17 00:00:00 2001 From: wehy <18174858434@163.com> Date: Tue, 20 Jan 2026 17:46:56 +0800 Subject: [PATCH 1/2] UCT/IB/UD: update stale dest ep_id and stale acks from before connection reset (#11092) * UCT/IB/UD: revise stale ep_id and stale acks from before connection reset * AUTHORS: update * UCT/IB/UD: unify codeline for readability * UCT/IB/UD: unify codeline for readability * TEST/GTEST/UCT/IB: added a test for checking dest ep id is being updated Signed-off-by: Roie Danino --------- Signed-off-by: Roie Danino Co-authored-by: Roie Danino (cherry picked from commit 0a4759cdb65abf5a1d9e689c03e0348af98c9699) --- AUTHORS | 1 + src/uct/ib/ud/base/ud_ep.c | 42 +++++++---- test/gtest/uct/ib/test_ud.cc | 133 +++++++++++++++++++++++++++++++++++ 3 files changed, 164 insertions(+), 12 deletions(-) diff --git a/AUTHORS b/AUTHORS index 9804584832a..a32d12a0ff7 100644 --- a/AUTHORS +++ b/AUTHORS @@ -115,6 +115,7 @@ Yuriy Shestakov Zhu Yanjun Zihao Zhao lzhang2 +michaelzli In addition we would like to acknowledge the following members of UCX community for their participation in annual face-to-face meeting, design discussions, and diff --git a/src/uct/ib/ud/base/ud_ep.c b/src/uct/ib/ud/base/ud_ep.c index 67615787719..c8503f49bfc 100644 --- a/src/uct/ib/ud/base/ud_ep.c +++ b/src/uct/ib/ud/base/ud_ep.c @@ -728,6 +728,15 @@ uct_ud_ep_process_ack(uct_ud_iface_t *iface, uct_ud_ep_t *ep, return; } + /* Ignore stale ACKs for unsent packets (e.g., after endpoint reset). + * Valid ACK PSN must be in (acked_psn, psn). */ + if (ucs_unlikely(UCT_UD_PSN_COMPARE(ep->tx.psn, <=, ack_psn))) { + ucs_debug("ep %p: ignoring invalid ack_psn=%u (tx.psn=%u acked_psn=%u)" + " - possibly stale from previous connection", + ep, ack_psn, ep->tx.psn, ep->tx.acked_psn); + return; + } + ep->tx.acked_psn = ack_psn; ucs_assertv(UCT_UD_PSN_COMPARE(ep->tx.acked_psn, <, ep->tx.psn), "ep %p: flags=0x%x acked_psn=%u must be smaller than" @@ -829,21 +838,30 @@ static void uct_ud_ep_rx_creq(uct_ud_iface_t *iface, uct_ud_neth_t *neth) ep->rx.ooo_pkts.head_sn = neth->psn; uct_ud_peer_copy(&ep->peer, ucs_unaligned_ptr(&ctl->peer)); uct_ud_ep_ctl_op_add(iface, ep, UCT_UD_EP_OP_CREP); - } else if (ep->dest_ep_id == UCT_UD_EP_NULL_ID) { - /* simultaneous CREQ */ + } else if (uct_ib_unpack_uint24(ctl->conn_req.ep_addr.ep_id) != ep->dest_ep_id) { + if (ep->dest_ep_id == UCT_UD_EP_NULL_ID) { + /* simultaneous CREQ */ + ucs_debug("simultaneous CREQ ep=%p" + "(iface=%p conn_sn=%d ep_id=%d, dest_ep_id=%d rx_psn=%u)", + ep, iface, ep->conn_sn, ep->ep_id, + ep->dest_ep_id, ep->rx.ooo_pkts.head_sn); + if (UCT_UD_PSN_COMPARE(ep->tx.psn, >, UCT_UD_INITIAL_PSN)) { + /* our own creq was sent, treat incoming creq as ack and remove our + * own from tx window + */ + uct_ud_ep_process_ack(iface, ep, UCT_UD_INITIAL_PSN, 0); + } + } else { + /* stale EP reuse */ + ucs_debug("iface=%p: detected stale EP reuse (ep=%p conn_sn=%d " + "old_dest_ep_id=%d new_ep_id=%d), updating dest_ep_id", + iface, ep, ep->conn_sn, ep->dest_ep_id, + uct_ib_unpack_uint24(ctl->conn_req.ep_addr.ep_id)); + } + /* Update dest_ep_id */ uct_ud_ep_set_dest_ep_id(ep, uct_ib_unpack_uint24(ctl->conn_req.ep_addr.ep_id)); ep->rx.ooo_pkts.head_sn = neth->psn; uct_ud_peer_copy(&ep->peer, ucs_unaligned_ptr(&ctl->peer)); - ucs_debug("simultaneous CREQ ep=%p" - "(iface=%p conn_sn=%d ep_id=%d, dest_ep_id=%d rx_psn=%u)", - ep, iface, ep->conn_sn, ep->ep_id, - ep->dest_ep_id, ep->rx.ooo_pkts.head_sn); - if (UCT_UD_PSN_COMPARE(ep->tx.psn, >, UCT_UD_INITIAL_PSN)) { - /* our own creq was sent, treat incoming creq as ack and remove our - * own from tx window - */ - uct_ud_ep_process_ack(iface, ep, UCT_UD_INITIAL_PSN, 0); - } uct_ud_ep_ctl_op_add(iface, ep, UCT_UD_EP_OP_CREP); } diff --git a/test/gtest/uct/ib/test_ud.cc b/test/gtest/uct/ib/test_ud.cc index b74252edcab..acdad330385 100644 --- a/test/gtest/uct/ib/test_ud.cc +++ b/test/gtest/uct/ib/test_ud.cc @@ -920,6 +920,139 @@ UCS_TEST_SKIP_COND_P(test_ud, ctls_loss, UCT_INSTANTIATE_UD_TEST_CASE(test_ud) +#if UCT_UD_EP_DEBUG_HOOKS +class test_ud_stale_ack : public test_ud { +public: + static ucs_status_t inject_stale_ack_psn(uct_ud_ep_t *ep, uct_ud_neth_t *neth) + { + if (m_stale_ack_psn_to_inject != 0) { + neth->ack_psn = m_stale_ack_psn_to_inject; + } + return UCS_OK; + } + + static ucs_status_t capture_rx_ack_psn(uct_ud_ep_t *ep, uct_ud_neth_t *neth) + { + m_received_ack_psn = neth->ack_psn; + return UCS_OK; + } + + static void set_stale_ack_psn(uct_ud_psn_t psn) { + m_stale_ack_psn_to_inject = psn; + } + + static uct_ud_psn_t get_stale_ack_psn() { + return m_stale_ack_psn_to_inject; + } + + static uct_ud_psn_t get_received_ack_psn() { + return m_received_ack_psn; + } + +private: + /* Stale ACK PSN to inject - simulates ACK from before endpoint reset */ + static volatile uct_ud_psn_t m_stale_ack_psn_to_inject; + /* Captured ACK PSN from received packet */ + static volatile uct_ud_psn_t m_received_ack_psn; +}; + +volatile uct_ud_psn_t test_ud_stale_ack::m_stale_ack_psn_to_inject = 0; +volatile uct_ud_psn_t test_ud_stale_ack::m_received_ack_psn = 0; + +UCS_TEST_SKIP_COND_P(test_ud_stale_ack, stale_ack_after_reset, + !check_caps(UCT_IFACE_FLAG_AM_SHORT)) { + constexpr uct_ud_psn_t STALE_ACK_PSN = 135; + constexpr uct_ud_psn_t WINDOW_SIZE = 1024; + + disable_async(m_e1); + disable_async(m_e2); + connect(); + set_tx_win(m_e1, WINDOW_SIZE); + set_tx_win(m_e2, WINDOW_SIZE); + + uct_ud_ep_t *ud_ep1 = ep(m_e1); + + /* Send some data to advance PSN */ + for (int i = 0; i < 5; i++) { + EXPECT_UCS_OK(tx(m_e1)); + } + flush(); + + /* Simulate endpoint reset on m_e1: + * Reset PSN to low values as if endpoint was just created. */ + uct_ud_enter(iface(m_e1)); + ud_ep1->tx.psn = 3; /* Next PSN to send */ + ud_ep1->tx.acked_psn = 0; /* Last ACKed PSN */ + ucs_queue_head_init(&ud_ep1->tx.window); /* Clear TX window */ + uct_ud_leave(iface(m_e1)); + + /* Set up TX hook on m_e2 to inject stale ack_psn in outgoing packets. + * This simulates a delayed/stale packet arriving after reset. */ + set_stale_ack_psn(STALE_ACK_PSN); + ep(m_e2)->tx.tx_hook = test_ud_stale_ack::inject_stale_ack_psn; + ep(m_e1)->rx.rx_hook = test_ud_stale_ack::capture_rx_ack_psn; + + /* m_e2 sends a packet. Due to the hook, it will contain ack_psn=135. */ + EXPECT_UCS_OK(tx(m_e2)); + short_progress_loop(); + + /* Verify the hook actually injected the stale ack_psn */ + EXPECT_EQ(STALE_ACK_PSN, get_received_ack_psn()) + << "Invalid value after the hook"; + + set_stale_ack_psn(0); + ep(m_e2)->tx.tx_hook = uct_ud_ep_null_hook; + ep(m_e1)->rx.rx_hook = uct_ud_ep_null_hook; + + /* Verify endpoint state wasn't corrupted by the stale ACK */ + EXPECT_EQ(0, ud_ep1->tx.acked_psn); /* Should NOT be STALE_ACK_PSN */ +} + +UCT_INSTANTIATE_UD_TEST_CASE(test_ud_stale_ack) + +/* Test that simulates a stale EP reuse, when an EP with an existing dest_ep_id receives a CREQ with + * a different ep_id, the dest_ep_id is updated. */ +UCS_TEST_SKIP_COND_P(test_ud, stale_dest_ep_id_update, + !check_caps(UCT_IFACE_FLAG_AM_SHORT)) { + constexpr uint32_t STALE_DEST_EP_ID = 0xBEEF; + constexpr uint32_t REMOTE_EP_ID = 1; + + /* Create a dummy EP on m_e1 first, so the actual EP will have ep_id=1 + * (not 0, which could be confused with NULL/default values) */ + m_e1->create_ep(0); + + /* Start connection from m_e1 to m_e2 - block m_e2's TX to delay CREP */ + iface(m_e2)->tx.available = 0; + + m_e1->connect_to_iface(1, *m_e2); + + /* Let CREQ be received and passive EP created on m_e2 */ + short_progress_loop(); + + /* m_e2 side: connect back, which will reuse the passive EP */ + m_e2->connect_to_iface(0, *m_e1); + + /* Now manually set a "stale" dest_ep_id on m_e2's EP to simulate + * the scenario where the peer had a different ep_id before reset */ + ep(m_e2)->dest_ep_id = STALE_DEST_EP_ID; + + /* Allow m_e2 to send - CREP will be sent, and any pending CREQs processed */ + iface(m_e2)->tx.available = 128; + + /* Wait for m_e2's dest_ep_id to be updated from the stale value */ + wait_for_value(&ep(m_e2)->dest_ep_id, REMOTE_EP_ID, true, + TEST_UD_LINGER_TIMEOUT_IN_SEC); + + EXPECT_EQ(REMOTE_EP_ID, ep(m_e1, REMOTE_EP_ID)->ep_id); + EXPECT_NE(UCT_UD_EP_NULL_ID, ep(m_e1, REMOTE_EP_ID)->dest_ep_id); + EXPECT_NE(STALE_DEST_EP_ID, ep(m_e2)->dest_ep_id) + << "dest_ep_id should have been updated from stale value"; + EXPECT_EQ(ep(m_e1, REMOTE_EP_ID)->ep_id, ep(m_e2)->dest_ep_id) + << "dest_ep_id should match the remote EP's actual ep_id"; + EXPECT_EQ(REMOTE_EP_ID, ep(m_e2)->dest_ep_id) + << "dest_ep_id should be 1 (not 0)"; +} +#endif class test_ud_peer_failure : public ud_base_test { public: From dc54d495b12c01f1233ec1b6adcb66a0a91ee9cf Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Thu, 22 Jan 2026 10:38:31 +0200 Subject: [PATCH 2/2] AUTHORS: added wehy Signed-off-by: Roie Danino --- AUTHORS | 1 + 1 file changed, 1 insertion(+) diff --git a/AUTHORS b/AUTHORS index a32d12a0ff7..60022b45073 100644 --- a/AUTHORS +++ b/AUTHORS @@ -116,6 +116,7 @@ Zhu Yanjun Zihao Zhao lzhang2 michaelzli +wehy <18174858434@163.com> In addition we would like to acknowledge the following members of UCX community for their participation in annual face-to-face meeting, design discussions, and