Hotfix for lost callbacks

We must protect the activation of the completion of a continuation request
to ensure that we don't miss any updates.

Signed-off-by: Joseph Schuchart <[email protected]>

Author: devreal

ompi/mpiext/continue/c/continuation.c

@@ -267,12 +267,19 @@ void ompi_continue_cont_release(ompi_continuation_t *cont, int rc)
         cont_req->cont_errorinfo.type            = OMPI_REQUEST_CONT;
     }
 
+    opal_atomic_lock(&cont_req->cont_lock);
     int num_active = OPAL_THREAD_ADD_FETCH32(&cont_req->cont_num_active, -1);
     if (num_active == 0) {
         opal_atomic_wmb();
-        /* signal that all continuations were found complete */
-        ompi_request_complete(&cont_req->super, true);
+	//opal_atomic_lock(&cont_req->cont_lock);
+	if (!REQUEST_COMPLETE(&cont_req->super)) {
+            /* signal that all continuations were found complete */
+	    //printf("COMPLETE cont_req %p cont %p\n", cont_req, cont);
+            ompi_request_complete(&cont_req->super, true);
+	}
+	//opal_atomic_unlock(&cont_req->cont_lock);
     }
+    opal_atomic_unlock(&cont_req->cont_lock);
 
     OBJ_RELEASE(cont_req);
 
@@ -618,17 +625,31 @@ ompi_continuation_t *ompi_continue_cont_create(
     /* signal that the continuation request has a new continuation */
     OBJ_RETAIN(cont_req);
 
-    OPAL_THREAD_ADD_FETCH32(&cont_req->cont_num_active, 1);
+    opal_atomic_lock(&cont_req->cont_lock);
+    int prev_num_active = OPAL_THREAD_ADD_FETCH32(&cont_req->cont_num_active, 1);
 
     /* if the continuation request was completed we mark it pending here */
-    if (REQUEST_COMPLETE(&cont_req->super)) {
+    //if (REQUEST_COMPLETE(&cont_req->super)) {
+    if (prev_num_active == 1) {
+	//printf("PENDING cont_req %p cont %p\n", cont_req, cont);
         if (using_threads) {
-            intptr_t tmp = (intptr_t)REQUEST_COMPLETED;
-            opal_atomic_compare_exchange_strong_ptr((intptr_t*)&cont_req->super.req_complete, &tmp, (intptr_t)REQUEST_PENDING);
+	    //opal_atomic_lock(&cont_req->cont_lock);
+	    //if (REQUEST_COMPLETE(&cont_req->super)) {
+                cont_req->super.req_complete = REQUEST_PENDING;
+	        cont_req->super.req_complete_cb = NULL;
+	    //}
+	    /* NOTE: atomic operations not required here, we're protected by the lock */
+            //intptr_t tmp = (intptr_t)REQUEST_COMPLETED;
+            //opal_atomic_compare_exchange_strong_ptr((intptr_t*)&cont_req->super.req_complete, &tmp, (intptr_t)REQUEST_PENDING);
+	    //tmp = (intptr_t)REQUEST_CB_COMPLETED;
+            //opal_atomic_compare_exchange_strong_ptr((intptr_t*)&cont_req->super.req_complete_cb, &tmp, (intptr_t)NULL);
+	    //opal_atomic_unlock(&cont_req->cont_lock);
         } else {
             cont_req->super.req_complete = REQUEST_PENDING;
+	    cont_req->super.req_complete_cb = NULL;
         }
     }
+    opal_atomic_unlock(&cont_req->cont_lock);
 
     /* if we don't have the requests we cannot handle oob errors,
      * so don't bother keeping the continuation around */
@@ -1005,6 +1026,7 @@ static int ompi_continue_request_start(size_t count, ompi_request_t** cont_req_p
         if (using_threads) {
             opal_atomic_lock(&cont_req->cont_lock);
         }
+	//OMPI_REQUEST_INIT(&cont_req->super, true);
         if (NULL == cont_req->cont_complete_list) {
             opal_list_join(&continuation_list,
                           opal_list_get_begin(&continuation_list),