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

src/ucp/core/ucp_ep.c

@@ -997,7 +997,7 @@ ucp_sa_data_v1_unpack(const ucp_wireup_sockaddr_data_base_t *sa_data,
         return UCS_ERR_UNSUPPORTED;
     }
 
-    *ep_init_flags_p = ucp_ep_err_mode_init_flags(sa_data->header);
+    *ep_init_flags_p = ucp_ep_err_mode_init_flags(sa_data->header, 0);
     *worker_addr_p   = sa_data_v1 + 1;
     return UCS_OK;
 }
@@ -1973,7 +1973,9 @@ ucp_ep_recovery_create_aux(ucp_ep_h ep, ucp_lane_index_t lane,
     ucp_context_h context    = worker->context;
     ucp_rsc_index_t lane_rsc = ucp_ep_get_rsc_index(ep, lane);
     unsigned ep_init_flags   =
-            ucp_ep_err_mode_init_flags(ucp_ep_config(ep)->key.err_mode) |
+            ucp_ep_err_mode_init_flags(
+                    ucp_ep_config(ep)->key.err_mode,
+                    remote_address->uuid == worker->uuid) |
             UCP_EP_INIT_RECOVERY;
     ucp_wireup_select_info_t select_info;
     const ucp_address_entry_t *peer_ae;
@@ -4796,13 +4798,14 @@ void ucp_ep_set_cfg_index(ucp_ep_h ep, ucp_worker_cfg_index_t cfg_index,
     ucp_ep_config_proto_init(ep->worker, cfg_index);
 }
 
-unsigned ucp_ep_err_mode_init_flags(ucp_err_handling_mode_t err_mode)
+unsigned ucp_ep_err_mode_init_flags(ucp_err_handling_mode_t err_mode,
+                                    int is_self)
 {
     switch (err_mode) {
     case UCP_ERR_HANDLING_MODE_NONE:
         return 0;
     case UCP_ERR_HANDLING_MODE_PEER:
-        return UCP_EP_INIT_ERR_MODE_PEER_FAILURE;
+        return is_self ? 0 : UCP_EP_INIT_ERR_MODE_PEER_FAILURE;
     case UCP_ERR_HANDLING_MODE_FAILOVER:
         return UCP_EP_INIT_ERR_MODE_FAILOVER_MASK;
     default:

src/ucp/core/ucp_request.c

@@ -385,6 +385,12 @@ static ucp_md_map_t ucp_request_get_invalidation_map(ucp_ep_h ep)
     ucp_lane_index_t i;
     ucp_md_map_t inv_map;
 
+    /* Same-process PEER EPs do not require cross-process RMA invalidation. */
+    if ((key->flags & UCP_EP_CONFIG_KEY_FLAG_SELF) &&
+        (key->err_mode == UCP_ERR_HANDLING_MODE_PEER)) {
+        return 0;
+    }
+
     for (i = 0, inv_map = 0;
          (key->rma_bw_lanes[i] != UCP_NULL_LANE) && (i < UCP_MAX_LANES); i++) {
         lane = key->rma_bw_lanes[i];

src/ucp/wireup/wireup.c

@@ -624,7 +624,7 @@ ucp_wireup_process_pre_request(ucp_worker_h worker, ucp_ep_h ep,
 {
     unsigned ep_init_flags = UCP_EP_INIT_CREATE_AM_LANE |
                              UCP_EP_INIT_CM_WIREUP_CLIENT |
-                             ucp_ep_err_mode_init_flags(msg->err_mode);
+                             ucp_ep_err_mode_init_flags(msg->err_mode, 0);
     unsigned addr_indices[UCP_MAX_LANES];
     int am_need_flush;
     ucs_status_t status;
@@ -670,7 +670,7 @@ ucp_wireup_process_request(ucp_worker_h worker, ucp_ep_h ep,
 {
     uint64_t remote_uuid      = remote_address->uuid;
     int send_reply            = 0;
-    unsigned ep_init_flags    = ucp_ep_err_mode_init_flags(msg->err_mode);
+    unsigned ep_init_flags    = ucp_ep_err_mode_init_flags(msg->err_mode, 0);
     ucp_tl_bitmap_t tl_bitmap = UCS_STATIC_BITMAP_ZERO_INITIALIZER;
     ucp_lane_index_t lanes2remote[UCP_MAX_LANES];
     unsigned addr_indices[UCP_MAX_LANES];
@@ -1976,6 +1976,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);
@@ -1984,7 +1985,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,
+    /* Preserve the requested error mode in the endpoint configuration, but do
+     * not impose peer-failure transport requirements on a same-worker EP. */
+    select_ep_init_flags =
+            (ep_init_flags & ~UCP_EP_INIT_ERR_MODE_FAILOVER_MASK) |
+            ucp_ep_err_mode_init_flags(
+                    key->err_mode,
+                    remote_address->uuid == ep->worker->uuid);
+    status = ucp_wireup_select_lanes(ep, select_ep_init_flags, *tl_bitmap,
                                      remote_address, addr_indices, key, 1);
     if (status != UCS_OK) {
         return status;
@@ -2471,7 +2479,8 @@ unsigned ucp_ep_init_flags(const ucp_worker_h worker,
     }
 
     return flags |
-           ucp_ep_err_mode_init_flags(ucp_ep_params_err_handling_mode(params));
+           ucp_ep_err_mode_init_flags(
+                   ucp_ep_params_err_handling_mode(params), 0);
 }
 
 double ucp_wireup_iface_lat_distance_v1(const ucp_worker_iface_t *wiface)

src/ucp/wireup/wireup.h

@@ -240,10 +240,12 @@ unsigned ucp_ep_init_flags(const ucp_worker_h worker,
  * @brief Convert error handling mode to endpoint initialization flags.
  *
  * @param [in] err_mode Error handling mode.
+ * @param [in] is_self  Whether the endpoint connects to the same worker.
  *
  * @return Endpoint initialization flags.
  */
- unsigned ucp_ep_err_mode_init_flags(ucp_err_handling_mode_t err_mode);
+unsigned ucp_ep_err_mode_init_flags(ucp_err_handling_mode_t err_mode,
+                                    int is_self);
 
 
 int ucp_wireup_connect_p2p(ucp_worker_h worker, ucp_rsc_index_t rsc_index,

test/gtest/ucp/test_ucp_peer_failure.cc

@@ -540,7 +540,9 @@ void test_ucp_peer_failure_inval::test_invalidate_rma_bw_lane(
     }
 
     if (force_invalidate_rma) {
-        ASSERT_FALSE(has_no_invalidate_rma);
+        if (!(config->key.flags & UCP_EP_CONFIG_KEY_FLAG_SELF)) {
+            EXPECT_FALSE(has_no_invalidate_rma);
+        }
     } else {
         ASSERT_TRUE(!has_transport("tcp") || has_no_invalidate_rma);
         ASSERT_TRUE(!has_transport("rc_x") || has_invalidate_rma);

test/gtest/ucp/test_ucp_wireup.cc

@@ -24,6 +24,21 @@ extern "C" {
 #include <ucs/sys/math.h>
 }
 
+class test_ucp_wireup_err_mode : public ucs::test {
+};
+
+UCS_TEST_F(test_ucp_wireup_err_mode, init_flags_self)
+{
+    EXPECT_EQ(UCP_EP_INIT_ERR_MODE_PEER_FAILURE,
+              ucp_ep_err_mode_init_flags(UCP_ERR_HANDLING_MODE_PEER, 0));
+    EXPECT_EQ(0,
+              ucp_ep_err_mode_init_flags(UCP_ERR_HANDLING_MODE_PEER, 1));
+    EXPECT_EQ(UCP_EP_INIT_ERR_MODE_FAILOVER_MASK,
+              ucp_ep_err_mode_init_flags(UCP_ERR_HANDLING_MODE_FAILOVER, 0));
+    EXPECT_EQ(UCP_EP_INIT_ERR_MODE_FAILOVER_MASK,
+              ucp_ep_err_mode_init_flags(UCP_ERR_HANDLING_MODE_FAILOVER, 1));
+}
+
 class test_ucp_wireup : public ucp_test {
 public:
     static void get_test_variants(std::vector<ucp_test_variant>& variants,
@@ -961,6 +976,28 @@ 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)
+{
+    EXPECT_FALSE(ep_iface_has_caps(sender(), "self",
+                                   UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE));
+
+    sender().connect(&receiver(), get_ep_params());
+
+    const ucp_ep_config_key_t &key = ucp_ep_config(sender().ep())->key;
+    EXPECT_EQ(UCP_ERR_HANDLING_MODE_PEER, key.err_mode);
+    EXPECT_TRUE(key.flags & UCP_EP_CONFIG_KEY_FLAG_SELF);
+}
+
+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());