Fix multiple issues with the collective requests.

This patch addresses most (if not all) @derbeyn concerns
expressed on #1015. I added checks for the requests allocation
in all functions, ompi_coll_base_free_reqs is called with the
right number of requests, I removed the unnecessary basic_module_comm_t
and use the base_module_comm_t instead, I remove all uses of the
COLL_BASE_BCAST_USE_BLOCKING define, and other minor fixes.

Author: bosilca

ompi/mca/coll/base/coll_base_alltoall.c

@@ -3,7 +3,7 @@
  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
  *                         University Research and Technology
  *                         Corporation.  All rights reserved.
- * Copyright (c) 2004-2015 The University of Tennessee and The University
+ * Copyright (c) 2004-2016 The University of Tennessee and The University
  *                         of Tennessee Research Foundation.  All rights
  *                         reserved.
  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
@@ -44,7 +44,7 @@ mca_coll_base_alltoall_intra_basic_inplace(const void *rbuf, int rcount,
     mca_coll_base_module_t *base_module = (mca_coll_base_module_t*) module;
     int i, j, size, rank, err = MPI_SUCCESS, line;
     OPAL_PTRDIFF_TYPE ext, gap;
-    MPI_Request *preq;
+    ompi_request_t **preq, **reqs;
     char *tmp_buffer;
     size_t max_size;
 
@@ -62,19 +62,20 @@ mca_coll_base_alltoall_intra_basic_inplace(const void *rbuf, int rcount,
     ompi_datatype_type_extent (rdtype, &ext);
     max_size = opal_datatype_span(&rdtype->super, rcount, &gap);
 
+    /* Initiate all send/recv to/from others. */
+    reqs = coll_base_comm_get_reqs(base_module->base_data, 2);
+    if( NULL == reqs ) { err = OMPI_ERR_OUT_OF_RESOURCE; line = __LINE__; goto error_hndl; }
+
     /* Allocate a temporary buffer */
     tmp_buffer = calloc (max_size, 1);
-    if (NULL == tmp_buffer) {
-      return OMPI_ERR_OUT_OF_RESOURCE;
-    }
+    if (NULL == tmp_buffer) { return OMPI_ERR_OUT_OF_RESOURCE; }
     tmp_buffer -= gap;
     max_size = ext * rcount;
 
     /* in-place alltoall slow algorithm (but works) */
     for (i = 0 ; i < size ; ++i) {
         for (j = i+1 ; j < size ; ++j) {
-            /* Initiate all send/recv to/from others. */
-            preq = coll_base_comm_get_reqs(base_module->base_data, size * 2);
+            preq = reqs;
 
             if (i == rank) {
                 /* Copy the data into the temporary buffer */
@@ -111,7 +112,7 @@ mca_coll_base_alltoall_intra_basic_inplace(const void *rbuf, int rcount,
             }
 
             /* Wait for the requests to complete */
-            err = ompi_request_wait_all (2, base_module->base_data->mcct_reqs, MPI_STATUSES_IGNORE);
+            err = ompi_request_wait_all (2, reqs, MPI_STATUSES_IGNORE);
             if (MPI_SUCCESS != err) { line = __LINE__; goto error_hndl; }
         }
     }
@@ -125,7 +126,7 @@ mca_coll_base_alltoall_intra_basic_inplace(const void *rbuf, int rcount,
                      "%s:%4d\tError occurred %d, rank %2d", __FILE__, line, err,
                      rank));
         (void)line;  // silence compiler warning
-        ompi_coll_base_free_reqs(base_module->base_data->mcct_reqs, 2);
+        ompi_coll_base_free_reqs(reqs, 2);
     }
 
     /* All done */
@@ -400,20 +401,20 @@ int ompi_coll_base_alltoall_intra_linear_sync(const void *sbuf, int scount,
 
     /* Post first batch or ireceive and isend requests  */
     for (nreqs = 0, nrreqs = 0, ri = (rank + 1) % size; nreqs < total_reqs;
-         ri = (ri + 1) % size, ++nreqs, ++nrreqs) {
-        error =
-            MCA_PML_CALL(irecv
-                         (prcv + (ptrdiff_t)ri * rext, rcount, rdtype, ri,
-                          MCA_COLL_BASE_TAG_ALLTOALL, comm, &reqs[nreqs]));
+         ri = (ri + 1) % size, ++nrreqs) {
+        nreqs++;
+        error = MCA_PML_CALL(irecv
+                             (prcv + (ptrdiff_t)ri * rext, rcount, rdtype, ri,
+                              MCA_COLL_BASE_TAG_ALLTOALL, comm, &reqs[nreqs]));
         if (MPI_SUCCESS != error) { line = __LINE__; goto error_hndl; }
     }
-    for ( nsreqs = 0, si =  (rank + size - 1) % size; nreqs < 2 * total_reqs;
-          si = (si + size - 1) % size, ++nreqs, ++nsreqs) {
-        error =
-            MCA_PML_CALL(isend
-                         (psnd + (ptrdiff_t)si * sext, scount, sdtype, si,
-                          MCA_COLL_BASE_TAG_ALLTOALL,
-                          MCA_PML_BASE_SEND_STANDARD, comm, &reqs[nreqs]));
+    for (nsreqs = 0, si =  (rank + size - 1) % size; nreqs < 2 * total_reqs;
+          si = (si + size - 1) % size, ++nsreqs) {
+        nreqs++;
+        error = MCA_PML_CALL(isend
+                             (psnd + (ptrdiff_t)si * sext, scount, sdtype, si,
+                              MCA_COLL_BASE_TAG_ALLTOALL,
+                              MCA_PML_BASE_SEND_STANDARD, comm, &reqs[nreqs]));
         if (MPI_SUCCESS != error) { line = __LINE__; goto error_hndl; }
     }
 
@@ -441,11 +442,10 @@ int ompi_coll_base_alltoall_intra_linear_sync(const void *sbuf, int scount,
             ncreqs++;
             if (completed < total_reqs) {
                 if (nrreqs < (size - 1)) {
-                    error =
-                        MCA_PML_CALL(irecv
-                                     (prcv + (ptrdiff_t)ri * rext, rcount, rdtype, ri,
-                                      MCA_COLL_BASE_TAG_ALLTOALL, comm,
-                                      &reqs[completed]));
+                    error = MCA_PML_CALL(irecv
+                                         (prcv + (ptrdiff_t)ri * rext, rcount, rdtype, ri,
+                                          MCA_COLL_BASE_TAG_ALLTOALL, comm,
+                                          &reqs[completed]));
                     if (MPI_SUCCESS != error) { line = __LINE__; goto error_hndl; }
                     ++nrreqs;
                     ri = (ri + 1) % size;
@@ -457,6 +457,7 @@ int ompi_coll_base_alltoall_intra_linear_sync(const void *sbuf, int scount,
                                           MCA_COLL_BASE_TAG_ALLTOALL,
                                           MCA_PML_BASE_SEND_STANDARD, comm,
                                           &reqs[completed]));
+                    if (MPI_SUCCESS != error) { line = __LINE__; goto error_hndl; }
                     ++nsreqs;
                     si = (si + size - 1) % size;
                 }
@@ -472,7 +473,7 @@ int ompi_coll_base_alltoall_intra_linear_sync(const void *sbuf, int scount,
                  "%s:%4d\tError occurred %d, rank %2d", __FILE__, line, error,
                  rank));
     (void)line;  // silence compiler warning
-    ompi_coll_base_free_reqs(reqs, 2 * total_reqs);
+    ompi_coll_base_free_reqs(reqs, nreqs);
     return error;
 }
 
@@ -610,14 +611,16 @@ int ompi_coll_base_alltoall_intra_basic_linear(const void *sbuf, int scount,
     /* Initiate all send/recv to/from others. */
 
     req = rreq = coll_base_comm_get_reqs(data, (size - 1) * 2);
+    if (NULL == req) { err = OMPI_ERR_OUT_OF_RESOURCE; line = __LINE__; goto err_hndl; }
 
     prcv = (char *) rbuf;
     psnd = (char *) sbuf;
 
     /* Post all receives first -- a simple optimization */
 
     for (nreqs = 0, i = (rank + 1) % size; i != rank;
-         i = (i + 1) % size, ++rreq, ++nreqs) {
+         i = (i + 1) % size, ++rreq) {
+        nreqs++;
         err = MCA_PML_CALL(irecv_init
                            (prcv + (ptrdiff_t)i * rcvinc, rcount, rdtype, i,
                            MCA_COLL_BASE_TAG_ALLTOALL, comm, rreq));
@@ -630,7 +633,8 @@ int ompi_coll_base_alltoall_intra_basic_linear(const void *sbuf, int scount,
      */
     sreq = rreq;
     for (i = (rank + size - 1) % size; i != rank;
-         i = (i + size - 1) % size, ++sreq, ++nreqs) {
+         i = (i + size - 1) % size, ++sreq) {
+        nreqs++;
         err = MCA_PML_CALL(isend_init
                            (psnd + (ptrdiff_t)i * sndinc, scount, sdtype, i,
                            MCA_COLL_BASE_TAG_ALLTOALL,

ompi/mca/coll/base/coll_base_alltoallv.c

@@ -3,7 +3,7 @@
  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
  *                         University Research and Technology
  *                         Corporation.  All rights reserved.
- * Copyright (c) 2004-2015 The University of Tennessee and The University
+ * Copyright (c) 2004-2016 The University of Tennessee and The University
  *                         of Tennessee Research Foundation.  All rights
  *                         reserved.
  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
@@ -44,7 +44,7 @@ mca_coll_base_alltoallv_intra_basic_inplace(const void *rbuf, const int *rcounts
 {
     mca_coll_base_module_t *base_module = (mca_coll_base_module_t*) module;
     int i, j, size, rank, err=MPI_SUCCESS;
-    MPI_Request *preq;
+    ompi_request_t **preq, **reqs;
     char *tmp_buffer;
     size_t max_size, rdtype_size;
     OPAL_PTRDIFF_TYPE ext, gap = 0;
@@ -75,11 +75,14 @@ mca_coll_base_alltoallv_intra_basic_inplace(const void *rbuf, const int *rcounts
     }
     tmp_buffer += gap;
 
+    /* Initiate all send/recv to/from others. */
+    reqs = preq = coll_base_comm_get_reqs(base_module->base_data, 2);
+    if( NULL == reqs ) { err = OMPI_ERR_OUT_OF_RESOURCE; goto error_hndl; }
+
     /* in-place alltoallv slow algorithm (but works) */
     for (i = 0 ; i < size ; ++i) {
         for (j = i+1 ; j < size ; ++j) {
-            /* Initiate all send/recv to/from others. */
-            preq = coll_base_comm_get_reqs(base_module->base_data, 2);
+            preq = reqs;
 
             if (i == rank && rcounts[j]) {
                 /* Copy the data into the temporary buffer */
@@ -116,7 +119,7 @@ mca_coll_base_alltoallv_intra_basic_inplace(const void *rbuf, const int *rcounts
             }
 
             /* Wait for the requests to complete */
-            err = ompi_request_wait_all (2, base_module->base_data->mcct_reqs, MPI_STATUSES_IGNORE);
+            err = ompi_request_wait_all (2, reqs, MPI_STATUSES_IGNORE);
             if (MPI_SUCCESS != err) { goto error_hndl; }
         }
     }
@@ -125,7 +128,7 @@ mca_coll_base_alltoallv_intra_basic_inplace(const void *rbuf, const int *rcounts
     /* Free the temporary buffer */
     free (tmp_buffer);
     if( MPI_SUCCESS != err ) {
-        ompi_coll_base_free_reqs(base_module->base_data->mcct_reqs, 2 );
+        ompi_coll_base_free_reqs(reqs, 2 );
     }
 
     /* All done */
@@ -207,7 +210,7 @@ ompi_coll_base_alltoallv_intra_basic_linear(const void *sbuf, const int *scounts
     int i, size, rank, err, nreqs;
     char *psnd, *prcv;
     ptrdiff_t sext, rext;
-    MPI_Request *preq;
+    ompi_request_t **preq, **reqs;
     mca_coll_base_module_t *base_module = (mca_coll_base_module_t*) module;
     mca_coll_base_comm_t *data = base_module->base_data;
 
@@ -243,22 +246,21 @@ ompi_coll_base_alltoallv_intra_basic_linear(const void *sbuf, const int *scounts
 
     /* Now, initiate all send/recv to/from others. */
     nreqs = 0;
-    preq = coll_base_comm_get_reqs(data, 2 * size);
+    reqs = preq = coll_base_comm_get_reqs(data, 2 * size);
+    if( NULL == reqs ) { err = OMPI_ERR_OUT_OF_RESOURCE; goto err_hndl; }
 
     /* Post all receives first */
     for (i = 0; i < size; ++i) {
         if (i == rank || 0 == rcounts[i]) {
             continue;
         }
 
+        ++nreqs;
         prcv = ((char *) rbuf) + (ptrdiff_t)rdisps[i] * rext;
         err = MCA_PML_CALL(irecv_init(prcv, rcounts[i], rdtype,
                                       i, MCA_COLL_BASE_TAG_ALLTOALLV, comm,
                                       preq++));
-        ++nreqs;
-        if (MPI_SUCCESS != err) {
-            goto err_hndl;
-        }
+        if (MPI_SUCCESS != err) { goto err_hndl; }
     }
 
     /* Now post all sends */
@@ -267,31 +269,30 @@ ompi_coll_base_alltoallv_intra_basic_linear(const void *sbuf, const int *scounts
             continue;
         }
 
+        ++nreqs;
         psnd = ((char *) sbuf) + (ptrdiff_t)sdisps[i] * sext;
         err = MCA_PML_CALL(isend_init(psnd, scounts[i], sdtype,
                                       i, MCA_COLL_BASE_TAG_ALLTOALLV,
                                       MCA_PML_BASE_SEND_STANDARD, comm,
                                       preq++));
-        ++nreqs;
-        if (MPI_SUCCESS != err) {
-            goto err_hndl;
-        }
+        if (MPI_SUCCESS != err) { goto err_hndl; }
     }
 
     /* Start your engines.  This will never return an error. */
-    MCA_PML_CALL(start(nreqs, data->mcct_reqs));
+    MCA_PML_CALL(start(nreqs, reqs));
 
     /* Wait for them all.  If there's an error, note that we don't care
      * what the error was -- just that there *was* an error.  The PML
      * will finish all requests, even if one or more of them fail.
      * i.e., by the end of this call, all the requests are free-able.
      * So free them anyway -- even if there was an error, and return the
      * error after we free everything. */
-    err = ompi_request_wait_all(nreqs, data->mcct_reqs,
-                                MPI_STATUSES_IGNORE);
+    err = ompi_request_wait_all(nreqs, reqs, MPI_STATUSES_IGNORE);
+    if( MPI_SUCCESS == err )
+        return MPI_SUCCESS;
  err_hndl:
     /* Free the requests in all cases as they are persistent */
-    ompi_coll_base_free_reqs(data->mcct_reqs, nreqs);
+    ompi_coll_base_free_reqs(reqs, nreqs);
 
     return err;
 }

ompi/mca/coll/base/coll_base_barrier.c

@@ -352,13 +352,16 @@ int ompi_coll_base_barrier_intra_basic_linear(struct ompi_communicator_t *comm,
 
     else {
         requests = coll_base_comm_get_reqs(module->base_data, size);
+        if( NULL == requests ) { err = OMPI_ERR_OUT_OF_RESOURCE; line = __LINE__; goto err_hndl; }
+
         for (i = 1; i < size; ++i) {
             err = MCA_PML_CALL(irecv(NULL, 0, MPI_BYTE, MPI_ANY_SOURCE,
                                      MCA_COLL_BASE_TAG_BARRIER, comm,
                                      &(requests[i])));
             if (MPI_SUCCESS != err) { line = __LINE__; goto err_hndl; }
         }
-        ompi_request_wait_all( size-1, requests+1, MPI_STATUSES_IGNORE );
+        err = ompi_request_wait_all( size-1, requests+1, MPI_STATUSES_IGNORE );
+        if (MPI_SUCCESS != err) { line = __LINE__; goto err_hndl; }
         requests = NULL;  /* we're done the requests array is clean */
 
         for (i = 1; i < size; ++i) {
@@ -376,7 +379,7 @@ int ompi_coll_base_barrier_intra_basic_linear(struct ompi_communicator_t *comm,
                   __FILE__, line, err, rank) );
     (void)line;  // silence compiler warning
     if( NULL != requests )
-        ompi_coll_base_free_reqs(requests, size-1);
+        ompi_coll_base_free_reqs(requests, size);
     return err;
 }
 /* copied function (with appropriate renaming) ends here */