Merge pull request #10760 from tvegas1/direct_nic_datapath

Direct NIC Datapath

Author: brminich

src/ucp/core/ucp_context.c

@@ -1190,6 +1190,7 @@ static void ucp_add_tl_resource_if_enabled(
         const uct_tl_resource_desc_t *resource, unsigned *num_resources_p,
         uint64_t dev_cfg_masks[], uint64_t *tl_cfg_mask)
 {
+    ucp_tl_md_t *md = &context->tl_mds[md_index];
     uint8_t rsc_flags;
     ucp_rsc_index_t dev_index, i;
 
@@ -1220,6 +1221,10 @@ static void ucp_add_tl_resource_if_enabled(
         }
         context->tl_rscs[context->num_tls].dev_index = dev_index;
 
+        if (resource->sys_device < UCP_MAX_SYS_DEVICES) {
+            md->sys_dev_map |= UCS_BIT(resource->sys_device);
+        }
+
         ++context->num_tls;
         ++(*num_resources_p);
     }
@@ -1429,8 +1434,9 @@ static ucs_status_t ucp_fill_tl_md(ucp_context_h context,
     ucs_status_t status;
 
     /* Initialize tl_md structure */
-    tl_md->cmpt_index = cmpt_index;
-    tl_md->rsc        = *md_rsc;
+    tl_md->cmpt_index  = cmpt_index;
+    tl_md->rsc         = *md_rsc;
+    tl_md->sys_dev_map = 0;
 
     /* Read MD configuration */
     status = uct_md_config_read(context->tl_cmpts[cmpt_index].cmpt, NULL, NULL,

src/ucp/core/ucp_context.h

@@ -326,6 +326,11 @@ typedef struct ucp_tl_md {
      * Global VA memory handle
      */
     uct_mem_h              gva_mr;
+
+    /**
+     * Set of known system devices associated to the MD
+     */
+    ucp_sys_dev_map_t      sys_dev_map;
 } ucp_tl_md_t;
 
 

src/ucp/core/ucp_mm.c

@@ -518,6 +518,28 @@ static ucs_status_t ucp_memh_register_gva(ucp_context_h context, ucp_mem_h memh,
     return UCS_OK;
 }
 
+static int ucp_memh_sys_dev_reachable(ucs_sys_device_t mem_sys_dev,
+                                      ucp_sys_dev_map_t sys_dev_map)
+{
+    ucs_sys_device_t sys_dev;
+
+    if (mem_sys_dev == UCS_SYS_DEVICE_ID_UNKNOWN) {
+        return 1;
+    }
+
+    /*
+     * If at least one sys_dev is not reachable, do not register on it
+     * as we cannot know in advance which device is going to be used.
+     */
+    ucs_for_each_bit(sys_dev, sys_dev_map) {
+        if (!ucs_topo_is_reachable(sys_dev, mem_sys_dev)) {
+            return 0;
+        }
+    }
+
+    return 1;
+}
+
 static ucs_status_t
 ucp_memh_register_internal(ucp_context_h context, ucp_mem_h memh,
                            ucp_md_map_t md_map, unsigned uct_flags,
@@ -538,6 +560,7 @@ ucp_memh_register_internal(ucp_context_h context, ucp_mem_h memh,
     void *reg_address;
     size_t reg_length;
     size_t reg_align;
+    ucp_sys_dev_map_t sys_dev_map;
 
     if (gva_enable) {
         status = ucp_memh_register_gva(context, memh, md_map);
@@ -565,7 +588,8 @@ ucp_memh_register_internal(ucp_context_h context, ucp_mem_h memh,
         (reg_md_map & context->dmabuf_reg_md_map)) {
         /* Query dmabuf file descriptor and offset */
         mem_attr.field_mask = UCT_MD_MEM_ATTR_FIELD_DMABUF_FD |
-                              UCT_MD_MEM_ATTR_FIELD_DMABUF_OFFSET;
+                              UCT_MD_MEM_ATTR_FIELD_DMABUF_OFFSET |
+                              UCT_MD_MEM_ATTR_FIELD_SYS_DEV;
         status = uct_md_mem_query(context->tl_mds[dmabuf_prov_md_index].md,
                                   address, length, &mem_attr);
         if (status != UCS_OK) {
@@ -574,12 +598,24 @@ ucp_memh_register_internal(ucp_context_h context, ucp_mem_h memh,
                     address, length, ucs_status_string(status));
         } else {
             ucs_trace("uct_md_mem_query(dmabuf address %p length %zu) returned "
-                      "fd %d offset %zu",
+                      "fd %d offset %zu sys_dev %u",
                       address, length, mem_attr.dmabuf_fd,
-                      mem_attr.dmabuf_offset);
+                      mem_attr.dmabuf_offset, mem_attr.sys_dev);
+
             dmabuf_md_map            = context->dmabuf_reg_md_map;
             reg_params.dmabuf_fd     = mem_attr.dmabuf_fd;
             reg_params.dmabuf_offset = mem_attr.dmabuf_offset;
+
+            /* Exclude any unreachable MD from registration */
+            ucs_for_each_bit(md_index, dmabuf_md_map) {
+                sys_dev_map = context->tl_mds[md_index].sys_dev_map;
+                if (!ucp_memh_sys_dev_reachable(mem_attr.sys_dev,
+                                                sys_dev_map)) {
+                    ucs_trace("md[%d] skipped: cannot reach mem_sys_dev=%u",
+                              md_index, mem_attr.sys_dev);
+                    reg_md_map &= ~UCS_BIT(md_index);
+                }
+            }
         }
     }
 
@@ -835,7 +871,7 @@ static ucs_status_t ucp_memh_init_uct_reg(ucp_context_h context, ucp_mem_h memh,
             goto err;
         }
 
-        ucp_memh_init_from_parent(memh, cache_md_map);
+        ucp_memh_init_from_parent(memh, memh->parent->md_map);
 
         status = ucp_memh_register(context, memh, reg_md_map, uct_flags,
                                    alloc_name);

src/ucp/proto/proto_common.c

@@ -576,6 +576,7 @@ ucp_proto_common_find_lanes(const ucp_proto_init_params_t *params,
     ucp_md_index_t md_index;
     ucp_lane_map_t lane_map;
     char lane_desc[64];
+    ucs_sys_device_t lane_sys_dev;
 
     if (max_lanes == 0) {
         return 0;
@@ -676,6 +677,15 @@ ucp_proto_common_find_lanes(const ucp_proto_init_params_t *params,
             }
         }
 
+        /* The two devices must also have internal reachability */
+        lane_sys_dev = context->tl_rscs[rsc_index].tl_rsc.sys_device;
+        if (!ucs_topo_is_reachable(lane_sys_dev, select_param->sys_dev)) {
+            ucs_trace("%s: no reachability between lane_sys_dev=%u and "
+                      "sys_dev=%u",
+                      lane_desc, lane_sys_dev, select_param->sys_dev);
+            continue;
+        }
+
         ucs_trace("%s: added as lane %d", lane_desc, lane);
         lanes[num_lanes++] = lane;
     }