fixes per George's review

Signed-off-by: Gilles Gouaillardet <gilles@rist.or.jp>

Author: ggouaillardet

ompi/mca/coll/base/coll_base_util.c

@@ -26,6 +26,7 @@
 #include "ompi/communicator/communicator.h"
 #include "ompi/mca/coll/base/coll_tags.h"
 #include "ompi/mca/coll/base/coll_base_functions.h"
+#include "ompi/mca/topo/base/base.h"
 #include "ompi/mca/pml/pml.h"
 #include "coll_base_util.h"
 
@@ -149,6 +150,14 @@ int ompi_coll_base_retain_op( ompi_request_t *req, ompi_op_t *op,
         retain = true;
     }
     if (OPAL_UNLIKELY(retain)) {
+        /* We need to consider two cases :
+         * - non blocking collectives:
+         *     the objects can be released when MPI_Wait() completes
+         *     and we use the req_complete_cb callback
+         * - persistent non blocking collectives:
+         *     the objects can only be released when the request is freed
+         *     (e.g. MPI_Request_free() completes) and we use req_free callback
+         */
         if (req->req_persistent) {
             request->cb.req_free = req->req_free;
             req->req_free = free_objs_callback;
@@ -192,11 +201,18 @@ int ompi_coll_base_retain_datatypes( ompi_request_t *req, ompi_datatype_t *stype
 
 static void release_vecs_callback(ompi_coll_base_nbc_request_t *request) {
     ompi_communicator_t *comm = request->super.req_mpi_object.comm;
-    int count = OMPI_COMM_IS_INTER(comm)?ompi_comm_remote_size(comm):ompi_comm_size(comm);
-    for (int i=0; i<count; i++) {
+    int scount, rcount;
+    if (OMPI_COMM_IS_TOPO(comm)) {
+        (void)mca_topo_base_neighbor_count (comm, &rcount, &scount);
+    } else {
+        scount = rcount = OMPI_COMM_IS_INTER(comm)?ompi_comm_remote_size(comm):ompi_comm_size(comm);
+    }
+    for (int i=0; i<scount; i++) {
         if (NULL != request->data.vecs.stypes && NULL != request->data.vecs.stypes[i]) {
             OMPI_DATATYPE_RELEASE(request->data.vecs.stypes[i]);
         }
+    }
+    for (int i=0; i<rcount; i++) {
         if (NULL != request->data.vecs.rtypes && NULL != request->data.vecs.rtypes[i]) {
             OMPI_DATATYPE_RELEASE(request->data.vecs.rtypes[i]);
         }
@@ -229,13 +245,20 @@ int ompi_coll_base_retain_datatypes_w( ompi_request_t *req,
     ompi_coll_base_nbc_request_t *request = (ompi_coll_base_nbc_request_t *)req;
     bool retain = false;
     ompi_communicator_t *comm = request->super.req_mpi_object.comm;
-    int count = OMPI_COMM_IS_INTER(comm)?ompi_comm_remote_size(comm):ompi_comm_size(comm);
+    int scount, rcount;
+    if (OMPI_COMM_IS_TOPO(comm)) {
+        (void)mca_topo_base_neighbor_count (comm, &rcount, &scount);
+    } else {
+        scount = rcount = OMPI_COMM_IS_INTER(comm)?ompi_comm_remote_size(comm):ompi_comm_size(comm);
+    }
 
-    for (int i=0; i<count; i++) {
+    for (int i=0; i<scount; i++) {
         if (NULL != stypes && NULL != stypes[i] && !ompi_datatype_is_predefined(stypes[i])) {
             OBJ_RETAIN(stypes[i]);
             retain = true;
         }
+    }
+    for (int i=0; i<rcount; i++) {
         if (NULL != rtypes && NULL != rtypes[i] && !ompi_datatype_is_predefined(rtypes[i])) {
             OBJ_RETAIN(rtypes[i]);
             retain = true;

ompi/mca/coll/libnbc/nbc.c

@@ -663,7 +663,7 @@ int NBC_Start(NBC_Handle *handle) {
   }
 
   OPAL_THREAD_LOCK(&mca_coll_libnbc_component.lock);
-  opal_list_append(&mca_coll_libnbc_component.active_requests, &(handle->super.super.super.super));
+  opal_list_append(&mca_coll_libnbc_component.active_requests, (opal_list_item_t *)handle);
   OPAL_THREAD_UNLOCK(&mca_coll_libnbc_component.lock);
 
   return OMPI_SUCCESS;