UCT: outstanding purge implementation

src/uct/api/v2/uct_v2.h

@@ -1195,7 +1195,11 @@ enum uct_iface_attr_field {
  */
         /* PUT capabilities */
 #define UCT_IFACE_FLAG_V2_PUT_SGL_ZCOPY       UCS_BIT(0)  /**< Zero-copy SGL put */
-#define UCT_IFACE_FLAG_V2_QUERY_TOKEN         UCS_BIT(1)  /**< Interface supports token query */
+#define UCT_IFACE_FLAG_V2_QUERY_TOKEN         UCS_BIT(1)  /**< @ref uct_iface_query_v2
+                                                               and @ref uct_ep_query
+                                                               support token query, and
+                                                               @ref uct_ep_outstanding_purge
+                                                               is supported. */
 /**
  * @}
  */
@@ -1888,6 +1892,18 @@ typedef struct {
 } uct_ep_outstanding_purge_params_t;
 
 
+/**
+ * @ingroup UCT_RESOURCE
+ * @brief Purge outstanding (undelivered) operations from an endpoint.
+ *
+ * @ref uct_ep_outstanding_purge_params_t::cb is invoked once for each
+ * undelivered outstanding operation, in the original endpoint posting order.
+ */
+ucs_status_t
+uct_ep_outstanding_purge(uct_ep_h ep,
+                         const uct_ep_outstanding_purge_params_t *params);
+
+
 END_C_DECLS
 
 #endif

src/uct/base/uct_iface.c

@@ -571,8 +571,7 @@ ucs_status_t uct_single_device_resource(uct_md_h md, const char *dev_name,
     return UCS_OK;
 }
 
-ucs_status_t
-uct_iface_base_query_v2(uct_iface_h iface, uct_iface_attr_v2_t *iface_attr)
+void uct_iface_query_v2_init(uct_iface_h iface, uct_iface_attr_v2_t *iface_attr)
 {
     if (iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_CAP_FLAGS) {
         iface_attr->cap.flags = 0;
@@ -582,15 +581,28 @@ uct_iface_base_query_v2(uct_iface_h iface, uct_iface_attr_v2_t *iface_attr)
         iface_attr->max_put_sgl_zcopy_count = 0;
     }
 
-    if (!(iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_TX_TOKEN) &&
-        (iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_TX_TOKEN_LENGTH)) {
+    if (iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_TX_TOKEN_LENGTH) {
         iface_attr->tx_token_length = 0;
     }
 
-    if (!(iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_RX_TOKEN) &&
-        (iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_RX_TOKEN_LENGTH)) {
+    if (iface_attr->field_mask & UCT_IFACE_ATTR_FIELD_RX_TOKEN_LENGTH) {
         iface_attr->rx_token_length = 0;
     }
+}
+
+ucs_status_t
+uct_iface_base_query_v2(uct_iface_h iface, uct_iface_attr_v2_t *iface_attr)
+{
+    const uint64_t token_mask = UCT_IFACE_ATTR_FIELD_TX_TOKEN_LENGTH |
+                                UCT_IFACE_ATTR_FIELD_RX_TOKEN_LENGTH |
+                                UCT_IFACE_ATTR_FIELD_TX_TOKEN |
+                                UCT_IFACE_ATTR_FIELD_RX_TOKEN;
+
+    uct_iface_query_v2_init(iface, iface_attr);
+
+    if (ucs_test_flags(iface_attr->field_mask, token_mask)) {
+        return UCS_ERR_UNSUPPORTED;
+    }
 
     return UCS_OK;
 }
@@ -691,6 +703,7 @@ UCS_CLASS_INIT_FUNC(uct_base_iface_t, uct_iface_ops_t *ops,
     ucs_assert(internal_ops->iface_vfs_refresh != NULL);
     ucs_assert(internal_ops->ep_query != NULL);
     ucs_assert(internal_ops->ep_invalidate != NULL);
+    ucs_assert(internal_ops->ep_outstanding_purge != NULL);
 
     UCS_STATIC_ASSERT(ucs_offsetof(uct_base_iface_t, internal_ops) ==
                       sizeof(uct_iface_t));
@@ -834,6 +847,15 @@ ucs_status_t uct_ep_invalidate(uct_ep_h ep,
     return iface->internal_ops->ep_invalidate(ep, params);
 }
 
+ucs_status_t
+uct_ep_outstanding_purge(uct_ep_h ep,
+                         const uct_ep_outstanding_purge_params_t *params)
+{
+    const uct_base_iface_t *iface = ucs_derived_of(ep->iface, uct_base_iface_t);
+
+    return iface->internal_ops->ep_outstanding_purge(ep, params);
+}
+
 void uct_ep_set_iface(uct_ep_h ep, uct_iface_t *iface)
 {
     ep->iface = iface;

src/uct/base/uct_iface.h

@@ -914,6 +914,9 @@ void uct_base_iface_progress_enable_cb(uct_base_iface_t *iface,
 
 void uct_base_iface_progress_disable(uct_iface_h tl_iface, unsigned flags);
 
+void uct_iface_query_v2_init(uct_iface_h iface,
+                             uct_iface_attr_v2_t *iface_attr);
+
 ucs_status_t
 uct_iface_base_query_v2(uct_iface_h iface, uct_iface_attr_v2_t *iface_attr);
 

src/uct/cuda/cuda_copy/cuda_copy_iface.c

@@ -272,7 +272,8 @@ static uct_iface_internal_ops_t uct_cuda_copy_iface_internal_ops = {
     .ep_connect_to_ep_v2   = (uct_ep_connect_to_ep_v2_func_t)ucs_empty_function_return_unsupported,
     .iface_is_reachable_v2 = uct_cuda_copy_iface_is_reachable_v2,
     .ep_is_connected       = uct_base_ep_is_connected,
-    .ep_get_device_ep      = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported
+    .ep_get_device_ep      = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported,
+    .ep_outstanding_purge  = (uct_ep_outstanding_purge_func_t)ucs_empty_function_return_unsupported
 };
 
 static uct_cuda_ctx_rsc_t * uct_cuda_copy_ctx_rsc_create(uct_iface_h tl_iface)

src/uct/cuda/cuda_ipc/cuda_ipc_iface.c

@@ -507,10 +507,11 @@ static uct_iface_internal_ops_t uct_cuda_ipc_iface_internal_ops = {
     .ep_is_connected        = uct_cuda_ipc_ep_is_connected,
     .ep_get_device_ep       = uct_cuda_ipc_ep_get_device_ep,
 #if CUDA_VERSION >= 13000
-    .ep_put_sgl_zcopy       = uct_cuda_ipc_ep_put_sgl_zcopy
+    .ep_put_sgl_zcopy       = uct_cuda_ipc_ep_put_sgl_zcopy,
 #else
-    .ep_put_sgl_zcopy       = (uct_ep_put_sgl_zcopy_func_t)ucs_empty_function_return_unsupported
+    .ep_put_sgl_zcopy       = (uct_ep_put_sgl_zcopy_func_t)ucs_empty_function_return_unsupported,
 #endif
+    .ep_outstanding_purge = (uct_ep_outstanding_purge_func_t)ucs_empty_function_return_unsupported
 };
 
 static uct_cuda_ctx_rsc_t * uct_cuda_ipc_ctx_rsc_create(uct_iface_h tl_iface)

src/uct/cuda/gdr_copy/gdr_copy_iface.c

@@ -218,7 +218,8 @@ static uct_iface_internal_ops_t uct_gdr_copy_iface_internal_ops = {
     .ep_connect_to_ep_v2   = (uct_ep_connect_to_ep_v2_func_t)ucs_empty_function_return_unsupported,
     .iface_is_reachable_v2 = uct_gdr_copy_iface_is_reachable_v2,
     .ep_is_connected       = uct_base_ep_is_connected,
-    .ep_get_device_ep      = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported
+    .ep_get_device_ep      = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported,
+    .ep_outstanding_purge  = (uct_ep_outstanding_purge_func_t)ucs_empty_function_return_unsupported
 };
 
 static UCS_CLASS_INIT_FUNC(uct_gdr_copy_iface_t, uct_md_h md, uct_worker_h worker,

src/uct/gaudi/gaudi_gdr/gaudi_gdr_iface.c

@@ -97,7 +97,9 @@ static uct_iface_internal_ops_t uct_gaudi_gdr_iface_internal_ops = {
     .iface_is_reachable_v2 = (uct_iface_is_reachable_v2_func_t)
             ucs_empty_function_return_zero,
     .ep_is_connected       = (uct_ep_is_connected_func_t)
-            ucs_empty_function_return_zero_int
+            ucs_empty_function_return_zero_int,
+    .ep_outstanding_purge  = (uct_ep_outstanding_purge_func_t)
+            ucs_empty_function_return_unsupported
 };
 
 static UCS_CLASS_INIT_FUNC(uct_gaudi_gdr_iface_t, uct_md_h md,

src/uct/ib/efa/srd/srd_iface.c

@@ -236,6 +236,8 @@ static uct_ib_iface_ops_t uct_srd_iface_ops = {
         .iface_is_reachable_v2 = uct_ib_iface_is_reachable_v2,
         .ep_is_connected       = uct_srd_ep_is_connected,
         .ep_get_device_ep      = (uct_ep_get_device_ep_func_t)
+            ucs_empty_function_return_unsupported,
+        .ep_outstanding_purge  = (uct_ep_outstanding_purge_func_t)
             ucs_empty_function_return_unsupported
     },
     .create_cq      = uct_ib_verbs_create_cq,

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

@@ -1452,7 +1452,8 @@ static uct_rc_iface_ops_t uct_dc_mlx5_iface_ops = {
             .ep_connect_to_ep_v2    = (uct_ep_connect_to_ep_v2_func_t)ucs_empty_function_return_unsupported,
             .iface_is_reachable_v2  = uct_dc_mlx5_iface_is_reachable_v2,
             .ep_is_connected        = uct_dc_mlx5_ep_is_connected,
-            .ep_get_device_ep       = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported
+            .ep_get_device_ep       = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported,
+            .ep_outstanding_purge = (uct_ep_outstanding_purge_func_t)ucs_empty_function_return_unsupported
         },
         .create_cq      = uct_rc_mlx5_iface_common_create_cq,
         .destroy_cq     = uct_rc_mlx5_iface_common_destroy_cq,

src/uct/ib/mlx5/gdaki/gdaki.c

@@ -1068,6 +1068,7 @@ static uct_rc_iface_ops_t uct_rc_gdaki_internal_ops = {
             .iface_is_reachable_v2  = (uct_iface_is_reachable_v2_func_t)ucs_empty_function_return_one_int,
             .ep_is_connected        = uct_rc_gdaki_ep_is_connected,
             .ep_get_device_ep       = uct_rc_gdaki_ep_get_device_ep,
+            .ep_outstanding_purge   = (uct_ep_outstanding_purge_func_t)ucs_empty_function_return_unsupported
         },
         .create_cq  = uct_rc_gdaki_create_cq,
         .destroy_cq = (uct_ib_iface_destroy_cq_func_t)ucs_empty_function_return_success,

src/uct/ib/mlx5/gga/gga_mlx5.c

@@ -737,7 +737,8 @@ static uct_rc_iface_ops_t uct_gga_mlx5_iface_ops = {
             .ep_connect_to_ep_v2    = uct_gga_mlx5_ep_connect_to_ep_v2,
             .iface_is_reachable_v2  = uct_gga_mlx5_iface_is_reachable_v2,
             .ep_is_connected        = uct_gga_mlx5_ep_is_connected,
-            .ep_get_device_ep       = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported
+            .ep_get_device_ep       = (uct_ep_get_device_ep_func_t)ucs_empty_function_return_unsupported,
+            .ep_outstanding_purge   = (uct_ep_outstanding_purge_func_t)ucs_empty_function_return_unsupported
         },
         .create_cq      = uct_rc_mlx5_iface_common_create_cq,
         .destroy_cq     = uct_rc_mlx5_iface_common_destroy_cq,