UCP/WIREUP: Implement common copying of lanes function for futher re-use

src/ucp/wireup/wireup.c

@@ -701,12 +701,12 @@ void ucp_wireup_assign_lane(ucp_ep_h ep, ucp_lane_index_t lane, uct_ep_h uct_ep,
         ucs_assert(ucp_wireup_ep_test(ep->uct_eps[lane]));
         ucs_trace("ep %p: wireup uct_ep[%d]=%p next set to %p%s", ep, lane,
                   ep->uct_eps[lane], uct_ep, info);
-        ucp_wireup_ep_set_next_ep(ep->uct_eps[lane], uct_ep);
+        ucp_wireup_ep_set_next_ep(ep->uct_eps[lane], uct_ep, 1, 1);
         ucp_wireup_ep_remote_connected(ep->uct_eps[lane]);
     }
 }
 
-static uct_ep_h ucp_wireup_extract_lane(ucp_ep_h ep, ucp_lane_index_t lane)
+uct_ep_h ucp_wireup_extract_lane(ucp_ep_h ep, ucp_lane_index_t lane)
 {
     uct_ep_h uct_ep = ep->uct_eps[lane];
 
@@ -868,6 +868,10 @@ ucs_status_t ucp_wireup_resolve_proxy_lanes(ucp_ep_h ep)
                               iface_attr->cap.am.max_bcopy);
         }
 
+        ucs_assert((ucp_wireup_ep(ep->uct_eps[proxy_lane]) == NULL) ||
+                   (ucp_wireup_ep(ep->uct_eps[proxy_lane])->flags &
+                    UCP_WIREUP_EP_FLAG_LOCAL_CONNECTED));
+
         /* Create a signaling ep to the proxy lane */
         if (proxy_lane == lane) {
             /* If proxy is to the same lane, temporarily remove the existing
@@ -876,7 +880,6 @@ ucs_status_t ucp_wireup_resolve_proxy_lanes(ucp_ep_h ep)
              * proxy, so ucp_wireup_extract_lane() handles both cases.
              */
             uct_ep = ucp_wireup_extract_lane(ep, proxy_lane);
-            ucs_assert_always(uct_ep != NULL);
             status = ucp_signaling_ep_create(ep, uct_ep, 1, &signaling_ep);
             if (status != UCS_OK) {
                 /* coverity[leaked_storage] */
@@ -1223,7 +1226,7 @@ ucs_status_t ucp_wireup_connect_remote(ucp_ep_h ep, ucp_lane_index_t lane)
     uct_ep_pending_purge(uct_ep, ucp_wireup_connect_remote_purge_cb, &tmp_q);
 
     /* the wireup ep should use the existing [am_lane] as next_ep */
-    ucp_wireup_ep_set_next_ep(ep->uct_eps[lane], uct_ep);
+    ucp_wireup_ep_set_next_ep(ep->uct_eps[lane], uct_ep, 1, 1);
 
     if (!(ep->flags & UCP_EP_FLAG_CONNECT_REQ_QUEUED)) {
         status = ucp_wireup_send_request(ep);

src/ucp/wireup/wireup.h

@@ -132,6 +132,8 @@ ucs_status_t ucp_signaling_ep_create(ucp_ep_h ucp_ep, uct_ep_h uct_ep,
 void ucp_wireup_assign_lane(ucp_ep_h ep, ucp_lane_index_t lane, uct_ep_h uct_ep,
                             const char *info);
 
+uct_ep_h ucp_wireup_extract_lane(ucp_ep_h ep, ucp_lane_index_t lane);
+
 ucs_status_t
 ucp_wireup_connect_lane(ucp_ep_h ep, unsigned ep_init_flags,
                         ucp_lane_index_t lane, unsigned path_index,

src/ucp/wireup/wireup_cm.c

@@ -245,24 +245,75 @@ ucp_cm_client_connect_prog_arg_free(ucp_cm_client_connect_progress_arg_t *arg)
     ucs_free(arg);
 }
 
-static void ucp_cm_client_restore_ep(ucp_wireup_ep_t *wireup_cm_ep,
-                                     ucp_ep_h ucp_ep)
+/**
+ * Copies lanes from the one UCP EP to the another UCP EP. The function
+ * creates new WIREUP EPs for all lanes in @to_ep and sets UCT EP of
+ * the TLs from @from_ep. Both EPs have to be created and initalized.
+ * @to_ep should not have initialized the lanes by UCT EPs that will be
+ * overwritten by UCT EPs from the @from_ep's lanes.
+ * After copying UCT EPs, @to_ep should try to reconfigure lanes and
+ * some of the copied UCT EPs could be re-used in the new configuration.
+ *
+ * @param [in] to_ep               UCP EP handle to copy the lanes to.
+ * @param [in] from_ep             UCP EP handle to copy the lanes from.
+ * @param [in] change_ownership    Make WIREUP EPs in @to_ep are owner for
+ *                                 copied UCT EPs from @from_ep.
+ */
+static void ucp_cm_copy_ep_lanes(ucp_ep_h to_ep, ucp_ep_h from_ep,
+                                 int change_ownership)
 {
-    ucp_ep_h tmp_ep = wireup_cm_ep->tmp_ep;
-    ucp_wireup_ep_t *w_ep;
+    int to_is_owner, from_is_owner;
     ucp_lane_index_t lane_idx;
+    ucs_status_t status;
+    uct_ep_h uct_ep;
+    int is_local_connected;
+    ucp_wireup_ep_t *from_wireup_ep;
 
-    for (lane_idx = 0; lane_idx < ucp_ep_num_lanes(tmp_ep); ++lane_idx) {
-        if (tmp_ep->uct_eps[lane_idx] != NULL) {
-            ucs_assert(ucp_ep->uct_eps[lane_idx] == NULL);
-            ucp_ep->uct_eps[lane_idx] = tmp_ep->uct_eps[lane_idx];
-            w_ep = ucs_derived_of(ucp_ep->uct_eps[lane_idx], ucp_wireup_ep_t);
-            w_ep->super.ucp_ep = ucp_ep;
+    to_is_owner   = change_ownership;
+    from_is_owner = !change_ownership;
+
+    for (lane_idx = 0; lane_idx < ucp_ep_num_lanes(from_ep); ++lane_idx) {
+        if ((lane_idx == ucp_ep_get_cm_lane(from_ep)) ||
+            (from_ep->uct_eps[lane_idx] == NULL)) {
+            continue;
+        }
+
+        ucs_assert_always(to_ep->uct_eps[lane_idx] == NULL);
+
+        uct_ep = ucp_wireup_extract_lane(from_ep, lane_idx);
+        if (uct_ep == NULL) {
+            /* UCT EP could be NULL only for non-P2P TLs */
+            ucs_assert(ucp_worker_is_tl_2iface(from_ep->worker,
+                                               ucp_ep_config(from_ep)->key.
+                                               lanes[lane_idx].rsc_index));
+            continue;
         }
-    }
 
-    ucp_ep_destroy_base(tmp_ep); /* not needed anymore */
-    wireup_cm_ep->tmp_ep = NULL;
+        from_wireup_ep = ucp_wireup_ep(from_ep->uct_eps[lane_idx]);
+
+        is_local_connected = (from_wireup_ep == NULL) ||
+                             (from_wireup_ep->flags &
+                              UCP_WIREUP_EP_FLAG_LOCAL_CONNECTED);
+
+        status = ucp_wireup_ep_create(to_ep, &to_ep->uct_eps[lane_idx]);
+        if (status != UCS_OK) {
+            ucs_fatal("%p: failed to create WIREUP EP to wrap %p UCT EP: %s",
+                      to_ep, uct_ep, ucs_status_string(status));
+        }
+
+        ucp_wireup_ep_set_next_ep(to_ep->uct_eps[lane_idx], uct_ep,
+                                  to_is_owner, is_local_connected);
+
+        if (from_wireup_ep == NULL) {
+            /* from_ep must be the owner of the UCT EP in this case */
+            ucs_assert(from_is_owner);
+            from_ep->uct_eps[lane_idx] = uct_ep;
+        } else {
+            ucs_assert(from_wireup_ep->super.is_owner);
+            ucp_wireup_ep_set_next_ep(from_ep->uct_eps[lane_idx], uct_ep,
+                                      from_is_owner, is_local_connected);
+        }
+    }
 }
 
 /*
@@ -275,7 +326,7 @@ static unsigned ucp_cm_client_connect_progress(void *arg)
     ucp_worker_h worker                                = ucp_ep->worker;
     ucp_context_h context                              = worker->context;
     uct_ep_h uct_cm_ep                                 = ucp_ep_get_cm_uct_ep(ucp_ep);
-    ucp_wireup_ep_t *wireup_ep;
+    ucp_wireup_ep_t *cm_wireup_ep;
     ucp_unpacked_address_t addr;
     uint64_t tl_bitmap;
     ucp_rsc_index_t dev_index;
@@ -286,9 +337,9 @@ static unsigned ucp_cm_client_connect_progress(void *arg)
 
     UCS_ASYNC_BLOCK(&worker->async);
 
-    wireup_ep = ucp_ep_get_cm_wireup_ep(ucp_ep);
-    ucs_assert(wireup_ep != NULL);
-    ucs_assert(wireup_ep->ep_init_flags & UCP_EP_INIT_CM_WIREUP_CLIENT);
+    cm_wireup_ep = ucp_ep_get_cm_wireup_ep(ucp_ep);
+    ucs_assert(cm_wireup_ep != NULL);
+    ucs_assert(cm_wireup_ep->ep_init_flags & UCP_EP_INIT_CM_WIREUP_CLIENT);
 
     status = ucp_address_unpack(worker, progress_arg->sa_data + 1,
                                 UCP_ADDRESS_PACK_FLAG_IFACE_ADDR |
@@ -311,14 +362,14 @@ static unsigned ucp_cm_client_connect_progress(void *arg)
     ucp_ep_update_remote_id(ucp_ep, progress_arg->sa_data->ep_id);
 
     /* Get tl bitmap from tmp_ep, because it contains initial configuration. */
-    tl_bitmap = ucp_ep_get_tl_bitmap(wireup_ep->tmp_ep);
+    tl_bitmap = ucp_ep_get_tl_bitmap(cm_wireup_ep->tmp_ep);
     ucs_assert(tl_bitmap != 0);
     rsc_index = ucs_ffs64(tl_bitmap);
     dev_index = context->tl_rscs[rsc_index].dev_index;
 
     /* Restore initial configuration from tmp_ep created for packing local
      * addresses. */
-    ucp_cm_client_restore_ep(wireup_ep, ucp_ep);
+    ucp_cm_copy_ep_lanes(ucp_ep, cm_wireup_ep->tmp_ep, 1);
 
 #ifdef ENABLE_ASSERT
     ucs_for_each_bit(rsc_index, tl_bitmap) {
@@ -327,7 +378,7 @@ static unsigned ucp_cm_client_connect_progress(void *arg)
 #endif
 
     tl_bitmap = ucp_context_dev_idx_tl_bitmap(context, dev_index);
-    status    = ucp_wireup_init_lanes(ucp_ep, wireup_ep->ep_init_flags,
+    status    = ucp_wireup_init_lanes(ucp_ep, cm_wireup_ep->ep_init_flags,
                                       tl_bitmap, &addr, addr_indices);
     if (status != UCS_OK) {
         goto out_free_addr;
@@ -351,7 +402,7 @@ static unsigned ucp_cm_client_connect_progress(void *arg)
     ucs_free(addr.address_list);
 out:
     if (status != UCS_OK) {
-        ucp_worker_set_ep_failed(worker, ucp_ep, &wireup_ep->super.super,
+        ucp_worker_set_ep_failed(worker, ucp_ep, &cm_wireup_ep->super.super,
                                  ucp_ep_get_cm_lane(ucp_ep), status);
     }
 
@@ -634,7 +685,7 @@ ucs_status_t ucp_ep_client_cm_connect_start(ucp_ep_h ucp_ep,
         return status;
     }
 
-    ucp_wireup_ep_set_next_ep(&wireup_ep->super.super, cm_ep);
+    ucp_wireup_ep_set_next_ep(&wireup_ep->super.super, cm_ep, 1, 1);
     ucp_ep_flush_state_reset(ucp_ep);
 
     return UCS_OK;
@@ -960,7 +1011,7 @@ ucs_status_t ucp_ep_cm_connect_server_lane(ucp_ep_h ep,
         return status;
     }
 
-    ucp_wireup_ep_set_next_ep(ep->uct_eps[lane], uct_ep);
+    ucp_wireup_ep_set_next_ep(ep->uct_eps[lane], uct_ep, 1, 1);
     return UCS_OK;
 }
 

src/ucp/wireup/wireup_ep.c

@@ -439,7 +439,7 @@ ucs_status_t ucp_wireup_ep_connect(uct_ep_h uct_ep, unsigned ep_init_flags,
         goto err;
     }
 
-    ucp_proxy_ep_set_uct_ep(&wireup_ep->super, next_ep, 1);
+    ucp_wireup_ep_set_next_ep(uct_ep, next_ep, 1, 0);
 
     ucs_debug("ep %p: created next_ep %p to %s using " UCT_TL_RESOURCE_DESC_FMT,
               ucp_ep, wireup_ep->super.uct_ep, ucp_ep_peer_name(ucp_ep),
@@ -647,14 +647,17 @@ ucs_status_t ucp_wireup_ep_connect_to_sockaddr(uct_ep_h uct_ep,
     return status;
 }
 
-void ucp_wireup_ep_set_next_ep(uct_ep_h uct_ep, uct_ep_h next_ep)
+void ucp_wireup_ep_set_next_ep(uct_ep_h uct_ep, uct_ep_h next_ep,
+                               int is_owner, int is_local_connected)
 {
     ucp_wireup_ep_t *wireup_ep = ucp_wireup_ep(uct_ep);
 
     ucs_assert(wireup_ep != NULL);
     ucs_assert(wireup_ep->super.uct_ep == NULL);
-    wireup_ep->flags |= UCP_WIREUP_EP_FLAG_LOCAL_CONNECTED;
-    ucp_proxy_ep_set_uct_ep(&wireup_ep->super, next_ep, 1);
+    if (is_local_connected) {
+        wireup_ep->flags |= UCP_WIREUP_EP_FLAG_LOCAL_CONNECTED;
+    }
+    ucp_proxy_ep_set_uct_ep(&wireup_ep->super, next_ep, is_owner);
 }
 
 uct_ep_h ucp_wireup_ep_extract_next_ep(uct_ep_h uct_ep)

src/ucp/wireup/wireup_ep.h

@@ -82,7 +82,8 @@ ucs_status_t
 ucp_wireup_ep_connect_aux(ucp_wireup_ep_t *wireup_ep, unsigned ep_init_flags,
                           const ucp_unpacked_address_t *remote_address);
 
-void ucp_wireup_ep_set_next_ep(uct_ep_h uct_ep, uct_ep_h next_ep);
+void ucp_wireup_ep_set_next_ep(uct_ep_h uct_ep, uct_ep_h next_ep,
+                               int is_owner, int is_local_connected);
 
 uct_ep_h ucp_wireup_ep_extract_next_ep(uct_ep_h uct_ep);