Request handling: fix MPI_Waitany and MPI_Waitsome

Author: artpol84

ompi/request/req_wait.c

@@ -15,6 +15,7 @@
  * Copyright (c) 2012      Oak Ridge National Labs.  All rights reserved.
  * Copyright (c) 2016      Los Alamos National Security, LLC. All rights
  *                         reserved.
+ * Copyright (c) 2016      Mellanox Technologies. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -107,6 +108,10 @@ int ompi_request_default_wait_any(size_t count,
             assert(REQUEST_COMPLETE(request));
             completed = i;
             *index = i;
+            /* set signalled flag so we won't 
+             * block in WAIT_SYNC_RELEASE 
+             */
+            WAIT_SYNC_SIGNALLED(&sync);
             goto after_sync_wait;
         }
     }
@@ -116,7 +121,8 @@ int ompi_request_default_wait_any(size_t count,
         if (MPI_STATUS_IGNORE != status) {
             *status = ompi_status_empty;
         }
-        WAIT_SYNC_RELEASE(&sync);
+        /* No signal-in-flight can be in this case */
+        WAIT_SYNC_RELEASE_NOWAIT(&sync);
         return rc;
     }
 
@@ -361,7 +367,8 @@ int ompi_request_default_wait_some(size_t count,
     ompi_request_t **rptr = NULL;
     ompi_request_t *request = NULL;
     ompi_wait_sync_t sync;
-
+    size_t sync_sets = 0, sync_unsets = 0;
+    
     WAIT_SYNC_INIT(&sync, 1);
 
     *outcount = 0;
@@ -384,12 +391,15 @@ int ompi_request_default_wait_some(size_t count,
             /* If the request is completed go ahead and mark it as such */
             assert( REQUEST_COMPLETE(request) );
             num_requests_done++;
+        } else {
+            sync_sets++;
         }
     }
 
     if(num_requests_null_inactive == count) {
         *outcount = MPI_UNDEFINED;
-        WAIT_SYNC_RELEASE(&sync);
+        /* nobody will signall us */
+        WAIT_SYNC_RELEASE_NOWAIT(&sync);
         return rc;
     }
 
@@ -418,9 +428,19 @@ int ompi_request_default_wait_some(size_t count,
         if( !OPAL_ATOMIC_CMPSET_PTR(&request->req_complete, &sync, REQUEST_PENDING) ) {
             indices[num_requests_done] = i;
             num_requests_done++;
+        } else {
+            /* request wasn't finished during this call */
+            sync_unsets++;
         }
     }
 
+    if( sync_sets == sync_unsets ){
+        /* nobody knows about us,
+         * set signa-in-progress flag to false
+         */
+        WAIT_SYNC_SIGNALLED(&sync);
+    }
+
     WAIT_SYNC_RELEASE(&sync);
 
     *outcount = num_requests_done;

opal/threads/wait_sync.h

@@ -43,21 +43,32 @@ typedef struct ompi_wait_sync_t {
  * the critical path. */
 #define WAIT_SYNC_RELEASE(sync)                       \
     if (opal_using_threads()) {                       \
-        while ((sync)->signaling) {                  \
+        while ((sync)->signaling) {                   \
             continue;                                 \
         }                                             \
         pthread_cond_destroy(&(sync)->condition);     \
         pthread_mutex_destroy(&(sync)->lock);         \
     }
 
+#define WAIT_SYNC_RELEASE_NOWAIT(sync)                \
+    if (opal_using_threads()) {                       \
+        pthread_cond_destroy(&(sync)->condition);     \
+        pthread_mutex_destroy(&(sync)->lock);         \
+    }
+
+
 #define WAIT_SYNC_SIGNAL(sync)                        \
     if (opal_using_threads()) {                       \
         pthread_mutex_lock(&(sync->lock));            \
         pthread_cond_signal(&sync->condition);        \
         pthread_mutex_unlock(&(sync->lock));          \
-        sync->signaling = false;                     \
+        sync->signaling = false;                      \
     }
 
+#define WAIT_SYNC_SIGNALLED(sync){                    \
+        (sync)->signaling = false;                    \
+}        
+
 OPAL_DECLSPEC int sync_wait_mt(ompi_wait_sync_t *sync);
 static inline int sync_wait_st (ompi_wait_sync_t *sync)
 {