UCP/UCT/IB: Add per-endpoint traffic class (DSCP) for RC/RoCE

Add an optional per-endpoint traffic class programmed onto the RC mlx5 QP,
so endpoints can be tagged with a distinct DSCP (RoCEv2) or GRH traffic
class (IB) for fabric QoS.

- API: ucp_ep_params.ep_traffic_class + UCP_EP_PARAM_FIELD_EP_TRAFFIC_CLASS
- UCT: uct_ep_connect_to_ep_params_t.ep_traffic_class +
  UCT_EP_CONNECT_TO_EP_PARAM_FIELD_EP_TRAFFIC_CLASS
- Plumb UCP ep params -> wireup -> rc mlx5 devx QP (primary_address_path.dscp
  for RoCEv2, .tclass for IB); default behavior unchanged when unset
- Sentinels UCP_EP_NO_TCLASS / UCT_EP_NO_TCLASS
- Java binding constant

Author: ybenvidia

AUTHORS

@@ -129,6 +129,8 @@ Xu Yifeng <yifeng.xyf@alibaba-inc.com>
 Yaser Afshar <yaser.afshar@intel.com>
 Yihua Xu <yihua.xu@intel.com>
 Yiltan Hassan Temucin <yiltan.temucin@amd.com>
+Yoel Benabou <ybenabou@nvidia.com>
+Yoel Benabou <ybenabou@nvidia.com>
 Yossi Itigin <yosefe@nvidia.com>
 Yuriy Shestakov <yuriis@mellanox.com>
 Zhenlong Ma <zhenlongm@nvidia.com>

bindings/java/src/main/java/org/openucx/jucx/ucp/UcpConstants.java

@@ -86,6 +86,7 @@ public class UcpConstants {
     static long UCP_EP_PARAM_FIELD_LOCAL_SOCK_ADDR;
     static long UCP_EP_PARAM_FIELD_CONN_REQUEST;
     static long UCP_EP_PARAM_FIELD_NAME;
+    static long UCP_EP_PARAM_FIELD_EP_TRAFFIC_CLASS;
 
     /**
      * UCP error handling mode.

bindings/java/src/main/native/ucp_constants.cc

@@ -65,6 +65,7 @@ Java_org_openucx_jucx_ucp_UcpConstants_loadConstants(JNIEnv *env, jclass cls)
     JUCX_DEFINE_LONG_CONSTANT(UCP_EP_PARAM_FIELD_LOCAL_SOCK_ADDR);
     JUCX_DEFINE_LONG_CONSTANT(UCP_EP_PARAM_FIELD_CONN_REQUEST);
     JUCX_DEFINE_LONG_CONSTANT(UCP_EP_PARAM_FIELD_NAME);
+    JUCX_DEFINE_LONG_CONSTANT(UCP_EP_PARAM_FIELD_EP_TRAFFIC_CLASS);
 
     // UCP error handling mode
     JUCX_DEFINE_INT_CONSTANT(UCP_ERR_HANDLING_MODE_PEER);

src/ucp/api/ucp.h

@@ -268,7 +268,10 @@ enum ucp_ep_params_field {
     /**< Connection request field */
     UCP_EP_PARAM_FIELD_CONN_REQUEST      = UCS_BIT(6),
     UCP_EP_PARAM_FIELD_NAME              = UCS_BIT(7), /**< Endpoint name */
-    UCP_EP_PARAM_FIELD_LOCAL_SOCK_ADDR   = UCS_BIT(8)  /**< Local socket Address */
+    UCP_EP_PARAM_FIELD_LOCAL_SOCK_ADDR   = UCS_BIT(
+            8), /**< Local socket Address */
+    UCP_EP_PARAM_FIELD_EP_TRAFFIC_CLASS  = UCS_BIT(
+            9) /**< Endpoint traffic class */
 };
 
 

src/ucp/api/ucp_def.h

@@ -779,6 +779,13 @@ typedef struct ucp_ep_params {
      */
     const char              *name;
 
+    /**
+     * Traffic class for this endpoint.
+     * This setting is optional. To enable it, the corresponding - @ref
+     * UCP_EP_PARAM_FIELD_EP_TRAFFIC_CLASS bit in the field mask must be set.
+     */
+    uint8_t                 ep_traffic_class;
+
     /**
      * The sockaddr to bind locally. Specifies the associated network device
      * to bind locally to establish new connections.
@@ -787,7 +794,6 @@ typedef struct ucp_ep_params {
      * UCP_EP_PARAM_FIELD_LOCAL_SOCK_ADDR bit in the field mask must be set.
      */
     ucs_sock_addr_t         local_sockaddr;
-
 } ucp_ep_params_t;
 
 

src/ucp/core/ucp_ep.c

@@ -667,6 +667,13 @@ ucp_ep_adjust_params(ucp_ep_h ep, const ucp_ep_params_t *params)
         ep->flags |= UCP_EP_FLAG_USER_DATA_PARAM;
     }
 
+    if (params->field_mask & UCP_EP_PARAM_FIELD_EP_TRAFFIC_CLASS) {
+        ep->ep_traffic_class = params->ep_traffic_class;
+        ep->flags           |= UCP_EP_FLAG_EP_TRAFFIC_CLASS;
+    } else {
+        ep->ep_traffic_class = UCP_EP_NO_TCLASS;
+    }
+
     return UCS_OK;
 }
 

src/ucp/core/ucp_ep.h

@@ -104,7 +104,8 @@ enum {
                                                         while merging pending queues */
     UCP_EP_FLAG_CONNECT_PRE_REQ_QUEUED = UCS_BIT(9), /* Pre-Connection request was queued */
     UCP_EP_FLAG_CLOSED                 = UCS_BIT(10),/* EP was closed */
-    /* 11 bit is vacant for a flag */
+    UCP_EP_FLAG_EP_TRAFFIC_CLASS = UCS_BIT(11), /* Endpoint traffic class */
+    /* 12 bit is vacant for a flag */
     UCP_EP_FLAG_ERR_HANDLER_INVOKED    = UCS_BIT(12),/* error handler was called */
     UCP_EP_FLAG_INTERNAL               = UCS_BIT(13),/* the internal EP which holds
                                                         temporary wireup configuration or
@@ -581,6 +582,7 @@ typedef struct ucp_ep {
     /* Transports for every lane */
     uct_ep_h                      uct_eps[UCP_MAX_FAST_PATH_LANES];
     ucp_ep_ext_t                  *ext;                   /* Endpoint extension */
+    uint8_t                       ep_traffic_class; /* Endpoint traffic class */
 
 #if ENABLE_DEBUG_DATA
     char                          peer_name[UCP_WORKER_ADDRESS_NAME_MAX];

src/ucp/core/ucp_types.h

@@ -60,6 +60,8 @@ typedef uint16_t                     ucp_worker_cfg_index_t;
 #define UCP_WORKER_MAX_RKEY_CONFIG   UINT16_MAX
 #define UCP_WORKER_CFG_INDEX_NULL    UINT16_MAX
 
+#define UCP_EP_NO_TCLASS ((uint8_t)-1)
+
 
 /* Forward declarations */
 typedef struct ucp_request            ucp_request_t;

src/ucp/wireup/wireup.c

@@ -480,7 +480,9 @@ ucp_wireup_connect_local(ucp_ep_h ep,
         }
 
         status = ucp_wireup_ep_connect_to_ep_v2(ucp_ep_get_lane(ep, lane),
-                                                address_entry, ep_entry);
+                                                address_entry, ep_entry,
+                                                ep->ep_traffic_class,
+                                                ep->flags);
         if (status != UCS_OK) {
             goto out;
         }

src/ucp/wireup/wireup_ep.c

@@ -513,6 +513,14 @@ ucs_status_t ucp_wireup_ep_connect(uct_ep_h uct_ep, unsigned ep_init_flags,
                                UCT_EP_PARAM_FIELD_PATH_INDEX;
     uct_ep_params.path_index = path_index;
     uct_ep_params.iface      = ucp_worker_iface(worker, rsc_index)->iface;
+
+    if (ucp_ep->flags & UCP_EP_FLAG_EP_TRAFFIC_CLASS) {
+        uct_ep_params.field_mask      |= UCT_EP_PARAM_FIELD_EP_TRAFFIC_CLASS;
+        uct_ep_params.ep_traffic_class = ucp_ep->ep_traffic_class;
+    } else {
+        uct_ep_params.ep_traffic_class = UCT_EP_NO_TCLASS;
+    }
+
     status = uct_ep_create(&uct_ep_params, &next_ep);
     if (status != UCS_OK) {
         /* make Coverity happy */
@@ -695,13 +703,19 @@ void ucp_wireup_eps_pending_extract(ucp_ep_t *ucp_ep, ucs_queue_head_t *queue)
 ucs_status_t
 ucp_wireup_ep_connect_to_ep_v2(uct_ep_h tl_ep,
                                const ucp_address_entry_t *address_entry,
-                               const ucp_address_entry_ep_addr_t *ep_entry)
+                               const ucp_address_entry_ep_addr_t *ep_entry,
+                               uint8_t ep_traffic_class, ucp_ep_flags_t flags)
 {
     const uct_ep_connect_to_ep_params_t param = {
-        .field_mask = UCT_EP_CONNECT_TO_EP_PARAM_FIELD_DEVICE_ADDR_LENGTH |
-                      UCT_EP_CONNECT_TO_EP_PARAM_FIELD_EP_ADDR_LENGTH,
+        .field_mask =
+                UCT_EP_CONNECT_TO_EP_PARAM_FIELD_DEVICE_ADDR_LENGTH |
+                UCT_EP_CONNECT_TO_EP_PARAM_FIELD_EP_ADDR_LENGTH |
+                ((flags & UCP_EP_FLAG_EP_TRAFFIC_CLASS) ?
+                         UCT_EP_CONNECT_TO_EP_PARAM_FIELD_EP_TRAFFIC_CLASS :
+                         0),
         .device_addr_length = address_entry->dev_addr_len,
-        .ep_addr_length     = ep_entry->len
+        .ep_addr_length     = ep_entry->len,
+        .ep_traffic_class   = ep_traffic_class
     };
     ucp_wireup_ep_t *wireup_ep                = ucp_wireup_ep(tl_ep);