From 60acc905b1667831029ca33602625b340a445f0d Mon Sep 17 00:00:00 2001 From: Evgeny Leksikov Date: Wed, 14 Oct 2026 15:26:59 +0300 Subject: [PATCH 1/5] UCP: Exclude device atomics from failover endpoints. NIC hardware atomics have no replay path, so failover lane selection leaves those transports out. Token failover is rejected when IB indirect atomics are enabled; auto selects software PSN recovery. --- src/ucp/core/ucp_context.c | 76 +++++++++++++++++++--- src/ucp/wireup/select.c | 25 +++++++ test/gtest/ucp/test_ucp_fault_tolerance.cc | 2 + test/gtest/ucp/test_ucp_wireup.cc | 22 +++++++ 4 files changed, 116 insertions(+), 9 deletions(-) diff --git a/src/ucp/core/ucp_context.c b/src/ucp/core/ucp_context.c index 12f4a5487f4..91b0eed5a86 100644 --- a/src/ucp/core/ucp_context.c +++ b/src/ucp/core/ucp_context.c @@ -482,11 +482,13 @@ static ucs_config_field_t ucp_context_config_table[] = { "Failover method for endpoints created with\n" "UCP_ERR_HANDLING_MODE_FAILOVER.\n" " auto - auto selected recovery protocol. Transport selection is\n" - " unchanged.\n" + " unchanged. Selects software PSN when UCX_IB_INDIRECT_ATOMIC\n" + " is enabled.\n" " sw - force software PSN based recovery protocol. Transport\n" " selection is unchanged.\n" " token - force token based retransmits. Transports that do not support\n" - " the required capabilities are disabled.", + " the required capabilities are disabled. Rejected when\n" + " UCX_IB_INDIRECT_ATOMIC is enabled.", ucs_offsetof(ucp_context_config_t, failover_mode), UCS_CONFIG_TYPE_ENUM(ucp_failover_modes)}, @@ -2213,6 +2215,64 @@ static void ucp_fill_resources_reg_md_map_update(ucp_context_h context) } } +/* Effective UCX_IB_INDIRECT_ATOMIC: IB MD default and environment, then + * cached ucp_config_modify() overrides. */ +static int ucp_context_query_ib_indirect_atomic(ucp_context_h context) +{ + int enabled = 1; + ucp_rsc_index_t cmpt_index; + uct_md_config_t *md_config; + ucs_status_t status; + char value[8]; + + for (cmpt_index = 0; cmpt_index < context->num_cmpts; ++cmpt_index) { + if (strcmp(context->tl_cmpts[cmpt_index].attr.name, "ib") != 0) { + continue; + } + + status = uct_md_config_read(context->tl_cmpts[cmpt_index].cmpt, + NULL, NULL, &md_config); + if (status != UCS_OK) { + return 1; + } + + ucp_apply_uct_config_list(context, md_config); + status = uct_config_get(md_config, "INDIRECT_ATOMIC", value, + sizeof(value)); + if ((status != UCS_OK) || + (ucs_config_sscanf_bool(value, &enabled, NULL) == 0)) { + enabled = 1; + } + + uct_config_release(md_config); + return enabled; + } + + return 0; +} + +/* Indirect atomics have no token replay. */ +static ucs_status_t ucp_context_adjust_failover_mode(ucp_context_h context) +{ + int enabled = ucp_context_query_ib_indirect_atomic(context); + + if (!enabled) { + return UCS_OK; + } + + if (context->config.ext.failover_mode == UCP_FAILOVER_MODE_TOKEN) { + ucs_error("UCX_FAILOVER_MODE=token requires " + "UCX_IB_INDIRECT_ATOMIC=n"); + return UCS_ERR_INVALID_PARAM; + } else if (context->config.ext.failover_mode == UCP_FAILOVER_MODE_AUTO) { + ucs_debug("ib indirect atomic is enabled, failover mode auto " + "selects software psn recovery"); + context->config.ext.failover_mode = UCP_FAILOVER_MODE_SW; + } + + return UCS_OK; +} + static ucs_status_t ucp_fill_resources(ucp_context_h context, const ucp_config_t *config) { @@ -2368,6 +2428,11 @@ ucp_fill_resources(ucp_context_h context, const ucp_config_t *config) ucp_context_log_tl_info(context, &all_rscs); + status = ucp_context_adjust_failover_mode(context); + if (status != UCS_OK) { + goto err_free_resources; + } + out_release_components: ucs_array_cleanup_dynamic(&all_rscs); uct_release_component_list(uct_components); @@ -2733,13 +2798,6 @@ static ucs_status_t ucp_fill_config(ucp_context_h context, context->config.worker_fence_mode = context->config.ext.fence_mode; } - /* Token based recovery is not implemented yet, so automatic selection - * uses the software PSN recovery protocol. */ - if (context->config.ext.failover_mode == UCP_FAILOVER_MODE_AUTO) { - context->config.ext.failover_mode = UCP_FAILOVER_MODE_SW; - ucs_debug("failover mode auto selects software psn recovery"); - } - context->config.progress_wrapper_enabled = ucs_log_is_enabled(UCS_LOG_LEVEL_TRACE_REQ) || ucp_context_usage_tracker_enabled(context); diff --git a/src/ucp/wireup/select.c b/src/ucp/wireup/select.c index 2a91666c22e..fc404548e75 100644 --- a/src/ucp/wireup/select.c +++ b/src/ucp/wireup/select.c @@ -1332,6 +1332,27 @@ double ucp_wireup_amo_score_func(const ucp_worker_iface_t *wiface, wiface->attr.overhead); } +/* Failover has no replay path for NIC hardware atomics. */ +static void +ucp_wireup_exclude_device_amo_tls(ucp_worker_h worker, + ucp_tl_bitmap_t *tl_bitmap) +{ + ucp_rsc_index_t rsc_index; + + UCS_STATIC_BITMAP_FOR_EACH_BIT(rsc_index, tl_bitmap) { + if (!(ucp_worker_iface_get_attr(worker, rsc_index)->cap.flags & + UCT_IFACE_FLAG_ATOMIC_DEVICE)) { + continue; + } + + ucs_trace(UCT_TL_RESOURCE_DESC_FMT + " : disabled, failover does not use device atomics", + UCT_TL_RESOURCE_DESC_ARG( + &worker->context->tl_rscs[rsc_index].tl_rsc)); + UCS_STATIC_BITMAP_RESET(tl_bitmap, rsc_index); + } +} + static ucs_status_t ucp_wireup_add_amo_lanes(const ucp_wireup_select_params_t *select_params, ucp_wireup_select_context_t *select_ctx) @@ -1371,6 +1392,10 @@ ucp_wireup_add_amo_lanes(const ucp_wireup_select_params_t *select_params, } } + if (ep_init_flags & UCP_EP_INIT_ERR_MODE_FAILOVER) { + ucp_wireup_exclude_device_amo_tls(worker, &tl_bitmap); + } + return ucp_wireup_add_memaccess_lanes(select_params, ep_init_flags, &criteria, UCS_MEMORY_TYPE_HOST, tl_bitmap, UCP_LANE_TYPE_AMO, diff --git a/test/gtest/ucp/test_ucp_fault_tolerance.cc b/test/gtest/ucp/test_ucp_fault_tolerance.cc index 824c32ef516..71170f60140 100644 --- a/test/gtest/ucp/test_ucp_fault_tolerance.cc +++ b/test/gtest/ucp/test_ucp_fault_tolerance.cc @@ -932,6 +932,8 @@ class DISABLED_test_ucp_fault_tolerance_tf : public test_ucp_fault_tolerance { public: DISABLED_test_ucp_fault_tolerance_tf() { modify_config("FAILOVER_MODE", "token"); + /* Token recovery stays selected only while indirect atomics are off. */ + modify_config("IB_INDIRECT_ATOMIC", "n"); } }; diff --git a/test/gtest/ucp/test_ucp_wireup.cc b/test/gtest/ucp/test_ucp_wireup.cc index cda2f1e1991..ba770a16742 100644 --- a/test/gtest/ucp/test_ucp_wireup.cc +++ b/test/gtest/ucp/test_ucp_wireup.cc @@ -1615,6 +1615,28 @@ UCS_TEST_P(test_ucp_wireup_amo, relese_key_after_flush) { UCP_INSTANTIATE_TEST_CASE(test_ucp_wireup_amo) +static void test_ucp_wireup_failover_err_cb(void *arg, ucp_ep_h ep, + ucs_status_t status) +{ +} + +UCS_TEST_P(test_ucp_wireup_fallback_amo, failover_no_device_amo) +{ + ucp_ep_params_t params = get_ep_params(); + + params.field_mask |= UCP_EP_PARAM_FIELD_ERR_HANDLING_MODE | + UCP_EP_PARAM_FIELD_ERR_HANDLER; + params.err_mode = UCP_ERR_HANDLING_MODE_FAILOVER; + params.err_handler.cb = test_ucp_wireup_failover_err_cb; + params.err_handler.arg = NULL; + + create_entity(false); + create_entity(true); + sender().connect(&receiver(), params); + EXPECT_FALSE(use_device_amo(sender().ep())); + test_ucp_wireup::cleanup(); +} + UCS_TEST_P(test_ucp_wireup_fallback_amo, different_amo_types) { std::vector tls; From de87c843076e1e88df912394e614717fb74081a6 Mon Sep 17 00:00:00 2001 From: Evgeny Leksikov Date: Wed, 14 Oct 2026 15:49:05 +0300 Subject: [PATCH 2/5] TEST/UCP: Select token failover with indirect atomics off. IB_INDIRECT_ATOMIC is a memory-domain setting, so the fixture publishes UCX_IB_INDIRECT_ATOMIC. A plain modify_config aborts because the key is not a UCS global option, and token mode is rejected while the knob is on. --- test/gtest/ucp/test_ucp_fault_tolerance.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/gtest/ucp/test_ucp_fault_tolerance.cc b/test/gtest/ucp/test_ucp_fault_tolerance.cc index 71170f60140..8d5d6ac739e 100644 --- a/test/gtest/ucp/test_ucp_fault_tolerance.cc +++ b/test/gtest/ucp/test_ucp_fault_tolerance.cc @@ -932,8 +932,8 @@ class DISABLED_test_ucp_fault_tolerance_tf : public test_ucp_fault_tolerance { public: DISABLED_test_ucp_fault_tolerance_tf() { modify_config("FAILOVER_MODE", "token"); - /* Token recovery stays selected only while indirect atomics are off. */ - modify_config("IB_INDIRECT_ATOMIC", "n"); + /* IB MD key; set UCX_IB_INDIRECT_ATOMIC so token mode stays selected. */ + modify_config("IB_INDIRECT_ATOMIC", "n", SETENV_IF_NOT_EXIST); } }; From 7a7a32a04fc0e1b739d1d509f4d59d5fbf038d19 Mon Sep 17 00:00:00 2001 From: Evgeny Leksikov Date: Wed, 14 Oct 2026 17:24:13 +0300 Subject: [PATCH 3/5] UCP: Check indirect atomics only for a selected IB domain. Walk the excluded atomic lanes with the UCP transport bitmap iterator. --- src/ucp/core/ucp_context.c | 9 ++++++--- src/ucp/wireup/select.c | 2 +- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/src/ucp/core/ucp_context.c b/src/ucp/core/ucp_context.c index 91b0eed5a86..8e40119f2e8 100644 --- a/src/ucp/core/ucp_context.c +++ b/src/ucp/core/ucp_context.c @@ -2215,17 +2215,20 @@ static void ucp_fill_resources_reg_md_map_update(ucp_context_h context) } } -/* Effective UCX_IB_INDIRECT_ATOMIC: IB MD default and environment, then - * cached ucp_config_modify() overrides. */ +/* Effective UCX_IB_INDIRECT_ATOMIC on a selected IB memory domain. IB MD + * default and environment, then cached ucp_config_modify() overrides. + * Returns 0 when no IB domain was selected. */ static int ucp_context_query_ib_indirect_atomic(ucp_context_h context) { int enabled = 1; + ucp_md_index_t md_index; ucp_rsc_index_t cmpt_index; uct_md_config_t *md_config; ucs_status_t status; char value[8]; - for (cmpt_index = 0; cmpt_index < context->num_cmpts; ++cmpt_index) { + for (md_index = 0; md_index < context->num_mds; ++md_index) { + cmpt_index = context->tl_mds[md_index].cmpt_index; if (strcmp(context->tl_cmpts[cmpt_index].attr.name, "ib") != 0) { continue; } diff --git a/src/ucp/wireup/select.c b/src/ucp/wireup/select.c index fc404548e75..66a2fd41843 100644 --- a/src/ucp/wireup/select.c +++ b/src/ucp/wireup/select.c @@ -1339,7 +1339,7 @@ ucp_wireup_exclude_device_amo_tls(ucp_worker_h worker, { ucp_rsc_index_t rsc_index; - UCS_STATIC_BITMAP_FOR_EACH_BIT(rsc_index, tl_bitmap) { + UCP_TL_BITMAP_FOR_EACH_BIT(rsc_index, tl_bitmap) { if (!(ucp_worker_iface_get_attr(worker, rsc_index)->cap.flags & UCT_IFACE_FLAG_ATOMIC_DEVICE)) { continue; From b6788941b202ea88d47c4990cb40f9d409e28825 Mon Sep 17 00:00:00 2001 From: Evgeny Leksikov Date: Wed, 14 Oct 2026 17:41:46 +0300 Subject: [PATCH 4/5] UCP/RMA: Note mixed failover endpoints in the software atomic TODO. A failover endpoint sends software atomics while the same worker can still select device atomics for its other endpoints. --- src/ucp/rma/amo_sw.c | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/ucp/rma/amo_sw.c b/src/ucp/rma/amo_sw.c index 4e4cd433211..dfea01c57e4 100644 --- a/src/ucp/rma/amo_sw.c +++ b/src/ucp/rma/amo_sw.c @@ -239,10 +239,14 @@ UCS_PROFILE_FUNC(ucs_status_t, ucp_atomic_req_handler, (arg, data, length, am_fl UCT_IFACE_FLAG_ATOMIC_DEVICE))) { ucs_error("Unsupported: got software atomic request while device atomics are selected on worker %p", worker); - /* TODO: this situation will be possible then CM wireup is implemented - * and CM lane is bound to suboptimal device, then need to execute - * AMO on fastest resource from worker->atomic_tls using loopback - * EP and continue SW AMO protocol */ + /* TODO: This situation may occur when: + * - The CM lane is bound to a suboptimal device + * - An EP with error handling mode failover is used alongside + * other EPs + * In this case, the AMO should be executed on the fastest + * resource from worker->atomic_tls using the loopback EP, and + * the SW AMO protocol should continue. + */ } if (atomicreqh->req.req_id == UCS_PTR_MAP_KEY_INVALID) { From 421a0a630ff545c47557f7558fd8839c819bc0eb Mon Sep 17 00:00:00 2001 From: Evgeny Leksikov Date: Thu, 15 Oct 2026 01:13:08 +0300 Subject: [PATCH 5/5] UCP/RMA: Cover software atomics on a failover endpoint. Select the software atomic protocols under failover and post an atomic from the wireup test. --- src/ucp/rma/amo_sw.c | 7 +++-- test/gtest/ucp/test_ucp_wireup.cc | 51 +++++++++++++++++++++++++++++-- 2 files changed, 53 insertions(+), 5 deletions(-) diff --git a/src/ucp/rma/amo_sw.c b/src/ucp/rma/amo_sw.c index dfea01c57e4..41b765594d0 100644 --- a/src/ucp/rma/amo_sw.c +++ b/src/ucp/rma/amo_sw.c @@ -434,7 +434,8 @@ static void ucp_proto_amo_sw_probe(const ucp_proto_init_params_t *init_params, .super.send_op = UCT_EP_OP_AM_BCOPY, .super.memtype_op = UCT_EP_OP_GET_SHORT, .super.flags = flags | UCP_PROTO_COMMON_INIT_FLAG_SINGLE_FRAG | - UCP_PROTO_COMMON_INIT_FLAG_CAP_SEG_SIZE, + UCP_PROTO_COMMON_INIT_FLAG_CAP_SEG_SIZE | + UCP_PROTO_COMMON_INIT_FLAG_FAILOVER, .super.exclude_map = ~UCP_MAX_FAST_PATH_LANES_MASK, .super.reg_mem_info = ucp_mem_info_unknown, .lane_type = UCP_LANE_TYPE_AM, @@ -482,7 +483,7 @@ ucp_proto_t ucp_get_amo_post_proto = { .probe = ucp_proto_amo_sw_post_probe, .query = ucp_proto_single_query, .progress = {ucp_proto_amo_sw_progress_post}, - .abort = ucp_proto_abort_fatal_not_implemented, + .abort = ucp_proto_request_bcopy_abort, .reset = ucp_proto_request_bcopy_reset }; @@ -511,6 +512,6 @@ ucp_proto_t ucp_get_amo_fetch_proto = { .probe = ucp_proto_amo_sw_fetch_probe, .query = ucp_proto_single_query, .progress = {ucp_proto_amo_sw_progress_fetch}, - .abort = ucp_proto_abort_fatal_not_implemented, + .abort = ucp_proto_request_bcopy_id_abort, .reset = ucp_proto_request_bcopy_id_reset }; diff --git a/test/gtest/ucp/test_ucp_wireup.cc b/test/gtest/ucp/test_ucp_wireup.cc index ba770a16742..48c29777a01 100644 --- a/test/gtest/ucp/test_ucp_wireup.cc +++ b/test/gtest/ucp/test_ucp_wireup.cc @@ -88,6 +88,8 @@ class test_ucp_wireup : public ucp_test { bool ep_iface_has_caps(const entity& e, const std::string& tl, uint64_t caps); + static void unmap_memh(ucp_mem_h memh, ucp_context_h context); + protected: vec_type m_send_data; vec_type m_recv_data; @@ -98,8 +100,6 @@ class test_ucp_wireup : public ucp_test { private: static void stream_recv_completion(void *request, ucs_status_t status, size_t length); - - static void unmap_memh(ucp_mem_h memh, ucp_context_h context); }; void test_ucp_wireup::get_test_variants(std::vector& variants, @@ -1623,6 +1623,13 @@ static void test_ucp_wireup_failover_err_cb(void *arg, ucp_ep_h ep, UCS_TEST_P(test_ucp_wireup_fallback_amo, failover_no_device_amo) { ucp_ep_params_t params = get_ep_params(); + ucp_mem_map_params_t map_params; + ucp_request_param_t req_param; + ucp_mem_h memh; + ucp_rkey_h rkey; + void *req; + void *flush_req; + ucs_status_t status; params.field_mask |= UCP_EP_PARAM_FIELD_ERR_HANDLING_MODE | UCP_EP_PARAM_FIELD_ERR_HANDLER; @@ -1634,7 +1641,47 @@ UCS_TEST_P(test_ucp_wireup_fallback_amo, failover_no_device_amo) create_entity(true); sender().connect(&receiver(), params); EXPECT_FALSE(use_device_amo(sender().ep())); + + m_send_data[0] = 1; + clear_recv_data(); + + map_params.field_mask = UCP_MEM_MAP_PARAM_FIELD_ADDRESS | + UCP_MEM_MAP_PARAM_FIELD_LENGTH | + UCP_MEM_MAP_PARAM_FIELD_FLAGS; + map_params.address = &m_recv_data[0]; + map_params.length = m_recv_data.size() * sizeof(m_recv_data[0]); + map_params.flags = 0; + status = ucp_mem_map(receiver().ucph(), &map_params, &memh); + ASSERT_UCS_OK(status); + m_memh_receiver.reset(memh, unmap_memh, receiver().ucph()); + + rkey = get_rkey(sender().ep(), m_memh_receiver); + m_rkeys.push_back(ucs::handle(rkey, ucp_rkey_destroy)); + + { + scoped_log_handler slh(wrap_errors_logger); + + req_param.op_attr_mask = UCP_OP_ATTR_FIELD_DATATYPE; + req_param.datatype = ucp_dt_make_contig(sizeof(m_send_data[0])); + req = ucp_atomic_op_nbx(sender().ep(), + UCP_ATOMIC_OP_ADD, + &m_send_data[0], 1, + (uint64_t)&m_recv_data[0], + rkey, &req_param); + status = request_wait(req); + if (status == UCS_OK) { + /* Post completes locally; flush waits for the remote atomic. */ + flush_req = sender().flush_ep_nb(); + status = request_wait(flush_req); + } + } + + if (status == UCS_OK) { + EXPECT_EQ(m_send_data[0], m_recv_data[0]); + } + test_ucp_wireup::cleanup(); + EXPECT_EQ(UCS_OK, status); } UCS_TEST_P(test_ucp_wireup_fallback_amo, different_amo_types) {