coll/acoll: A few fixes and fallbacks

amd-nithyavs · MithunMohanKadavil · commit 07302f65aa63 · 2025-05-09T05:49:26.000+05:30
Code changes to remove a few coverity scan warnings, fallbacks to base
algorithms for custom datatypes.

Signed-off-by: Nithya V S &lt;Nithya.VS@amd.com&gt;
diff --git a/ompi/mca/coll/acoll/README b/ompi/mca/coll/acoll/README
@@ -8,9 +8,9 @@ $HEADER$
 
 ===========================================================================
 
-The collective component, AMD Coll (“acoll”), is a high-performant MPI collective component for the OpenMPI library that is optimized for AMD "Zen"-based processors. “acoll” is optimized for communications within a single node of AMD “Zen”-based processors and provides the following commonly used collective algorithms: boardcast (MPI_Bcast), allreduce (MPI_Allreduce), reduce (MPI_Reduce), gather (MPI_Gather), allgather (MPI_Allgather), and barrier (MPI_Barrier).
+The collective component, AMD Coll (“acoll”), is a high-performant MPI collective component for the OpenMPI library that is optimized for AMD "Zen"-based processors. “acoll” is optimized for communications within a single node of AMD “Zen”-based processors and provides the following commonly used collective algorithms: boardcast (MPI_Bcast), allreduce (MPI_Allreduce), reduce (MPI_Reduce), gather (MPI_Gather), allgather (MPI_Allgather), alltoall (MPI_Alltoall), and barrier (MPI_Barrier).
 
-At present, “acoll” has been tested with OpenMPI v5.0.2 and can be built as part of OpenMPI.
+At present, “acoll” has been tested with OpenMPI main branch and can be built as part of OpenMPI.
 
 To run an application with acoll, use the following command line parameters
 -              mpirun <common mpi runtime parameters> --mca coll acoll,tuned,libnbc,basic --mca coll_acoll_priority 40 <executable>
diff --git a/ompi/mca/coll/acoll/coll_acoll_allgather.c b/ompi/mca/coll/acoll/coll_acoll_allgather.c
@@ -29,7 +29,7 @@ static inline int log_sg_bcast_intra(void *buff, size_t count, struct ompi_datat
                                      mca_coll_base_module_t *module, ompi_request_t **preq,
                                      int *nreqs)
 {
-    int msb_pos, sub_rank, peer, err;
+    int msb_pos, sub_rank, peer, err = MPI_SUCCESS;
     int i, mask;
     int end_sg, end_peer;
 
@@ -92,7 +92,7 @@ static inline int lin_sg_bcast_intra(void *buff, size_t count, struct ompi_datat
                                      int *nreqs)
 {
     int peer;
-    int err;
+    int err = MPI_SUCCESS;
     int sg_end;
 
     sg_end = sg_start + sg_size - 1;
diff --git a/ompi/mca/coll/acoll/coll_acoll_allreduce.c b/ompi/mca/coll/acoll/coll_acoll_allreduce.c
@@ -450,7 +450,21 @@ int mca_coll_acoll_allreduce_intra(const void *sbuf, void *rbuf, size_t count,
     ompi_datatype_type_size(dtype, &dsize);
     total_dsize = dsize * count;
 
-    if (1 == size) {
+    /* Disable shm/xpmem based optimizations if: */
+    /* - datatype is not a predefined type */
+    /* - it's a gpu buffer */
+    uint64_t flags = 0;
+    int dev_id;
+    bool is_opt = true;
+    if (!OMPI_COMM_CHECK_ASSERT_NO_ACCEL_BUF(comm)) {
+        if (!ompi_datatype_is_predefined(dtype)
+            || (0 < opal_accelerator.check_addr(sbuf, &dev_id, &flags))
+            || (0 < opal_accelerator.check_addr(rbuf, &dev_id, &flags))) {
+            is_opt = false;
+        }
+    }
+
+    if ((1 == size) && is_opt) {
         if (MPI_IN_PLACE != sbuf) {
             memcpy((char *) rbuf, sbuf, total_dsize);
         }
@@ -486,7 +500,7 @@ int mca_coll_acoll_allreduce_intra(const void *sbuf, void *rbuf, size_t count,
         if (total_dsize < 32) {
             return ompi_coll_base_allreduce_intra_recursivedoubling(sbuf, rbuf, count, dtype, op,
                                                                     comm, module);
-        } else if (total_dsize < 512) {
+        } else if ((total_dsize < 512) && is_opt) {
             return mca_coll_acoll_allreduce_small_msgs_h(sbuf, rbuf, count, dtype, op, comm, module,
                                                          subc, 1);
         } else if (total_dsize <= 2048) {
@@ -505,7 +519,7 @@ int mca_coll_acoll_allreduce_intra(const void *sbuf, void *rbuf, size_t count,
             }
         } else if (total_dsize < 4194304) {
 #ifdef HAVE_XPMEM_H
-            if (((subc->xpmem_use_sr_buf != 0) || (subc->xpmem_buf_size > 2 * total_dsize)) && (subc->without_xpmem != 1)) {
+            if (((subc->xpmem_use_sr_buf != 0) || (subc->xpmem_buf_size > 2 * total_dsize)) && (subc->without_xpmem != 1) && is_opt) {
                 return mca_coll_acoll_allreduce_xpmem_f(sbuf, rbuf, count, dtype, op, comm, module, subc);
             } else {
                 return ompi_coll_base_allreduce_intra_redscat_allgather(sbuf, rbuf, count, dtype,
@@ -517,7 +531,7 @@ int mca_coll_acoll_allreduce_intra(const void *sbuf, void *rbuf, size_t count,
 #endif
         } else if (total_dsize <= 16777216) {
 #ifdef HAVE_XPMEM_H
-            if (((subc->xpmem_use_sr_buf != 0) || (subc->xpmem_buf_size > 2 * total_dsize)) && (subc->without_xpmem != 1)) {
+            if (((subc->xpmem_use_sr_buf != 0) || (subc->xpmem_buf_size > 2 * total_dsize)) && (subc->without_xpmem != 1) && is_opt) {
                 mca_coll_acoll_reduce_xpmem_h(sbuf, rbuf, count, dtype, op, comm, module, subc);
                 return mca_coll_acoll_bcast(rbuf, count, dtype, 0, comm, module);
             } else {
@@ -530,7 +544,7 @@ int mca_coll_acoll_allreduce_intra(const void *sbuf, void *rbuf, size_t count,
 #endif
         } else {
 #ifdef HAVE_XPMEM_H
-            if (((subc->xpmem_use_sr_buf != 0) || (subc->xpmem_buf_size > 2 * total_dsize)) && (subc->without_xpmem != 1)) {
+            if (((subc->xpmem_use_sr_buf != 0) || (subc->xpmem_buf_size > 2 * total_dsize)) && (subc->without_xpmem != 1) && is_opt) {
                 return mca_coll_acoll_allreduce_xpmem_f(sbuf, rbuf, count, dtype, op, comm, module, subc);
             } else {
                 return ompi_coll_base_allreduce_intra_redscat_allgather(sbuf, rbuf, count, dtype,
diff --git a/ompi/mca/coll/acoll/coll_acoll_alltoall.c b/ompi/mca/coll/acoll/coll_acoll_alltoall.c
@@ -529,14 +529,8 @@ int mca_coll_acoll_alltoall
         struct ompi_communicator_t *split_comm;
 
         /* Select the right split_comm. */
-        int pow2_idx = -2;
-        int tmp_grp_split_f = grp_split_f;
-        while (tmp_grp_split_f > 0)
-        {
-            pow2_idx += 1;
-            tmp_grp_split_f = tmp_grp_split_f / 2;
-        }
-        split_comm = subc->split_comm[pow2_idx];
+        int comm_idx = grp_split_f > 2 ? opal_cube_dim(grp_split_f/2) : 0;
+        split_comm = subc->split_comm[comm_idx];
 
         error = mca_coll_acoll_base_alltoall_dispatcher
                         (sbuf, (grp_split_f * scount), sdtype,
diff --git a/ompi/mca/coll/acoll/coll_acoll_bcast.c b/ompi/mca/coll/acoll/coll_acoll_bcast.c
@@ -686,6 +686,18 @@ int mca_coll_acoll_bcast(void *buff, size_t count, struct ompi_datatype_t *datat
     if (size <= 2)
         no_sg = 1;
 
+    /* Disable shm based bcast if: */
+    /* - datatype is not a predefined type */
+    /* - it's a gpu buffer */
+    uint64_t flags = 0;
+    int dev_id;
+    if (!OMPI_COMM_CHECK_ASSERT_NO_ACCEL_BUF(comm)) {
+        if (!ompi_datatype_is_predefined(datatype)
+            || (0 < opal_accelerator.check_addr(buff, &dev_id, &flags))) {
+            use_shm = 0;
+        }
+    }
+
     coll_acoll_bcast_subcomms(comm, subc, subcomms, subc_roots, root, num_nodes, use_0, no_sg,
                               use_numa, use_socket);
 
diff --git a/ompi/mca/coll/acoll/coll_acoll_reduce.c b/ompi/mca/coll/acoll/coll_acoll_reduce.c
@@ -119,7 +119,7 @@ static inline int coll_acoll_reduce_topo(const void *sbuf, void *rbuf, size_t co
                                     MCA_COLL_BASE_TAG_REDUCE, MCA_PML_BASE_SEND_STANDARD,
                                     subc->base_comm[ind1][ind2]));
             if (ret != MPI_SUCCESS) {
-                free(pml_buffer);
+                free(free_buffer);
                 if (NULL != tmp_rbuf) {
                     coll_acoll_buf_free(reserve_mem_rbuf_reduce, tmp_rbuf);
                 }
@@ -134,7 +134,7 @@ static inline int coll_acoll_reduce_topo(const void *sbuf, void *rbuf, size_t co
                 ret = MCA_PML_CALL(recv(pml_buffer, count, dtype, i, MCA_COLL_BASE_TAG_REDUCE,
                                         subc->base_comm[ind1][ind2], MPI_STATUS_IGNORE));
                 if (ret != MPI_SUCCESS) {
-                    free(pml_buffer);
+                    free(free_buffer);
                     return ret;
                 }
                 ompi_op_reduce(op, pml_buffer, rbuf, count, dtype);
@@ -143,8 +143,8 @@ static inline int coll_acoll_reduce_topo(const void *sbuf, void *rbuf, size_t co
     }
 
     /* if local root, reduce at root */
-    if (is_base && (sz > 1)) {
-        free(pml_buffer);
+    if (is_base) {
+        free(free_buffer);
         if (rank != root && NULL != tmp_rbuf) {
             coll_acoll_buf_free(reserve_mem_rbuf_reduce, tmp_rbuf);
         }
@@ -329,6 +329,20 @@ int mca_coll_acoll_reduce_intra(const void *sbuf, void *rbuf, size_t count,
                                                     module, 0, 0);
     }
 
+    /* Disable shm/xpmem based optimizations if: */
+    /* - datatype is not a predefined type */
+    /* - it's a gpu buffer */
+    uint64_t flags = 0;
+    int dev_id;
+    bool is_opt = true;
+    if (!OMPI_COMM_CHECK_ASSERT_NO_ACCEL_BUF(comm)) {
+        if (!ompi_datatype_is_predefined(dtype)
+            || (0 < opal_accelerator.check_addr(sbuf, &dev_id, &flags))
+            || (0 < opal_accelerator.check_addr(rbuf, &dev_id, &flags))) {
+            is_opt = false;
+        }
+    }
+
     ompi_datatype_type_size(dtype, &dsize);
     total_dsize = dsize * count;
 
@@ -374,7 +388,7 @@ int mca_coll_acoll_reduce_intra(const void *sbuf, void *rbuf, size_t count,
                   && (acoll_module->reserve_mem_s).reserve_mem_allocate
                   && ((acoll_module->reserve_mem_s).reserve_mem_size >= total_dsize))
                  || ((0 == subc->xpmem_use_sr_buf) && (subc->xpmem_buf_size > 2 * total_dsize)))
-                && (subc->without_xpmem != 1)) {
+                && (subc->without_xpmem != 1) && is_opt) {
                 return mca_coll_acoll_reduce_xpmem(sbuf, rbuf, count, dtype, op, root, comm,
                                                    module, subc);
             } else {
diff --git a/ompi/mca/coll/acoll/coll_acoll_utils.h b/ompi/mca/coll/acoll/coll_acoll_utils.h
@@ -101,13 +101,15 @@ static inline int check_and_create_subc(ompi_communicator_t *comm,
         if (NULL == acoll_module->subc) {
             return OMPI_ERR_OUT_OF_RESOURCE;
         }
-    }
-
-    /* Check if subcomms structure is already created for the communicator */
-    for (int i = 0; i < num_subc; i++) {
-        if (acoll_module->subc[i]->cid == cid) {
-            *subc_ptr = acoll_module->subc[i];
-            return MPI_SUCCESS;
+    } else {
+        /* Check if subcomms structure is already created for the communicator */
+        for (int i = 0; i < num_subc; i++) {
+            if (NULL != acoll_module->subc[i]) {
+                if (acoll_module->subc[i]->cid == cid) {
+                    *subc_ptr = acoll_module->subc[i];
+                    return MPI_SUCCESS;
+                }
+            }
         }
     }
 
