Merge pull request #1365 from hjelmn/osc_pt2pt_v2.x

osc/pt2pt threading fixes

Author: jsquyres

ompi/mca/osc/pt2pt/osc_pt2pt.h

@@ -121,7 +121,7 @@ struct ompi_osc_pt2pt_peer_t {
     int32_t passive_incoming_frag_count;
 
     /** peer flags */
-    int32_t flags;
+    volatile int32_t flags;
 };
 typedef struct ompi_osc_pt2pt_peer_t ompi_osc_pt2pt_peer_t;
 
@@ -144,11 +144,15 @@ static inline bool ompi_osc_pt2pt_peer_eager_active (ompi_osc_pt2pt_peer_t *peer
 
 static inline void ompi_osc_pt2pt_peer_set_flag (ompi_osc_pt2pt_peer_t *peer, int32_t flag, bool value)
 {
-    if (value) {
-        peer->flags |= flag;
-    } else {
-        peer->flags &= ~flag;
-    }
+    int32_t peer_flags, new_flags;
+    do {
+        peer_flags = peer->flags;
+        if (value) {
+            new_flags = peer_flags | flag;
+        } else {
+            new_flags = peer_flags & ~flag;
+        }
+    } while (!OPAL_ATOMIC_CMPSET_32 (&peer->flags, peer_flags, new_flags));
 }
 
 static inline void ompi_osc_pt2pt_peer_set_locked (ompi_osc_pt2pt_peer_t *peer, bool value)
@@ -937,13 +941,14 @@ static inline bool ompi_osc_pt2pt_access_epoch_active (ompi_osc_pt2pt_module_t *
 static inline bool ompi_osc_pt2pt_peer_sends_active (ompi_osc_pt2pt_module_t *module, int rank)
 {
     ompi_osc_pt2pt_sync_t *sync;
+    ompi_osc_pt2pt_peer_t *peer;
 
-    sync = ompi_osc_pt2pt_module_sync_lookup (module, rank, NULL);
+    sync = ompi_osc_pt2pt_module_sync_lookup (module, rank, &peer);
     if (!sync) {
         return false;
     }
 
-    return sync->eager_send_active;
+    return sync->eager_send_active || ompi_osc_pt2pt_peer_eager_active (peer);
 }
 
 END_C_DECLS

ompi/mca/osc/pt2pt/osc_pt2pt_comm.c

@@ -116,7 +116,7 @@ static inline int ompi_osc_pt2pt_put_self (ompi_osc_pt2pt_sync_t *pt2pt_sync, co
     int ret;
 
     /* if we are in active target mode wait until all post messages arrive */
-    ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+    ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
 
     ret = ompi_datatype_sndrcv ((void *)source, source_count, source_datatype,
                                 target, target_count, target_datatype);
@@ -140,7 +140,7 @@ static inline int ompi_osc_pt2pt_get_self (ompi_osc_pt2pt_sync_t *pt2pt_sync, vo
     int ret;
 
     /* if we are in active target mode wait until all post messages arrive */
-    ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+    ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
 
     ret = ompi_datatype_sndrcv (source, source_count, source_datatype,
                                 target, target_count, target_datatype);
@@ -162,7 +162,7 @@ static inline int ompi_osc_pt2pt_cas_self (ompi_osc_pt2pt_sync_t *pt2pt_sync, co
         ((unsigned long) target_disp * module->disp_unit);
 
     /* if we are in active target mode wait until all post messages arrive */
-    ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+    ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
 
     ompi_osc_pt2pt_accumulate_lock (module);
 
@@ -186,7 +186,7 @@ static inline int ompi_osc_pt2pt_acc_self (ompi_osc_pt2pt_sync_t *pt2pt_sync, co
     int ret;
 
     /* if we are in active target mode wait until all post messages arrive */
-    ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+    ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
 
     ompi_osc_pt2pt_accumulate_lock (module);
 
@@ -336,7 +336,7 @@ static inline int ompi_osc_pt2pt_put_w_req (const void *origin_addr, int origin_
 
     if (is_long_msg) {
         /* wait for eager sends to be active before starting a long put */
-        ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+        ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
     }
 
     OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
@@ -495,7 +495,7 @@ ompi_osc_pt2pt_accumulate_w_req (const void *origin_addr, int origin_count,
 
     if (is_long_msg) {
         /* wait for synchronization before posting a long message */
-        ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+        ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
     }
 
     header = (ompi_osc_pt2pt_header_acc_t*) ptr;
@@ -802,7 +802,7 @@ static inline int ompi_osc_pt2pt_rget_internal (void *origin_addr, int origin_co
 
     if (!release_req) {
         /* wait for epoch to begin before starting rget operation */
-        ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+        ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
     }
 
     header = (ompi_osc_pt2pt_header_get_t*) ptr;
@@ -968,7 +968,7 @@ int ompi_osc_pt2pt_rget_accumulate_internal (const void *origin_addr, int origin
 
     if (!release_req) {
         /* wait for epoch to begin before starting operation */
-        ompi_osc_pt2pt_sync_wait (pt2pt_sync);
+        ompi_osc_pt2pt_sync_wait_expected (pt2pt_sync);
     }
 
     /* optimize the self case. TODO: optimize the local case */

ompi/mca/osc/pt2pt/osc_pt2pt_data_move.c

@@ -1546,12 +1546,6 @@ static inline int process_frag (ompi_osc_pt2pt_module_t *module,
                 ret = process_acc_long (module, frag->source, &header->acc);
                 break;
 
-            case OMPI_OSC_PT2PT_HDR_TYPE_LOCK_REQ:
-                ret = ompi_osc_pt2pt_process_lock(module, frag->source, &header->lock);
-                if (OPAL_LIKELY(OMPI_SUCCESS == ret)) {
-                    ret = sizeof (header->lock);
-                }
-                break;
             case OMPI_OSC_PT2PT_HDR_TYPE_UNLOCK_REQ:
                 ret = process_unlock(module, frag->source, &header->unlock);
                 break;
@@ -1655,6 +1649,9 @@ int ompi_osc_pt2pt_process_receive (ompi_osc_pt2pt_receive_t *recv)
     case OMPI_OSC_PT2PT_HDR_TYPE_POST:
         osc_pt2pt_incoming_post (module, source);
         break;
+    case OMPI_OSC_PT2PT_HDR_TYPE_LOCK_REQ:
+        ompi_osc_pt2pt_process_lock(module, source, (ompi_osc_pt2pt_header_lock_t *) base_header);
+        break;
     case OMPI_OSC_PT2PT_HDR_TYPE_LOCK_ACK:
         ompi_osc_pt2pt_process_lock_ack(module, (ompi_osc_pt2pt_header_lock_ack_t *) base_header);
         break;

ompi/mca/osc/pt2pt/osc_pt2pt_frag.c

@@ -151,7 +151,6 @@ int ompi_osc_pt2pt_frag_flush_target (ompi_osc_pt2pt_module_t *module, int targe
     return ret;
 }
 
-
 int ompi_osc_pt2pt_frag_flush_all (ompi_osc_pt2pt_module_t *module)
 {
     int ret = OMPI_SUCCESS;

ompi/mca/osc/pt2pt/osc_pt2pt_frag.h

@@ -41,9 +41,9 @@ struct ompi_osc_pt2pt_frag_t {
 typedef struct ompi_osc_pt2pt_frag_t ompi_osc_pt2pt_frag_t;
 OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_frag_t);
 
-extern int ompi_osc_pt2pt_frag_start(ompi_osc_pt2pt_module_t *module, ompi_osc_pt2pt_frag_t *buffer);
-extern int ompi_osc_pt2pt_frag_flush_target(ompi_osc_pt2pt_module_t *module, int target);
-extern int ompi_osc_pt2pt_frag_flush_all(ompi_osc_pt2pt_module_t *module);
+int ompi_osc_pt2pt_frag_start(ompi_osc_pt2pt_module_t *module, ompi_osc_pt2pt_frag_t *buffer);
+int ompi_osc_pt2pt_frag_flush_target(ompi_osc_pt2pt_module_t *module, int target);
+int ompi_osc_pt2pt_frag_flush_all(ompi_osc_pt2pt_module_t *module);
 
 static inline int ompi_osc_pt2pt_frag_finish (ompi_osc_pt2pt_module_t *module,
                                               ompi_osc_pt2pt_frag_t* buffer)

ompi/mca/osc/pt2pt/osc_pt2pt_passive_target.c

@@ -122,6 +122,12 @@ int ompi_osc_pt2pt_lock_remote (ompi_osc_pt2pt_module_t *module, int target, omp
 
     int ret;
 
+    OPAL_THREAD_LOCK(&peer->lock);
+    if (ompi_osc_pt2pt_peer_locked (peer)) {
+        OPAL_THREAD_UNLOCK(&peer->lock);
+        return OMPI_SUCCESS;
+    }
+
     (void) OPAL_THREAD_ADD32(&lock->sync_expected, 1);
 
     assert (lock->type == OMPI_OSC_PT2PT_SYNC_TYPE_LOCK);
@@ -137,17 +143,15 @@ int ompi_osc_pt2pt_lock_remote (ompi_osc_pt2pt_module_t *module, int target, omp
     lock_req.lock_ptr = (uint64_t) (uintptr_t) lock;
     OSC_PT2PT_HTON(&lock_req, module, target);
 
-    ret = ompi_osc_pt2pt_control_send (module, target, &lock_req, sizeof (lock_req));
+    ret = ompi_osc_pt2pt_control_send_unbuffered (module, target, &lock_req, sizeof (lock_req));
     if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
-        return ret;
-    }
-
-    /* make sure the request gets sent, so we can start eager sending... */
-    ret = ompi_osc_pt2pt_frag_flush_target (module, target);
-    if (OPAL_LIKELY(OMPI_SUCCESS == ret)) {
+        OPAL_THREAD_ADD32(&lock->sync_expected, -1);
+    } else {
         ompi_osc_pt2pt_peer_set_locked (peer, true);
     }
 
+    OPAL_THREAD_UNLOCK(&peer->lock);
+
     return ret;
 }
 
@@ -316,6 +320,8 @@ static int ompi_osc_pt2pt_lock_internal (int lock_type, int target, int assert,
         if (OPAL_UNLIKELY(NULL == lock)) {
             return OMPI_ERR_OUT_OF_RESOURCE;
         }
+
+        lock->peer_list.peer = ompi_osc_pt2pt_peer_lookup (module, target);
     } else {
         lock = &module->all_sync;
     }

ompi/mca/osc/pt2pt/osc_pt2pt_sync.h

@@ -74,10 +74,10 @@ struct ompi_osc_pt2pt_sync_t {
     int num_peers;
 
     /** number of synchronization messages expected */
-    int32_t sync_expected;
+    volatile int32_t sync_expected;
 
     /** eager sends are active to all peers in this access epoch */
-    bool eager_send_active;
+    volatile bool eager_send_active;
 
     /** communication has started on this epoch */
     bool epoch_active;
@@ -175,7 +175,7 @@ static inline void ompi_osc_pt2pt_sync_expected (ompi_osc_pt2pt_sync_t *sync)
 static inline void ompi_osc_pt2pt_sync_reset (ompi_osc_pt2pt_sync_t *sync)
 {
     sync->type = OMPI_OSC_PT2PT_SYNC_TYPE_NONE;
-    sync->eager_send_active = 0;
+    sync->eager_send_active = false;
     sync->epoch_active = 0;
     sync->peer_list.peers = NULL;
     sync->sync.pscw.group = NULL;