UCT: fix CI

Author: JeynmannZ

src/ucp/core/ucp_worker.c

@@ -492,44 +492,42 @@ static ucs_status_t
 ucp_worker_iface_handle_uct_ep_failure(ucp_ep_h ucp_ep, ucp_lane_index_t lane,
                                        uct_ep_h uct_ep, ucs_status_t ep_status)
 {
-    ucp_wireup_ep_t *wireup_ep;
     ucs_status_t status = UCS_OK;
+    ucp_wireup_ep_t *wireup_ep;
     int ft_available;
 
     ft_available = ucp_worker_iface_ft_available(ucp_ep, uct_ep);
-    if (ucp_ep->flags & UCP_EP_FLAG_FAILED) {
-        if (!ft_available) {
+    if (!ft_available) {
+        status = ucp_worker_uct_ep_outstanding_purge(uct_ep, ep_status);
+    } else {
+        status = uct_ep_invalidate(uct_ep, NULL);
+        if (status != UCS_OK) {
+            ucs_error("failed to invalidate UCT EP %p: %s", uct_ep,
+                      ucs_status_string(status));
             status = ucp_worker_uct_ep_outstanding_purge(uct_ep, ep_status);
         }
+    }
 
+    if (ucp_ep->flags & UCP_EP_FLAG_FAILED) {
         /* No pending operations should be scheduled */
         uct_ep_pending_purge(uct_ep, ucp_destroyed_ep_pending_purge, ucp_ep);
         return status;
     }
 
-    if (ft_available) {
-        status = uct_ep_invalidate(uct_ep, NULL);
-        if (status != UCS_OK) {
-            ucs_error("failed to invalidate UCT EP %p: %s", uct_ep,
-                      ucs_status_string(status));
-            return status;
-        }
-    }
-
     wireup_ep = ucp_wireup_ep(ucp_ep_get_lane(ucp_ep, lane));
     if ((wireup_ep == NULL) ||
         !ucp_wireup_aux_ep_is_owner(wireup_ep, uct_ep) ||
         !ucp_ep_is_local_connected(ucp_ep)) {
         /* Failure on NON-AUX EP or failure on AUX EP before it sent its address
          * means failure on the UCP EP */
         ucp_ep_set_lanes_failed(ucp_ep, UCS_BIT(lane), ep_status);
-        goto out;
+        return status;
     }
 
     if (wireup_ep->flags & UCP_WIREUP_EP_FLAG_READY) {
         /* @ref ucp_wireup_ep_progress was scheduled, wireup ep and its
          * pending requests have to be handled there */
-        goto out;
+        return status;
     }
 
     /**
@@ -541,12 +539,6 @@ ucp_worker_iface_handle_uct_ep_failure(ucp_ep_h ucp_ep, ucp_lane_index_t lane,
     ucp_wireup_ep_discard_aux_ep(wireup_ep, UCT_FLUSH_FLAG_CANCEL,
                                  ucp_destroyed_ep_pending_purge, ucp_ep);
     ucp_wireup_remote_connected(ucp_ep);
-
-out:
-    if (!ft_available) {
-        status = ucp_worker_uct_ep_outstanding_purge(uct_ep, ep_status);
-    }
-
     return status;
 }
 
@@ -606,8 +598,9 @@ ucp_worker_iface_error_handler(void *arg, uct_ep_h uct_ep, ucs_status_t status)
     if (ucp_worker_is_uct_ep_discarding(worker, uct_ep)) {
         ucs_debug("UCT EP %p is being discarded on UCP Worker %p",
                   uct_ep, worker);
-        status = ucp_worker_uct_ep_outstanding_purge(uct_ep, status);
+        (void)ucp_worker_uct_ep_outstanding_purge(uct_ep, status);
         ucp_discard_lane_ff(uct_ep);
+        status = UCS_OK;
         goto out;
     }
 

src/uct/ib/mlx5/rc/rc_mlx5_ep.c

@@ -807,31 +807,18 @@ ucs_status_t uct_rc_mlx5_base_ep_invalidate(uct_ep_h tl_ep,
 {
     UCT_RC_MLX5_BASE_EP_DECL(tl_ep, iface, ep);
     uct_ib_mlx5_txwq_t *txwq = &ep->tx.wq;
-    uint16_t ft_ci = txwq->ft_ci;
+    uint16_t ft_ci           = txwq->ft_ci;
     uint16_t outstanding;
-    int failover_arm;
-    ucs_status_t status;
 
-    failover_arm = !(ep->super.ext_flags &
-                     UCT_RC_EP_EXT_FLAG_FAILOVER_ARMED);
-    if (failover_arm) {
+    if (!(ep->super.ext_flags & UCT_RC_EP_EXT_FLAG_FAILOVER_ARMED)) {
         outstanding = txwq->bb_max - uct_rc_txqp_available(&ep->super.txqp);
-        ft_ci       = txwq->prev_sw_pi - outstanding;
+        txwq->ft_ci = txwq->prev_sw_pi - outstanding;
+        ep->super.ext_flags |= UCT_RC_EP_EXT_FLAG_FAILOVER_ARMED;
         ucs_assert(ft_ci == txwq->hw_ci);
     }
 
-    status = uct_ib_mlx5_modify_qp_state(&iface->super.super, &txwq->super,
-                                         IBV_QPS_ERR);
-    if ((status != UCS_OK) || !failover_arm) {
-        return status;
-    }
-
-    txwq->ft_ci          = ft_ci;
-    ep->super.ext_flags |= UCT_RC_EP_EXT_FLAG_FAILOVER_ARMED;
-
-    ucs_debug("ep %p armed failover WQE range (%u, %u) posted index %u", ep,
-              txwq->ft_ci, txwq->sw_pi, txwq->nnop_pi);
-    return UCS_OK;
+    return uct_ib_mlx5_modify_qp_state(&iface->super.super, &txwq->super,
+                                       IBV_QPS_ERR);
 }
 
 ucs_status_t uct_rc_mlx5_base_ep_fc_ctrl(uct_ep_t *tl_ep, unsigned op,

src/uct/ib/mlx5/rc/rc_mlx5_iface.c

@@ -206,7 +206,8 @@ void uct_rc_mlx5_iface_handle_failure(uct_ib_iface_t *ib_iface, void *arg,
     }
 
     base_iface = ucs_derived_of(&iface->super.super.super, uct_base_iface_t);
-    if (base_iface->err_handler == NULL) {
+    is_flush_cancel = ep->super.flags & UCT_RC_EP_FLAG_FLUSH_CANCEL;
+    if ((base_iface->err_handler == NULL) || is_flush_cancel) {
         uct_rc_txqp_purge_outstanding(iface, &ep->super.txqp, ep_status, pi, 0);
     }
 
@@ -219,20 +220,14 @@ void uct_rc_mlx5_iface_handle_failure(uct_ib_iface_t *ib_iface, void *arg,
         goto out;
     }
 
-    is_flush_cancel  = ep->super.flags & UCT_RC_EP_FLAG_FLUSH_CANCEL;
     ep->super.flags |= UCT_RC_EP_FLAG_ERR_HANDLER_INVOKED;
-    uct_rc_fc_restore_wnd(iface, &ep->super.fc);
-
-    if (is_flush_cancel && (base_iface->err_handler == NULL)) {
+    if (is_flush_cancel) {
         goto out;
     }
 
+    uct_rc_fc_restore_wnd(iface, &ep->super.fc);
     status = uct_iface_handle_ep_err(&iface->super.super.super,
                                      &ep->super.super.super, ep_status);
-    if (is_flush_cancel) {
-        goto out;
-    }
-
     log_lvl = uct_base_iface_failure_log_level(&ib_iface->super, status,
                                                ep_status);