UCP/WIREUP: Relax peer-failure criteria for same-worker endpoints

When an endpoint is wired to the same UCP worker (same unpacked address
UUID), there is no independent remote peer for cross-worker RMA. Skip
requiring UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE and
UCT_MD_FLAG_INVALIDATE_RMA in RMA BW wireup criteria in that case. In
ucp_request_get_invalidation_map(), return an empty invalidation map for
UCP_EP_CONFIG_KEY_FLAG_SELF so RMA BW lanes without MD invalidate
support (e.g. cuda_copy) remain valid for same-worker / memtype EPs with
peer-failure error handling. Extend ucp_wireup_fill_peer_err_criteria()
and ucp_wireup_fill_aux_criteria() with worker + unpacked address so
auxiliary and other lane selection paths apply the same rule.

Author: pentschev

src/ucp/core/ucp_request.c

@@ -379,6 +379,11 @@ 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-worker EPs (loopback, memtype): no cross-worker RMA to invalidate. */
+    if (key->flags & UCP_EP_CONFIG_KEY_FLAG_SELF) {
+        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/select.c

@@ -1042,9 +1042,18 @@ static double ucp_wireup_rma_score_func(const ucp_worker_iface_t *wiface,
 }
 
 static void ucp_wireup_fill_peer_err_criteria(ucp_wireup_criteria_t *criteria,
-                                              unsigned ep_init_flags)
+                                              unsigned ep_init_flags,
+                                              ucp_worker_h worker,
+                                              const ucp_unpacked_address_t *unpacked_addr)
 {
     if (ep_init_flags & UCP_EP_INIT_ERR_MODE_PEER_FAILURE) {
+        /* No independent remote worker when connecting an EP to itself (loopback,
+         * memtype EPs, etc.): peer-failure iface caps and MD invalidation are
+         * relaxed in wireup; see ucp_request_get_invalidation_map(). */
+        if ((unpacked_addr != NULL) && (unpacked_addr->uuid == worker->uuid)) {
+            return;
+        }
+
         criteria->local_iface_flags.mandatory |=
                 UCT_IFACE_FLAG_ERRHANDLE_PEER_FAILURE;
         /* transport selection procedure will check additionally for KA or EP check
@@ -1068,7 +1077,9 @@ static double ucp_wireup_aux_score_func(const ucp_worker_iface_t *wiface,
 
 static void ucp_wireup_fill_aux_criteria(ucp_wireup_criteria_t *criteria,
                                          unsigned ep_init_flags,
-                                         uint64_t mandatory_flags)
+                                         uint64_t mandatory_flags,
+                                         ucp_worker_h worker,
+                                         const ucp_unpacked_address_t *unpacked_addr)
 {
     criteria->title          = "auxiliary";
     criteria->local_md_flags = 0;
@@ -1092,7 +1103,8 @@ static void ucp_wireup_fill_aux_criteria(ucp_wireup_criteria_t *criteria,
     criteria->tl_rsc_flags       = UCP_TL_RSC_FLAG_AUX; /* Can use aux transports */
     criteria->lane_type          = UCP_LANE_TYPE_LAST;
 
-    ucp_wireup_fill_peer_err_criteria(criteria, ep_init_flags);
+    ucp_wireup_fill_peer_err_criteria(criteria, ep_init_flags, worker,
+                                      unpacked_addr);
 }
 
 static void ucp_wireup_criteria_init(ucp_wireup_criteria_t *criteria)
@@ -1185,7 +1197,9 @@ ucp_wireup_add_rma_lanes(const ucp_wireup_select_params_t *select_params,
                                      UCT_IFACE_FLAG_PENDING, 0);
     }
     criteria.calc_score             = ucp_wireup_rma_score_func;
-    ucp_wireup_fill_peer_err_criteria(&criteria, ep_init_flags);
+    ucp_wireup_fill_peer_err_criteria(&criteria, ep_init_flags,
+                                      select_params->ep->worker,
+                                      select_params->address);
 
     tl_bitmap = ucp_tl_bitmap_max;
     ucs_memory_type_for_each(mem_type) {
@@ -1239,7 +1253,9 @@ ucp_wireup_add_amo_lanes(const ucp_wireup_select_params_t *select_params,
     criteria.calc_score         = ucp_wireup_amo_score_func;
     ucp_wireup_init_select_flags(&criteria.local_iface_flags,
                                  UCT_IFACE_FLAG_PENDING, 0);
-    ucp_wireup_fill_peer_err_criteria(&criteria, ep_init_flags);
+    ucp_wireup_fill_peer_err_criteria(&criteria, ep_init_flags,
+                                      select_params->ep->worker,
+                                      select_params->address);
     ucp_context_uct_atomic_iface_flags(context, &criteria.remote_atomic_flags);
 
     /* We can use only non-p2p resources or resources which are explicitly
@@ -1487,9 +1503,10 @@ ucp_wireup_add_am_lane(const ucp_wireup_select_params_t *select_params,
                                      UCT_IFACE_FLAG_AM_BCOPY, 0);
         ucp_wireup_init_select_flags(&criteria.remote_iface_flags,
                                      UCP_ADDR_IFACE_FLAG_AM_SYNC, 0);
-        ucp_wireup_fill_peer_err_criteria(&criteria,
-                                          ucp_wireup_ep_init_flags(select_params,
-                                                                   select_ctx));
+        ucp_wireup_fill_peer_err_criteria(
+                &criteria,
+                ucp_wireup_ep_init_flags(select_params, select_ctx),
+                select_params->ep->worker, select_params->address);
 
         if (ucs_test_all_flags(ucp_ep_get_context_features(select_params->ep),
                                UCP_FEATURE_TAG | UCP_FEATURE_WAKEUP)) {
@@ -1925,7 +1942,8 @@ ucp_wireup_add_am_bw_lanes(const ucp_wireup_select_params_t *select_params,
                                  UCP_ADDR_IFACE_FLAG_AM_SYNC, 0);
     ucp_wireup_init_select_flags(&bw_info.criteria.local_iface_flags,
                                  UCT_IFACE_FLAG_AM_BCOPY, 0);
-    ucp_wireup_fill_peer_err_criteria(&bw_info.criteria, ep_init_flags);
+    ucp_wireup_fill_peer_err_criteria(&bw_info.criteria, ep_init_flags,
+                                      ep->worker, select_params->address);
 
     if (ucs_test_all_flags(ucp_ep_get_context_features(ep),
                            UCP_FEATURE_TAG | UCP_FEATURE_WAKEUP)) {
@@ -2090,7 +2108,8 @@ ucp_wireup_add_rma_bw_lanes(const ucp_wireup_select_params_t *select_params,
     bw_info.criteria.calc_score = ucp_wireup_rma_bw_score_func;
     ucp_wireup_init_select_flags(&bw_info.criteria.local_iface_flags,
                                  UCT_IFACE_FLAG_PENDING, 0);
-    ucp_wireup_fill_peer_err_criteria(&bw_info.criteria, ep_init_flags);
+    ucp_wireup_fill_peer_err_criteria(&bw_info.criteria, ep_init_flags,
+                                       ep->worker, select_params->address);
 
     if (ucs_test_all_flags(ucp_ep_get_context_features(ep),
                            UCP_FEATURE_TAG | UCP_FEATURE_WAKEUP)) {
@@ -2139,9 +2158,11 @@ ucp_wireup_add_rma_bw_lanes(const ucp_wireup_select_params_t *select_params,
 
     /* If error handling is requested and we have RNDV, we require memory
      * invalidation support to provide correct data integrity in case of error.
+     * Skip for endpoints wired to the same worker (loopback / memtype EPs).
      */
     if ((ep_init_flags & UCP_EP_INIT_ERR_MODE_PEER_FAILURE) &&
-        ucp_context_rndv_is_enabled(context)) {
+        ucp_context_rndv_is_enabled(context) &&
+        (select_params->address->uuid != ep->worker->uuid)) {
         bw_info.criteria.local_md_flags |= UCT_MD_FLAG_INVALIDATE_RMA;
     }
 
@@ -2290,7 +2311,8 @@ ucp_wireup_select_wireup_msg_lane(ucp_worker_h worker,
                                   const ucp_address_entry_t *address_list,
                                   const ucp_wireup_lane_desc_t *lane_descs,
                                   ucp_lane_index_t num_lanes,
-                                  ucp_lane_index_t am_lane)
+                                  ucp_lane_index_t am_lane,
+                                  const ucp_unpacked_address_t *unpacked_addr)
 {
     ucp_context_h context          = worker->context;
     ucp_lane_index_t p2p_lane      = UCP_NULL_LANE;
@@ -2307,7 +2329,8 @@ ucp_wireup_select_wireup_msg_lane(ucp_worker_h worker,
     }
 
     ucp_wireup_fill_aux_criteria(&criteria, ep_init_flags,
-                                 UCP_ADDR_IFACE_FLAG_CB_ASYNC);
+                                 UCP_ADDR_IFACE_FLAG_CB_ASYNC, worker,
+                                 unpacked_addr);
     for (lane = 0; lane < num_lanes; ++lane) {
         if (lane_descs[lane].rsc_index == UCP_NULL_RESOURCE) {
             continue;
@@ -2421,7 +2444,8 @@ ucp_wireup_add_keepalive_lane(const ucp_wireup_select_params_t *select_params,
     /* Keepalive can also use auxiliary transports */
     criteria.tl_rsc_flags       = UCP_TL_RSC_FLAG_AUX;
     criteria.lane_type          = UCP_LANE_TYPE_KEEPALIVE;
-    ucp_wireup_fill_peer_err_criteria(&criteria, ep_init_flags);
+    ucp_wireup_fill_peer_err_criteria(&criteria, ep_init_flags, worker,
+                                      select_params->address);
 
     status = ucp_wireup_select_transport(select_ctx, select_params, &criteria,
                                          *tl_bitmap, UINT64_MAX, UINT64_MAX,
@@ -2770,7 +2794,8 @@ ucp_wireup_construct_lanes(const ucp_wireup_select_params_t *select_params,
                                                                    select_ctx),
                                           select_params->address->address_list,
                                           select_ctx->lane_descs,
-                                          key->num_lanes, key->am_lane);
+                                          key->num_lanes, key->am_lane,
+                                          select_params->address);
     }
 
     for (i = 0; key->rma_bw_lanes[i] != UCP_NULL_LANE; i++) {
@@ -2875,7 +2900,8 @@ ucp_wireup_select_aux_transport(ucp_ep_h ep, unsigned ep_init_flags,
 
     /* Select auxiliary transport that supports async active message callback */
     ucp_wireup_fill_aux_criteria(&criteria, ep_init_flags,
-                                 UCP_ADDR_IFACE_FLAG_CB_ASYNC);
+                                 UCP_ADDR_IFACE_FLAG_CB_ASYNC, ep->worker,
+                                 remote_address);
     status = ucp_wireup_select_transport(&select_ctx, &select_params, &criteria,
                                          ucp_tl_bitmap_max, UINT64_MAX,
                                          UINT64_MAX, UINT64_MAX, 0,
@@ -2886,7 +2912,8 @@ ucp_wireup_select_aux_transport(ucp_ep_h ep, unsigned ep_init_flags,
 
     /* Fallback to an auxiliary transport without async active message callback
      * requirement */
-    ucp_wireup_fill_aux_criteria(&criteria, ep_init_flags, 0);
+    ucp_wireup_fill_aux_criteria(&criteria, ep_init_flags, 0, ep->worker,
+                                 remote_address);
     return ucp_wireup_select_transport(&select_ctx, &select_params, &criteria,
                                        ucp_tl_bitmap_max, UINT64_MAX,
                                        UINT64_MAX, UINT64_MAX, 1, select_info);