Merge branch 'master' into dscp-prio

Author: ybenvidia

src/ucp/core/ucp_context.c

@@ -461,7 +461,7 @@ static ucs_config_field_t ucp_context_config_table[] = {
    ucs_offsetof(ucp_context_config_t, dynamic_tl_progress_factor),
    UCS_CONFIG_TYPE_TIME_UNITS},
 
-  {"RESOLVE_REMOTE_EP_ID", "n",
+  {"RESOLVE_REMOTE_EP_ID", "auto",
    "Defines whether resolving remote endpoint ID is required or not when\n"
    "creating a local endpoint. 'auto' means resolving remote endpoint ID only\n"
    "in case of error handling and keepalive enabled.",

src/ucp/core/ucp_ep.c

@@ -1826,13 +1826,250 @@ ucs_status_t ucp_ep_reconfig_clear_failed_lanes(ucp_ep_h ep,
     return UCS_OK;
 }
 
-/* Prepare failed lanes for recovery; return the subset to pack into
- * LANES_ADDR_REQUEST. */
+/* Install an empty wireup proxy on the lane, replacing any failed-stub UCT
+ * EP left by a preceding ucp_ep_failover_reconfig(). The inner UCT EP is
+ * attached separately by ucp_ep_recovery_set_next_ep(). */
+static ucs_status_t
+ucp_ep_recovery_install_wireup_ep(ucp_ep_h ep, ucp_lane_index_t lane)
+{
+    uct_ep_h old_uct_ep = ucp_ep_get_lane(ep, lane);
+    uct_ep_h wireup_ep_uct;
+    ucs_status_t status;
+
+    ucs_assert(old_uct_ep != NULL);
+    if (ucp_wireup_ep_test(old_uct_ep)) {
+        return UCS_OK;
+    }
+
+    if (!ucp_is_uct_ep_failed(old_uct_ep)) {
+        /* The lane has been already recovered */
+        return UCS_ERR_NO_PROGRESS;
+    }
+
+    status = ucp_wireup_ep_create(ep, &wireup_ep_uct);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    ucs_trace("ep %p: recovery lane[%d] %p -> wireup_ep %p", ep, lane,
+              old_uct_ep, wireup_ep_uct);
+    ucp_ep_set_lane(ep, lane, wireup_ep_uct);
+    uct_ep_destroy(old_uct_ep);
+    return UCS_OK;
+}
+
+/* Idempotently ensure the lane's wireup proxy has an inner transport EP. */
+static ucs_status_t
+ucp_ep_recovery_set_next_ep(ucp_ep_h ep, ucp_lane_index_t lane,
+                            const ucp_address_entry_t *ae)
+{
+    const ucp_ep_config_key_t *cfg_key = &ucp_ep_config(ep)->key;
+    const ucp_rsc_index_t rsc_index    = cfg_key->lanes[lane].rsc_index;
+    const unsigned path_index          = cfg_key->lanes[lane].path_index;
+    uct_ep_h proxy                     = ucp_ep_get_lane(ep, lane);
+    ucp_wireup_ep_t *wireup_ep         = ucp_wireup_ep(proxy);
+    uct_ep_h next_ep;
+    ucs_status_t status;
+
+    ucs_assertv(wireup_ep != NULL, "ep %p lane %d: expected wireup proxy",
+                ep, lane);
+
+    if (wireup_ep->super.uct_ep != NULL) {
+        return UCS_OK;
+    }
+
+    if (ae == NULL) {
+        ucs_assert(ucp_ep_is_lane_p2p(ep, lane));
+        return ucp_wireup_ep_connect(proxy, 0, rsc_index, path_index, 0, NULL);
+    }
+
+    status = ucp_wireup_iface_ep_create(ucp_worker_iface(ep->worker, rsc_index),
+                                        ae, path_index, &next_ep);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    ucp_wireup_ep_set_next_ep(proxy, next_ep, rsc_index);
+    return UCS_OK;
+}
+
+/**
+ * @brief Find a CONNECT_TO_IFACE address entry for the specified lane.
+ *
+ * Returns the reachable entry whose memory domain index and system device
+ * match the lane's recorded destination, to keep the rebuilt lane bound to
+ * the same peer device. Returns NULL if no such entry exists.
+ */
+static const ucp_address_entry_t *
+ucp_ep_recovery_find_iface_addr(ucp_ep_h ep, ucp_lane_index_t lane,
+                                const ucp_unpacked_address_t *remote_address)
+{
+    const ucp_ep_config_key_lane_t *cfg_lane =
+            &ucp_ep_config(ep)->key.lanes[lane];
+    const ucp_rsc_index_t rsc_index          = cfg_lane->rsc_index;
+    const ucp_address_entry_t *ae;
+
+    ucp_unpacked_address_for_each(ae, remote_address) {
+        if ((ae->iface_addr != NULL) &&
+            (ae->md_index == cfg_lane->dst_md_index) &&
+            (ae->sys_dev  == cfg_lane->dst_sys_dev) &&
+            ucp_wireup_is_reachable(ep, 0, rsc_index, ae, NULL, 0)) {
+            return ae;
+        }
+    }
+
+    return NULL;
+}
+
+/* Rebuild one CONNECT_TO_IFACE lane: find peer iface addr -> install empty
+ * wireup proxy -> create fully-connected inner UCT EP and attach it -> mark
+ * ready. */
+static ucs_status_t
+ucp_ep_recovery_rebuild_iface_lane(
+        ucp_ep_h ep, ucp_lane_index_t lane,
+        const ucp_unpacked_address_t *remote_address)
+{
+    const ucp_address_entry_t *ae;
+    ucs_status_t status;
+
+    ae = ucp_ep_recovery_find_iface_addr(ep, lane, remote_address);
+    if (ae == NULL) {
+        ucs_debug("ep %p: no remote iface address for lane %d", ep, lane);
+        return UCS_ERR_UNREACHABLE;
+    }
+
+    status = ucp_ep_recovery_install_wireup_ep(ep, lane);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    status = ucp_ep_recovery_set_next_ep(ep, lane, ae);
+    if (status != UCS_OK) {
+        ucs_diag("ep %p: set_next_ep failed for recovery iface lane %d: %s",
+                 ep, lane, ucs_status_string(status));
+        return status;
+    }
+
+    ucp_wireup_update_flags(ep, UCS_BIT(lane),
+                            UCP_WIREUP_EP_FLAG_READY |
+                            UCP_WIREUP_EP_FLAG_REMOTE_CONNECTED);
+    ucs_debug("ep %p: recovered iface-lane[%d] via rsc[%d]", ep, lane,
+              ucp_ep_get_rsc_index(ep, lane));
+    return UCS_OK;
+}
+
+/* Rebuild one p2p (CONNECT_TO_EP) lane. Flow: find peer ep_addr ->
+ * install empty wireup proxy -> ensure iface-only inner UCT EP ->
+ * connect_to_ep_v2 against peer ep_addr -> mark ready. */
+static ucs_status_t
+ucp_ep_recovery_rebuild_p2p_lane(
+        ucp_ep_h ep, ucp_lane_index_t lane,
+        const ucp_unpacked_address_t *remote_address)
+{
+    const ucp_address_entry_t *address_entry;
+    const ucp_address_entry_ep_addr_t *ep_entry;
+    ucp_lane_index_t remote_lane;
+    ucs_status_t status;
+
+    /* Symmetric assumption: the peer's lane index mirrors ours.
+     * ucp_address_pack() with lanes2remote==NULL stamps the sender-side local
+     * lane as the remote_lane tag in the ep_addr, so we look up by our own
+     * lane index. */
+    remote_lane = lane;
+    status = ucp_wireup_find_remote_p2p_addr(ep, remote_lane, remote_address,
+                                             &address_entry, &ep_entry);
+    if (status != UCS_OK) {
+        ucs_debug("ep %p: no remote ep_addr for p2p lane %d", ep, lane);
+        return status;
+    }
+
+    status = ucp_ep_recovery_install_wireup_ep(ep, lane);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    status = ucp_ep_recovery_set_next_ep(ep, lane, NULL);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    status = ucp_wireup_ep_connect_to_ep_v2(ucp_ep_get_lane(ep, lane),
+                                            address_entry, ep_entry);
+    if (status != UCS_OK) {
+        ucs_diag("ep %p: connect_to_ep_v2 failed for recovery p2p lane %d: %s",
+                 ep, lane, ucs_status_string(status));
+        return status;
+    }
+
+    ucp_wireup_update_flags(ep, UCS_BIT(lane),
+                            UCP_WIREUP_EP_FLAG_READY |
+                            UCP_WIREUP_EP_FLAG_REMOTE_CONNECTED);
+    ucs_debug("ep %p: recovered p2p-lane[%d] via rsc[%d]", ep, lane,
+              ucp_ep_get_rsc_index(ep, lane));
+    return UCS_OK;
+}
+
+/* For each lane in `lanes_to_rebuild`, bring its UCT resources back online.
+ * Dispatches to the iface-connect or p2p helper based on lane type. Returns
+ * the bitmap of lanes successfully rebuilt; lanes not in the returned bitmap
+ * stay UCP_LANE_TYPE_FAILED and will be retried on a future recovery round. */
+ucp_lane_map_t
+ucp_ep_recovery_rebuild_lanes(ucp_ep_h ep, ucp_lane_map_t lanes_to_rebuild,
+                              const ucp_unpacked_address_t *remote_address)
+{
+    ucp_lane_map_t rebuilt = 0;
+    ucp_lane_index_t lane;
+    ucs_status_t status;
+
+    ucs_for_each_bit(lane, lanes_to_rebuild) {
+        if (ucp_ep_is_lane_p2p(ep, lane)) {
+            status = ucp_ep_recovery_rebuild_p2p_lane(ep, lane,
+                                                      remote_address);
+        } else {
+            status = ucp_ep_recovery_rebuild_iface_lane(ep, lane,
+                                                        remote_address);
+        }
+
+        if (status == UCS_OK) {
+            rebuilt |= UCS_BIT(lane);
+        }
+    }
+
+    return rebuilt;
+}
+
+/* Replace failed UCT stubs with wireup proxies for the given lane set and,
+ * for p2p lanes, create their iface-only inner transport EP so that the
+ * local ep_addr is available to ucp_address_pack() at REQUEST-send time.
+ * Returns the bitmap of lanes that ended up with the full proxy state
+ * required to appear in LANES_ADDR_REQUEST. Lanes that failed to reach
+ * that state are excluded so they don't get packed (which would try to
+ * read addresses from a half-built proxy). */
 static ucp_lane_map_t
 ucp_ep_recovery_prepare_lanes(ucp_ep_h ep, ucp_lane_map_t lanes)
 {
-    /* TODO: lane rebuild is not implemented yet. */
-    return 0;
+    ucp_lane_map_t lane_map            = 0;
+    ucp_lane_index_t lane;
+
+    ucs_for_each_bit(lane, lanes) {
+        if (ucp_ep_recovery_install_wireup_ep(ep, lane) != UCS_OK) {
+            continue;
+        }
+
+        /* For p2p lanes the ep_addr has to exist before we pack the
+         * REQUEST (ucp_address_pack iterates p2p_lanes and calls
+         * uct_ep_get_address on each). On failure the lane stays with an
+         * empty proxy and is skipped from this round; the next recovery
+         * round will retry. */
+        if (ucp_ep_is_lane_p2p(ep, lane) &&
+            (ucp_ep_recovery_set_next_ep(ep, lane, NULL) != UCS_OK)) {
+            continue;
+        }
+
+        lane_map |= UCS_BIT(lane);
+    }
+
+    return lane_map;
 }
 
 /* Send a LANES_ADDR_REQUEST for the currently failed lanes. */
@@ -1957,6 +2194,11 @@ int ucp_ep_recovery_progress(ucp_ep_h ep)
 
     failed &= ~recovered;
     if (failed == 0) {
+        /* All failed lanes recovered - drop the retry state so a later
+         * round (failed == 0) does not trip the recovery_arg == NULL
+         * assertion. */
+        ucs_free(ep->ext->recovery_arg);
+        ep->ext->recovery_arg = NULL;
         goto done;
     }
 

src/ucp/core/ucp_ep.h

@@ -1017,6 +1017,15 @@ ucs_status_t ucp_ep_recovery_arm(ucp_ep_h ep);
 int ucp_ep_recovery_progress(ucp_ep_h ep);
 
 
+/**
+ * Rebuild UCT endpoints for the given set of currently-failed lanes using the
+ * peer addresses in @a remote_address. Returns the bitmap of lanes rebuilt.
+ */
+ucp_lane_map_t
+ucp_ep_recovery_rebuild_lanes(ucp_ep_h ep, ucp_lane_map_t lanes_to_rebuild,
+                              const ucp_unpacked_address_t *remote_address);
+
+
 /**
  * Mark a subset of lanes as UCP_LANE_TYPE_FAILED and arm recovery.
  */

src/ucp/proto/proto_multi.c

@@ -160,8 +160,6 @@ static ucp_sys_dev_map_t ucp_proto_multi_init_flush_sys_dev_mask(
     return UCS_BIT(key->sys_dev);
 }
 
-/* Pack a device's PCI bus id (BDF) into a comparable key. Devices without a
- * bus id sort last, ordered by sys_dev to keep a deterministic total order. */
 static ucs_bus_id_bit_rep_t
 ucp_proto_multi_sys_dev_bus_id_key(ucs_sys_device_t sys_dev)
 {
@@ -177,14 +175,22 @@ ucp_proto_multi_sys_dev_bus_id_key(ucs_sys_device_t sys_dev)
 static int ucp_proto_multi_sys_dev_cmp(const void *pa, const void *pb,
                                        void *UCS_V_UNUSED arg)
 {
-    const ucs_sys_device_t a   = *(const ucs_sys_device_t*)pa;
-    const ucs_sys_device_t b   = *(const ucs_sys_device_t*)pb;
-    ucs_bus_id_bit_rep_t key_a = ucp_proto_multi_sys_dev_bus_id_key(a);
-    ucs_bus_id_bit_rep_t key_b = ucp_proto_multi_sys_dev_bus_id_key(b);
-
-    /* ascending order by PCI bus id so every rank on the node observes the
-     * same ordering regardless of local device discovery order */
-    return (key_a > key_b) - (key_a < key_b);
+    const ucs_sys_device_t a    = *(const ucs_sys_device_t*)pa;
+    const ucs_sys_device_t b    = *(const ucs_sys_device_t*)pb;
+    ucs_bus_id_bit_rep_t key_a  = ucp_proto_multi_sys_dev_bus_id_key(a);
+    ucs_bus_id_bit_rep_t key_b  = ucp_proto_multi_sys_dev_bus_id_key(b);
+    uintptr_t user_value_a, user_value_b;
+
+    /* Sort by topology identity so every rank on the node observes the same
+     * ordering regardless of local device discovery order. */
+    if (key_a != key_b) {
+        return (key_a > key_b) - (key_a < key_b);
+    }
+
+    user_value_a = ucs_topo_sys_device_get_user_value(a);
+    user_value_b = ucs_topo_sys_device_get_user_value(b);
+
+    return (user_value_a > user_value_b) - (user_value_a < user_value_b);
 }
 
 static UCS_F_ALWAYS_INLINE ucs_sys_dev_distance_t

src/ucp/rndv/proto_rndv.c

@@ -140,6 +140,44 @@ ucp_proto_rndv_md_map_to_remote(const ucp_proto_rndv_ctrl_init_params_t *params,
     return remote_md_map;
 }
 
+/*
+ * Estimate remote buffer flags from the packed rkey MD map. This is used only
+ * for protocol estimation and assumes local and remote MDs have the same
+ * required memory flags.
+ */
+static uint8_t
+ucp_proto_rndv_rkey_mem_flags_estimate(const ucp_proto_init_params_t *params)
+{
+    ucp_context_h context = params->worker->context;
+    const ucp_rkey_config_key_t *rkey_config_key = params->rkey_config_key;
+    const ucp_ep_config_key_lane_t *lane_cfg;
+    ucp_md_index_t md_index;
+    uint8_t mem_flags = 0;
+
+    ucs_assert(rkey_config_key != NULL);
+
+    ucs_carray_for_each(lane_cfg, params->ep_config_key->lanes,
+                        params->ep_config_key->num_lanes) {
+        if ((lane_cfg->rsc_index == UCP_NULL_RESOURCE) ||
+            (lane_cfg->dst_md_index == UCP_NULL_RESOURCE)) {
+            continue;
+        }
+
+        if (!(rkey_config_key->md_map & UCS_BIT(lane_cfg->dst_md_index))) {
+            continue;
+        }
+
+        /*
+         * Derive UCS_MEM_FLAG_REGISTRABLE from matching local MDs which
+         * require it and whose remote MDs are present in the rkey.
+         */
+        md_index   = context->tl_rscs[lane_cfg->rsc_index].md_index;
+        mem_flags |= context->tl_mds[md_index].attr.required_mem_flags;
+    }
+
+    return mem_flags;
+}
+
 /*
  * Init protocols that can be used by the remote peer
  * (assuming peer has the same configuration)
@@ -297,7 +335,7 @@ static ucp_proto_select_param_t ucp_proto_rndv_remote_select_param_init(
          */
         mem_info.sys_dev = init_params->rkey_config_key->sys_dev;
         mem_info.type    = init_params->rkey_config_key->mem_type;
-        mem_info.flags   = UCS_MEM_FLAG_REGISTRABLE;
+        mem_info.flags   = ucp_proto_rndv_rkey_mem_flags_estimate(init_params);
         ucp_proto_select_param_init(&remote_select_param, params->remote_op_id,
                                     op_attr_mask, 0, UCP_DATATYPE_CONTIG,
                                     &mem_info, 1);