[svn-r6919] completion counter cleanup (adds MPID_cc_t)

When compiled for fine-grained threading, the completion counter serves
as a form of lockfree signalling.  As such, atomic access and memory
barriers must be used to ensure correctness.

In per-object mode, this code also contains valgrind client request annotations
to inform Helgrind/DRD/TSan about the lockfree signalling pattern.

No reviewer.

src/include/mpiimpl.h

@@ -1290,6 +1290,10 @@ typedef void (MPIR_Grequest_f77_cancel_function)(void *, int*, int *);
 typedef void (MPIR_Grequest_f77_free_function)(void *, int *); 
 typedef void (MPIR_Grequest_f77_query_function)(void *, MPI_Status *, int *); 
 
+
+/* see mpiimplthread.h for the def of MPID_cc_t and related functions/macros */
+#define MPID_Request_is_complete(req_) (MPID_cc_is_complete((req_)->cc_ptr))
+
 /*S
   MPID_Request - Description of the Request data structure
 
@@ -1309,11 +1313,11 @@ typedef struct MPID_Request {
     MPIU_OBJECT_HEADER; /* adds handle and ref_count fields */
     MPID_Request_kind_t kind;
     /* completion counter */
-    volatile int cc;
+    MPID_cc_t cc;
     /* pointer to the completion counter */
     /* This is necessary for the case when an operation is described by a 
        list of requests */
-    int volatile *cc_ptr;
+    MPID_cc_t *cc_ptr;
     /* A comm is needed to find the proper error handler */
     MPID_Comm *comm;
     /* Status is needed for wait/test/recv */

src/include/mpiimplthread.h

@@ -731,6 +731,71 @@ enum MPIU_Nest_mutexes {
     } while (0);                                               \
     MPIU_THREAD_CHECK_END
 
+
+/* MT FIXME the following description is almost right, but it needs minor
+ * updates and revision to account for the COMPLETION CS and other issues in the
+ * request */
+/* The fine-grained locking discipline for requests is unfortunately complicated:
+ *
+ * (1) Raw allocation and deallocation of requests is protected internally by
+ * the HANDLEALLOC critical section.  This is currently the same as the HANDLE
+ * CS, not sure why we have both...
+ *
+ * (2) Once allocated, a directly allocated request is intially held exclusively
+ * by a single thread.  Direct allocation is common for send requests, but recv
+ * requests are usually created differently.
+ *
+ * (3) Most receive requests are created as the result of a call to FDP_or_AEU
+ * or FDU_or_AEP.  Calls to these functions (along with the other receive queue
+ * functions) must be inside a MSGQUEUE CS.  This CS protects the queue data
+ * structures as well as any fields inside the requests while they are in the
+ * queue.  For example, assume a call to FDU_or_AEP, as in MPID_Recv.  If the
+ * FDU case hits, the MSGQUEUE CS may be released immediately after the call.
+ * If the AEP case hits, however, the MSGQUEUE CS must remain held until any
+ * request field manipulation (such as dev.recv_pending_count) is complete.
+ *
+ * (4) In both the send and receive request cases, there is usually a particular
+ * thread in some upper-level code (e.g. MPI_Send) with interest in the
+ * completion of the request.  This may or may not be a thread that is also
+ * making progress on this request (often not).  The upper level code must not
+ * attempt to access any request fields (such as the status) until completion is
+ * signalled by the lower layer.
+ *
+ * (5) Once removed from the receive queue, the request is once again
+ * exclusively owned by the dequeuing thread.  From here, the dequeuing thread
+ * may do whatever it wants with the request without holding any CS, until it
+ * signals the request's completion.  Signalling completion indicates that the
+ * thread in the upper layer polling on it may access the rest of the fields in
+ * the request.  This completion signalling is lock-free and must be implemented
+ * carefully to work correctly in the face of optimizing compilers and CPUs.
+ * The upper-level thread now wholly owns the request until it is deallocated.
+ *
+ * (6) In ch3:nemesis at least, multithreaded access to send requests is managed
+ * by the MPIDCOMM (progress engine) CS.  The completion signalling pattern
+ * applies here (think MPI_Isend/MPI_Wait).
+ *
+ * (7) Request cancellation is tricky-ish.  For send cancellation, it is
+ * possible that the completion counter is actually *incremented* because a
+ * pkt is sent to the recipient asking for remote cancellation.  By asking for
+ * cancellation (of any kind of req), the upper layer gives up its exclusive
+ * access to the request and must wait for the completion counter to drop to 0
+ * before exclusively accessing the request fields.
+ *
+ * The completion counter is a reference count, much like the object liveness
+ * reference count.  However it differs from a normal refcount because of
+ * guarantees in the MPI Standard.  Applications must not attempt to complete
+ * (wait/test/free) a given request concurrently in two separate threads.  So
+ * checking for cc==0 is safe because only one thread is ever allowed to make
+ * that check.
+ *
+ * A non-zero completion count must always be accompanied by a normal reference
+ * that is logically held by the progress engine.  Similarly, once the
+ * completion counter drops to zero, the progress engine is expected to release
+ * its reference.
+ */
+/* lock ordering: if MPIDCOMM+MSGQUEUE must be aquired at the same time, then
+ * the order should be to acquire MPIDCOMM first, then MSGQUEUE.  Release in
+ * reverse order. */
 #define MPIU_THREAD_CS_ENTER_MSGQUEUE(context_)           \
     do {                                                  \
         MPIU_THREAD_CS_ENTER_LOCKNAME_CHECKED(msgq_mutex) \
@@ -806,5 +871,107 @@ enum MPIU_Nest_mutexes {
 #define MPIU_THREAD_CS_YIELD(_name,_context)
 #endif /* MPICH_IS_THREADED */
 
+
+/* define a type for the completion counter */
+#if defined(MPICH_IS_THREADED)
+# if MPIU_THREAD_GRANULARITY == MPIU_THREAD_GRANULARITY_GLOBAL
+
+/* memory barriers aren't needed in this impl, because all access to completion
+ * counters is done while holding the ALLFUNC critical section */
+typedef volatile int MPID_cc_t;
+#  define MPID_cc_set(cc_ptr_, val_) (*(cc_ptr_)) = (val_)
+#  define MPID_cc_is_complete(cc_ptr_) (0 == *(cc_ptr_))
+#define MPID_cc_decr(cc_ptr_, incomplete_)     \
+do {                                           \
+    *(incomplete_) = --(*(req_)->cc_ptr);      \
+} while (0)
+#define MPID_cc_incr(cc_ptr_, was_incomplete_) \
+do {                                           \
+    *(was_incomplete_) = (*(req_)->cc_ptr)++;  \
+} while (0)
+
+# elif MPIU_THREAD_GRANULARITY == MPIU_THREAD_GRANULARITY_PER_OBJECT
+
+typedef OPA_int_t MPID_cc_t;
+
+/* implies no barrier, since this routine should only be used for request
+ * initialization */
+static inline void MPID_cc_set(MPID_cc_t *cc_ptr, int val)
+{
+    if (val == 0) {
+        /* values other than 0 do not enforce any ordering, and therefore do not
+         * start a HB arc */
+        /* MT FIXME using cc_set in this way is sloppy.  Sometimes the caller
+         * really does know that the cc value may cleared, but more likely this
+         * is just a hack to avoid the work of figuring out what the cc value
+         * currently is and decrementing it instead. */
+        /* barrier ensures that any state written before indicating completion is
+         * seen by the thread polling on the cc.  If OPA adds store-release
+         * semantics, we can convert to that instead. */
+        OPA_write_barrier();
+        MPL_VG_ANNOTATE_HAPPENS_BEFORE(cc_ptr);
+    }
+#if defined(MPL_VG_AVAILABLE)
+    /* MT subtle: store_int is actually safe to use, but Helgrind/DRD/TSan all
+     * view the store/load pair as a race.  Using an atomic operation for the
+     * store side makes all three happy.  DRD & TSan also support
+     * ANNOTATE_BENIGN_RACE, but Helgrind does not. */
+    OPA_swap_int(cc_ptr, val);
+#else
+    OPA_store_int(cc_ptr, val);
+#endif
+}
+
+ATTRIBUTE((unused))
+static MPIU_DBG_INLINE_KEYWORD int MPID_cc_is_complete(MPID_cc_t *cc_ptr)
+{
+    int complete;
+    complete = (0 == OPA_load_int(cc_ptr));
+    if (complete) {
+        MPL_VG_ANNOTATE_HAPPENS_AFTER(cc_ptr);
+        OPA_read_barrier();
+    }
+    return complete;
+}
+
+/* incomplete_==TRUE iff the cc > 0 after the decr */
+#define MPID_cc_decr(cc_ptr_, incomplete_)                                  \
+    do {                                                                    \
+        OPA_write_barrier();                                                \
+        MPL_VG_ANNOTATE_HAPPENS_BEFORE(cc_ptr_);                            \
+        *(incomplete_) = !OPA_decr_and_test_int(cc_ptr_);                   \
+        /* TODO check if this HA is actually necessary */                   \
+        if (!*(incomplete_)) {                                              \
+            MPL_VG_ANNOTATE_HAPPENS_AFTER(cc_ptr_);                         \
+        }                                                                   \
+    } while (0)
+
+/* MT FIXME does this need a HB/HA annotation?  This macro is only used for
+ * cancel_send right now. */
+/* was_incomplete_==TRUE iff the cc==0 before the decr */
+#define MPID_cc_incr(cc_ptr_, was_incomplete_)                              \
+    do {                                                                    \
+        *(was_incomplete_) = OPA_fetch_and_incr_int(cc_ptr_);               \
+    } while (0)
+
+# else
+#  error "unexpected thread granularity"
+# endif /* granularity */
+#else /* !defined(MPICH_IS_THREADED) */
+
+typedef int MPID_cc_t;
+# define MPID_cc_set(cc_ptr_, val_) (*(cc_ptr_)) = (val_)
+# define MPID_cc_is_complete(cc_ptr_) (0 == *(cc_ptr_))
+#define MPID_cc_decr(cc_ptr_, incomplete_)     \
+do {                                           \
+    *(incomplete_) = --(*(req_)->cc_ptr);      \
+} while (0)
+#define MPID_cc_incr(cc_ptr_, was_incomplete_) \
+do {                                           \
+    *(was_incomplete_) = (*(req_)->cc_ptr)++;  \
+} while (0)
+
+#endif /* defined(MPICH_IS_THREADED) */
+
 #endif /* !defined(MPIIMPLTHREAD_H_INCLUDED) */
 

src/mpi/coll/helper_fns.c

@@ -238,12 +238,12 @@ int MPIC_Sendrecv_replace(void *buf, int count, MPI_Datatype datatype,
         /* --END ERROR HANDLING-- */
     }
 
-    if (*sreq->cc_ptr != 0 || *rreq->cc_ptr != 0)
+    if (!MPID_Request_is_complete(sreq) || !MPID_Request_is_complete(rreq))
     {
         MPID_Progress_state progress_state;
 
         MPID_Progress_start(&progress_state);
-        while (*sreq->cc_ptr != 0 || *rreq->cc_ptr != 0)
+        while (!MPID_Request_is_complete(sreq) || !MPID_Request_is_complete(rreq))
         {
             mpi_errno = MPID_Progress_wait(&progress_state);
             if (mpi_errno != MPI_SUCCESS)
@@ -508,12 +508,12 @@ int MPIC_Wait(MPID_Request * request_ptr)
     MPIDI_STATE_DECL(MPID_STATE_MPIC_WAIT);
 
     MPIDI_PT2PT_FUNC_ENTER(MPID_STATE_MPIC_WAIT);
-    if ((*(request_ptr)->cc_ptr) != 0)
+    if (!MPID_Request_is_complete(request_ptr))
     {
 	MPID_Progress_state progress_state;
 	
 	MPID_Progress_start(&progress_state);
-	while((*(request_ptr)->cc_ptr) != 0)
+        while (!MPID_Request_is_complete(request_ptr))
 	{
 	    mpi_errno = MPID_Progress_wait(&progress_state);
 	    if (mpi_errno) { MPIU_ERR_POP(mpi_errno); }

src/mpi/pt2pt/cancel.c

@@ -57,8 +57,11 @@ int MPIR_Cancel_impl(MPID_Request *request_ptr)
                     if (mpi_errno) MPIU_ERR_POP(mpi_errno);
 		} else {
 		    /* This is needed for persistent Bsend requests */
+                    /* FIXME why do we directly access the partner request's
+                     * cc field?  shouldn't our cc_ptr be set to the address
+                     * of the partner req's cc field? */
                     mpi_errno = MPIR_Grequest_cancel(request_ptr->partner_request,
-                                                     (request_ptr->partner_request->cc == 0));
+                                                     MPID_cc_is_complete(&request_ptr->partner_request->cc));
                     if (mpi_errno) MPIU_ERR_POP(mpi_errno);
 		}
 	    } else {
@@ -80,7 +83,7 @@ int MPIR_Cancel_impl(MPID_Request *request_ptr)
 
 	case MPID_UREQUEST:
 	{
-            mpi_errno = MPIR_Grequest_cancel(request_ptr, (request_ptr->cc == 0));
+            mpi_errno = MPIR_Grequest_cancel(request_ptr, MPID_cc_is_complete(&request_ptr->cc));
             if (mpi_errno) MPIU_ERR_POP(mpi_errno);
 	    break;
 	}

src/mpi/pt2pt/greq_start.c

@@ -150,6 +150,8 @@ int MPI_Grequest_start( MPI_Grequest_query_function *query_fn,
 #   endif /* HAVE_ERROR_CHECKING */
 
     /* ... body of routine ...  */
+
+    /* MT FIXME this routine is not thread-safe in the non-global case */
     
     lrequest_ptr = MPID_Request_create();
     /* --BEGIN ERROR HANDLING-- */
@@ -166,7 +168,7 @@ int MPI_Grequest_start( MPI_Grequest_query_function *query_fn,
     lrequest_ptr->kind                 = MPID_UREQUEST;
     MPIU_Object_set_ref( lrequest_ptr, 1 );
     lrequest_ptr->cc_ptr               = &lrequest_ptr->cc;
-    lrequest_ptr->cc                   = 1;
+    MPID_cc_set(lrequest_ptr->cc_ptr, 1);
     lrequest_ptr->comm                 = NULL;
     lrequest_ptr->cancel_fn            = cancel_fn;
     lrequest_ptr->free_fn              = free_fn;

src/mpi/pt2pt/mpir_request.c

@@ -21,12 +21,12 @@ int MPIR_Progress_wait_request(MPID_Request *req)
 {
     int mpi_errno = MPI_SUCCESS;
 
-    if ((*(req)->cc_ptr) != 0)
+    if (!MPID_Request_is_complete(req))
     {
         MPID_Progress_state progress_state;
 
         MPID_Progress_start(&progress_state);
-        while ((*(req)->cc_ptr) != 0)
+        while (!MPID_Request_is_complete(req))
         {
             mpi_errno = MPID_Progress_wait(&progress_state);
             if (mpi_errno != MPI_SUCCESS)
@@ -96,7 +96,7 @@ int MPIR_Request_complete(MPI_Request * request, MPID_Request * request_ptr,
 		MPID_Request * prequest_ptr = request_ptr->partner_request;
 
 		/* reset persistent request to inactive state */
-		request_ptr->cc = 0;
+                MPID_cc_set(&request_ptr->cc, 0);
 		request_ptr->cc_ptr = &request_ptr->cc;
 		request_ptr->partner_request = NULL;
 		
@@ -174,7 +174,7 @@ int MPIR_Request_complete(MPI_Request * request, MPID_Request * request_ptr,
 		MPID_Request * prequest_ptr = request_ptr->partner_request;
 
 		/* reset persistent request to inactive state */
-		request_ptr->cc = 0;
+                MPID_cc_set(&request_ptr->cc, 0);
 		request_ptr->cc_ptr = &request_ptr->cc;
 		request_ptr->partner_request = NULL;
 		
@@ -565,7 +565,7 @@ int MPIR_Grequest_progress_poke(int count,
      * request classes those grequests are members of */
     for (i=0, j=0, n_classes=1, n_native=0, n_greq=0; i< count; i++)
     {
-	if (request_ptrs[i] == NULL || *request_ptrs[i]->cc_ptr == 0) continue;
+	if (request_ptrs[i] == NULL || MPID_Request_is_complete(request_ptrs[i])) continue;
 	if (request_ptrs[i]->kind == MPID_UREQUEST)
 	{
 	    n_greq += 1;
@@ -590,7 +590,7 @@ int MPIR_Grequest_progress_poke(int count,
 	{
 	    if (request_ptrs[i] != NULL && 
                 request_ptrs[i]->kind == MPID_UREQUEST && 
-                *request_ptrs[i]->cc_ptr != 0 &&
+                !MPID_Request_is_complete(request_ptrs[i]) &&
                 request_ptrs[i]->poll_fn != NULL)
             {
 		mpi_errno = (request_ptrs[i]->poll_fn)(request_ptrs[i]->grequest_extra_state, &(array_of_statuses[i]));
@@ -680,7 +680,7 @@ int MPIR_Grequest_waitall(int count, MPID_Request * const * request_ptrs)
     {
         /* skip over requests we're not interested in */
         if (request_ptrs[i] == NULL ||
-            *request_ptrs[i]->cc_ptr == 0 ||
+            MPID_Request_is_complete(request_ptrs[i]) ||
             request_ptrs[i]->kind != MPID_UREQUEST ||
             request_ptrs[i]->wait_fn == NULL)
         {
@@ -689,21 +689,25 @@ int MPIR_Grequest_waitall(int count, MPID_Request * const * request_ptrs)
 
         mpi_error = (request_ptrs[i]->wait_fn)(1, &request_ptrs[i]->grequest_extra_state, 0, NULL);
         if (mpi_error) MPIU_ERR_POP(mpi_error);
-        MPIU_Assert(*request_ptrs[i]->cc_ptr == 0);
+        MPIU_Assert(MPID_Request_is_complete(request_ptrs[i]));
     }
 
     MPID_Progress_start(&progress_state);
     for (i = 0; i < count; ++i)
     {
-        if (request_ptrs[i] == NULL || *request_ptrs[i]->cc_ptr == 0 || request_ptrs[i]->kind != MPID_UREQUEST)
+        if (request_ptrs[i] == NULL ||
+            MPID_Request_is_complete(request_ptrs[i]) ||
+            request_ptrs[i]->kind != MPID_UREQUEST)
+        {
             continue;
+        }
         /* We have a greq that doesn't have a wait function; some other
            thread will cause completion via MPI_Grequest_complete().  Rather
            than waste the time by simply yielding the processor to the
            other thread, we'll make progress on regular requests too.  The
            progress engine should permit the other thread to run at some
            point. */
-        while (*request_ptrs[i]->cc_ptr != 0)
+        while (MPID_Request_is_complete(request_ptrs[i]))
         {
             mpi_error = MPID_Progress_wait(&progress_state);
             if (mpi_error != MPI_SUCCESS)

src/mpi/pt2pt/recv.c

@@ -144,12 +144,12 @@ int MPI_Recv(void *buf, int count, MPI_Datatype datatype, int source, int tag,
     
     /* If a request was returned, then we need to block until the request is 
        complete */
-    if ((*(request_ptr)->cc_ptr) != 0)
+    if (!MPID_Request_is_complete(request_ptr))
     {
 	MPID_Progress_state progress_state;
 	    
 	MPID_Progress_start(&progress_state);
-	while((*(request_ptr)->cc_ptr) != 0)
+        while (!MPID_Request_is_complete(request_ptr))
 	{
 	    /* MT: Progress_wait may release the SINGLE_CS while it
 	       waits */

src/mpi/pt2pt/request_get_status.c

@@ -105,13 +105,13 @@ int MPI_Request_get_status(MPI_Request request, int *flag, MPI_Status *status)
 
     /* ... body of routine ...  */
 
-    if (*request_ptr->cc_ptr != 0) {
+    if (!MPID_Request_is_complete(request_ptr)) {
 	/* request not complete. poke the progress engine. Req #3130 */
 	mpi_errno = MPID_Progress_test();
 	if (mpi_errno != MPI_SUCCESS) goto fn_fail;
     }
     
-    if (*request_ptr->cc_ptr == 0)
+    if (MPID_Request_is_complete(request_ptr))
     {
 	switch(request_ptr->kind)
 	{

src/mpi/pt2pt/rsend.c

@@ -129,12 +129,12 @@ int MPI_Rsend(void *buf, int count, MPI_Datatype datatype, int dest, int tag,
 
     /* If a request was returned, then we need to block until the request 
        is complete */
-    if ((*(request_ptr)->cc_ptr) != 0)
+    if (!MPID_Request_is_complete(request_ptr))
     {
 	MPID_Progress_state progress_state;
 	    
 	MPID_Progress_start(&progress_state);
-	while((*(request_ptr)->cc_ptr) != 0)
+        while (!MPID_Request_is_complete(request_ptr))
 	{
 	    mpi_errno = MPID_Progress_wait(&progress_state);
 	    if (mpi_errno != MPI_SUCCESS)

src/mpi/pt2pt/send.c

@@ -134,12 +134,12 @@ int MPI_Send(void *buf, int count, MPI_Datatype datatype, int dest, int tag,
 
     /* If a request was returned, then we need to block until the request 
        is complete */
-    if ((*(request_ptr)->cc_ptr) != 0)
+    if (!MPID_Request_is_complete(request_ptr))
     {
 	MPID_Progress_state progress_state;
 	    
 	MPID_Progress_start(&progress_state);
-	while((*(request_ptr)->cc_ptr) != 0)
+        while (!MPID_Request_is_complete(request_ptr))
 	{
 	    mpi_errno = MPID_Progress_wait(&progress_state);
 	    if (mpi_errno != MPI_SUCCESS)

src/mpi/pt2pt/sendrecv.c

@@ -171,12 +171,12 @@ int MPI_Sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 	/* --END ERROR HANDLING-- */
     }
 
-    if (*sreq->cc_ptr != 0 || *rreq->cc_ptr != 0)
+    if (!MPID_Request_is_complete(sreq) || !MPID_Request_is_complete(rreq))
     {
 	MPID_Progress_state progress_state;
 	
 	MPID_Progress_start(&progress_state);
-	while (*sreq->cc_ptr != 0 || *rreq->cc_ptr != 0)
+        while (!MPID_Request_is_complete(sreq) || !MPID_Request_is_complete(rreq))
 	{
 	    mpi_errno = MPID_Progress_wait(&progress_state);
 	    if (mpi_errno != MPI_SUCCESS)

src/mpi/pt2pt/sendrecv_rep.c

@@ -175,12 +175,12 @@ int MPI_Sendrecv_replace(void *buf, int count, MPI_Datatype datatype,
 	    /* --END ERROR HANDLING-- */
 	}
 	
-	if (*sreq->cc_ptr != 0 || *rreq->cc_ptr != 0)
+        if (!MPID_Request_is_complete(sreq) || !MPID_Request_is_complete(rreq))
 	{
 	    MPID_Progress_state progress_state;
 	
 	    MPID_Progress_start(&progress_state);
-	    while (*sreq->cc_ptr != 0 || *rreq->cc_ptr != 0)
+            while (!MPID_Request_is_complete(sreq) || !MPID_Request_is_complete(rreq))
 	    {
 		mpi_errno = MPID_Progress_wait(&progress_state);
 		if (mpi_errno != MPI_SUCCESS)

src/mpi/pt2pt/test.c

@@ -128,7 +128,7 @@ int MPI_Test(MPI_Request *request, int *flag, MPI_Status *status)
 	if (mpi_errno != MPI_SUCCESS) goto fn_fail;
     }
     
-    if (*request_ptr->cc_ptr == 0)
+    if (MPID_Request_is_complete(request_ptr))
     {
 	mpi_errno = MPIR_Request_complete(request, request_ptr, status, 
 					  &active_flag);

src/mpi/pt2pt/testall.c

@@ -169,7 +169,7 @@ int MPI_Testall(int count, MPI_Request array_of_requests[], int *flag,
 			    &(array_of_statuses[i]));
 	    if (mpi_errno != MPI_SUCCESS) goto fn_fail;
 	}
-	if (request_ptrs[i] != NULL && *request_ptrs[i]->cc_ptr == 0)
+	if (request_ptrs[i] != NULL && MPID_Request_is_complete(request_ptrs[i]))
 	{
 	    n_completed++;
 	    if (MPIR_Request_get_error(request_ptrs[i]) != MPI_SUCCESS)
@@ -186,7 +186,7 @@ int MPI_Testall(int count, MPI_Request array_of_requests[], int *flag,
 	{
 	    if (request_ptrs[i] != NULL)
 	    {
-		if (*request_ptrs[i]->cc_ptr == 0) 
+                if (MPID_Request_is_complete(request_ptrs[i]))
 		{
 		    n_completed ++;
 		    status_ptr = (array_of_statuses != MPI_STATUSES_IGNORE) ? &array_of_statuses[i] : MPI_STATUS_IGNORE;

src/mpi/pt2pt/testany.c

@@ -170,7 +170,7 @@ int MPI_Testany(int count, MPI_Request array_of_requests[], int *index,
 			    status);
 	    if (mpi_errno != MPI_SUCCESS) goto fn_fail;
 	}
-	if (request_ptrs[i] != NULL && *request_ptrs[i]->cc_ptr == 0)
+	if (request_ptrs[i] != NULL && MPID_Request_is_complete(request_ptrs[i]))
 	{
 	    mpi_errno = MPIR_Request_complete(&array_of_requests[i], 
 					      request_ptrs[i], 

src/mpi/pt2pt/testsome.c

@@ -171,7 +171,7 @@ int MPI_Testsome(int incount, MPI_Request array_of_requests[], int *outcount,
 			    array_of_statuses);
 	    if (mpi_errno != MPI_SUCCESS) goto fn_fail;
 	}
-	if (request_ptrs[i] != NULL && *request_ptrs[i]->cc_ptr == 0)
+	if (request_ptrs[i] != NULL && MPID_Request_is_complete(request_ptrs[i]))
 	{
 	    status_ptr = (array_of_statuses != MPI_STATUSES_IGNORE) ? &array_of_statuses[n_active] : MPI_STATUS_IGNORE;
 	    rc = MPIR_Request_complete(&array_of_requests[i], request_ptrs[i],

src/mpi/pt2pt/wait.c

@@ -104,12 +104,12 @@ int MPI_Wait(MPI_Request *request, MPI_Status *status)
 
     /* ... body of routine ... */
     
-    if ((*(request_ptr)->cc_ptr) != 0)
+    if (!MPID_Request_is_complete(request_ptr))
     {
 	MPID_Progress_state progress_state;
 	    
 	MPID_Progress_start(&progress_state);
-	while((*(request_ptr)->cc_ptr) != 0)
+        while (!MPID_Request_is_complete(request_ptr))
 	{
 	    mpi_errno = MPIR_Grequest_progress_poke(1, &request_ptr, status);
 	    if (request_ptr->kind == MPID_UREQUEST && 

src/mpi/pt2pt/waitall.c

@@ -186,7 +186,7 @@ int MPI_Waitall(int count, MPI_Request array_of_requests[],
         }
         
         /* wait for ith request to complete */
-        while (*request_ptrs[i]->cc_ptr != 0)
+        while (!MPID_Request_is_complete(request_ptrs[i]))
         {
             /* generalized requests should already be finished */
             MPIU_Assert(request_ptrs[i]->kind != MPID_UREQUEST);

src/mpi/pt2pt/waitany.c

@@ -164,7 +164,7 @@ int MPI_Waitany(int count, MPI_Request array_of_requests[], int *index,
 		mpi_errno = (request_ptrs[i]->poll_fn)(request_ptrs[i]->grequest_extra_state, status);
 		if (mpi_errno != MPI_SUCCESS) goto fn_progress_end_fail;
 	    }
-	    if (*request_ptrs[i]->cc_ptr == 0)
+            if (MPID_Request_is_complete(request_ptrs[i]))
 	    {
 		mpi_errno = MPIR_Request_complete(&array_of_requests[i], 
 						  request_ptrs[i], status, 

src/mpi/pt2pt/waitsome.c

@@ -196,7 +196,7 @@ int MPI_Waitsome(int incount, MPI_Request array_of_requests[],
 	if (mpi_errno != MPI_SUCCESS) goto fn_fail;
 	for (i = 0; i < incount; i++)
 	{
-	    if (request_ptrs[i] != NULL && *request_ptrs[i]->cc_ptr == 0)
+            if (request_ptrs[i] != NULL && MPID_Request_is_complete(request_ptrs[i]))
 	    {
 		status_ptr = (array_of_statuses != MPI_STATUSES_IGNORE) ? &array_of_statuses[n_active] : MPI_STATUS_IGNORE;
 		rc = MPIR_Request_complete(&array_of_requests[i], request_ptrs[i], status_ptr, &active_flag);