From 50be3cf9e2510ebe7a4fd0637555af8ae8fdd0fb Mon Sep 17 00:00:00 2001 From: Dmitrii Date: Tue, 3 Nov 2026 20:39:12 +0100 Subject: [PATCH] UCT/IB/MLX5/RC: Complete SGL op with ep error status on peer failure An SGL operation posts one WQE per element, and only the last WQE is signaled. Its send op was kept on the last WQE, so the peer failure purge, which completes the operations up to the failed WQE with the endpoint error status, missed it when an element other than the last one failed, and the operation was completed later by the QP flush with UCS_ERR_CANCELED. Keep the send op on the first WQE of the chain, so a failure of any of its WQEs completes the operation with the endpoint error status, like an operation of a single WQE. Without a failure it is still completed by the CQE of the last WQE. Add test_uct_peer_failure_sgl_zcopy on rc_mlx5 and cuda_ipc: a failure of the first, a middle or the last element of an SGL put/get, SGL puts with a zero-length element, an SGL put posted after a failed put_short, a put_zcopy posted after a failed SGL put, and completion without a failure. uct_test::check_caps() gets optional required v2 caps for the SGL caps check. --- src/uct/ib/mlx5/rc/rc_mlx5.inl | 10 +- test/gtest/uct/test_peer_failure.cc | 365 ++++++++++++++++++++++++++++ test/gtest/uct/uct_test.cc | 5 +- test/gtest/uct/uct_test.h | 3 +- 4 files changed, 378 insertions(+), 5 deletions(-) diff --git a/src/uct/ib/mlx5/rc/rc_mlx5.inl b/src/uct/ib/mlx5/rc/rc_mlx5.inl index 4b855c02c1b..177d231334d 100644 --- a/src/uct/ib/mlx5/rc/rc_mlx5.inl +++ b/src/uct/ib/mlx5/rc/rc_mlx5.inl @@ -516,6 +516,7 @@ uct_rc_mlx5_base_ep_sgl_zcopy_post(uct_rc_mlx5_base_ep_t *ep, uint16_t op_flags, uct_completion_t *comp) { uct_ib_mlx5_txwq_t *txwq = &ep->tx.wq; + const uint16_t first_pi = txwq->sw_pi; size_t total = 0; struct mlx5_wqe_ctrl_seg *ctrl = NULL; uint32_t num_packets = 0; @@ -535,7 +536,7 @@ uct_rc_mlx5_base_ep_sgl_zcopy_post(uct_rc_mlx5_base_ep_t *ep, ucs_assert(count > 0); - pi = txwq->sw_pi; + pi = first_pi; ucs_assert(!(txwq->flags & UCT_IB_MLX5_TXWQ_FLAG_FAILED)); ucs_assert(ucs_div_round_up(data_wqe_size, MLX5_SEND_WQE_BB) == 1); @@ -591,8 +592,13 @@ uct_rc_mlx5_base_ep_sgl_zcopy_post(uct_rc_mlx5_base_ep_t *ep, uct_ib_mlx5_txwq_ring_doorbell(txwq, ctrl, txwq->sw_pi, 1); uct_rc_mlx5_txwq_add_psn(txwq, IBV_QPT_RC, num_packets); + /* Keep the send op on the first WQE of the chain, so a failure reported for + * any WQE of the chain completes the operation with the error status, like + * for an operation of a single WQE, but otherwise it is completed by the + * CQE of the last WQE, since only it is signaled, while the others are + * unsignaled. */ uct_rc_txqp_add_send_comp_sgl_always( - &iface->super, &ep->super.txqp, handler, comp, txwq->sig_pi, + &iface->super, &ep->super.txqp, handler, comp, first_pi, UCT_RC_IFACE_SEND_OP_FLAG_ZCOPY | op_flags, buffers, lengths, count, total); return total; diff --git a/test/gtest/uct/test_peer_failure.cc b/test/gtest/uct/test_peer_failure.cc index 3b74fef8527..5b89d54eb54 100644 --- a/test/gtest/uct/test_peer_failure.cc +++ b/test/gtest/uct/test_peer_failure.cc @@ -1593,3 +1593,368 @@ UCS_TEST_SKIP_COND_P(test_uct_peer_failure_rma_zcopy, get, } _UCT_INSTANTIATE_TEST_CASE(test_uct_peer_failure_rma_zcopy, cma) + +class test_uct_peer_failure_sgl_zcopy : public test_uct_peer_failure +{ +public: + static const size_t SGL_COUNT = 4; + static const size_t SGL_LENGTH = 256; + static const size_t NO_INDEX = SIZE_MAX; + + /* Fail one element of the SGL by accessing memory outside of the remote + * buffer, the whole operation must complete with the endpoint error + * status. The element at empty_index is of zero length */ + void test_sgl_zcopy_peer_failure(bool is_put_op, size_t failed_index, + size_t empty_index = NO_INDEX) + { + uint64_t remote_addrs[SGL_COUNT]; + uct_rkey_t rkeys[SGL_COUNT]; + op_completion comp; + + skip_no_op_failure(); + + mapped_buffer localbuf(SGL_COUNT * SGL_LENGTH, 0, *m_sender); + mapped_buffer remotebuf(SGL_COUNT * SGL_LENGTH, 0, + *m_receivers.front()); + + set_remote_sgl(remotebuf, failed_index, remote_addrs, rkeys); + + { + scoped_log_handler slh(wrap_errors_logger); + + ASSERT_EQ(UCS_INPROGRESS, + post_sgl_zcopy(is_put_op, localbuf, remote_addrs, rkeys, + &comp, empty_index)); + wait_for_flag(&comp.done); + flush(); + } + + ASSERT_EQ(1u, comp.done); + ASSERT_EQ(1ul, m_failed_eps.count(ep0())); + EXPECT_EQ(m_failed_eps[ep0()], comp.uct.status) + << ucs_status_string(comp.uct.status); + } + + /* Fail the first element of an SGL put, which is of zero length, so its + * WQE has no data segment. The remote address of a zero-length element is + * not accessed, so the failure is caused by killing the receiver */ + void test_sgl_zcopy_empty_element_failure() + { + uint64_t remote_addrs[SGL_COUNT] = {}; + uct_rkey_t rkeys[SGL_COUNT] = {}; + op_completion comp; + + skip_no_op_failure(); + + mapped_buffer localbuf(SGL_COUNT * SGL_LENGTH, 0, *m_sender); + + { + scoped_log_handler slh(wrap_errors_logger); + + kill_receiver(); + ASSERT_EQ(UCS_INPROGRESS, + post_sgl_zcopy(true, localbuf, remote_addrs, rkeys, + &comp, 0)); + wait_for_flag(&comp.done); + flush(); + } + + ASSERT_EQ(1u, comp.done); + ASSERT_EQ(1ul, m_failed_eps.count(ep0())); + EXPECT_EQ(m_failed_eps[ep0()], comp.uct.status) + << ucs_status_string(comp.uct.status); + } + + /* Fail a zero-length put_short posted right before an SGL put. The SGL put + * must not be completed by the failure of the put_short WQE, since its own + * WQEs are not reported yet, but by the flush of the QP */ + void test_sgl_zcopy_after_failed_put_short() + { + /* The receiver is killed, so the remote memory is not accessed */ + uint64_t remote_addrs[SGL_COUNT] = {}; + uct_rkey_t rkeys[SGL_COUNT] = {}; + op_completion comp; + + skip_no_op_failure(); + + mapped_buffer localbuf(SGL_COUNT * SGL_LENGTH, 0, *m_sender); + + { + scoped_log_handler slh(wrap_errors_logger); + + kill_receiver(); + ASSERT_UCS_OK(uct_ep_put_short(ep0(), NULL, 0, 0, 0)); + ASSERT_EQ(UCS_INPROGRESS, + post_sgl_zcopy(true, localbuf, remote_addrs, rkeys, + &comp)); + wait_for_flag(&comp.done); + flush(); + } + + ASSERT_EQ(1u, comp.done); + ASSERT_EQ(1ul, m_failed_eps.count(ep0())); + EXPECT_EQ(UCS_ERR_CANCELED, comp.uct.status) + << ucs_status_string(comp.uct.status); + } + + /* Fail the first element of an SGL put followed by a put_zcopy. The SGL put + * completes with the endpoint error status, while the put_zcopy is not + * completed by this failure, but by the flush of the QP */ + void test_put_zcopy_after_failed_sgl_zcopy() + { + uint64_t remote_addrs[SGL_COUNT]; + uct_rkey_t rkeys[SGL_COUNT]; + op_completion sgl_comp, zcopy_comp; + + skip_no_op_failure(); + + mapped_buffer localbuf(SGL_COUNT * SGL_LENGTH, 0, *m_sender); + mapped_buffer remotebuf(SGL_COUNT * SGL_LENGTH, 0, + *m_receivers.front()); + + set_remote_sgl(remotebuf, 0, remote_addrs, rkeys); + + { + scoped_log_handler slh(wrap_errors_logger); + + ASSERT_EQ(UCS_INPROGRESS, + post_sgl_zcopy(true, localbuf, remote_addrs, rkeys, + &sgl_comp)); + init_completion(&zcopy_comp); + ASSERT_EQ(UCS_INPROGRESS, + uct_ep_put_zcopy(ep0(), localbuf.iov(), 1, + remotebuf.addr(), remotebuf.rkey(), + &zcopy_comp.uct)); + wait_for_flag(&sgl_comp.done); + wait_for_flag(&zcopy_comp.done); + flush(); + } + + ASSERT_EQ(1u, sgl_comp.done); + ASSERT_EQ(1u, zcopy_comp.done); + ASSERT_EQ(1ul, m_failed_eps.count(ep0())); + EXPECT_EQ(m_failed_eps[ep0()], sgl_comp.uct.status) + << ucs_status_string(sgl_comp.uct.status); + EXPECT_EQ(UCS_ERR_CANCELED, zcopy_comp.uct.status) + << ucs_status_string(zcopy_comp.uct.status); + } + + /* Without a failure the operation completes with UCS_OK by the CQE of the + * last WQE of the chain, so the data of all its elements is in place. The + * element at empty_index is of zero length, so its data is not moved */ + void test_sgl_zcopy_completion(bool is_put_op, + size_t empty_index = NO_INDEX) + { + const uint64_t local_seed = 1; + const uint64_t remote_seed = 2; + uint64_t remote_addrs[SGL_COUNT]; + uct_rkey_t rkeys[SGL_COUNT]; + op_completion comp; + + mapped_buffer localbuf(SGL_COUNT * SGL_LENGTH, local_seed, *m_sender, 0, + mem_type()); + mapped_buffer remotebuf(SGL_COUNT * SGL_LENGTH, remote_seed, + *m_receivers.front(), 0, mem_type()); + + set_remote_sgl(remotebuf, NO_INDEX, remote_addrs, rkeys); + ASSERT_EQ(UCS_INPROGRESS, + post_sgl_zcopy(is_put_op, localbuf, remote_addrs, rkeys, + &comp, empty_index)); + wait_for_flag(&comp.done); + + ASSERT_EQ(1u, comp.done); + EXPECT_UCS_OK(comp.uct.status); + if (is_put_op) { + /* Put completion does not imply remote completion */ + flush(); + } + + /* The buffers are filled with different patterns, so they are equal at + * the offset of an element only if its data was moved */ + for (size_t i = 0; i < SGL_COUNT; ++i) { + const size_t offset = i * SGL_LENGTH; + + EXPECT_EQ(i != empty_index, + mem_buffer::compare( + UCS_PTR_BYTE_OFFSET(localbuf.ptr(), offset), + UCS_PTR_BYTE_OFFSET(remotebuf.ptr(), offset), + SGL_LENGTH, mem_type(), mem_type())) + << "element " << i; + } + + EXPECT_EQ(0ul, m_failed_eps.count(ep0())); + } + +private: + struct op_completion { + uct_completion_t uct; + unsigned done; + }; + + static void completion_cb(uct_completion_t *self) + { + ucs_container_of(self, op_completion, uct)->done = 1; + } + + static void init_completion(op_completion *comp) + { + comp->uct.func = completion_cb; + comp->uct.count = 1; + comp->uct.status = UCS_OK; + comp->done = 0; + } + + /* cuda_ipc does not report errors of posted operations, and a copy to an + * invalid remote address is a fatal CUDA error */ + void skip_no_op_failure() const + { + if (has_cuda_ipc()) { + UCS_TEST_SKIP_R("cuda_ipc does not fail posted operations"); + } + } + + ucs_memory_type_t mem_type() const + { + return has_cuda_ipc() ? UCS_MEMORY_TYPE_CUDA : UCS_MEMORY_TYPE_HOST; + } + + /* The element at failed_index points outside of the remote buffer */ + static void set_remote_sgl(const mapped_buffer &remotebuf, + size_t failed_index, uint64_t *remote_addrs, + uct_rkey_t *rkeys) + { + for (size_t i = 0; i < SGL_COUNT; ++i) { + remote_addrs[i] = (i == failed_index) ? + 0 : (remotebuf.addr() + (i * SGL_LENGTH)); + rkeys[i] = remotebuf.rkey(); + } + } + + ucs_status_t post_sgl_zcopy(bool is_put_op, const mapped_buffer &localbuf, + const uint64_t *remote_addrs, + const uct_rkey_t *rkeys, op_completion *comp, + size_t empty_index = NO_INDEX) + { + void *buffers[SGL_COUNT]; + size_t lengths[SGL_COUNT]; + uct_mem_h memhs[SGL_COUNT]; + + for (size_t i = 0; i < SGL_COUNT; ++i) { + buffers[i] = UCS_PTR_BYTE_OFFSET(localbuf.ptr(), i * SGL_LENGTH); + lengths[i] = (i == empty_index) ? 0 : SGL_LENGTH; + memhs[i] = localbuf.memh(); + } + + init_completion(comp); + if (is_put_op) { + return uct_ep_put_sgl_zcopy(ep0(), buffers, lengths, memhs, + remote_addrs, rkeys, NULL, NULL, + SGL_COUNT, &comp->uct); + } + + return uct_ep_get_sgl_zcopy(ep0(), buffers, lengths, memhs, + remote_addrs, rkeys, NULL, NULL, SGL_COUNT, + &comp->uct); + } +}; + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, put_fail_first, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_peer_failure(true, 0); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, put_fail_middle, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_peer_failure(true, SGL_COUNT / 2); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, put_fail_last, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_peer_failure(true, SGL_COUNT - 1); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, get_fail_first, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_GET_SGL_ZCOPY)) +{ + test_sgl_zcopy_peer_failure(false, 0); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, get_fail_middle, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_GET_SGL_ZCOPY)) +{ + test_sgl_zcopy_peer_failure(false, SGL_COUNT / 2); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, get_fail_last, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_GET_SGL_ZCOPY)) +{ + test_sgl_zcopy_peer_failure(false, SGL_COUNT - 1); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, + put_after_failed_put_short, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE | + UCT_IFACE_FLAG_PUT_SHORT, + 0, UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_after_failed_put_short(); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, + put_zcopy_after_failed_sgl, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE | + UCT_IFACE_FLAG_PUT_ZCOPY, + 0, UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_put_zcopy_after_failed_sgl_zcopy(); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, put_completion, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_completion(true); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, get_completion, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_GET_SGL_ZCOPY)) +{ + test_sgl_zcopy_completion(false); +} + +/* Only put allows elements of zero length */ +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, put_fail_empty, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_empty_element_failure(); +} + +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, put_fail_after_empty, + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_peer_failure(true, SGL_COUNT / 2, 0); +} + +/* TODO: run on cuda_ipc when it supports zero-length SGL elements */ +UCS_TEST_SKIP_COND_P(test_uct_peer_failure_sgl_zcopy, put_completion_empty, + has_cuda_ipc() || + !check_caps(UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE, 0, + UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY)) +{ + test_sgl_zcopy_completion(true, SGL_COUNT / 2); +} + +_UCT_INSTANTIATE_TEST_CASE(test_uct_peer_failure_sgl_zcopy, rc_mlx5) +_UCT_INSTANTIATE_TEST_CASE(test_uct_peer_failure_sgl_zcopy, cuda_ipc) diff --git a/test/gtest/uct/uct_test.cc b/test/gtest/uct/uct_test.cc index 374a7d6237e..a88fc371af4 100644 --- a/test/gtest/uct/uct_test.cc +++ b/test/gtest/uct/uct_test.cc @@ -504,13 +504,14 @@ bool uct_test::is_caps_supported(uint64_t required_flags) { return ret; } -bool uct_test::check_caps(uint64_t required_flags, uint64_t invalid_flags) { +bool uct_test::check_caps(uint64_t required_flags, uint64_t invalid_flags, + uint64_t required_v2_flags) { FOR_EACH_ENTITY(iter) { if (!(*iter)->check_caps(required_flags, invalid_flags)) { return false; } } - return true; + return (required_v2_flags == 0) || check_caps_v2(required_v2_flags); } bool uct_test::check_caps_v2(uint64_t required_flags) { diff --git a/test/gtest/uct/uct_test.h b/test/gtest/uct/uct_test.h index 5550524b6e2..74b01b773d0 100644 --- a/test/gtest/uct/uct_test.h +++ b/test/gtest/uct/uct_test.h @@ -397,7 +397,8 @@ class uct_test : public testing::TestWithParam, virtual bool has_gpu() const; bool is_caps_supported(uint64_t required_flags); - bool check_caps(uint64_t required_flags, uint64_t invalid_flags = 0); + bool check_caps(uint64_t required_flags, uint64_t invalid_flags = 0, + uint64_t required_v2_flags = 0); bool check_caps_v2(uint64_t required_flags); void check_caps_skip(uint64_t required_flags, uint64_t invalid_flags = 0); void check_skip_flush_remote();