Merge pull request #7228 from devreal/progress-returns

Harmonize return values of progress callbacks

Author: bosilca

ompi/communicator/comm_request.c

@@ -100,6 +100,7 @@ static int ompi_comm_request_progress (void)
 {
     ompi_comm_request_t *request, *next;
     static opal_atomic_int32_t progressing = 0;
+    int completed = 0;
 
     /* don't allow re-entry */
     if (opal_atomic_swap_32 (&progressing, 1)) {
@@ -126,6 +127,7 @@ static int ompi_comm_request_progress (void)
                     }
                     ompi_request_free (&subreq);
                     request_item->subreq_count--;
+                    completed++;
                 } else {
                     item_complete = false;
                     break;
@@ -163,7 +165,7 @@ static int ompi_comm_request_progress (void)
     opal_mutex_unlock (&ompi_comm_request_mutex);
     progressing = 0;
 
-    return 1;
+    return completed;
 }
 
 void ompi_comm_request_start (ompi_comm_request_t *request)

ompi/mca/coll/libnbc/coll_libnbc_component.c

@@ -427,6 +427,7 @@ ompi_coll_libnbc_progress(void)
 {
     ompi_coll_libnbc_request_t* request, *next;
     int res;
+    int completed = 0;
 
     if (0 == opal_list_get_size (&mca_coll_libnbc_component.active_requests)) {
         /* no requests -- nothing to do. do not grab a lock */
@@ -464,14 +465,15 @@ ompi_coll_libnbc_progress(void)
                 if(!request->super.super.req_persistent || !REQUEST_COMPLETE(&request->super.super)) {
             	    ompi_request_complete(&request->super.super, true);
                 }
+                completed++;
             }
             OPAL_THREAD_LOCK(&mca_coll_libnbc_component.lock);
         }
         libnbc_in_progress = false;
     }
     OPAL_THREAD_UNLOCK(&mca_coll_libnbc_component.lock);
 
-    return 0;
+    return completed;
 }
 
 

ompi/mca/mtl/psm2/mtl_psm2.c

@@ -403,7 +403,7 @@ int ompi_mtl_psm2_progress( void ) {
     mca_mtl_psm2_request_t* mtl_psm2_request;
     psm2_mq_status2_t psm2_status;
     psm2_mq_req_t req;
-    int completed = 1;
+    int completed = 0;
 
     do {
         OPAL_THREAD_LOCK(&mtl_psm2_mq_mutex);
@@ -469,5 +469,5 @@ int ompi_mtl_psm2_progress( void ) {
     opal_show_help("help-mtl-psm2.txt",
 		   "error polling network", true,
 		   psm2_error_get_string(err));
-    return 1;
+    return OMPI_ERROR;
 }

ompi/mca/osc/pt2pt/osc_pt2pt_component.c

@@ -153,6 +153,7 @@ static int component_register (void)
 
 static int component_progress (void)
 {
+    int completed = 0;
     int pending_count = opal_list_get_size (&mca_osc_pt2pt_component.pending_operations);
     int recv_count = opal_list_get_size (&mca_osc_pt2pt_component.pending_receives);
     ompi_osc_pt2pt_pending_t *pending, *next;
@@ -167,6 +168,7 @@ static int component_progress (void)
             }
 
             (void) ompi_osc_pt2pt_process_receive (recv);
+            completed++;
         }
     }
 
@@ -194,12 +196,13 @@ static int component_progress (void)
             if (OMPI_SUCCESS == ret) {
                 opal_list_remove_item (&mca_osc_pt2pt_component.pending_operations, &pending->super);
                 OBJ_RELEASE(pending);
+                completed++;
             }
         }
         OPAL_THREAD_UNLOCK(&mca_osc_pt2pt_component.pending_operations_lock);
     }
 
-    return 1;
+    return completed;
 }
 
 static int

ompi/mca/pml/ucx/pml_ucx.c

@@ -515,8 +515,7 @@ int mca_pml_ucx_enable(bool enable)
 
 int mca_pml_ucx_progress(void)
 {
-    ucp_worker_progress(ompi_pml_ucx.ucp_worker);
-    return OMPI_SUCCESS;
+    return ucp_worker_progress(ompi_pml_ucx.ucp_worker);
 }
 
 int mca_pml_ucx_add_comm(struct ompi_communicator_t* comm)

ompi/request/grequest.c

@@ -202,14 +202,13 @@ int ompi_grequest_invoke_query(ompi_request_t *request,
     int rc = OMPI_SUCCESS;
     ompi_grequest_t *g = (ompi_grequest_t*) request;
 
-    /* MPI-2:8.2 does not say what to do with the return value from
-       the query function (i.e., the int return value from the C
-       function or the ierr argument from the Fortran function).
-       Making the command decision here to ignore it.  If the handler
-       wants to pass an error back, it should set it in the MPI_ERROR
-       field in the status (which is always kept, regardless if the
-       top-level function was invoked with MPI_STATUS[ES]_IGNORE or
-       not).  */
+    /* MPI-3 mandates that the return value from the query function
+     * (i.e., the int return value from the C function or the ierr
+     * argument from the Fortran function) must be returned to the
+     * user. Thus, if the return of the query function is not MPI_SUCCESS
+     * we will update the MPI_ERROR field. Otherwise, the MPI_ERROR
+     * field is untouched (or left to the discretion of the query function).
+     */
     if (NULL != g->greq_query.c_query) {
         if (g->greq_funcs_are_c) {
             rc = g->greq_query.c_query(g->greq_state, status);
@@ -221,7 +220,9 @@ int ompi_grequest_invoke_query(ompi_request_t *request,
             rc = OMPI_FINT_2_INT(ierr);
         }
     }
-
+    if( MPI_SUCCESS != rc ) {
+        status->MPI_ERROR = rc;
+    }
     return rc;
 }
 

ompi/request/grequestx.c

@@ -34,6 +34,7 @@ static opal_mutex_t lock;
 
 static int grequestx_progress(void) {
     ompi_grequest_t *request, *next;
+    int completed = 0;
 
     OPAL_THREAD_LOCK(&lock);
     if (!in_progress) {
@@ -43,18 +44,17 @@ static int grequestx_progress(void) {
             MPI_Status status;
             OPAL_THREAD_UNLOCK(&lock);
             request->greq_poll.c_poll(request->greq_state, &status);
+            OPAL_THREAD_LOCK(&lock);
             if (REQUEST_COMPLETE(&request->greq_base)) {
-                OPAL_THREAD_LOCK(&lock);
                 opal_list_remove_item(&requests, &request->greq_base.super.super);
-                OPAL_THREAD_UNLOCK(&lock);
+                completed++;
             }
-            OPAL_THREAD_LOCK(&lock);
         }
         in_progress = false;
     }
     OPAL_THREAD_UNLOCK(&lock);
 
-    return OMPI_SUCCESS;
+    return completed;
 }
 
 int ompi_grequestx_start(

opal/mca/btl/uct/btl_uct_component.c

@@ -564,6 +564,7 @@ static int mca_btl_uct_tl_progress (mca_btl_uct_tl_t *tl, int starting_index)
 static int mca_btl_uct_component_progress_pending (mca_btl_uct_module_t *uct_btl)
 {
     mca_btl_uct_base_frag_t *frag, *next;
+    int completed = 0;
     size_t count;
 
     if (0 == (count = opal_list_get_size (&uct_btl->pending_frags))) {
@@ -580,11 +581,13 @@ static int mca_btl_uct_component_progress_pending (mca_btl_uct_module_t *uct_btl
 
         if (OPAL_SUCCESS > mca_btl_uct_send_frag (uct_btl, frag, false)) {
             opal_list_prepend (&uct_btl->pending_frags, (opal_list_item_t *) frag);
+        } else {
+            completed++;
         }
     }
     OPAL_THREAD_UNLOCK(&uct_btl->lock);
 
-    return OPAL_SUCCESS;
+    return completed;
 }
 
 /**

opal/mca/common/ucx/common_ucx_wpool.c

@@ -279,10 +279,11 @@ void opal_common_ucx_wpool_finalize(opal_common_ucx_wpool_t *wpool)
     return;
 }
 
-OPAL_DECLSPEC void
+OPAL_DECLSPEC int
 opal_common_ucx_wpool_progress(opal_common_ucx_wpool_t *wpool)
 {
     _winfo_list_item_t *item = NULL, *next = NULL;
+    int completed = 0, progressed = 0;
 
     /* Go over all active workers and progress them
      * TODO: may want to have some partitioning to progress only part of
@@ -297,14 +298,19 @@ opal_common_ucx_wpool_progress(opal_common_ucx_wpool_t *wpool)
                 opal_list_remove_item(&wpool->active_workers, &item->super);
                 _winfo_reset(winfo);
                 opal_list_append(&wpool->idle_workers, &item->super);
+                completed++;
             } else {
                 /* Progress worker until there are existing events */
-                while(ucp_worker_progress(winfo->worker));
+                do {
+                    progressed = ucp_worker_progress(winfo->worker);
+                    completed += progressed;
+                } while (progressed);
             }
             opal_mutex_unlock(&winfo->mutex);
         }
         opal_mutex_unlock(&wpool->mutex);
     }
+    return completed;
 }
 
 static int

opal/mca/common/ucx/common_ucx_wpool.h

@@ -165,7 +165,7 @@ OPAL_DECLSPEC void opal_common_ucx_wpool_free(opal_common_ucx_wpool_t *wpool);
 OPAL_DECLSPEC int opal_common_ucx_wpool_init(opal_common_ucx_wpool_t *wpool,
                                              int proc_world_size, bool enable_mt);
 OPAL_DECLSPEC void opal_common_ucx_wpool_finalize(opal_common_ucx_wpool_t *wpool);
-OPAL_DECLSPEC void opal_common_ucx_wpool_progress(opal_common_ucx_wpool_t *wpool);
+OPAL_DECLSPEC int opal_common_ucx_wpool_progress(opal_common_ucx_wpool_t *wpool);
 
 /* Manage Communication context */
 OPAL_DECLSPEC int opal_common_ucx_wpctx_create(opal_common_ucx_wpool_t *wpool, int comm_size,