Merge pull request #8711 from awlauria/account_for_more_than_2_btls_v5.0.x

v5.0.x: osc/rdma: Account for more than 2 btls.

Author: gpaulsen

ompi/mca/osc/rdma/osc_rdma.h

@@ -55,7 +55,7 @@
 
 #define RANK_ARRAY_COUNT(module) ((ompi_comm_size ((module)->comm) + (module)->node_count - 1) / (module)->node_count)
 
-#define MCA_OSC_RDMA_MAX_USED_BTLS 2
+#define MCA_OSC_RDMA_BTLS_SIZE_INIT 4
 
 enum {
     OMPI_OSC_RDMA_LOCKING_TWO_LEVEL,
@@ -260,7 +260,8 @@ struct ompi_osc_rdma_module_t {
      * non-RDMA BTLs. The typical usage is btl/sm + btl/tcp. In the future this
      * could be used to support multiple RDMA-capable BTLs but the memory registration
      * paths will need to be updated to pack/unpack multiple registration handles. */
-    struct mca_btl_base_module_t *selected_btls[MCA_OSC_RDMA_MAX_USED_BTLS];
+    struct mca_btl_base_module_t **selected_btls;
+    uint8_t selected_btls_size;
     uint8_t btls_in_use;
 
     /** Only true if one BTL is in use. Memory registration is only supported when
@@ -634,4 +635,14 @@ static inline mca_btl_base_module_t *ompi_osc_rdma_selected_btl (ompi_osc_rdma_m
     return module->selected_btls[btl_index];
 }
 
+__opal_attribute_always_inline__
+static inline void ompi_osc_rdma_selected_btl_insert (ompi_osc_rdma_module_t *module, struct mca_btl_base_module_t *btl, uint8_t btl_index) {
+    if(btl_index == module->selected_btls_size) {
+        module->selected_btls_size *= 2;
+        module->selected_btls = realloc(module->selected_btls, module->selected_btls_size * sizeof(struct mca_btl_base_module_t *));
+        assert(NULL != module->selected_btls);
+    }
+    module->selected_btls[btl_index] = btl;
+}
+
 #endif /* OMPI_OSC_RDMA_H */

ompi/mca/osc/rdma/osc_rdma_component.c

@@ -947,7 +947,7 @@ static int ompi_osc_rdma_query_alternate_btls (ompi_communicator_t *comm, ompi_o
             ++btls_found;
             if (module) {
                 mca_btl_base_am_rdma_init(item->btl_module);
-                module->selected_btls[module->btls_in_use++] = item->btl_module;
+                ompi_osc_rdma_selected_btl_insert(module, item->btl_module, module->btls_in_use++);
             }
 
         }
@@ -974,7 +974,7 @@ static int ompi_osc_rdma_query_btls (ompi_communicator_t *comm, ompi_osc_rdma_mo
     btls_to_use = opal_argv_split (ompi_osc_rdma_btl_names, ',');
 
     if (module) {
-        module->selected_btls[0] = NULL;
+        ompi_osc_rdma_selected_btl_insert(module, NULL, 0);
         module->btls_in_use = 0;
         module->use_memory_registration = false;
     }
@@ -1001,7 +1001,7 @@ static int ompi_osc_rdma_query_btls (ompi_communicator_t *comm, ompi_osc_rdma_mo
 
     if (NULL != selected_btl) {
         if (module) {
-            module->selected_btls[0] = selected_btl;
+            ompi_osc_rdma_selected_btl_insert(module, selected_btl, 0);
             module->btls_in_use = 1;
             module->use_memory_registration = selected_btl->btl_register_mem != NULL;
         }
@@ -1118,7 +1118,7 @@ static int ompi_osc_rdma_query_btls (ompi_communicator_t *comm, ompi_osc_rdma_mo
     }
 
     if (module) {
-        module->selected_btls[0] = selected_btl;
+        ompi_osc_rdma_selected_btl_insert(module, selected_btl, 0);
         module->btls_in_use = 1;
         module->use_memory_registration = selected_btl->btl_register_mem != NULL;
     }
@@ -1334,6 +1334,9 @@ static int ompi_osc_rdma_component_select (struct ompi_win_t *win, void **base,
     module->acc_use_amo = mca_osc_rdma_component.acc_use_amo;
     module->network_amo_max_count = mca_osc_rdma_component.network_amo_max_count;
 
+    module->selected_btls_size = MCA_OSC_RDMA_BTLS_SIZE_INIT;
+    module->selected_btls = calloc(module->selected_btls_size, sizeof(struct mca_btl_base_module_t *));
+
     module->all_sync.module = module;
 
     module->flavor = flavor;

ompi/mca/osc/rdma/osc_rdma_module.c

@@ -141,6 +141,7 @@ int ompi_osc_rdma_free(ompi_win_t *win)
     free (module->peer_array);
     free (module->outstanding_lock_array);
     free (module->free_after);
+    free (module->selected_btls);
     free (module);
 
     return OMPI_SUCCESS;