osc/pt2pt: fix threading issues

This commit fixes a number of threading issues discovered in
osc/pt2pt. This includes:

 - Lock the synchronization object not the module in osc_pt2pt_start.
   This fixes a race between the start function and processing post
   messages.

 - Always lock before calling cond_broadcast. Fixes a race between
   the waiting thread and signaling thread.

 - Make all atomically updated values volatile.

 - Make the module lock recursive to protect against some deadlock
   conditions. Will roll this back once the locks have been
   re-designed.

 - Mark incoming complete *after* completing an accumulate not
   before. This was causing an incorrect answer under certain
   conditions.

Signed-off-by: Nathan Hjelm <[email protected]>

Author: hjelmn

ompi/mca/osc/pt2pt/osc_pt2pt.h

@@ -118,7 +118,7 @@ struct ompi_osc_pt2pt_peer_t {
     opal_list_t queued_frags;
 
     /** number of fragments incomming (negative - expected, positive - unsynchronized) */
-    int32_t passive_incoming_frag_count;
+    volatile int32_t passive_incoming_frag_count;
 
     /** peer flags */
     volatile int32_t flags;
@@ -198,7 +198,7 @@ struct ompi_osc_pt2pt_module_t {
     int disp_unit;
 
     /** Mutex lock protecting module data */
-    opal_mutex_t lock;
+    opal_recursive_mutex_t lock;
 
     /** condition variable associated with lock */
     opal_condition_t cond;
@@ -214,19 +214,13 @@ struct ompi_osc_pt2pt_module_t {
     uint32_t *epoch_outgoing_frag_count;
 
     /** cyclic counter for a unique tage for long messages. */
-    uint32_t tag_counter;
+    volatile uint32_t tag_counter;
 
-    /* Number of outgoing fragments that have completed since the
-       begining of time */
-    volatile uint32_t outgoing_frag_count;
-    /* Next outgoing fragment count at which we want a signal on cond */
-    volatile uint32_t outgoing_frag_signal_count;
+    /** number of outgoing fragments still to be completed */
+    volatile int32_t outgoing_frag_count;
 
-    /* Number of incoming fragments that have completed since the
-       begining of time */
-    volatile uint32_t active_incoming_frag_count;
-    /* Next incoming buffer count at which we want a signal on cond */
-    volatile uint32_t active_incoming_frag_signal_count;
+    /** number of incoming fragments */
+    volatile int32_t active_incoming_frag_count;
 
     /** Number of targets locked/being locked */
     unsigned int passive_target_access_epoch;
@@ -239,7 +233,7 @@ struct ompi_osc_pt2pt_module_t {
 
     /** Number of "count" messages from the remote complete group
         we've received */
-    int32_t num_complete_msgs;
+    volatile int32_t num_complete_msgs;
 
     /* ********************* LOCK data ************************ */
 
@@ -264,7 +258,12 @@ struct ompi_osc_pt2pt_module_t {
 
     /* enforce accumulate semantics */
     opal_atomic_lock_t accumulate_lock;
-    opal_list_t        pending_acc;
+
+    /** accumulate operations pending the accumulation lock */
+    opal_list_t pending_acc;
+
+    /** lock for pending_acc */
+    opal_mutex_t pending_acc_lock;
 
     /** Lock for garbage collection lists */
     opal_mutex_t gc_lock;
@@ -512,23 +511,29 @@ int ompi_osc_pt2pt_progress_pending_acc (ompi_osc_pt2pt_module_t *module);
  */
 static inline void mark_incoming_completion (ompi_osc_pt2pt_module_t *module, int source)
 {
+    int32_t new_value;
+
     if (MPI_PROC_NULL == source) {
         OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
-                             "mark_incoming_completion marking active incoming complete. count = %d. signal = %d",
-                             (int) module->active_incoming_frag_count + 1, module->active_incoming_frag_signal_count));
-        OPAL_THREAD_ADD32((int32_t *) &module->active_incoming_frag_count, 1);
-        if (module->active_incoming_frag_count >= module->active_incoming_frag_signal_count) {
+                             "mark_incoming_completion marking active incoming complete. module %p, count = %d",
+                             (void *) module, (int) module->active_incoming_frag_count + 1));
+        new_value = OPAL_THREAD_ADD32(&module->active_incoming_frag_count, 1);
+        if (new_value >= 0) {
+            OPAL_THREAD_LOCK(&module->lock);
             opal_condition_broadcast(&module->cond);
+            OPAL_THREAD_UNLOCK(&module->lock);
         }
     } else {
         ompi_osc_pt2pt_peer_t *peer = ompi_osc_pt2pt_peer_lookup (module, source);
 
         OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
-                             "mark_incoming_completion marking passive incoming complete. source = %d, count = %d",
-                             source, (int) peer->passive_incoming_frag_count + 1));
-        OPAL_THREAD_ADD32((int32_t *) &peer->passive_incoming_frag_count, 1);
-        if (0 == peer->passive_incoming_frag_count) {
+                             "mark_incoming_completion marking passive incoming complete. module %p, source = %d, count = %d",
+                             (void *) module, source, (int) peer->passive_incoming_frag_count + 1));
+        new_value = OPAL_THREAD_ADD32((int32_t *) &peer->passive_incoming_frag_count, 1);
+        if (0 == new_value) {
+            OPAL_THREAD_LOCK(&module->lock);
             opal_condition_broadcast(&module->cond);
+            OPAL_THREAD_UNLOCK(&module->lock);
         }
     }
 }
@@ -548,9 +553,13 @@ static inline void mark_incoming_completion (ompi_osc_pt2pt_module_t *module, in
  */
 static inline void mark_outgoing_completion (ompi_osc_pt2pt_module_t *module)
 {
-    OPAL_THREAD_ADD32((int32_t *) &module->outgoing_frag_count, 1);
-    if (module->outgoing_frag_count >= module->outgoing_frag_signal_count) {
+    int32_t new_value = OPAL_THREAD_ADD32((int32_t *) &module->outgoing_frag_count, 1);
+    OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
+                         "mark_outgoing_completion: outgoing_frag_count = %d", new_value));
+    if (new_value >= 0) {
+        OPAL_THREAD_LOCK(&module->lock);
         opal_condition_broadcast(&module->cond);
+        OPAL_THREAD_UNLOCK(&module->lock);
     }
 }
 
@@ -568,7 +577,7 @@ static inline void mark_outgoing_completion (ompi_osc_pt2pt_module_t *module)
  */
 static inline void ompi_osc_signal_outgoing (ompi_osc_pt2pt_module_t *module, int target, int count)
 {
-    OPAL_THREAD_ADD32((int32_t *) &module->outgoing_frag_signal_count, count);
+    OPAL_THREAD_ADD32((int32_t *) &module->outgoing_frag_count, -count);
     if (MPI_PROC_NULL != target) {
         OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                              "ompi_osc_signal_outgoing_passive: target = %d, count = %d, total = %d", target,

ompi/mca/osc/pt2pt/osc_pt2pt_active_target.c

@@ -168,7 +168,6 @@ int ompi_osc_pt2pt_fence(int assert, ompi_win_t *win)
                                                           MPI_SUM, module->comm,
                                                           module->comm->c_coll.coll_reduce_scatter_block_module);
     if (OMPI_SUCCESS != ret) {
-        OPAL_THREAD_UNLOCK(&module->lock);
         return ret;
     }
 
@@ -181,11 +180,10 @@ int ompi_osc_pt2pt_fence(int assert, ompi_win_t *win)
                          incoming_reqs));
 
     /* set our complete condition for incoming requests */
-    module->active_incoming_frag_signal_count += incoming_reqs;
+    OPAL_THREAD_ADD32(&module->active_incoming_frag_count, -incoming_reqs);
 
     /* wait for completion */
-    while (module->outgoing_frag_count != module->outgoing_frag_signal_count ||
-           module->active_incoming_frag_count < module->active_incoming_frag_signal_count) {
+    while (module->outgoing_frag_count < 0 || module->active_incoming_frag_count < 0) {
         opal_condition_wait(&module->cond, &module->lock);
     }
 
@@ -196,10 +194,10 @@ int ompi_osc_pt2pt_fence(int assert, ompi_win_t *win)
     }
 
     module->all_sync.epoch_active = false;
-
-    opal_condition_broadcast (&module->cond);
     OPAL_THREAD_UNLOCK(&module->lock);
 
+    module->comm->c_coll.coll_barrier (module->comm, module->comm->c_coll.coll_barrier_module);
+
     OPAL_OUTPUT_VERBOSE((25, ompi_osc_base_framework.framework_output,
                          "osc pt2pt: fence end: %d", ret));
 
@@ -212,11 +210,11 @@ int ompi_osc_pt2pt_start (ompi_group_t *group, int assert, ompi_win_t *win)
     ompi_osc_pt2pt_module_t *module = GET_MODULE(win);
     ompi_osc_pt2pt_sync_t *sync = &module->all_sync;
 
-    OPAL_THREAD_LOCK(&module->lock);
+    OPAL_THREAD_LOCK(&sync->lock);
 
     /* check if we are already in an access epoch */
     if (ompi_osc_pt2pt_access_epoch_active (module)) {
-        OPAL_THREAD_UNLOCK(&module->lock);
+        OPAL_THREAD_UNLOCK(&sync->lock);
         return OMPI_ERR_RMA_SYNC;
     }
 
@@ -249,7 +247,7 @@ int ompi_osc_pt2pt_start (ompi_group_t *group, int assert, ompi_win_t *win)
     if (0 == ompi_group_size (group)) {
         /* nothing more to do. this is an empty start epoch */
         sync->eager_send_active = true;
-        OPAL_THREAD_UNLOCK(&module->lock);
+        OPAL_THREAD_UNLOCK(&sync->lock);
         return OMPI_SUCCESS;
     }
 
@@ -258,12 +256,11 @@ int ompi_osc_pt2pt_start (ompi_group_t *group, int assert, ompi_win_t *win)
     /* translate the group ranks into the communicator */
     sync->peer_list.peers = ompi_osc_pt2pt_get_peers (module, group);
     if (NULL == sync->peer_list.peers) {
-        OPAL_THREAD_UNLOCK(&module->lock);
+        OPAL_THREAD_UNLOCK(&sync->lock);
         return OMPI_ERR_OUT_OF_RESOURCE;
     }
 
     if (!(assert & MPI_MODE_NOCHECK)) {
-        OPAL_THREAD_LOCK(&sync->lock);
         for (int i = 0 ; i < sync->num_peers ; ++i) {
             ompi_osc_pt2pt_peer_t *peer = sync->peer_list.peers[i];
 
@@ -276,7 +273,6 @@ int ompi_osc_pt2pt_start (ompi_group_t *group, int assert, ompi_win_t *win)
                 ompi_osc_pt2pt_peer_set_unex (peer, false);
             }
         }
-        OPAL_THREAD_UNLOCK(&sync->lock);
     } else {
         sync->sync_expected = 0;
     }
@@ -295,7 +291,7 @@ int ompi_osc_pt2pt_start (ompi_group_t *group, int assert, ompi_win_t *win)
                          "ompi_osc_pt2pt_start complete. eager sends active: %d",
                          sync->eager_send_active));
 
-    OPAL_THREAD_UNLOCK(&module->lock);
+    OPAL_THREAD_UNLOCK(&sync->lock);
     return OMPI_SUCCESS;
 }
 
@@ -313,14 +309,14 @@ int ompi_osc_pt2pt_complete (ompi_win_t *win)
     OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                          "ompi_osc_pt2pt_complete entering..."));
 
-    OPAL_THREAD_LOCK(&module->lock);
+    OPAL_THREAD_LOCK(&sync->lock);
     if (OMPI_OSC_PT2PT_SYNC_TYPE_PSCW != sync->type) {
-        OPAL_THREAD_UNLOCK(&module->lock);
+        OPAL_THREAD_UNLOCK(&sync->lock);
         return OMPI_ERR_RMA_SYNC;
     }
 
     /* wait for all the post messages */
-    ompi_osc_pt2pt_sync_wait (sync);
+    ompi_osc_pt2pt_sync_wait_nolock (sync);
 
     /* phase 1 cleanup sync object */
     group = sync->sync.pscw.group;
@@ -330,8 +326,7 @@ int ompi_osc_pt2pt_complete (ompi_win_t *win)
 
     /* need to reset the sync here to avoid processing incorrect post messages */
     ompi_osc_pt2pt_sync_reset (sync);
-
-    OPAL_THREAD_UNLOCK(&module->lock);
+    OPAL_THREAD_UNLOCK(&sync->lock);
 
     OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                          "ompi_osc_pt2pt_complete all posts received. sending complete messages..."));
@@ -403,7 +398,7 @@ int ompi_osc_pt2pt_complete (ompi_win_t *win)
     OPAL_THREAD_LOCK(&module->lock);
     /* wait for outgoing requests to complete.  Don't wait for incoming, as
        we're only completing the access epoch, not the exposure epoch */
-    while (module->outgoing_frag_count != module->outgoing_frag_signal_count) {
+    while (module->outgoing_frag_count < 0) {
         opal_condition_wait(&module->cond, &module->lock);
     }
 
@@ -513,15 +508,13 @@ int ompi_osc_pt2pt_wait (ompi_win_t *win)
     }
 
     OPAL_OUTPUT_VERBOSE((25, ompi_osc_base_framework.framework_output,
-                         "ompi_osc_pt2pt_wait entering..."));
+                         "ompi_osc_pt2pt_wait entering... module %p", (void *) module));
 
     OPAL_THREAD_LOCK(&module->lock);
-    while (0 != module->num_complete_msgs ||
-             module->active_incoming_frag_count != module->active_incoming_frag_signal_count) {
-        OPAL_OUTPUT_VERBOSE((25, ompi_osc_base_framework.framework_output, "num_complete_msgs = %d, "
-                             "active_incoming_frag_count = %d, active_incoming_frag_signal_count = %d",
-                             module->num_complete_msgs, module->active_incoming_frag_count,
-                             module->active_incoming_frag_signal_count));
+    while (0 != module->num_complete_msgs || module->active_incoming_frag_count < 0) {
+        OPAL_OUTPUT_VERBOSE((25, ompi_osc_base_framework.framework_output, "module %p, num_complete_msgs = %d, "
+                             "active_incoming_frag_count = %d", (void *) module, module->num_complete_msgs,
+                             module->active_incoming_frag_count));
         opal_condition_wait(&module->cond, &module->lock);
     }
 
@@ -554,21 +547,15 @@ int ompi_osc_pt2pt_test (ompi_win_t *win, int *flag)
 
     OPAL_THREAD_LOCK(&(module->lock));
 
-    if (0 != module->num_complete_msgs ||
-           module->active_incoming_frag_count != module->active_incoming_frag_signal_count) {
+    if (0 != module->num_complete_msgs || module->active_incoming_frag_count < 0) {
         *flag = 0;
-        ret = OMPI_SUCCESS;
     } else {
         *flag = 1;
 
         group = module->pw_group;
         module->pw_group = NULL;
 
-        OPAL_THREAD_UNLOCK(&(module->lock));
-
         OBJ_RELEASE(group);
-
-        return OMPI_SUCCESS;
     }
 
     OPAL_THREAD_UNLOCK(&(module->lock));
@@ -580,15 +567,19 @@ void osc_pt2pt_incoming_complete (ompi_osc_pt2pt_module_t *module, int source, i
 {
     OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                          "osc pt2pt:  process_complete got complete message from %d. expected fragment count %d. "
-                         "current signal count %d. current incomming count: %d. expected complete msgs: %d",
-                         source, frag_count, module->active_incoming_frag_signal_count,
-                         module->active_incoming_frag_count, module->num_complete_msgs));
+                         "current incomming count: %d. expected complete msgs: %d", source,
+                         frag_count, module->active_incoming_frag_count, module->num_complete_msgs));
 
     /* the current fragment is not part of the frag_count so we need to add it here */
-    OPAL_THREAD_ADD32((int32_t *) &module->active_incoming_frag_signal_count, frag_count);
+    OPAL_THREAD_ADD32(&module->active_incoming_frag_count, -frag_count);
 
-    if (0 == OPAL_THREAD_ADD32((int32_t *) &module->num_complete_msgs, 1)) {
+    /* make sure the signal count is written before changing the complete message count */
+    opal_atomic_wmb ();
+
+    if (0 == OPAL_THREAD_ADD32(&module->num_complete_msgs, 1)) {
+        OPAL_THREAD_LOCK(&module->lock);
         opal_condition_broadcast (&module->cond);
+        OPAL_THREAD_UNLOCK(&module->lock);
     }
 }
 

ompi/mca/osc/pt2pt/osc_pt2pt_comm.c

@@ -501,6 +501,9 @@ ompi_osc_pt2pt_accumulate_w_req (const void *origin_addr, int origin_count,
 
         is_long_msg = true;
         tag = get_tag (module);
+    } else {
+        /* still need to set the tag for the active/passive logic on the target */
+        tag = !!(module->passive_target_access_epoch);
     }
 
     if (is_long_msg) {
@@ -523,6 +526,7 @@ ompi_osc_pt2pt_accumulate_w_req (const void *origin_addr, int origin_count,
     header->count = target_count;
     header->displacement = target_disp;
     header->op = op->o_f_to_c_index;
+    header->tag = tag;
     ptr += sizeof (*header);
 
     do {
@@ -565,7 +569,6 @@ ompi_osc_pt2pt_accumulate_w_req (const void *origin_addr, int origin_count,
             }
         } else {
             header->base.type = OMPI_OSC_PT2PT_HDR_TYPE_ACC_LONG;
-            header->tag = tag;
             osc_pt2pt_hton(header, proc);
 
             OPAL_OUTPUT_VERBOSE((25, ompi_osc_base_framework.framework_output,

ompi/mca/osc/pt2pt/osc_pt2pt_component.c

@@ -314,12 +314,13 @@ component_select(struct ompi_win_t *win, void **base, size_t size, int disp_unit
            sizeof(ompi_osc_base_module_t));
 
     /* initialize the objects, so that always free in cleanup */
-    OBJ_CONSTRUCT(&module->lock, opal_mutex_t);
+    OBJ_CONSTRUCT(&module->lock, opal_recursive_mutex_t);
     OBJ_CONSTRUCT(&module->cond, opal_condition_t);
     OBJ_CONSTRUCT(&module->locks_pending, opal_list_t);
     OBJ_CONSTRUCT(&module->locks_pending_lock, opal_mutex_t);
     OBJ_CONSTRUCT(&module->outstanding_locks, opal_hash_table_t);
     OBJ_CONSTRUCT(&module->pending_acc, opal_list_t);
+    OBJ_CONSTRUCT(&module->pending_acc_lock, opal_mutex_t);
     OBJ_CONSTRUCT(&module->buffer_gc, opal_list_t);
     OBJ_CONSTRUCT(&module->gc_lock, opal_mutex_t);
     OBJ_CONSTRUCT(&module->all_sync, ompi_osc_pt2pt_sync_t);