UCT/IB/MLX5: cleanup

Author: JeynmannZ

src/uct/api/v2/uct_v2.h

@@ -412,11 +412,17 @@ typedef enum {
  * @brief Flags used by @ref uct_ep_invalidate.
  */
 typedef enum {
+    /** Move the endpoint QP to the error state. */
+    UCT_EP_INVALIDATE_FLAG_MODIFY_QP_TO_ERR     = UCS_BIT(0),
+
     /**
-     * Do not complete outstanding operations when the endpoint transitions to
-     * the error state. The caller is responsible for completing them.
+     * Suppress completion of outstanding operations by the normal endpoint
+     * error path. The caller transfers outstanding-operation ownership to an
+     * external provider and is responsible for completing them. This state
+     * remains active until endpoint cancellation or destruction. This flag
+     * does not modify QP state.
      */
-    UCT_EP_INVALIDATE_FLAG_SUPPRESS_COMPLETIONS = UCS_BIT(0)
+    UCT_EP_INVALIDATE_FLAG_SUPPRESS_COMPLETIONS = UCS_BIT(1)
 } uct_ep_invalidate_flags_t;
 
 
@@ -1372,10 +1378,11 @@ ucs_status_t uct_ep_query(uct_ep_h ep, uct_ep_attr_t *ep_attr);
  * @ingroup UCT_RESOURCE
  * @brief Invalidate the endpoint.
  *
- * This routine invalidates the endpoint and moves it to the error state.
- * All the incomplete and subsequent operations on the endpoint will be
- * completed with error, unless
- * @ref UCT_EP_INVALIDATE_FLAG_SUPPRESS_COMPLETIONS is specified.
+ * This routine invalidates the endpoint according to the requested flags.
+ * @ref UCT_EP_INVALIDATE_FLAG_MODIFY_QP_TO_ERR moves a QP-based endpoint to
+ * the error state. Incomplete operations are completed with error unless
+ * @ref UCT_EP_INVALIDATE_FLAG_SUPPRESS_COMPLETIONS transfers their ownership
+ * to the caller.
  *
  * @param [in]  ep         Endpoint to invalidate.
  * @param [in]  params     Operation parameters, see @ref
@@ -1893,8 +1900,7 @@ typedef void (*uct_ep_outstanding_purge_callback_t)(
 typedef enum {
     UCT_EP_OUTSTANDING_FIELD_RX_TOKEN = UCS_BIT(0),
     UCT_EP_OUTSTANDING_FIELD_CB       = UCS_BIT(1),
-    UCT_EP_OUTSTANDING_FIELD_ARG      = UCS_BIT(2),
-    UCT_EP_OUTSTANDING_FIELD_STATUS   = UCS_BIT(3)
+    UCT_EP_OUTSTANDING_FIELD_ARG      = UCS_BIT(2)
 } uct_ep_outstanding_purge_field_t;
 
 
@@ -1922,20 +1928,18 @@ typedef struct {
      * Valid when @ref UCT_EP_OUTSTANDING_FIELD_ARG is set.
      */
     void                                *arg;
-
-    /** Completion status for purging all outstanding operations. */
-    ucs_status_t                        status;
 } uct_ep_outstanding_purge_params_t;
 
 
 /**
  * @ingroup UCT_RESOURCE
  * @brief Purge outstanding (undelivered) operations from an endpoint.
  *
- * If only @ref UCT_EP_OUTSTANDING_FIELD_STATUS is specified, all outstanding
- * operations are completed with the supplied error status. Otherwise,
- * @ref uct_ep_outstanding_purge_params_t::cb is invoked once for each
- * undelivered outstanding operation, in the original endpoint posting order.
+ * The transport delegates outstanding-operation classification to its
+ * external provider. The provider invokes @ref
+ * uct_ep_outstanding_purge_params_t::cb once for each undelivered operation,
+ * in the original endpoint posting order. No transport-local status purge is
+ * provided by this API.
  */
 ucs_status_t
 uct_ep_outstanding_purge(uct_ep_h ep,

src/uct/base/uct_iface.h

@@ -324,23 +324,6 @@ typedef ucs_status_t (*uct_ep_outstanding_purge_func_t)(
         uct_ep_h ep, const uct_ep_outstanding_purge_params_t *params);
 
 
-static UCS_F_ALWAYS_INLINE ucs_status_t uct_ep_outstanding_purge_get_status(
-        const uct_ep_outstanding_purge_params_t *params, ucs_status_t *status_p)
-{
-    if (params == NULL) {
-        return UCS_ERR_INVALID_PARAM;
-    }
-
-    if ((params->field_mask != UCT_EP_OUTSTANDING_FIELD_STATUS) ||
-        !UCS_STATUS_IS_ERR(params->status)) {
-        return UCS_ERR_UNSUPPORTED;
-    }
-
-    *status_p = params->status;
-    return UCS_OK;
-}
-
-
 /* Internal operations, not exposed by the external API */
 typedef struct uct_iface_internal_ops {
     uct_iface_query_v2_func_t        iface_query_v2;

src/uct/ib/mlx5/dc/dc_mlx5_ep.c

@@ -847,6 +847,19 @@ ucs_status_t uct_dc_mlx5_ep_invalidate(uct_ep_h tl_ep,
 {
     uct_dc_mlx5_ep_t *ep = ucs_derived_of(tl_ep, uct_dc_mlx5_ep_t);
 
+    if ((params == NULL) ||
+        !(params->field_mask & UCT_EP_INVALIDATE_PARAM_FIELD_FLAGS)) {
+        return UCS_ERR_INVALID_PARAM;
+    }
+
+    if ((params->flags & UCT_EP_INVALIDATE_FLAG_SUPPRESS_COMPLETIONS)) {
+        return UCS_ERR_UNSUPPORTED;
+    }
+
+    if (!(params->flags & UCT_EP_INVALIDATE_FLAG_MODIFY_QP_TO_ERR)) {
+        return UCS_OK;
+    }
+
     if (ep->dci == UCT_DC_MLX5_EP_NO_DCI) {
         ep->flags |= UCT_DC_MLX5_EP_FLAG_INVALIDATED;
         return UCS_OK;

src/uct/ib/mlx5/rc/rc_mlx5.h

@@ -268,9 +268,6 @@ ucs_status_t uct_rc_mlx5_ep_get_address(uct_ep_h tl_ep, uct_ep_addr_t *addr);
 
 ucs_status_t uct_rc_mlx5_base_ep_query(uct_ep_h tl_ep, uct_ep_attr_t *ep_attr);
 
-ucs_status_t uct_rc_mlx5_ep_outstanding_purge(
-        uct_ep_h tl_ep, const uct_ep_outstanding_purge_params_t *params);
-
 void uct_rc_mlx5_ep_update_tx_res(uct_ep_h tl_ep);
 
 unsigned uct_rc_mlx5_ep_cleanup_qp(void *arg);

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

@@ -64,39 +64,6 @@ void uct_rc_mlx5_ep_update_tx_res(uct_ep_h tl_ep)
 }
 
 
-ucs_status_t uct_rc_mlx5_ep_outstanding_purge(
-        uct_ep_h tl_ep, const uct_ep_outstanding_purge_params_t *params)
-{
-    UCT_RC_MLX5_BASE_EP_DECL(tl_ep, iface, ep);
-    ucs_status_t purge_status;
-    ucs_status_t status;
-
-    if ((params != NULL) &&
-        (params->field_mask & UCT_EP_OUTSTANDING_FIELD_STATUS)) {
-        status = uct_ep_outstanding_purge_get_status(params, &purge_status);
-        if (status != UCS_OK) {
-            return status;
-        }
-
-        if (ep->flags & UCT_RC_MLX5_EP_FLAG_SUPPRESS_COMPLETIONS) {
-            uct_rc_mlx5_ep_update_tx_res(tl_ep);
-            ep->flags &= ~UCT_RC_MLX5_EP_FLAG_SUPPRESS_COMPLETIONS;
-        }
-    } else {
-        status = uct_ib_mlx5_ext_ep_outstanding_purge(tl_ep, params);
-        if (status != UCS_OK) {
-            return status;
-        }
-
-        purge_status = UCS_ERR_CANCELED;
-    }
-
-    uct_rc_txqp_purge_outstanding(&iface->super, &ep->super.txqp, purge_status,
-                                  ep->tx.wq.sw_pi, 0);
-    return UCS_OK;
-}
-
-
 static ucs_status_t UCS_F_ALWAYS_INLINE uct_rc_mlx5_base_ep_put_short_inline(
         uct_ep_h tl_ep, const void *buffer, unsigned length,
         uint64_t remote_addr, uct_rkey_t rkey)
@@ -849,24 +816,31 @@ ucs_status_t uct_rc_mlx5_ep_invalidate(uct_ep_h tl_ep,
 {
     UCT_RC_MLX5_EP_DECL(tl_ep, iface, ep);
     uct_ib_mlx5_txwq_t *txwq = &ep->super.tx.wq;
+    ucs_status_t status;
 
-    if ((params != NULL) &&
-        (params->field_mask & UCT_EP_INVALIDATE_PARAM_FIELD_FLAGS) &&
-        (params->flags & UCT_EP_INVALIDATE_FLAG_SUPPRESS_COMPLETIONS) &&
+    if ((params == NULL) ||
+        !(params->field_mask & UCT_EP_INVALIDATE_PARAM_FIELD_FLAGS)) {
+        return UCS_ERR_INVALID_PARAM;
+    }
+
+    if (params->flags & UCT_EP_INVALIDATE_FLAG_MODIFY_QP_TO_ERR) {
+        status = uct_ib_mlx5_modify_qp_state(&iface->super.super, &txwq->super,
+                                             IBV_QPS_ERR);
+        if (status != UCS_OK) {
+            return status;
+        }
+    }
+
+    if ((params->flags & UCT_EP_INVALIDATE_FLAG_SUPPRESS_COMPLETIONS) &&
         !(ep->super.super.flags & UCT_RC_EP_FLAG_FLUSH_CANCEL) &&
         !(ep->super.flags & UCT_RC_MLX5_EP_FLAG_SUPPRESS_COMPLETIONS)) {
         ep->super.flags |= UCT_RC_MLX5_EP_FLAG_SUPPRESS_COMPLETIONS;
         txwq->ft_ci      = txwq->hw_ci;
-        ucs_assert(txwq->ft_ci ==
-                   (txwq->prev_sw_pi -
-                    (txwq->bb_max -
-                     uct_rc_txqp_available(&ep->super.super.txqp))));
         ucs_debug("ep %p suppress completions WQE range (%u, %u) next token %u",
                   ep, txwq->ft_ci, txwq->sw_pi, txwq->next_token);
     }
 
-    return uct_ib_mlx5_modify_qp_state(&iface->super.super, &txwq->super,
-                                       IBV_QPS_ERR);
+    return UCS_OK;
 }
 
 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

@@ -1140,7 +1140,7 @@ static uct_rc_iface_ops_t uct_rc_mlx5_iface_ops = {
             .ep_is_connected        = uct_rc_mlx5_base_ep_is_connected,
             .ep_get_device_ep       = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported,
             .ep_put_sgl_zcopy       = uct_rc_mlx5_ep_put_sgl_zcopy,
-            .ep_outstanding_purge   = uct_rc_mlx5_ep_outstanding_purge
+            .ep_outstanding_purge   = uct_ib_mlx5_ext_ep_outstanding_purge
         },
         .create_cq      = uct_rc_mlx5_iface_common_create_cq,
         .destroy_cq     = uct_rc_mlx5_iface_common_destroy_cq,

test/gtest/ucp/test_ucp_fault_tolerance.cc

@@ -53,6 +53,15 @@ class test_ucp_fault_tolerance : public test_ucp_memheap {
 protected:
     static constexpr uint16_t AM_ID = 0;
 
+    static ucs_status_t invalidate_uct_ep(uct_ep_h ep)
+    {
+        uct_ep_invalidate_params_t params = {};
+
+        params.field_mask = UCT_EP_INVALIDATE_PARAM_FIELD_FLAGS;
+        params.flags      = UCT_EP_INVALIDATE_FLAG_MODIFY_QP_TO_ERR;
+        return uct_ep_invalidate(ep, &params);
+    }
+
     enum {
         GOOD_EP_INDEX = 0,      /* Index for good endpoint */
         INJECTED_EP_INDEX = 1   /* Index for failure-injected endpoint */
@@ -248,7 +257,7 @@ class test_ucp_fault_tolerance : public test_ucp_memheap {
             std::vector<ucs_status_ptr_t> status_ptrs;
             ucp_lane_index_t lane = lanes[lane_idx];
             uct_ep_h uct_ep_for_injection = ucp_ep_get_lane(ucp_ep_for_injection, lane);
-            ucs_status_t status = uct_ep_invalidate(uct_ep_for_injection, 0);
+            ucs_status_t status = invalidate_uct_ep(uct_ep_for_injection);
             if (status == UCS_ERR_UNSUPPORTED) {
                 UCS_TEST_SKIP_R("uct_ep_invalidate is not supported");
             }
@@ -322,7 +331,7 @@ class test_ucp_fault_tolerance : public test_ucp_memheap {
                 break;
             }
 
-            status = uct_ep_invalidate(uct_ep_for_injection, 0);
+            status = invalidate_uct_ep(uct_ep_for_injection);
             if (status == UCS_ERR_UNSUPPORTED) {
                 UCS_TEST_SKIP_R("uct_ep_invalidate is not supported");
             }
@@ -404,7 +413,7 @@ class test_ucp_fault_tolerance : public test_ucp_memheap {
         for (size_t lane_idx = 0; lane_idx < rma_bw_lanes.size() - 1; ++lane_idx) {
             ucp_lane_index_t lane = rma_bw_lanes[lane_idx];
             uct_ep_h uct_ep_for_injection = ucp_ep_get_lane(ucp_ep_for_injection, lane);
-            status = uct_ep_invalidate(uct_ep_for_injection, 0);
+            status = invalidate_uct_ep(uct_ep_for_injection);
             if (status == UCS_ERR_UNSUPPORTED) {
                 UCS_TEST_SKIP_R("uct_ep_invalidate is not supported");
             }