UCT/GDAKI: Add endpoint (#10825)

Author: Artemy-Mellanox

src/uct/ib/base/ib_iface.c

@@ -1684,8 +1684,7 @@ UCS_CLASS_INIT_FUNC(uct_ib_iface_t, uct_iface_ops_t *tl_ops,
                             dev->stats)
                      UCS_STATS_ARG(params->mode.device.dev_name));
 
-    status = uct_ib_device_find_port(dev, params->mode.device.dev_name,
-                                     &port_num);
+    status = uct_ib_device_find_port(dev, init_attr->dev_name, &port_num);
     if (status != UCS_OK) {
         goto err;
     }

src/uct/ib/base/ib_iface.h

@@ -272,6 +272,7 @@ typedef struct uct_ib_iface_init_attr {
     unsigned    max_rd_atomic;
     uint8_t     cqe_zip_sizes[UCT_IB_DIR_LAST];
     uint16_t    tx_moderation;           /* TX CQ moderation */
+    const char  *dev_name;               /* Device Name */
 } uct_ib_iface_init_attr_t;
 
 

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

@@ -472,6 +472,7 @@ static UCS_CLASS_INIT_FUNC(uct_srd_iface_t, uct_md_h md, uct_worker_h worker,
     init_attr.rx_hdr_len            = sizeof(uct_srd_hdr_t);
     init_attr.seg_size              = ucs_min(mtu, config->super.seg_size);
     init_attr.qp_type               = IBV_QPT_DRIVER;
+    init_attr.dev_name              = params->mode.device.dev_name;
 
     UCS_CLASS_CALL_SUPER_INIT(uct_ib_iface_t, &uct_srd_iface_tl_ops,
                               &uct_srd_iface_ops, md, worker, params,

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

@@ -1608,6 +1608,7 @@ static UCS_CLASS_INIT_FUNC(uct_dc_mlx5_iface_t, uct_md_h tl_md, uct_worker_h wor
     init_attr.fc_req_size   = sizeof(uct_dc_fc_request_t);
     init_attr.max_rd_atomic = md->max_rd_atomic_dc;
     init_attr.tx_moderation = 0; /* disable tx moderation for dcs */
+    init_attr.dev_name      = params->mode.device.dev_name;
 
     if (md->flags & UCT_IB_MLX5_MD_FLAG_DC_TM) {
         init_attr.flags  |= UCT_IB_TM_SUPPORTED;

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

@@ -34,12 +34,305 @@ ucs_config_field_t uct_rc_gdaki_iface_config_table[] = {
   {NULL}
 };
 
+
+static UCS_CLASS_INIT_FUNC(uct_rc_gdaki_ep_t, const uct_ep_params_t *params)
+{
+    uct_rc_gdaki_iface_t *iface = ucs_derived_of(params->iface,
+                                                 uct_rc_gdaki_iface_t);
+    uct_ib_mlx5_md_t *md = ucs_derived_of(iface->super.super.super.super.md,
+                                          uct_ib_mlx5_md_t);
+    uct_ib_iface_init_attr_t init_attr = {};
+    uct_ib_mlx5_qp_attr_t qp_attr      = {};
+    ucs_status_t status;
+
+    UCS_CLASS_CALL_SUPER_INIT(uct_base_ep_t, &iface->super.super.super.super);
+
+    init_attr.cq_len[UCT_IB_DIR_TX] = iface->super.super.config.tx_qp_len *
+                                      UCT_IB_MLX5_MAX_BB;
+    init_attr.flags                 = UCT_IB_CQ_IGNORE_OVERRUN;
+    status = uct_ib_mlx5_devx_create_cq(&iface->super.super.super,
+                                        UCT_IB_DIR_TX, &init_attr, &self->cq, 0,
+                                        0);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    uct_rc_iface_fill_attr(&iface->super.super, &qp_attr.super,
+                           iface->super.super.config.tx_qp_len, NULL);
+    qp_attr.mmio_mode     = UCT_IB_MLX5_MMIO_MODE_DB;
+    qp_attr.super.srq_num = 0;
+    status = uct_ib_mlx5_devx_create_qp(&iface->super.super.super, &self->cq,
+                                        &self->cq, &self->qp.super, &self->qp,
+                                        &qp_attr);
+    if (status != UCS_OK) {
+        goto err_qp;
+    }
+
+    return UCS_OK;
+
+err_qp:
+    uct_ib_mlx5_devx_destroy_cq(md, &self->cq);
+    return status;
+}
+
+static UCS_CLASS_CLEANUP_FUNC(uct_rc_gdaki_ep_t)
+{
+    uct_rc_gdaki_iface_t *iface = ucs_derived_of(self->super.super.iface,
+                                                 uct_rc_gdaki_iface_t);
+    uct_ib_mlx5_md_t *md = ucs_derived_of(iface->super.super.super.super.md,
+                                          uct_ib_mlx5_md_t);
+
+    uct_ib_mlx5_devx_destroy_qp(md, &self->qp.super);
+    uct_ib_mlx5_devx_destroy_cq(md, &self->cq);
+}
+
+UCS_CLASS_DEFINE(uct_rc_gdaki_ep_t, uct_base_ep_t);
+UCS_CLASS_DEFINE_NEW_FUNC(uct_rc_gdaki_ep_t, uct_ep_t, const uct_ep_params_t *);
+UCS_CLASS_DEFINE_DELETE_FUNC(uct_rc_gdaki_ep_t, uct_ep_t);
+
+static ucs_status_t
+uct_rc_gdaki_ep_get_address(uct_ep_h tl_ep, uct_ep_addr_t *addr)
+{
+    uct_rc_gdaki_ep_t *ep = ucs_derived_of(tl_ep, uct_rc_gdaki_ep_t);
+    uct_rc_mlx5_base_ep_address_t *rc_addr = (void*)addr;
+
+    uct_ib_pack_uint24(rc_addr->qp_num, ep->qp.super.qp_num);
+    return UCS_OK;
+}
+
+static ucs_status_t uct_rc_gdaki_iface_get_address(uct_iface_h tl_iface,
+                                                   uct_iface_addr_t *addr)
+{
+    *(uint8_t*)addr = UCT_RC_MLX5_IFACE_ADDR_TYPE_BASIC;
+    return UCS_OK;
+}
+
+static ucs_status_t
+uct_rc_gdaki_ep_connect_to_ep_v2(uct_ep_h ep,
+                                 const uct_device_addr_t *device_addr,
+                                 const uct_ep_addr_t *ep_addr,
+                                 const uct_ep_connect_to_ep_params_t *params)
+{
+    uct_rc_gdaki_ep_t *gdaki_ep     = ucs_derived_of(ep, uct_rc_gdaki_ep_t);
+    uct_rc_gdaki_iface_t *iface     = ucs_derived_of(ep->iface,
+                                                     uct_rc_gdaki_iface_t);
+    const uct_ib_address_t *ib_addr = (void*)device_addr;
+    const uct_rc_mlx5_base_ep_address_t *rc_addr = (void*)ep_addr;
+    uint8_t path_index                           = 0;
+    struct ibv_ah_attr ah_attr;
+    enum ibv_mtu path_mtu;
+    uint32_t dest_qp_num;
+    ucs_status_t status;
+
+    status = uct_ib_iface_fill_ah_attr_from_addr(&iface->super.super.super,
+                                                 ib_addr, path_index, &ah_attr,
+                                                 &path_mtu);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    ucs_assert(path_mtu != UCT_IB_ADDRESS_INVALID_PATH_MTU);
+    dest_qp_num = uct_ib_unpack_uint24(rc_addr->qp_num);
+
+    return uct_rc_mlx5_iface_common_devx_connect_qp(
+            &iface->super, &gdaki_ep->qp.super, dest_qp_num, &ah_attr, path_mtu,
+            path_index, iface->super.super.config.max_rd_atomic);
+}
+
+int uct_rc_gdaki_ep_is_connected(uct_ep_h tl_ep,
+                                 const uct_ep_is_connected_params_t *params)
+{
+    uct_rc_gdaki_ep_t *ep       = ucs_derived_of(tl_ep, uct_rc_gdaki_ep_t);
+    uct_rc_gdaki_iface_t *iface = ucs_derived_of(ep->super.super.iface,
+                                                 uct_rc_gdaki_iface_t);
+    uint32_t addr_qp            = 0;
+    uct_rc_mlx5_base_ep_address_t *rc_addr;
+    ucs_status_t status;
+    struct ibv_ah_attr ah_attr;
+    uint32_t qp_num;
+    union ibv_gid *rgid;
+    const uct_ib_address_t *ib_addr;
+
+    status = uct_ib_mlx5_query_qp_peer_info(&iface->super.super.super,
+                                            &ep->qp.super, &ah_attr, &qp_num);
+    if (status != UCS_OK) {
+        return 0;
+    }
+
+    /* TODO unite code with uct_rc_mlx5_base_ep_is_connected */
+    if (params->field_mask & UCT_EP_IS_CONNECTED_FIELD_EP_ADDR) {
+        rc_addr = (uct_rc_mlx5_base_ep_address_t*)params->ep_addr;
+        addr_qp = uct_ib_unpack_uint24(rc_addr->qp_num);
+    }
+
+    if ((addr_qp != 0) && (qp_num != addr_qp)) {
+        return 0;
+    }
+
+    rgid    = (ah_attr.is_global) ? &ah_attr.grh.dgid : NULL;
+    ib_addr = (const uct_ib_address_t*)params->device_addr;
+    return uct_ib_iface_is_same_device(ib_addr, ah_attr.dlid, rgid);
+}
+
+static ucs_status_t
+uct_rc_gdaki_iface_query(uct_iface_h tl_iface, uct_iface_attr_t *iface_attr)
+{
+    uct_rc_gdaki_iface_t *iface = ucs_derived_of(tl_iface,
+                                                 uct_rc_gdaki_iface_t);
+    ucs_status_t status;
+
+    status = uct_ib_iface_query(&iface->super.super.super, 0, iface_attr);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    /* TODO:
+     *  - add UCT_IFACE_FLAG_PUT_BATCH
+     *  - PENDING and PUT_ZCOPY will be needed to establish rma_bw lanes
+     *  - As this lane does not really support PUT_ZCOPY and PENDING, this could be
+     *    causing issue when trying to send standard PUT. Eventually we must probably
+     *    introduce another type of lane (rma_batch#x).
+     */
+    iface_attr->cap.flags      = UCT_IFACE_FLAG_CONNECT_TO_EP;
+    iface_attr->ep_addr_len    = sizeof(uct_rc_mlx5_base_ep_address_t);
+    iface_attr->iface_addr_len = sizeof(uint8_t);
+    iface_attr->overhead       = UCT_RC_MLX5_IFACE_OVERHEAD;
+
+    iface_attr->cap.put.min_zcopy = 0;
+    iface_attr->cap.put.max_zcopy =
+            uct_ib_iface_port_attr(&iface->super.super.super)->max_msg_sz;
+    return UCS_OK;
+}
+
+ucs_status_t
+uct_rc_gdaki_create_cq(uct_ib_iface_t *ib_iface, uct_ib_dir_t dir,
+                       const uct_ib_iface_init_attr_t *init_attr,
+                       int preferred_cpu, size_t inl)
+{
+    uct_rc_gdaki_iface_t *iface = ucs_derived_of(ib_iface,
+                                                 uct_rc_gdaki_iface_t);
+
+    iface->super.cq[dir].type = UCT_IB_MLX5_OBJ_TYPE_NULL;
+    return UCS_OK;
+}
+
+static UCS_CLASS_DECLARE_NEW_FUNC(uct_rc_gdaki_iface_t, uct_iface_t, uct_md_h,
+                                  uct_worker_h, const uct_iface_params_t*,
+                                  const uct_iface_config_t*);
+
+static UCS_CLASS_DECLARE_DELETE_FUNC(uct_rc_gdaki_iface_t, uct_iface_t);
+
+static uct_rc_iface_ops_t uct_rc_gdaki_internal_ops = {
+    .super = {
+        .super = {
+            .iface_estimate_perf   = uct_ib_iface_estimate_perf,
+            .iface_vfs_refresh     = (uct_iface_vfs_refresh_func_t)ucs_empty_function,
+            .ep_query              = (uct_ep_query_func_t)ucs_empty_function_return_unsupported,
+            .ep_invalidate         = (uct_ep_invalidate_func_t)ucs_empty_function_return_unsupported,
+            .ep_connect_to_ep_v2   = uct_rc_gdaki_ep_connect_to_ep_v2,
+            .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,
+        },
+        .create_cq  = uct_rc_gdaki_create_cq,
+        .destroy_cq = (uct_ib_iface_destroy_cq_func_t)ucs_empty_function_return_success,
+    },
+    .init_rx    = (uct_rc_iface_init_rx_func_t)ucs_empty_function_return_success,
+    .cleanup_rx = (uct_rc_iface_cleanup_rx_func_t)
+            ucs_empty_function_return_success,
+};
+
+static uct_iface_ops_t uct_rc_gdaki_iface_tl_ops = {
+    .ep_flush          = uct_base_ep_flush,
+    .ep_fence          = uct_base_ep_fence,
+    .ep_create         = UCS_CLASS_NEW_FUNC_NAME(uct_rc_gdaki_ep_t),
+    .ep_destroy        = UCS_CLASS_DELETE_FUNC_NAME(uct_rc_gdaki_ep_t),
+    .ep_get_address    = uct_rc_gdaki_ep_get_address,
+    .ep_connect_to_ep  = uct_base_ep_connect_to_ep,
+    .ep_pending_purge  = (uct_ep_pending_purge_func_t)ucs_empty_function,
+    .iface_close       = UCS_CLASS_DELETE_FUNC_NAME(uct_rc_gdaki_iface_t),
+    .iface_query       = uct_rc_gdaki_iface_query,
+    .iface_get_address = uct_rc_gdaki_iface_get_address,
+    .iface_get_device_address = uct_ib_iface_get_device_address,
+    .iface_is_reachable       = uct_base_iface_is_reachable,
+    .iface_flush              = (uct_iface_flush_func_t)
+            ucs_empty_function_return_success,
+    .iface_fence              = (uct_iface_fence_func_t)
+            ucs_empty_function_return_unsupported,
+    .iface_progress_enable    = (uct_iface_progress_enable_func_t)
+            ucs_empty_function_return_unsupported,
+    .iface_progress_disable   = (uct_iface_progress_disable_func_t)
+            ucs_empty_function_return_unsupported,
+    .iface_progress           = (uct_iface_progress_func_t)
+            ucs_empty_function_return_unsupported,
+};
+
+
 static UCS_CLASS_INIT_FUNC(uct_rc_gdaki_iface_t, uct_md_h tl_md,
                            uct_worker_h worker,
                            const uct_iface_params_t *params,
                            const uct_iface_config_t *tl_config)
 {
-    return UCS_ERR_NOT_IMPLEMENTED;
+    uct_rc_gdaki_iface_config_t *config =
+            ucs_derived_of(tl_config, uct_rc_gdaki_iface_config_t);
+    uct_ib_mlx5_md_t *md = ucs_derived_of(tl_md, uct_ib_mlx5_md_t);
+    uct_ib_iface_init_attr_t init_attr = {};
+    UCS_STRING_BUFFER_ONSTACK(strb, 64);
+    char *gpu_name, *ib_name;
+    char pci_addr[UCS_SYS_BDF_NAME_MAX];
+    ucs_status_t status;
+    doca_error_t derr;
+    int cuda_id;
+
+    status = uct_rc_mlx5_dp_ordering_ooo_init(md, &self->super,
+                                              md->dp_ordering_cap.rc,
+                                              &config->mlx5, "gdaki");
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    ucs_string_buffer_appendf(&strb, "%s", params->mode.device.dev_name);
+    gpu_name = ucs_string_buffer_next_token(&strb, NULL, "-");
+    ib_name  = ucs_string_buffer_next_token(&strb, gpu_name, "-");
+
+    init_attr.seg_size = config->super.super.seg_size;
+    init_attr.qp_type  = IBV_QPT_RC;
+    init_attr.dev_name = ib_name;
+
+    UCS_CLASS_CALL_SUPER_INIT(uct_rc_mlx5_iface_common_t,
+                              &uct_rc_gdaki_iface_tl_ops,
+                              &uct_rc_gdaki_internal_ops, tl_md, worker, params,
+                              &config->super, &config->mlx5, &init_attr);
+
+    if (memcmp(gpu_name, UCT_DEV_CUDA_NAME, UCT_DEV_CUDA_NAME_LEN)) {
+        ucs_error("wrong device name: %s\n", gpu_name);
+        return status;
+    }
+
+    cuda_id = atoi(gpu_name + UCT_DEV_CUDA_NAME_LEN);
+    status = UCT_CUDADRV_FUNC_LOG_ERR(cuDeviceGetPCIBusId(
+                    pci_addr, UCS_SYS_BDF_NAME_MAX, cuda_id));
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    derr = doca_gpu_create(pci_addr, &self->gpu_dev);
+    if (derr != DOCA_SUCCESS) {
+        status = UCS_ERR_IO_ERROR;
+        ucs_error("doca_gpu_create failed: %s %s", doca_error_get_descr(derr),
+                  pci_addr);
+        return status;
+    }
+
+    status = UCT_CUDADRV_FUNC_LOG_ERR(cuDeviceGet(&self->cuda_dev, cuda_id));
+    if (status != UCS_OK) {
+        goto err_doca;
+    }
+
+    return UCS_OK;
+
+err_doca:
+    doca_gpu_destroy(self->gpu_dev);
+    return status;
 }
 
 static UCS_CLASS_CLEANUP_FUNC(uct_rc_gdaki_iface_t)
@@ -53,6 +346,8 @@ static UCS_CLASS_DEFINE_NEW_FUNC(uct_rc_gdaki_iface_t, uct_iface_t, uct_md_h,
                                  uct_worker_h, const uct_iface_params_t*,
                                  const uct_iface_config_t*);
 
+static UCS_CLASS_DEFINE_DELETE_FUNC(uct_rc_gdaki_iface_t, uct_iface_t);
+
 static ucs_status_t
 uct_gdaki_query_tl_devices(uct_md_h md, uct_tl_device_resource_t **tl_devices_p,
                            unsigned *num_tl_devices_p)
@@ -94,8 +389,9 @@ uct_gdaki_query_tl_devices(uct_md_h md, uct_tl_device_resource_t **tl_devices_p,
         }
 
         snprintf(tl_devices[num_tl_devices].name,
-                 sizeof(tl_devices[num_tl_devices].name), "cuda%d-%s", device,
-                 uct_ib_device_name(&ib_md->dev));
+                 sizeof(tl_devices[num_tl_devices].name), "cuda%d-%s:%d",
+                 device, uct_ib_device_name(&ib_md->dev),
+                 ib_md->dev.first_port);
         tl_devices[num_tl_devices].type       = UCT_DEVICE_TYPE_NET;
         tl_devices[num_tl_devices].sys_device = dev;
         num_tl_devices++;

src/uct/ib/mlx5/gdaki/gdaki.h

@@ -6,14 +6,28 @@
 #ifndef UCT_GDAKI_IFACE_H
 #define UCT_GDAKI_IFACE_H
 
+#include <uct/ib/mlx5/rc/rc_mlx5_common.h>
 #include <uct/base/uct_iface.h>
 
 #include <cuda.h>
 #include <doca_gpunetio.h>
 
+
+#define UCT_DEV_CUDA_NAME "cuda"
+#define UCT_DEV_CUDA_NAME_LEN 4
+
+
 typedef struct uct_rc_gdaki_iface {
-    struct doca_gpu *gpu_dev;
-    CUdevice        cuda_dev;
+    uct_rc_mlx5_iface_common_t super;
+    struct doca_gpu            *gpu_dev;
+    CUdevice                   cuda_dev;
 } uct_rc_gdaki_iface_t;
 
+
+typedef struct uct_rc_gdaki_ep {
+    uct_base_ep_t      super;
+    uct_ib_mlx5_cq_t   cq;
+    uct_ib_mlx5_txwq_t qp;
+} uct_rc_gdaki_ep_t;
+
 #endif /* UCT_GDAKI_IFACE_H */