Merge pull request #1321 from hjelmn/v2.x_osc_pt2pt_fixes_2s

v2.x osc/pt2pt fixes

Author: jsquyres

ompi/mca/osc/pt2pt/osc_pt2pt.h

@@ -12,7 +12,7 @@
  *                         reserved.
  * Copyright (c) 2010      Cisco Systems, Inc.  All rights reserved.
  * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
- * Copyright (c) 2015      Research Organization for Information Science
+ * Copyright (c) 2015-2016 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2016      FUJITSU LIMITED.  All rights reserved.
  * $COPYRIGHT$
@@ -47,6 +47,7 @@
 BEGIN_C_DECLS
 
 struct ompi_osc_pt2pt_frag_t;
+struct ompi_osc_pt2pt_receive_t;
 
 struct ompi_osc_pt2pt_component_t {
     /** Extend the basic osc component interface */
@@ -61,6 +62,9 @@ struct ompi_osc_pt2pt_component_t {
     /** module count */
     int module_count;
 
+    /** number of buffers per window */
+    int receive_count;
+
     /** free list of ompi_osc_pt2pt_frag_t structures */
     opal_free_list_t frags;
 
@@ -76,11 +80,26 @@ struct ompi_osc_pt2pt_component_t {
     /** List of operations that need to be processed */
     opal_list_t pending_operations;
 
+    /** List of receives to be processed */
+    opal_list_t pending_receives;
+
+    /** Lock for pending_receives */
+    opal_mutex_t pending_receives_lock;
+
     /** Is the progress function enabled? */
     bool progress_enable;
 };
 typedef struct ompi_osc_pt2pt_component_t ompi_osc_pt2pt_component_t;
 
+enum {
+    /** peer has sent an unexpected post message (no matching start) */
+    OMPI_OSC_PT2PT_PEER_FLAG_UNEX = 1,
+    /** eager sends are active on this peer */
+    OMPI_OSC_PT2PT_PEER_FLAG_EAGER = 2,
+    /** peer has been locked (on-demand locking for lock_all) */
+    OMPI_OSC_PT2PT_PEER_FLAG_LOCK = 4,
+};
+
 
 struct ompi_osc_pt2pt_peer_t {
     /** make this an opal object */
@@ -101,13 +120,54 @@ struct ompi_osc_pt2pt_peer_t {
     /** number of fragments incomming (negative - expected, positive - unsynchronized) */
     int32_t passive_incoming_frag_count;
 
-    /** unexpected post message arrived */
-    bool unexpected_post;
+    /** peer flags */
+    int32_t flags;
 };
 typedef struct ompi_osc_pt2pt_peer_t ompi_osc_pt2pt_peer_t;
 
 OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_peer_t);
 
+static inline bool ompi_osc_pt2pt_peer_locked (ompi_osc_pt2pt_peer_t *peer)
+{
+    return !!(peer->flags & OMPI_OSC_PT2PT_PEER_FLAG_LOCK);
+}
+
+static inline bool ompi_osc_pt2pt_peer_unex (ompi_osc_pt2pt_peer_t *peer)
+{
+    return !!(peer->flags & OMPI_OSC_PT2PT_PEER_FLAG_UNEX);
+}
+
+static inline bool ompi_osc_pt2pt_peer_eager_active (ompi_osc_pt2pt_peer_t *peer)
+{
+    return !!(peer->flags & OMPI_OSC_PT2PT_PEER_FLAG_EAGER);
+}
+
+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;
+    }
+}
+
+static inline void ompi_osc_pt2pt_peer_set_locked (ompi_osc_pt2pt_peer_t *peer, bool value)
+{
+    ompi_osc_pt2pt_peer_set_flag (peer, OMPI_OSC_PT2PT_PEER_FLAG_LOCK, value);
+}
+
+static inline void ompi_osc_pt2pt_peer_set_unex (ompi_osc_pt2pt_peer_t *peer, bool value)
+{
+    ompi_osc_pt2pt_peer_set_flag (peer, OMPI_OSC_PT2PT_PEER_FLAG_UNEX, value);
+}
+
+static inline void ompi_osc_pt2pt_peer_set_eager_active (ompi_osc_pt2pt_peer_t *peer, bool value)
+{
+    ompi_osc_pt2pt_peer_set_flag (peer, OMPI_OSC_PT2PT_PEER_FLAG_EAGER, value);
+}
+
+OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_peer_t);
+
 /** Module structure.  Exactly one of these is associated with each
     PT2PT window */
 struct ompi_osc_pt2pt_module_t {
@@ -192,8 +252,11 @@ struct ompi_osc_pt2pt_module_t {
     /** origin side list of locks currently outstanding */
     opal_hash_table_t outstanding_locks;
 
-    unsigned char *incoming_buffer;
-    ompi_request_t *frag_request;
+    /** receive fragments */
+    struct ompi_osc_pt2pt_receive_t *recv_frags;
+
+    /** number of receive fragments */
+    unsigned int recv_frag_count;
 
     /* enforce accumulate semantics */
     opal_atomic_lock_t accumulate_lock;
@@ -243,6 +306,15 @@ struct ompi_osc_pt2pt_pending_t {
 typedef struct ompi_osc_pt2pt_pending_t ompi_osc_pt2pt_pending_t;
 OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_pending_t);
 
+struct ompi_osc_pt2pt_receive_t {
+    opal_list_item_t super;
+    ompi_osc_pt2pt_module_t *module;
+    ompi_request_t *pml_request;
+    void *buffer;
+};
+typedef struct ompi_osc_pt2pt_receive_t ompi_osc_pt2pt_receive_t;
+OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_receive_t);
+
 #define GET_MODULE(win) ((ompi_osc_pt2pt_module_t*) win->w_osc_module)
 
 extern bool ompi_osc_pt2pt_no_locks;
@@ -409,6 +481,8 @@ int ompi_osc_pt2pt_component_irecv(ompi_osc_pt2pt_module_t *module,
                                   int tag,
                                   struct ompi_communicator_t *comm);
 
+int ompi_osc_pt2pt_lock_remote (ompi_osc_pt2pt_module_t *module, int target, ompi_osc_pt2pt_sync_t *lock);
+
 /**
  * ompi_osc_pt2pt_progress_pending_acc:
  *
@@ -823,6 +897,12 @@ static inline void ompi_osc_pt2pt_module_lock_remove (struct ompi_osc_pt2pt_modu
 static inline ompi_osc_pt2pt_sync_t *ompi_osc_pt2pt_module_sync_lookup (ompi_osc_pt2pt_module_t *module, int target,
                                                                         struct ompi_osc_pt2pt_peer_t **peer)
 {
+    ompi_osc_pt2pt_peer_t *tmp;
+
+    if (NULL == peer) {
+        peer = &tmp;
+    }
+
     OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                          "osc/pt2pt: looking for synchronization object for target %d", target));
 
@@ -840,8 +920,9 @@ static inline ompi_osc_pt2pt_sync_t *ompi_osc_pt2pt_module_sync_lookup (ompi_osc
 
         /* fence epoch is now active */
         module->all_sync.epoch_active = true;
-        if (peer) {
-            *peer = ompi_osc_pt2pt_peer_lookup (module, target);
+        *peer = ompi_osc_pt2pt_peer_lookup (module, target);
+        if (OMPI_OSC_PT2PT_SYNC_TYPE_LOCK == module->all_sync.type && !ompi_osc_pt2pt_peer_locked (*peer)) {
+            (void) ompi_osc_pt2pt_lock_remote (module, target, &module->all_sync);
         }
 
         return &module->all_sync;

ompi/mca/osc/pt2pt/osc_pt2pt_active_target.c

@@ -261,13 +261,13 @@ int ompi_osc_pt2pt_start (ompi_group_t *group, int assert, ompi_win_t *win)
         for (int i = 0 ; i < sync->num_peers ; ++i) {
             ompi_osc_pt2pt_peer_t *peer = sync->peer_list.peers[i];
 
-            if (peer->unexpected_post) {
+            if (ompi_osc_pt2pt_peer_unex (peer)) {
                 /* the peer already sent a post message for this pscw access epoch */
                 OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                                      "found unexpected post from %d",
                                      peer->rank));
                 OPAL_THREAD_ADD32 (&sync->sync_expected, -1);
-                peer->unexpected_post = false;
+                ompi_osc_pt2pt_peer_set_unex (peer, false);
             }
         }
         OPAL_THREAD_UNLOCK(&sync->lock);
@@ -600,7 +600,7 @@ void osc_pt2pt_incoming_post (ompi_osc_pt2pt_module_t *module, int source)
                              "received unexpected post message from %d for future PSCW synchronization",
                              source));
 
-        peer->unexpected_post = true;
+        ompi_osc_pt2pt_peer_set_unex (peer, true);
         OPAL_THREAD_UNLOCK(&sync->lock);
     } else {
         OPAL_THREAD_UNLOCK(&sync->lock);

ompi/mca/osc/pt2pt/osc_pt2pt_component.c

@@ -14,7 +14,7 @@
  * Copyright (c) 2006-2008 University of Houston.  All rights reserved.
  * Copyright (c) 2010      Oracle and/or its affiliates.  All rights reserved.
  * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
- * Copyright (c) 2015      Research Organization for Information Science
+ * Copyright (c) 2015-2016 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * $COPYRIGHT$
  *
@@ -142,44 +142,62 @@ static int component_register (void)
 					    NULL, 0, 0, OPAL_INFO_LVL_9, MCA_BASE_VAR_SCOPE_READONLY,
 					    &mca_osc_pt2pt_component.buffer_size);
 
+    mca_osc_pt2pt_component.receive_count = 4;
+    (void) mca_base_component_var_register (&mca_osc_pt2pt_component.super.osc_version, "receive_count",
+                                            "Number of receives to post for each window for incoming fragments "
+                                            "(default: 4)", MCA_BASE_VAR_TYPE_UNSIGNED_INT, NULL, 0, 0, OPAL_INFO_LVL_4,
+                                            MCA_BASE_VAR_SCOPE_READONLY, &mca_osc_pt2pt_component.receive_count);
+
     return OMPI_SUCCESS;
 }
 
 static int component_progress (void)
 {
-    int count = opal_list_get_size (&mca_osc_pt2pt_component.pending_operations);
+    int pending_count = opal_list_get_size (&mca_osc_pt2pt_component.pending_operations);
+    int recv_count = opal_list_get_size (&mca_osc_pt2pt_component.pending_receives);
     ompi_osc_pt2pt_pending_t *pending, *next;
 
-    if (0 == count) {
-	return 0;
+    if (recv_count) {
+        for (int i = 0 ; i < recv_count ; ++i) {
+            OPAL_THREAD_LOCK(&mca_osc_pt2pt_component.pending_receives_lock);
+            ompi_osc_pt2pt_receive_t *recv = (ompi_osc_pt2pt_receive_t *) opal_list_remove_first (&mca_osc_pt2pt_component.pending_receives);
+            OPAL_THREAD_UNLOCK(&mca_osc_pt2pt_component.pending_receives_lock);
+            if (NULL == recv) {
+                break;
+            }
+
+            (void) ompi_osc_pt2pt_process_receive (recv);
+        }
     }
 
     /* process one incoming request */
-    OPAL_THREAD_LOCK(&mca_osc_pt2pt_component.pending_operations_lock);
-    OPAL_LIST_FOREACH_SAFE(pending, next, &mca_osc_pt2pt_component.pending_operations, ompi_osc_pt2pt_pending_t) {
-	int ret;
-
-	switch (pending->header.base.type) {
-	case OMPI_OSC_PT2PT_HDR_TYPE_FLUSH_REQ:
-	    ret = ompi_osc_pt2pt_process_flush (pending->module, pending->source,
-					       &pending->header.flush);
-	    break;
-	case OMPI_OSC_PT2PT_HDR_TYPE_UNLOCK_REQ:
-	    ret = ompi_osc_pt2pt_process_unlock (pending->module, pending->source,
-						&pending->header.unlock);
-	    break;
-	default:
-	    /* shouldn't happen */
-	    assert (0);
-	    abort ();
-	}
-
-	if (OMPI_SUCCESS == ret) {
-	    opal_list_remove_item (&mca_osc_pt2pt_component.pending_operations, &pending->super);
-	    OBJ_RELEASE(pending);
-	}
+    if (pending_count) {
+        OPAL_THREAD_LOCK(&mca_osc_pt2pt_component.pending_operations_lock);
+        OPAL_LIST_FOREACH_SAFE(pending, next, &mca_osc_pt2pt_component.pending_operations, ompi_osc_pt2pt_pending_t) {
+            int ret;
+
+            switch (pending->header.base.type) {
+            case OMPI_OSC_PT2PT_HDR_TYPE_FLUSH_REQ:
+                ret = ompi_osc_pt2pt_process_flush (pending->module, pending->source,
+                                                    &pending->header.flush);
+                break;
+            case OMPI_OSC_PT2PT_HDR_TYPE_UNLOCK_REQ:
+                ret = ompi_osc_pt2pt_process_unlock (pending->module, pending->source,
+                                                     &pending->header.unlock);
+                break;
+            default:
+                /* shouldn't happen */
+                assert (0);
+                abort ();
+            }
+
+            if (OMPI_SUCCESS == ret) {
+                opal_list_remove_item (&mca_osc_pt2pt_component.pending_operations, &pending->super);
+                OBJ_RELEASE(pending);
+            }
+        }
+        OPAL_THREAD_UNLOCK(&mca_osc_pt2pt_component.pending_operations_lock);
     }
-    OPAL_THREAD_UNLOCK(&mca_osc_pt2pt_component.pending_operations_lock);
 
     return 1;
 }
@@ -193,6 +211,8 @@ component_init(bool enable_progress_threads,
     OBJ_CONSTRUCT(&mca_osc_pt2pt_component.lock, opal_mutex_t);
     OBJ_CONSTRUCT(&mca_osc_pt2pt_component.pending_operations, opal_list_t);
     OBJ_CONSTRUCT(&mca_osc_pt2pt_component.pending_operations_lock, opal_mutex_t);
+    OBJ_CONSTRUCT(&mca_osc_pt2pt_component.pending_receives, opal_list_t);
+    OBJ_CONSTRUCT(&mca_osc_pt2pt_component.pending_receives_lock, opal_mutex_t);
 
     OBJ_CONSTRUCT(&mca_osc_pt2pt_component.modules,
                   opal_hash_table_t);
@@ -253,6 +273,8 @@ component_finalize(void)
     OBJ_DESTRUCT(&mca_osc_pt2pt_component.requests);
     OBJ_DESTRUCT(&mca_osc_pt2pt_component.pending_operations);
     OBJ_DESTRUCT(&mca_osc_pt2pt_component.pending_operations_lock);
+    OBJ_DESTRUCT(&mca_osc_pt2pt_component.pending_receives);
+    OBJ_DESTRUCT(&mca_osc_pt2pt_component.pending_receives_lock);
 
     return OMPI_SUCCESS;
 }
@@ -385,11 +407,6 @@ component_select(struct ompi_win_t *win, void **base, size_t size, int disp_unit
     /* sync memory - make sure all initialization completed */
     opal_atomic_mb();
 
-    module->incoming_buffer = malloc (mca_osc_pt2pt_component.buffer_size + sizeof (ompi_osc_pt2pt_frag_header_t));
-    if (OPAL_UNLIKELY(NULL == module->incoming_buffer)) {
-	goto cleanup;
-    }
-
     ret = ompi_osc_pt2pt_frag_start_receive (module);
     if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
 	goto cleanup;
@@ -449,13 +466,33 @@ ompi_osc_pt2pt_get_info(struct ompi_win_t *win, struct ompi_info_t **info_used)
 
 OBJ_CLASS_INSTANCE(ompi_osc_pt2pt_pending_t, opal_list_item_t, NULL, NULL);
 
+static void ompi_osc_pt2pt_receive_construct (ompi_osc_pt2pt_receive_t *recv)
+{
+    recv->buffer = NULL;
+    recv->pml_request = NULL;
+}
+
+static void ompi_osc_pt2pt_receive_destruct (ompi_osc_pt2pt_receive_t *recv)
+{
+    free (recv->buffer);
+    if (recv->pml_request && MPI_REQUEST_NULL != recv->pml_request) {
+        recv->pml_request->req_complete_cb = NULL;
+        ompi_request_cancel (recv->pml_request);
+        ompi_request_free (&recv->pml_request);
+    }
+}
+
+OBJ_CLASS_INSTANCE(ompi_osc_pt2pt_receive_t, opal_list_item_t,
+                   ompi_osc_pt2pt_receive_construct,
+                   ompi_osc_pt2pt_receive_destruct);
+
 static void ompi_osc_pt2pt_peer_construct (ompi_osc_pt2pt_peer_t *peer)
 {
     OBJ_CONSTRUCT(&peer->queued_frags, opal_list_t);
     OBJ_CONSTRUCT(&peer->lock, opal_mutex_t);
     peer->active_frag = NULL;
     peer->passive_incoming_frag_count = 0;
-    peer->unexpected_post = false;
+    peer->flags = 0;
 }
 
 static void ompi_osc_pt2pt_peer_destruct (ompi_osc_pt2pt_peer_t *peer)