UCT/CUDA/COPY: Qualify cuda_copy for RNDV with peer-failure EH

src/ucp/core/ucp_request.c

@@ -390,6 +390,15 @@ static ucp_md_map_t ucp_request_get_invalidation_map(ucp_ep_h ep)
         lane = key->rma_bw_lanes[i];
 
         if (!ucp_ep_is_lane_p2p(ep, lane)) {
+            /* Same-worker PEER selection may include MDs without RMA
+             * invalidation, while capable MDs still need it. */
+            if ((key->flags & UCP_EP_CONFIG_KEY_FLAG_SELF) &&
+                (key->err_mode == UCP_ERR_HANDLING_MODE_PEER) &&
+                !(ucp_ep_md_attr(ep, lane)->flags &
+                  UCT_MD_FLAG_INVALIDATE_RMA)) {
+                continue;
+            }
+
             ucs_assert(ucp_ep_get_iface_attr(ep, lane)->cap.flags &
                        UCT_IFACE_FLAG_GET_ZCOPY);
             ucs_assert(ucp_ep_md_attr(ep, lane)->flags &

src/ucp/wireup/wireup.c

@@ -2200,6 +2200,7 @@ ucp_wireup_try_select_lanes(ucp_ep_h ep, unsigned ep_init_flags,
                             unsigned *addr_indices, ucp_ep_config_key_t *key,
                             ucp_rsc_index_t *dst_md_storage)
 {
+    unsigned select_ep_init_flags;
     ucs_status_t status;
 
     ucp_ep_config_key_reset(key);
@@ -2208,7 +2209,14 @@ ucp_wireup_try_select_lanes(ucp_ep_h ep, unsigned ep_init_flags,
     key->dst_version  = remote_address->dst_version;
     key->dst_md_cmpts = dst_md_storage;
 
-    status = ucp_wireup_select_lanes(ep, ep_init_flags, *tl_bitmap,
+    select_ep_init_flags = ep_init_flags;
+    if ((key->err_mode == UCP_ERR_HANDLING_MODE_PEER) &&
+        (remote_address->uuid == ep->worker->uuid)) {
+        /* Same-worker PEER EPs do not require peer-failure transports. */
+        select_ep_init_flags &= ~UCP_EP_INIT_ERR_MODE_PEER_FAILURE;
+    }
+
+    status = ucp_wireup_select_lanes(ep, select_ep_init_flags, *tl_bitmap,
                                      remote_address, addr_indices, key, 1);
     if (status != UCS_OK) {
         return status;

test/gtest/ucp/test_ucp_peer_failure.cc

@@ -513,7 +513,8 @@ class test_ucp_peer_failure_inval : public ucp_test {
 
     void init()
     {
-        create_entity();
+        /* Use a different worker to keep the invalidation requirement tested. */
+        ucp_test::init();
         sender().connect(&receiver(), get_ep_params(), 0);
     }
 };

test/gtest/ucp/test_ucp_wireup.cc

@@ -961,6 +961,52 @@ class test_ucp_wireup_errh_peer : public test_ucp_wireup_1sided
     static void err_cb(void *, ucp_ep_h, ucs_status_t) {}
 };
 
+class test_ucp_wireup_errh_peer_self : public test_ucp_wireup_errh_peer
+{
+public:
+    void init() override {
+        test_ucp_wireup::init();
+    }
+};
+
+UCS_TEST_P(test_ucp_wireup_errh_peer_self, config)
+{
+    ucp_ep_params_t ep_params = get_ep_params();
+    ucp_address_t *address;
+    ucp_ep_h ep;
+    size_t address_length;
+    ucs_status_t status;
+
+    EXPECT_FALSE(ep_iface_has_caps(sender(), "self",
+                                   UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE));
+
+    status = ucp_worker_get_address(receiver().worker(), &address,
+                                    &address_length);
+    ASSERT_UCS_OK(status);
+
+    ep_params.field_mask |= UCP_EP_PARAM_FIELD_REMOTE_ADDRESS;
+    ep_params.address     = address;
+    {
+        scoped_log_handler slh(hide_errors_logger);
+        status = ucp_ep_create(sender().worker(), &ep_params, &ep);
+    }
+
+    ucp_worker_release_address(receiver().worker(), address);
+    ASSERT_UCS_OK(status);
+
+    const ucp_ep_config_key_t &key = ucp_ep_config(ep)->key;
+    EXPECT_EQ(UCP_ERR_HANDLING_MODE_PEER, key.err_mode);
+    EXPECT_TRUE(key.flags & UCP_EP_CONFIG_KEY_FLAG_SELF);
+    EXPECT_GT(key.num_lanes, 0);
+    for (ucp_lane_index_t lane = 0; lane < key.num_lanes; ++lane) {
+        EXPECT_STREQ("self", ucp_ep_get_tl_rsc(ep, lane)->tl_name);
+    }
+
+    disconnect(ep);
+}
+
+UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_wireup_errh_peer_self, self, "self")
+
 UCS_TEST_P(test_ucp_wireup_errh_peer, msg_after_ep_create) {
     receiver().connect(&sender(), get_ep_params());