From f44378aeaca4f9075dde05b6ac08d042e5453e46 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Wed, 26 Aug 2026 14:14:57 -0700 Subject: [PATCH 01/15] UCT/RC: Preserve pending order for multi-credit SGL puts --- src/uct/ib/mlx5/rc/rc_mlx5_ep.c | 1 + src/uct/ib/rc/base/rc_ep.c | 20 +++--- src/uct/ib/rc/base/rc_ep.h | 25 ++++++- test/gtest/ucp/test_ucp_rma.cc | 115 +++++++++++++++++++++++++++++++- 4 files changed, 150 insertions(+), 11 deletions(-) diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_ep.c b/src/uct/ib/mlx5/rc/rc_mlx5_ep.c index 95e48a643ba..75f0380b5bc 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_ep.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_ep.c @@ -243,6 +243,7 @@ uct_rc_mlx5_base_ep_put_sgl_zcopy(uct_ep_h tl_ep, void * const *buffers, "put_sgl_zcopy"); } + UCT_RC_CHECK_PENDING(&ep->super); UCT_RC_CHECK_CQE_VALUE_RET(&iface->super, &ep->super, UCS_ERR_NO_RESOURCE, count - 1); diff --git a/src/uct/ib/rc/base/rc_ep.c b/src/uct/ib/rc/base/rc_ep.c index 36177b3b6a7..8061acf39e4 100644 --- a/src/uct/ib/rc/base/rc_ep.c +++ b/src/uct/ib/rc/base/rc_ep.c @@ -147,10 +147,10 @@ UCS_CLASS_INIT_FUNC(uct_rc_ep_t, uct_rc_iface_t *iface, uint32_t qp_num, return status; } - self->path_index = UCT_EP_PARAMS_GET_PATH_INDEX(params); - self->flags = 0; - self->txqp_reserve = 0; - self->cq_reserve = 0; + self->path_index = UCT_EP_PARAMS_GET_PATH_INDEX(params); + self->flags = 0; + self->in_pending = 0; + uct_rc_ep_reset_reserve(self); status = uct_rc_fc_init(&self->fc, iface UCS_STATS_ARG(self->super.stats)); if (status != UCS_OK) { @@ -316,7 +316,8 @@ ucs_status_t uct_rc_ep_pending_add(uct_ep_h tl_ep, uct_pending_req_t *n, uct_rc_iface_t *iface = ucs_derived_of(tl_ep->iface, uct_rc_iface_t); uct_rc_ep_t *ep = ucs_derived_of(tl_ep, uct_rc_ep_t); - if (uct_rc_ep_have_tx_cqe_avail(ep) && + if (!UCT_RC_EP_MUST_BLOCK_SEND(ep) && + uct_rc_ep_have_tx_cqe_avail(ep) && uct_rc_ep_has_tx_resources(ep) && uct_rc_iface_has_tx_resources(iface)) { return UCS_ERR_BUSY; @@ -345,10 +346,11 @@ ucs_arbiter_cb_result_t uct_rc_ep_process_pending(ucs_arbiter_t *arbiter, uct_rc_iface_t *iface = ucs_derived_of(ep->super.super.iface, uct_rc_iface_t); ucs_status_t status; - status = uct_rc_iface_invoke_pending_cb(iface, req); + ep->in_pending = 1; + status = uct_rc_iface_invoke_pending_cb(iface, req); + ep->in_pending = 0; if (status == UCS_OK) { - ep->txqp_reserve = 0; - ep->cq_reserve = 0; + uct_rc_ep_reset_reserve(ep); return UCS_ARBITER_CB_RESULT_REMOVE_ELEM; } else if (status == UCS_INPROGRESS) { return UCS_ARBITER_CB_RESULT_NEXT_GROUP; @@ -434,6 +436,8 @@ void uct_rc_ep_pending_purge(uct_ep_h tl_ep, uct_pending_purge_callback_t cb, ucs_arbiter_group_purge(&iface->tx.arbiter, &ep->arb_group, uct_rc_ep_arbiter_purge_cb, &args); + + uct_rc_ep_reset_reserve(ep); } ucs_status_t uct_rc_ep_fc_grant(uct_pending_req_t *self) diff --git a/src/uct/ib/rc/base/rc_ep.h b/src/uct/ib/rc/base/rc_ep.h index ea3f12b7ce0..52ed1126cc5 100644 --- a/src/uct/ib/rc/base/rc_ep.h +++ b/src/uct/ib/rc/base/rc_ep.h @@ -90,10 +90,25 @@ enum { #define UCT_RC_DEFINE_ATOMIC_HANDLER_FUNC_NAME(_num_bits, _is_be) \ uct_rc_ep_atomic_handler_##_num_bits##_be##_is_be +#define UCT_RC_EP_MUST_BLOCK_SEND(_ep) \ + (!(_ep)->in_pending && \ + (!ucs_arbiter_group_is_empty(&(_ep)->arb_group) || \ + (((_ep)->cq_reserve | (_ep)->txqp_reserve) != 0))) + +#define UCT_RC_CHECK_PENDING_RET(_ep, _ret) \ + if (ucs_unlikely(UCT_RC_EP_MUST_BLOCK_SEND(_ep))) { \ + UCS_STATS_UPDATE_COUNTER((_ep)->super.stats, UCT_EP_STAT_NO_RES, 1); \ + return _ret; \ + } + +#define UCT_RC_CHECK_PENDING(_ep) \ + UCT_RC_CHECK_PENDING_RET(_ep, UCS_ERR_NO_RESOURCE) + /* * Check for send resources */ #define UCT_RC_CHECK_CQE_RET(_iface, _ep, _ret) \ + UCT_RC_CHECK_PENDING_RET(_ep, _ret) \ /* tx_moderation == 0 for TLs which don't support it */ \ if (ucs_unlikely((_iface)->tx.cq_available <= \ (signed)(_iface)->config.tx_moderation)) { \ @@ -201,8 +216,7 @@ enum { #define UCT_RC_CHECK_RES_AND_FC(_iface, _ep, _am_id) \ UCT_RC_CHECK_RES(_iface, _ep) \ - UCT_RC_CHECK_FC(_iface, _ep, _am_id) \ - uct_rc_iface_check_pending(_iface, &(_ep)->arb_group); + UCT_RC_CHECK_FC(_iface, _ep, _am_id) /* this is a common type for all rc and dc transports */ struct uct_rc_txqp { @@ -233,6 +247,7 @@ struct uct_rc_ep { uint16_t cq_reserve; uint8_t path_index; uint8_t flags; + uint8_t in_pending; }; @@ -354,6 +369,12 @@ static UCS_F_ALWAYS_INLINE void uct_rc_fc_restore_wnd(uct_rc_iface_t *iface, UCS_STATS_SET_COUNTER(fc->stats, UCT_RC_FC_STAT_FC_WND, fc->fc_wnd); } +static UCS_F_ALWAYS_INLINE void uct_rc_ep_reset_reserve(uct_rc_ep_t *ep) +{ + ep->txqp_reserve = 0; + ep->cq_reserve = 0; +} + static UCS_F_ALWAYS_INLINE int uct_rc_ep_has_tx_resources(uct_rc_ep_t *ep) { uct_rc_iface_t *iface = ucs_derived_of(ep->super.super.iface, uct_rc_iface_t); diff --git a/test/gtest/ucp/test_ucp_rma.cc b/test/gtest/ucp/test_ucp_rma.cc index ab2a233491b..93be5f7cff3 100755 --- a/test/gtest/ucp/test_ucp_rma.cc +++ b/test/gtest/ucp/test_ucp_rma.cc @@ -7,6 +7,9 @@ #include "test_ucp_memheap.h" +#include +#include + extern "C" { #include #include /* for UCP_MEM_IS_ACCESSIBLE_FROM_CPU */ @@ -1003,7 +1006,8 @@ UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_ep_based_fence, all, "all") class test_ucp_rma_sgl : public test_ucp_rma { public: static void get_base_variants(std::vector& variants) { - add_variant(variants, UCP_FEATURE_RMA); + add_variant(variants, UCP_FEATURE_RMA | UCP_FEATURE_AM, + MULTI_THREAD_WORKER); } static void get_test_variants(std::vector& variants) { @@ -1658,4 +1662,113 @@ UCS_TEST_SKIP_COND_P(test_ucp_rma_sgl, put_without_proto, REMOTE_MASK_DEFAULT, 2, UCS_ERR_UNSUPPORTED); } +UCS_TEST_P(test_ucp_rma_sgl, put_am_pending) +{ + /* size, count, repeat, num_threads */ + static const size_t cases[][4] = { + {4096, 8, 3, 1}, + {32768, 64, 3, 2}, + {1000000, 100, 3, 4}, + {40, 1000, 1, 4}, + {40, 1000, 1, 8} + }; + static const char am_msg[] = "am"; + static const unsigned am_id = 1; + + auto am_cb = [](void *, const void *, size_t, void *, size_t, + const ucp_am_recv_param_t *) -> ucs_status_t { + return UCS_OK; + }; + ucp_am_handler_param_t am_handler = {}; + am_handler.field_mask = UCP_AM_HANDLER_PARAM_FIELD_ID | + UCP_AM_HANDLER_PARAM_FIELD_CB; + am_handler.id = am_id; + am_handler.cb = am_cb; + ASSERT_UCS_OK(ucp_worker_set_am_recv_handler(receiver().worker(), + &am_handler)); + + std::atomic stop_progress{false}; + std::thread progress_thread([&]() { + while (!stop_progress.load(std::memory_order_relaxed)) { + ucp_worker_progress(sender().worker()); + ucp_worker_progress(receiver().worker()); + } + }); + + auto stop_progress_thread = [&]() { + stop_progress = true; + progress_thread.join(); + }; + + auto wait_req = [&](void *req) { + if (req == NULL) { + return; + } + + if (!UCS_PTR_IS_PTR(req)) { + EXPECT_UCS_OK(UCS_PTR_STATUS(req)); + return; + } + + while (ucp_request_check_status(req) == UCS_INPROGRESS) { + ucp_worker_progress(sender().worker()); + ucp_worker_progress(receiver().worker()); + } + + EXPECT_UCS_OK(ucp_request_check_status(req)); + ucp_request_free(req); + }; + + for (size_t c = 0; c < ucs_static_array_size(cases); c++) { + const size_t size = cases[c][0]; + const size_t count = cases[c][1]; + const size_t repeat = cases[c][2]; + const size_t num_threads = cases[c][3]; + + sgl_ctx ctx; + init_sgl_ctx(ctx, count, size); + + ucp_dt_local_sgl_t local = make_local_sgl( + ctx, LOCAL_MASK_DEFAULT | UCP_DT_LOCAL_SGL_FIELD_MEMHS); + ucp_dt_remote_sgl_t remote = make_remote_sgl(ctx, REMOTE_MASK_DEFAULT); + ucp_request_param_t param = make_sgl_param(&remote, count); + + for (size_t r = 0; r < repeat; r++) { + std::vector threads; + threads.reserve(num_threads); + + for (size_t t = 0; t < num_threads; t++) { + threads.emplace_back([&]() { + void *put_req = sgl_op_nbx(SGL_OP_PUT, &local, count, + UCP_REMOTE_ADDR_INVALID, + UCP_RKEY_INVALID, ¶m); + void *flush_req = sender().flush_ep_nb(); + + ucp_request_param_t am_param = {}; + am_param.op_attr_mask = UCP_OP_ATTR_FIELD_FLAGS; + am_param.flags = UCP_AM_SEND_FLAG_EAGER; + void *am_req = ucp_am_send_nbx(sender().ep(), am_id, NULL, + 0, am_msg, sizeof(am_msg), + &am_param); + + wait_req(put_req); + wait_req(flush_req); + wait_req(am_req); + }); + } + + for (auto &thread : threads) { + thread.join(); + } + + if (HasFailure() || (num_errors() > 0)) { + stop_progress_thread(); + return; + } + } + } + + stop_progress_thread(); +} + UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_rma_sgl, all, "all") From dc05f89d3b326076893648e65ce0ead9a9e0395c Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Wed, 26 Aug 2026 14:42:03 -0700 Subject: [PATCH 02/15] GTEST/UCP: Scope SGL AM+MT worker to the put-AM pending test --- test/gtest/ucp/test_ucp_rma.cc | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/test/gtest/ucp/test_ucp_rma.cc b/test/gtest/ucp/test_ucp_rma.cc index 93be5f7cff3..64b84f7b3a2 100755 --- a/test/gtest/ucp/test_ucp_rma.cc +++ b/test/gtest/ucp/test_ucp_rma.cc @@ -1006,8 +1006,7 @@ UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_ep_based_fence, all, "all") class test_ucp_rma_sgl : public test_ucp_rma { public: static void get_base_variants(std::vector& variants) { - add_variant(variants, UCP_FEATURE_RMA | UCP_FEATURE_AM, - MULTI_THREAD_WORKER); + add_variant(variants, UCP_FEATURE_RMA); } static void get_test_variants(std::vector& variants) { @@ -1662,7 +1661,21 @@ UCS_TEST_SKIP_COND_P(test_ucp_rma_sgl, put_without_proto, REMOTE_MASK_DEFAULT, 2, UCS_ERR_UNSUPPORTED); } -UCS_TEST_P(test_ucp_rma_sgl, put_am_pending) +class test_ucp_rma_sgl_am_pending : public test_ucp_rma_sgl { +public: + static void get_base_variants(std::vector& variants) { + add_variant(variants, UCP_FEATURE_RMA | UCP_FEATURE_AM, + MULTI_THREAD_WORKER); + } + + static void get_test_variants(std::vector& variants) { + add_variant_memtypes(variants, get_base_variants, + UCS_BIT(UCS_MEMORY_TYPE_CUDA) | + UCS_BIT(UCS_MEMORY_TYPE_HOST)); + } +}; + +UCS_TEST_P(test_ucp_rma_sgl_am_pending, put) { /* size, count, repeat, num_threads */ static const size_t cases[][4] = { @@ -1772,3 +1785,4 @@ UCS_TEST_P(test_ucp_rma_sgl, put_am_pending) } UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_rma_sgl, all, "all") +UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_rma_sgl_am_pending, all, "all") From c5fcd2019754ae004993ffacb85952683534ad4c Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Wed, 26 Aug 2026 18:01:13 -0700 Subject: [PATCH 03/15] GTEST: Replace UCP SGL pending test with UCT RC test --- test/gtest/ucp/test_ucp_rma.cc | 127 --------------------------------- test/gtest/uct/ib/test_rc.cc | 62 +++++++++++++++- 2 files changed, 61 insertions(+), 128 deletions(-) diff --git a/test/gtest/ucp/test_ucp_rma.cc b/test/gtest/ucp/test_ucp_rma.cc index 64b84f7b3a2..ab2a233491b 100755 --- a/test/gtest/ucp/test_ucp_rma.cc +++ b/test/gtest/ucp/test_ucp_rma.cc @@ -7,9 +7,6 @@ #include "test_ucp_memheap.h" -#include -#include - extern "C" { #include #include /* for UCP_MEM_IS_ACCESSIBLE_FROM_CPU */ @@ -1661,128 +1658,4 @@ UCS_TEST_SKIP_COND_P(test_ucp_rma_sgl, put_without_proto, REMOTE_MASK_DEFAULT, 2, UCS_ERR_UNSUPPORTED); } -class test_ucp_rma_sgl_am_pending : public test_ucp_rma_sgl { -public: - static void get_base_variants(std::vector& variants) { - add_variant(variants, UCP_FEATURE_RMA | UCP_FEATURE_AM, - MULTI_THREAD_WORKER); - } - - static void get_test_variants(std::vector& variants) { - add_variant_memtypes(variants, get_base_variants, - UCS_BIT(UCS_MEMORY_TYPE_CUDA) | - UCS_BIT(UCS_MEMORY_TYPE_HOST)); - } -}; - -UCS_TEST_P(test_ucp_rma_sgl_am_pending, put) -{ - /* size, count, repeat, num_threads */ - static const size_t cases[][4] = { - {4096, 8, 3, 1}, - {32768, 64, 3, 2}, - {1000000, 100, 3, 4}, - {40, 1000, 1, 4}, - {40, 1000, 1, 8} - }; - static const char am_msg[] = "am"; - static const unsigned am_id = 1; - - auto am_cb = [](void *, const void *, size_t, void *, size_t, - const ucp_am_recv_param_t *) -> ucs_status_t { - return UCS_OK; - }; - ucp_am_handler_param_t am_handler = {}; - am_handler.field_mask = UCP_AM_HANDLER_PARAM_FIELD_ID | - UCP_AM_HANDLER_PARAM_FIELD_CB; - am_handler.id = am_id; - am_handler.cb = am_cb; - ASSERT_UCS_OK(ucp_worker_set_am_recv_handler(receiver().worker(), - &am_handler)); - - std::atomic stop_progress{false}; - std::thread progress_thread([&]() { - while (!stop_progress.load(std::memory_order_relaxed)) { - ucp_worker_progress(sender().worker()); - ucp_worker_progress(receiver().worker()); - } - }); - - auto stop_progress_thread = [&]() { - stop_progress = true; - progress_thread.join(); - }; - - auto wait_req = [&](void *req) { - if (req == NULL) { - return; - } - - if (!UCS_PTR_IS_PTR(req)) { - EXPECT_UCS_OK(UCS_PTR_STATUS(req)); - return; - } - - while (ucp_request_check_status(req) == UCS_INPROGRESS) { - ucp_worker_progress(sender().worker()); - ucp_worker_progress(receiver().worker()); - } - - EXPECT_UCS_OK(ucp_request_check_status(req)); - ucp_request_free(req); - }; - - for (size_t c = 0; c < ucs_static_array_size(cases); c++) { - const size_t size = cases[c][0]; - const size_t count = cases[c][1]; - const size_t repeat = cases[c][2]; - const size_t num_threads = cases[c][3]; - - sgl_ctx ctx; - init_sgl_ctx(ctx, count, size); - - ucp_dt_local_sgl_t local = make_local_sgl( - ctx, LOCAL_MASK_DEFAULT | UCP_DT_LOCAL_SGL_FIELD_MEMHS); - ucp_dt_remote_sgl_t remote = make_remote_sgl(ctx, REMOTE_MASK_DEFAULT); - ucp_request_param_t param = make_sgl_param(&remote, count); - - for (size_t r = 0; r < repeat; r++) { - std::vector threads; - threads.reserve(num_threads); - - for (size_t t = 0; t < num_threads; t++) { - threads.emplace_back([&]() { - void *put_req = sgl_op_nbx(SGL_OP_PUT, &local, count, - UCP_REMOTE_ADDR_INVALID, - UCP_RKEY_INVALID, ¶m); - void *flush_req = sender().flush_ep_nb(); - - ucp_request_param_t am_param = {}; - am_param.op_attr_mask = UCP_OP_ATTR_FIELD_FLAGS; - am_param.flags = UCP_AM_SEND_FLAG_EAGER; - void *am_req = ucp_am_send_nbx(sender().ep(), am_id, NULL, - 0, am_msg, sizeof(am_msg), - &am_param); - - wait_req(put_req); - wait_req(flush_req); - wait_req(am_req); - }); - } - - for (auto &thread : threads) { - thread.join(); - } - - if (HasFailure() || (num_errors() > 0)) { - stop_progress_thread(); - return; - } - } - } - - stop_progress_thread(); -} - UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_rma_sgl, all, "all") -UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_rma_sgl_am_pending, all, "all") diff --git a/test/gtest/uct/ib/test_rc.cc b/test/gtest/uct/ib/test_rc.cc index e9946df00c3..f26c7151cb6 100644 --- a/test/gtest/uct/ib/test_rc.cc +++ b/test/gtest/uct/ib/test_rc.cc @@ -1,5 +1,5 @@ /** -* Copyright (c) NVIDIA CORPORATION & AFFILIATES, 2001-2019. ALL RIGHTS RESERVED. +* Copyright (c) NVIDIA CORPORATION & AFFILIATES, 2001-2026. ALL RIGHTS RESERVED. * Copyright (C) UT-Battelle, LLC. 2016. ALL RIGHTS RESERVED. * Copyright (C) ARM Ltd. 2016.All rights reserved. * See file LICENSE for terms. @@ -9,6 +9,10 @@ #include #include +extern "C" { +#include +} + #ifdef HAVE_MLX5_DV extern "C" { #include @@ -142,6 +146,62 @@ UCS_TEST_P(test_rc, fence_am_short_consumed, "RC_FENCE=weak") EXPECT_EQ(rc_iface(m_e1)->tx.fi.fence_beat, fence_info->fence_beat); } +UCS_TEST_SKIP_COND_P(test_rc, sgl_pending_am_order, + !check_caps(UCT_IFACE_FLAG_AM_SHORT | + UCT_IFACE_FLAG_PENDING | + UCT_IFACE_FLAG_PUT_SHORT)) +{ + uct_iface_attr_v2_t attr = {}; + const size_t length = 64; + const size_t count = 2; + unsigned n; + ucs_status_t status; + uct_pending_req_t pend_req; + + attr.field_mask = UCT_IFACE_ATTR_FIELD_CAP_FLAGS | + UCT_IFACE_ATTR_FIELD_MAX_PUT_SGL_ZCOPY_COUNT; + ASSERT_UCS_OK(uct_iface_query_v2(m_e1->iface(), &attr)); + if (!(attr.cap.flags & UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY) || + (attr.max_put_sgl_zcopy_count < count)) { + UCS_TEST_SKIP_R("put_sgl_zcopy is not supported"); + } + + mapped_buffer s0(length, 0ul, *m_e1), s1(length, 0ul, *m_e1); + mapped_buffer r0(length, 0ul, *m_e2), r1(length, 0ul, *m_e2); + void *buffers[] = {s0.ptr(), s1.ptr()}; + size_t lengths[] = {length, length}; + uct_mem_h memhs[] = {s0.memh(), s1.memh()}; + uint64_t raddrs[] = {r0.addr(), r1.addr()}; + uct_rkey_t rkeys[] = {r0.rkey(), r1.rkey()}; + + n = 0; + while (uct_rc_txqp_available(&rc_ep(m_e1)->txqp) > (int16_t)(count - 1)) { + ASSERT_LT(n++, 1024u); + ASSERT_UCS_OK(uct_ep_put_short(m_e1->ep(0), s0.ptr(), 8, r0.addr(), + r0.rkey())); + } + + ASSERT_GT(uct_rc_txqp_available(&rc_ep(m_e1)->txqp), 0); + ASSERT_LT(uct_rc_txqp_available(&rc_ep(m_e1)->txqp), (int16_t)count); + ASSERT_GT(rc_iface(m_e1)->tx.cq_available, (int16_t)(count - 1)); + + status = uct_ep_put_sgl_zcopy(m_e1->ep(0), buffers, lengths, memhs, raddrs, + rkeys, NULL, NULL, count, NULL); + ASSERT_EQ(UCS_ERR_NO_RESOURCE, status); + + pend_req.func = [](uct_pending_req_t*) -> ucs_status_t { + return UCS_ERR_NO_RESOURCE; + }; + + ASSERT_UCS_OK(uct_ep_pending_add(m_e1->ep(0), &pend_req, 0)); + + EXPECT_EQ(UCS_ERR_NO_RESOURCE, + uct_ep_am_short(m_e1->ep(0), 0, 0, NULL, 0)); + + uct_ep_pending_purge(m_e1->ep(0), NULL, NULL); + flush(); +} + UCT_INSTANTIATE_RC_TEST_CASE(test_rc) From fdbc016689fcc25349f78faa9e7cd64ba531e332 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Wed, 26 Aug 2026 18:07:08 -0700 Subject: [PATCH 04/15] GTEST/UCT/RC: Cover pending check in put_sgl_zcopy --- test/gtest/uct/ib/test_rc.cc | 3 +++ 1 file changed, 3 insertions(+) diff --git a/test/gtest/uct/ib/test_rc.cc b/test/gtest/uct/ib/test_rc.cc index f26c7151cb6..3a942673a89 100644 --- a/test/gtest/uct/ib/test_rc.cc +++ b/test/gtest/uct/ib/test_rc.cc @@ -195,6 +195,9 @@ UCS_TEST_SKIP_COND_P(test_rc, sgl_pending_am_order, ASSERT_UCS_OK(uct_ep_pending_add(m_e1->ep(0), &pend_req, 0)); + EXPECT_EQ(UCS_ERR_NO_RESOURCE, + uct_ep_put_sgl_zcopy(m_e1->ep(0), buffers, lengths, memhs, + raddrs, rkeys, NULL, NULL, count, NULL)); EXPECT_EQ(UCS_ERR_NO_RESOURCE, uct_ep_am_short(m_e1->ep(0), 0, 0, NULL, 0)); From b8b396412333906aec7d206cf066de8c9a870f60 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Wed, 26 Aug 2026 18:26:29 -0700 Subject: [PATCH 05/15] UCT/RC: Block sends only when pending is queued --- src/uct/ib/rc/base/rc_ep.h | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/uct/ib/rc/base/rc_ep.h b/src/uct/ib/rc/base/rc_ep.h index 52ed1126cc5..25640c789ea 100644 --- a/src/uct/ib/rc/base/rc_ep.h +++ b/src/uct/ib/rc/base/rc_ep.h @@ -91,9 +91,7 @@ enum { uct_rc_ep_atomic_handler_##_num_bits##_be##_is_be #define UCT_RC_EP_MUST_BLOCK_SEND(_ep) \ - (!(_ep)->in_pending && \ - (!ucs_arbiter_group_is_empty(&(_ep)->arb_group) || \ - (((_ep)->cq_reserve | (_ep)->txqp_reserve) != 0))) + (!(_ep)->in_pending && !ucs_arbiter_group_is_empty(&(_ep)->arb_group)) #define UCT_RC_CHECK_PENDING_RET(_ep, _ret) \ if (ucs_unlikely(UCT_RC_EP_MUST_BLOCK_SEND(_ep))) { \ From 5b20bdae54ce2422e94187a8030b0932008f20d1 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Wed, 26 Aug 2026 20:21:35 -0700 Subject: [PATCH 06/15] GTEST/UCT/RC: Dispatch pending to send FC grant --- test/gtest/uct/ib/test_rc.cc | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/test/gtest/uct/ib/test_rc.cc b/test/gtest/uct/ib/test_rc.cc index 3a942673a89..aa32e8ac0ba 100644 --- a/test/gtest/uct/ib/test_rc.cc +++ b/test/gtest/uct/ib/test_rc.cc @@ -951,12 +951,11 @@ void test_rc_flow_control::test_pending_grant(int16_t wnd) wait_fc_hard_resend(m_e1); - /* Enable send capabilities of m_e2 and send short put message to force - * pending queue dispatch. Can't send AM message for that, because it may - * trigger reordering assert due to disable/enable entity hack. */ + /* Enable send capabilities of m_e2 and dispatch the pending queue. + * Can't send AM/PUT for that, because it is blocked while FC grant is + * queued. */ enable_entity(m_e2); - set_tx_moderation(m_e2, 0); - EXPECT_EQ(UCS_OK, uct_ep_put_short(m_e2->ep(0), NULL, 0, 0, 0)); + uct_rc_iface_arbiter_dispatch(rc_iface(m_e2)); /* Check that m_e1 got grant */ validate_grant(m_e1); From 2f34f631644d1950c7bb28b9a4220bf87b5a5f00 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Thu, 27 Aug 2026 02:02:03 -0700 Subject: [PATCH 07/15] GTEST/UCT: Drop redundant uct_v2.h include from test_rc --- test/gtest/uct/ib/test_rc.cc | 4 ---- 1 file changed, 4 deletions(-) diff --git a/test/gtest/uct/ib/test_rc.cc b/test/gtest/uct/ib/test_rc.cc index aa32e8ac0ba..9ea33b788a3 100644 --- a/test/gtest/uct/ib/test_rc.cc +++ b/test/gtest/uct/ib/test_rc.cc @@ -9,10 +9,6 @@ #include #include -extern "C" { -#include -} - #ifdef HAVE_MLX5_DV extern "C" { #include From 435a117002db8dced151b172c5a9a1fd56c10166 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Fri, 4 Sep 2026 09:06:22 -0700 Subject: [PATCH 08/15] UCT/RC: Reserve TX credits instead of blocking sends on pending --- src/uct/ib/mlx5/rc/rc_mlx5.inl | 2 +- src/uct/ib/mlx5/rc/rc_mlx5_ep.c | 1 - src/uct/ib/rc/base/rc_ep.c | 9 +--- src/uct/ib/rc/base/rc_ep.h | 64 ++++++++++++++++++---------- src/uct/ib/rc/verbs/rc_verbs_iface.c | 3 +- 5 files changed, 46 insertions(+), 33 deletions(-) diff --git a/src/uct/ib/mlx5/rc/rc_mlx5.inl b/src/uct/ib/mlx5/rc/rc_mlx5.inl index f28d4c9865c..431fb4d1f5f 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5.inl +++ b/src/uct/ib/mlx5/rc/rc_mlx5.inl @@ -1992,7 +1992,7 @@ uct_rc_mlx5_iface_poll_tx(uct_rc_mlx5_iface_common_t *iface, int poll_flags) ep, qp_num, hw_ci); uct_rc_mlx5_txqp_process_tx_cqe(&ep->super.txqp, cqe, hw_ci); - ucs_arbiter_group_schedule(&iface->super.tx.arbiter, &ep->super.arb_group); + uct_rc_ep_schedule_pending(&iface->super, &ep->super); uct_rc_mlx5_ep_update_tx_qp_res(ep, hw_ci); uct_rc_mlx5_iface_update_tx_cq_res(&iface->super, ep, hw_ci); uct_rc_iface_arbiter_dispatch(&iface->super); diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_ep.c b/src/uct/ib/mlx5/rc/rc_mlx5_ep.c index 7fc7cd92d3c..3effef7bd12 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_ep.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_ep.c @@ -275,7 +275,6 @@ uct_rc_mlx5_base_ep_put_sgl_zcopy(uct_ep_h tl_ep, void * const *buffers, "put_sgl_zcopy"); } - UCT_RC_CHECK_PENDING(&ep->super); UCT_RC_CHECK_CQE_VALUE_RET(&iface->super, &ep->super, UCS_ERR_NO_RESOURCE, count - 1); diff --git a/src/uct/ib/rc/base/rc_ep.c b/src/uct/ib/rc/base/rc_ep.c index 8061acf39e4..d2af80708be 100644 --- a/src/uct/ib/rc/base/rc_ep.c +++ b/src/uct/ib/rc/base/rc_ep.c @@ -149,7 +149,6 @@ UCS_CLASS_INIT_FUNC(uct_rc_ep_t, uct_rc_iface_t *iface, uint32_t qp_num, self->path_index = UCT_EP_PARAMS_GET_PATH_INDEX(params); self->flags = 0; - self->in_pending = 0; uct_rc_ep_reset_reserve(self); status = uct_rc_fc_init(&self->fc, iface UCS_STATS_ARG(self->super.stats)); @@ -316,8 +315,7 @@ ucs_status_t uct_rc_ep_pending_add(uct_ep_h tl_ep, uct_pending_req_t *n, uct_rc_iface_t *iface = ucs_derived_of(tl_ep->iface, uct_rc_iface_t); uct_rc_ep_t *ep = ucs_derived_of(tl_ep, uct_rc_ep_t); - if (!UCT_RC_EP_MUST_BLOCK_SEND(ep) && - uct_rc_ep_have_tx_cqe_avail(ep) && + if (uct_rc_ep_have_tx_cqe_avail(ep) && uct_rc_ep_has_tx_resources(ep) && uct_rc_iface_has_tx_resources(iface)) { return UCS_ERR_BUSY; @@ -346,11 +344,8 @@ ucs_arbiter_cb_result_t uct_rc_ep_process_pending(ucs_arbiter_t *arbiter, uct_rc_iface_t *iface = ucs_derived_of(ep->super.super.iface, uct_rc_iface_t); ucs_status_t status; - ep->in_pending = 1; - status = uct_rc_iface_invoke_pending_cb(iface, req); - ep->in_pending = 0; + status = uct_rc_iface_invoke_pending_cb(iface, req); if (status == UCS_OK) { - uct_rc_ep_reset_reserve(ep); return UCS_ARBITER_CB_RESULT_REMOVE_ELEM; } else if (status == UCS_INPROGRESS) { return UCS_ARBITER_CB_RESULT_NEXT_GROUP; diff --git a/src/uct/ib/rc/base/rc_ep.h b/src/uct/ib/rc/base/rc_ep.h index 25640c789ea..7cac1622d16 100644 --- a/src/uct/ib/rc/base/rc_ep.h +++ b/src/uct/ib/rc/base/rc_ep.h @@ -90,34 +90,20 @@ enum { #define UCT_RC_DEFINE_ATOMIC_HANDLER_FUNC_NAME(_num_bits, _is_be) \ uct_rc_ep_atomic_handler_##_num_bits##_be##_is_be -#define UCT_RC_EP_MUST_BLOCK_SEND(_ep) \ - (!(_ep)->in_pending && !ucs_arbiter_group_is_empty(&(_ep)->arb_group)) - -#define UCT_RC_CHECK_PENDING_RET(_ep, _ret) \ - if (ucs_unlikely(UCT_RC_EP_MUST_BLOCK_SEND(_ep))) { \ - UCS_STATS_UPDATE_COUNTER((_ep)->super.stats, UCT_EP_STAT_NO_RES, 1); \ - return _ret; \ - } - -#define UCT_RC_CHECK_PENDING(_ep) \ - UCT_RC_CHECK_PENDING_RET(_ep, UCS_ERR_NO_RESOURCE) - /* * Check for send resources */ #define UCT_RC_CHECK_CQE_RET(_iface, _ep, _ret) \ - UCT_RC_CHECK_PENDING_RET(_ep, _ret) \ /* tx_moderation == 0 for TLs which don't support it */ \ if (ucs_unlikely((_iface)->tx.cq_available <= \ - (signed)(_iface)->config.tx_moderation)) { \ + (signed)ucs_max((_iface)->config.tx_moderation, \ + (_ep)->cq_reserve))) { \ UCT_RC_CHECK_CQE_VALUE_RET(_iface, _ep, _ret, 0); \ (_ep)->txqp.unsignaled = RC_UNSIGNALED_INF; \ } #define UCT_RC_CHECK_CQE_VALUE_RET(_iface, _ep, _ret, _value) \ - if ((_iface)->tx.cq_available <= (signed)(_value)) { \ - (_ep)->cq_reserve = ucs_max((_ep)->cq_reserve, \ - (uint16_t)(_value)); \ + if (uct_rc_ep_reserve_cqe(_iface, _ep, (uint16_t)(_value))) { \ UCS_STATS_UPDATE_COUNTER((_iface)->stats, UCT_RC_IFACE_STAT_NO_CQE, 1); \ UCS_STATS_UPDATE_COUNTER((_ep)->super.stats, UCT_EP_STAT_NO_RES, 1); \ return _ret; \ @@ -127,9 +113,7 @@ enum { UCT_RC_CHECK_TXQP_VALUE_RET(_iface, _ep, _ret, 0) #define UCT_RC_CHECK_TXQP_VALUE_RET(_iface, _ep, _ret, _value) \ - if (uct_rc_txqp_available(&(_ep)->txqp) <= (_value)) { \ - (_ep)->txqp_reserve = ucs_max((_ep)->txqp_reserve, \ - (uint16_t)(_value)); \ + if (uct_rc_ep_reserve_txqp(_ep, (uint16_t)(_value))) { \ UCS_STATS_UPDATE_COUNTER((_ep)->txqp.stats, UCT_RC_TXQP_STAT_QP_FULL, 1); \ UCS_STATS_UPDATE_COUNTER((_ep)->super.stats, UCT_EP_STAT_NO_RES, 1); \ return _ret; \ @@ -214,7 +198,8 @@ enum { #define UCT_RC_CHECK_RES_AND_FC(_iface, _ep, _am_id) \ UCT_RC_CHECK_RES(_iface, _ep) \ - UCT_RC_CHECK_FC(_iface, _ep, _am_id) + UCT_RC_CHECK_FC(_iface, _ep, _am_id) \ + uct_rc_iface_check_pending(_iface, &(_ep)->arb_group); /* this is a common type for all rc and dc transports */ struct uct_rc_txqp { @@ -245,7 +230,6 @@ struct uct_rc_ep { uint16_t cq_reserve; uint8_t path_index; uint8_t flags; - uint8_t in_pending; }; @@ -373,6 +357,42 @@ static UCS_F_ALWAYS_INLINE void uct_rc_ep_reset_reserve(uct_rc_ep_t *ep) ep->cq_reserve = 0; } +static UCS_F_ALWAYS_INLINE void +uct_rc_ep_schedule_pending(uct_rc_iface_t *iface, uct_rc_ep_t *ep) +{ + if (ucs_arbiter_group_is_empty(&ep->arb_group)) { + uct_rc_ep_reset_reserve(ep); + } else { + ucs_arbiter_group_schedule_nonempty(&iface->tx.arbiter, &ep->arb_group); + } +} + +static UCS_F_ALWAYS_INLINE int +uct_rc_ep_reserve_txqp(uct_rc_ep_t *ep, uint16_t value) +{ + uint16_t reserve = ucs_max(ep->txqp_reserve, value); + + if (uct_rc_txqp_available(&ep->txqp) > reserve) { + return 0; + } + + ep->txqp_reserve = reserve; + return 1; +} + +static UCS_F_ALWAYS_INLINE int +uct_rc_ep_reserve_cqe(uct_rc_iface_t *iface, uct_rc_ep_t *ep, uint16_t value) +{ + uint16_t reserve = ucs_max(ep->cq_reserve, value); + + if (iface->tx.cq_available > (signed)reserve) { + return 0; + } + + ep->cq_reserve = reserve; + return 1; +} + static UCS_F_ALWAYS_INLINE int uct_rc_ep_has_tx_resources(uct_rc_ep_t *ep) { uct_rc_iface_t *iface = ucs_derived_of(ep->super.super.iface, uct_rc_iface_t); diff --git a/src/uct/ib/rc/verbs/rc_verbs_iface.c b/src/uct/ib/rc/verbs/rc_verbs_iface.c index f548cdd3968..76201c1232f 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_iface.c +++ b/src/uct/ib/rc/verbs/rc_verbs_iface.c @@ -156,8 +156,7 @@ uct_rc_verbs_iface_poll_tx(uct_rc_verbs_iface_t *iface) iface, wc[i].wr_id, ep, wc[i].qp_num, count); uct_rc_txqp_completion_desc(&ep->super.txqp, ep->txcnt.ci + count); - ucs_arbiter_group_schedule(&iface->super.tx.arbiter, - &ep->super.arb_group); + uct_rc_ep_schedule_pending(&iface->super, &ep->super); uct_rc_verbs_update_tx_res(&iface->super, ep, count); ucs_arbiter_dispatch(&iface->super.tx.arbiter, 1, uct_rc_ep_process_pending, NULL); From 377863a3272638547be60e6f57d488df99ad3d5b Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Mon, 19 Oct 2026 14:52:37 -0700 Subject: [PATCH 09/15] UCT/RC: Reset reserve on ep failure --- src/uct/ib/mlx5/rc/rc_mlx5_iface.c | 1 + src/uct/ib/rc/verbs/rc_verbs_iface.c | 1 + 2 files changed, 2 insertions(+) diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c index 4e7417b9d55..ede9f5c8cfb 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c @@ -209,6 +209,7 @@ void uct_rc_mlx5_iface_handle_failure(uct_ib_iface_t *ib_iface, void *arg, ucs_arbiter_group_purge(&iface->tx.arbiter, &ep->super.arb_group, uct_rc_ep_arbiter_purge_internal_cb, NULL); + uct_rc_ep_reset_reserve(&ep->super); uct_ib_mlx5_txwq_update_flags(&ep->tx.wq, UCT_IB_MLX5_TXWQ_FLAG_FAILED, 0); if (ep->super.flags & (UCT_RC_EP_FLAG_ERR_HANDLER_INVOKED | diff --git a/src/uct/ib/rc/verbs/rc_verbs_iface.c b/src/uct/ib/rc/verbs/rc_verbs_iface.c index 76201c1232f..c38f28a65aa 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_iface.c +++ b/src/uct/ib/rc/verbs/rc_verbs_iface.c @@ -101,6 +101,7 @@ static void uct_rc_verbs_handle_failure(uct_ib_iface_t *ib_iface, void *arg, ep->txcnt.ci + count, 0); ucs_arbiter_group_purge(&iface->tx.arbiter, &ep->super.arb_group, uct_rc_ep_arbiter_purge_internal_cb, NULL); + uct_rc_ep_reset_reserve(&ep->super); uct_rc_verbs_update_tx_res(iface, ep, count); if (ep->super.flags & (UCT_RC_EP_FLAG_ERR_HANDLER_INVOKED | From 2ea9451f5557ec7463ce65ea620ce1fb35ebf75c Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Mon, 19 Oct 2026 14:52:38 -0700 Subject: [PATCH 10/15] GTEST/UCT/RC: Fix DC pending dispatch in pending grant test --- test/gtest/uct/ib/test_dc.cc | 10 ++++++++++ test/gtest/uct/ib/test_rc.cc | 2 +- test/gtest/uct/ib/test_rc.h | 4 ++++ 3 files changed, 15 insertions(+), 1 deletion(-) diff --git a/test/gtest/uct/ib/test_dc.cc b/test/gtest/uct/ib/test_dc.cc index 7639f2ab353..8d601e95833 100644 --- a/test/gtest/uct/ib/test_dc.cc +++ b/test/gtest/uct/ib/test_dc.cc @@ -813,6 +813,16 @@ class test_dc_flow_control : public test_rc_flow_control { } } + virtual void dispatch_pending(entity *e) { + uct_dc_mlx5_iface_t *iface = ucs_derived_of(e->iface(), + uct_dc_mlx5_iface_t); + + for (uint8_t pool_index = 0; pool_index < iface->tx.num_dci_pools; + ++pool_index) { + uct_dc_mlx5_iface_progress_pending(iface, pool_index); + } + } + virtual void test_pending_grant(int16_t wnd) { test_rc_flow_control::test_pending_grant(wnd); diff --git a/test/gtest/uct/ib/test_rc.cc b/test/gtest/uct/ib/test_rc.cc index 5507456b410..09c911ea207 100644 --- a/test/gtest/uct/ib/test_rc.cc +++ b/test/gtest/uct/ib/test_rc.cc @@ -1242,7 +1242,7 @@ void test_rc_flow_control::test_pending_grant(int16_t wnd) * Can't send AM/PUT for that, because it is blocked while FC grant is * queued. */ enable_entity(m_e2); - uct_rc_iface_arbiter_dispatch(rc_iface(m_e2)); + dispatch_pending(m_e2); /* Check that m_e1 got grant */ validate_grant(m_e1); diff --git a/test/gtest/uct/ib/test_rc.h b/test/gtest/uct/ib/test_rc.h index 462c5068f49..250894cd952 100644 --- a/test/gtest/uct/ib/test_rc.h +++ b/test/gtest/uct/ib/test_rc.h @@ -83,6 +83,10 @@ class test_rc_flow_control : public test_rc { rc_iface(e)->tx.cq_available = cq_num; } + virtual void dispatch_pending(entity *e) { + uct_rc_iface_arbiter_dispatch(rc_iface(e)); + } + virtual void set_tx_moderation(entity *e, int val) { rc_iface(e)->config.tx_moderation = val; } From 035bc2ecb73f032feb7445c2c212d41e5b98f7e0 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Thu, 29 Oct 2026 16:50:55 -0700 Subject: [PATCH 11/15] GTEST/UCT/RC: Check TX reserve in SGL pending tests --- test/gtest/uct/ib/test_rc.cc | 61 +++++++++++++++++++++++++----------- test/gtest/uct/ib/test_rc.h | 2 ++ 2 files changed, 44 insertions(+), 19 deletions(-) diff --git a/test/gtest/uct/ib/test_rc.cc b/test/gtest/uct/ib/test_rc.cc index 7df6cfed64b..c8c786cc3d2 100644 --- a/test/gtest/uct/ib/test_rc.cc +++ b/test/gtest/uct/ib/test_rc.cc @@ -150,15 +150,11 @@ UCS_TEST_P(test_rc, fence_am_short_consumed, "RC_FENCE=weak") EXPECT_EQ(rc_iface(m_e1)->tx.fi.fence_beat, fence_info->fence_beat); } -UCS_TEST_SKIP_COND_P(test_rc, sgl_pending_am_order, - !check_caps(UCT_IFACE_FLAG_AM_SHORT | - UCT_IFACE_FLAG_PENDING | - UCT_IFACE_FLAG_PUT_SHORT)) +void test_rc::test_sgl_reserve(bool add_pending) { uct_iface_attr_v2_t attr = {}; const size_t length = 64; const size_t count = 2; - unsigned n; ucs_status_t status; uct_pending_req_t pend_req; @@ -178,26 +174,28 @@ UCS_TEST_SKIP_COND_P(test_rc, sgl_pending_am_order, uint64_t raddrs[] = {r0.addr(), r1.addr()}; uct_rkey_t rkeys[] = {r0.rkey(), r1.rkey()}; - n = 0; - while (uct_rc_txqp_available(&rc_ep(m_e1)->txqp) > (int16_t)(count - 1)) { - ASSERT_LT(n++, 1024u); + /* A single PUT SHORT consumes exactly one TXQP credit */ + const int16_t initial = uct_rc_txqp_available(&rc_ep(m_e1)->txqp); + for (int16_t i = initial; i > (int16_t)(count - 1); --i) { ASSERT_UCS_OK(uct_ep_put_short(m_e1->ep(0), s0.ptr(), 8, r0.addr(), r0.rkey())); } - ASSERT_GT(uct_rc_txqp_available(&rc_ep(m_e1)->txqp), 0); - ASSERT_LT(uct_rc_txqp_available(&rc_ep(m_e1)->txqp), (int16_t)count); + ASSERT_EQ(uct_rc_txqp_available(&rc_ep(m_e1)->txqp), (int16_t)(count - 1)); ASSERT_GT(rc_iface(m_e1)->tx.cq_available, (int16_t)(count - 1)); status = uct_ep_put_sgl_zcopy(m_e1->ep(0), buffers, lengths, memhs, raddrs, rkeys, NULL, NULL, count, NULL); ASSERT_EQ(UCS_ERR_NO_RESOURCE, status); + EXPECT_GT(rc_ep(m_e1)->txqp_reserve, 0); - pend_req.func = [](uct_pending_req_t*) -> ucs_status_t { - return UCS_ERR_NO_RESOURCE; - }; + if (add_pending) { + pend_req.func = [](uct_pending_req_t*) -> ucs_status_t { + return UCS_ERR_NO_RESOURCE; + }; - ASSERT_UCS_OK(uct_ep_pending_add(m_e1->ep(0), &pend_req, 0)); + ASSERT_UCS_OK(uct_ep_pending_add(m_e1->ep(0), &pend_req, 0)); + } EXPECT_EQ(UCS_ERR_NO_RESOURCE, uct_ep_put_sgl_zcopy(m_e1->ep(0), buffers, lengths, memhs, @@ -205,8 +203,32 @@ UCS_TEST_SKIP_COND_P(test_rc, sgl_pending_am_order, EXPECT_EQ(UCS_ERR_NO_RESOURCE, uct_ep_am_short(m_e1->ep(0), 0, 0, NULL, 0)); - uct_ep_pending_purge(m_e1->ep(0), NULL, NULL); + if (add_pending) { + /* The pending request never completes, so purge it to drop the + * reserve */ + uct_ep_pending_purge(m_e1->ep(0), NULL, NULL); + } + + /* Without pending requests, the reserve is dropped on TX completion */ flush(); + + EXPECT_EQ(0, rc_ep(m_e1)->txqp_reserve); + EXPECT_EQ(0, rc_ep(m_e1)->cq_reserve); +} + +UCS_TEST_SKIP_COND_P(test_rc, sgl_pending_am_order, + !check_caps(UCT_IFACE_FLAG_AM_SHORT | + UCT_IFACE_FLAG_PENDING | + UCT_IFACE_FLAG_PUT_SHORT)) +{ + test_sgl_reserve(true); +} + +UCS_TEST_SKIP_COND_P(test_rc, sgl_reserve_released_on_completion, + !check_caps(UCT_IFACE_FLAG_AM_SHORT | + UCT_IFACE_FLAG_PUT_SHORT)) +{ + test_sgl_reserve(false); } UCS_TEST_SKIP_COND_P(test_rc, relaxed_order_required_strong_fence, @@ -1412,11 +1434,12 @@ void test_rc_flow_control::test_pending_grant(int16_t wnd) wait_fc_hard_resend(m_e1); - /* Enable send capabilities of m_e2 and dispatch the pending queue. - * Can't send AM/PUT for that, because it is blocked while FC grant is - * queued. */ + /* Enable send capabilities of m_e2 and send short put message to force + * pending queue dispatch. Can't send AM message for that, because it may + * trigger reordering assert due to disable/enable entity hack. */ enable_entity(m_e2); - uct_rc_iface_arbiter_dispatch(rc_iface(m_e2)); + set_tx_moderation(m_e2, 0); + EXPECT_EQ(UCS_OK, uct_ep_put_short(m_e2->ep(0), NULL, 0, 0, 0)); /* Check that m_e1 got grant */ validate_grant(m_e1); diff --git a/test/gtest/uct/ib/test_rc.h b/test/gtest/uct/ib/test_rc.h index 462c5068f49..44c913aec33 100644 --- a/test/gtest/uct/ib/test_rc.h +++ b/test/gtest/uct/ib/test_rc.h @@ -50,6 +50,8 @@ class test_rc : public uct_test { void test_iface_ops(int cq_len); + void test_sgl_reserve(bool add_pending); + static ucs_status_t am_dummy_handler(void *arg, void *data, size_t length, unsigned flags) { return UCS_OK; From af88ade7683d4b71b14292899b94aa2774656289 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Thu, 29 Oct 2026 17:14:09 -0700 Subject: [PATCH 12/15] UCT/RC: Tidy reserve and schedule_pending helper signatures --- src/uct/ib/mlx5/rc/rc_mlx5.inl | 2 +- src/uct/ib/rc/base/rc_ep.h | 7 +++++-- src/uct/ib/rc/verbs/rc_verbs_iface.c | 2 +- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/src/uct/ib/mlx5/rc/rc_mlx5.inl b/src/uct/ib/mlx5/rc/rc_mlx5.inl index eacf5df085a..4b855c02c1b 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5.inl +++ b/src/uct/ib/mlx5/rc/rc_mlx5.inl @@ -2109,7 +2109,7 @@ uct_rc_mlx5_iface_poll_tx(uct_rc_mlx5_iface_common_t *iface, int poll_flags) ep, qp_num, hw_ci); uct_rc_mlx5_txqp_process_tx_cqe(&ep->super.txqp, cqe, hw_ci); - uct_rc_ep_schedule_pending(&iface->super, &ep->super); + uct_rc_ep_schedule_pending(&ep->super); uct_rc_mlx5_ep_update_tx_qp_res(ep, hw_ci); uct_rc_mlx5_iface_update_tx_cq_res(&iface->super, ep, hw_ci); uct_rc_iface_arbiter_dispatch(&iface->super); diff --git a/src/uct/ib/rc/base/rc_ep.h b/src/uct/ib/rc/base/rc_ep.h index 79d35492876..70bc97c153a 100644 --- a/src/uct/ib/rc/base/rc_ep.h +++ b/src/uct/ib/rc/base/rc_ep.h @@ -379,8 +379,10 @@ static UCS_F_ALWAYS_INLINE void uct_rc_ep_reset_reserve(uct_rc_ep_t *ep) } static UCS_F_ALWAYS_INLINE void -uct_rc_ep_schedule_pending(uct_rc_iface_t *iface, uct_rc_ep_t *ep) +uct_rc_ep_schedule_pending(uct_rc_ep_t *ep) { + uct_rc_iface_t *iface = ucs_derived_of(ep->super.super.iface, uct_rc_iface_t); + if (ucs_arbiter_group_is_empty(&ep->arb_group)) { uct_rc_ep_reset_reserve(ep); } else { @@ -402,7 +404,8 @@ uct_rc_ep_reserve_txqp(uct_rc_ep_t *ep, uint16_t value) } static UCS_F_ALWAYS_INLINE int -uct_rc_ep_reserve_cqe(uct_rc_iface_t *iface, uct_rc_ep_t *ep, uint16_t value) +uct_rc_ep_reserve_cqe(const uct_rc_iface_t *iface, uct_rc_ep_t *ep, + uint16_t value) { uint16_t reserve = ucs_max(ep->cq_reserve, value); diff --git a/src/uct/ib/rc/verbs/rc_verbs_iface.c b/src/uct/ib/rc/verbs/rc_verbs_iface.c index 76201c1232f..1af69f2b06d 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_iface.c +++ b/src/uct/ib/rc/verbs/rc_verbs_iface.c @@ -156,7 +156,7 @@ uct_rc_verbs_iface_poll_tx(uct_rc_verbs_iface_t *iface) iface, wc[i].wr_id, ep, wc[i].qp_num, count); uct_rc_txqp_completion_desc(&ep->super.txqp, ep->txcnt.ci + count); - uct_rc_ep_schedule_pending(&iface->super, &ep->super); + uct_rc_ep_schedule_pending(&ep->super); uct_rc_verbs_update_tx_res(&iface->super, ep, count); ucs_arbiter_dispatch(&iface->super.tx.arbiter, 1, uct_rc_ep_process_pending, NULL); From b2aa6c301d1e58d9e1ef31b59097c05e829c4946 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Thu, 29 Oct 2026 17:20:09 -0700 Subject: [PATCH 13/15] UCT/RC: Explain why pending success keeps the reserve --- src/uct/ib/rc/base/rc_ep.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/uct/ib/rc/base/rc_ep.c b/src/uct/ib/rc/base/rc_ep.c index 7298e60850c..c651398cf0c 100644 --- a/src/uct/ib/rc/base/rc_ep.c +++ b/src/uct/ib/rc/base/rc_ep.c @@ -424,6 +424,9 @@ ucs_arbiter_cb_result_t uct_rc_ep_process_pending(ucs_arbiter_t *arbiter, status = uct_rc_iface_invoke_pending_cb(iface, req); if (status == UCS_OK) { + /* Keep the reserve because other requests may still be queued, and new + * sends must not pass them. It is reset when the group is empty, on TX + * completion or purge. */ return UCS_ARBITER_CB_RESULT_REMOVE_ELEM; } else if (status == UCS_INPROGRESS) { return UCS_ARBITER_CB_RESULT_NEXT_GROUP; From 4dd65aecc566395be0aecba48c05fbe8b1e544d5 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Sun, 1 Nov 2026 14:35:52 -0800 Subject: [PATCH 14/15] GTEST/UCT/RC: Address review comments in pending grant test --- test/gtest/uct/ib/test_dc.cc | 6 +----- test/gtest/uct/ib/test_rc.cc | 5 ++--- 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/test/gtest/uct/ib/test_dc.cc b/test/gtest/uct/ib/test_dc.cc index 8d601e95833..7d2ca92f31d 100644 --- a/test/gtest/uct/ib/test_dc.cc +++ b/test/gtest/uct/ib/test_dc.cc @@ -806,11 +806,7 @@ class test_dc_flow_control : public test_rc_flow_control { ucs_array_length(&iface->tx.dci_pool[0].stack) = 0; iface->tx.dci_pool[0].stack_top = 0; - uint8_t pool_index; - for (pool_index = 0; pool_index < iface->tx.num_dci_pools; - ++pool_index) { - uct_dc_mlx5_iface_progress_pending(iface, pool_index); - } + dispatch_pending(e); } virtual void dispatch_pending(entity *e) { diff --git a/test/gtest/uct/ib/test_rc.cc b/test/gtest/uct/ib/test_rc.cc index 6a2a8f4ab5f..f21993350a2 100644 --- a/test/gtest/uct/ib/test_rc.cc +++ b/test/gtest/uct/ib/test_rc.cc @@ -1434,9 +1434,8 @@ void test_rc_flow_control::test_pending_grant(int16_t wnd) wait_fc_hard_resend(m_e1); - /* Enable send capabilities of m_e2 and dispatch the pending queue. - * Can't send AM/PUT for that, because it is blocked while FC grant is - * queued. */ + /* Enable send capabilities of m_e2 and dispatch the pending queue directly, + * so it does not depend on a send to generate a TX completion. */ enable_entity(m_e2); dispatch_pending(m_e2); From bfa0c7821137aae6574df0daae0e4c17dbf31330 Mon Sep 17 00:00:00 2001 From: Michal Shalev Date: Sun, 1 Nov 2026 15:30:57 -0800 Subject: [PATCH 15/15] UCT/RC: Keep reserve on failure while requests are queued --- src/uct/ib/mlx5/rc/rc_mlx5_iface.c | 2 +- src/uct/ib/rc/verbs/rc_verbs_iface.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c index bc9014f08be..8572ff7d725 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5_iface.c +++ b/src/uct/ib/mlx5/rc/rc_mlx5_iface.c @@ -209,7 +209,7 @@ void uct_rc_mlx5_iface_handle_failure(uct_ib_iface_t *ib_iface, void *arg, ucs_arbiter_group_purge(&iface->tx.arbiter, &ep->super.arb_group, uct_rc_ep_arbiter_purge_internal_cb, NULL); - uct_rc_ep_reset_reserve(&ep->super); + uct_rc_ep_schedule_pending(&ep->super); uct_ib_mlx5_txwq_update_flags(&ep->tx.wq, UCT_IB_MLX5_TXWQ_FLAG_FAILED, 0); if (ep->super.flags & (UCT_RC_EP_FLAG_ERR_HANDLER_INVOKED | diff --git a/src/uct/ib/rc/verbs/rc_verbs_iface.c b/src/uct/ib/rc/verbs/rc_verbs_iface.c index 771919907a9..d48c0bc3e11 100644 --- a/src/uct/ib/rc/verbs/rc_verbs_iface.c +++ b/src/uct/ib/rc/verbs/rc_verbs_iface.c @@ -101,7 +101,7 @@ static void uct_rc_verbs_handle_failure(uct_ib_iface_t *ib_iface, void *arg, ep->txcnt.ci + count, 0); ucs_arbiter_group_purge(&iface->tx.arbiter, &ep->super.arb_group, uct_rc_ep_arbiter_purge_internal_cb, NULL); - uct_rc_ep_reset_reserve(&ep->super); + uct_rc_ep_schedule_pending(&ep->super); uct_rc_verbs_update_tx_res(iface, ep, count); if (ep->super.flags & (UCT_RC_EP_FLAG_ERR_HANDLER_INVOKED |