Merge pull request open-mpi#725 from hjelmn/add_procs_threading

add_procs: add threading protection for dynamic add_procs

Author: hppritcha

ompi/group/group.h

@@ -333,25 +333,34 @@ int ompi_group_minloc (int list[], int length);
  */
 static inline struct ompi_proc_t *ompi_group_dense_lookup (ompi_group_t *group, const int peer_id, const bool allocate)
 {
+    ompi_proc_t *proc;
+
 #if OPAL_ENABLE_DEBUG
     if (peer_id >= group->grp_proc_count) {
         opal_output(0, "ompi_group_dense_lookup: invalid peer index (%d)", peer_id);
         return (struct ompi_proc_t *) NULL;
     }
 #endif
 
-    if (OPAL_UNLIKELY(ompi_proc_is_sentinel (group->grp_proc_pointers[peer_id]))) {
+    proc = group->grp_proc_pointers[peer_id];
+
+    if (OPAL_UNLIKELY(ompi_proc_is_sentinel (proc))) {
         if (!allocate) {
             return NULL;
         }
 
         /* replace sentinel value with an actual ompi_proc_t */
-        group->grp_proc_pointers[peer_id] =
-            (ompi_proc_t *) ompi_proc_for_name (ompi_proc_sentinel_to_name ((intptr_t) group->grp_proc_pointers[peer_id]));
-        OBJ_RETAIN(group->grp_proc_pointers[peer_id]);
+        ompi_proc_t *real_proc =
+            (ompi_proc_t *) ompi_proc_for_name (ompi_proc_sentinel_to_name ((intptr_t) proc));
+
+        if (opal_atomic_cmpset_ptr (group->grp_proc_pointers + peer_id, proc, real_proc)) {
+            OBJ_RETAIN(real_proc);
+        }
+
+        proc = real_proc;
     }
 
-    return group->grp_proc_pointers[peer_id];
+    return proc;
 }
 
 /*

ompi/mca/bml/base/base.h

@@ -63,10 +63,15 @@ OMPI_DECLSPEC  int mca_bml_base_ft_event(int state);
 OMPI_DECLSPEC extern mca_bml_base_component_t mca_bml_component;
 OMPI_DECLSPEC extern mca_bml_base_module_t mca_bml;
 OMPI_DECLSPEC extern mca_base_framework_t ompi_bml_base_framework;
+OMPI_DECLSPEC extern opal_mutex_t mca_bml_lock;
 
 static inline struct mca_bml_base_endpoint_t *mca_bml_base_get_endpoint (struct ompi_proc_t *proc) {
     if (OPAL_UNLIKELY(NULL == proc->proc_endpoints[OMPI_PROC_ENDPOINT_TAG_BML])) {
-        mca_bml.bml_add_proc (proc);
+        OPAL_THREAD_LOCK(&mca_bml_lock);
+        if (NULL == proc->proc_endpoints[OMPI_PROC_ENDPOINT_TAG_BML]) {
+            mca_bml.bml_add_proc (proc);
+        }
+        OPAL_THREAD_UNLOCK(&mca_bml_lock);
     }
 
     return (struct mca_bml_base_endpoint_t *) proc->proc_endpoints[OMPI_PROC_ENDPOINT_TAG_BML];

ompi/mca/bml/base/bml_base_frame.c

@@ -50,6 +50,8 @@ static bool mca_bml_base_srand;
 opal_rng_buff_t mca_bml_base_rand_buff;
 #endif
 
+opal_mutex_t mca_bml_lock = OPAL_MUTEX_STATIC_INIT;
+
 static int mca_bml_base_register(mca_base_register_flag_t flags)
 {
 #if OPAL_ENABLE_DEBUG_RELIABILITY

ompi/mca/pml/ob1/pml_ob1_comm.c

@@ -55,6 +55,7 @@ static void mca_pml_ob1_comm_construct(mca_pml_ob1_comm_t* comm)
 {
     OBJ_CONSTRUCT(&comm->wild_receives, opal_list_t);
     OBJ_CONSTRUCT(&comm->matching_lock, opal_mutex_t);
+    OBJ_CONSTRUCT(&comm->proc_lock, opal_mutex_t);
     comm->recv_sequence = 0;
     comm->procs = NULL;
     comm->last_probed = 0;
@@ -76,6 +77,7 @@ static void mca_pml_ob1_comm_destruct(mca_pml_ob1_comm_t* comm)
 
     OBJ_DESTRUCT(&comm->wild_receives);
     OBJ_DESTRUCT(&comm->matching_lock);
+    OBJ_DESTRUCT(&comm->proc_lock);
 }
 
 

ompi/mca/pml/ob1/pml_ob1_comm.h

@@ -1,3 +1,4 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
 /*
  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
  *                         University Research and Technology
@@ -9,6 +10,8 @@
  *                         University of Stuttgart.  All rights reserved.
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
+ * Copyright (c) 2015      Los Alamos National Security, LLC. All rights
+ *                         reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -51,13 +54,10 @@ OBJ_CLASS_DECLARATION(mca_pml_ob1_comm_proc_t);
  */
 struct mca_pml_comm_t {
     opal_object_t super;
-#if OPAL_ENABLE_MULTI_THREADS
     volatile uint32_t recv_sequence;  /**< recv request sequence number - receiver side */
-#else
-    uint32_t recv_sequence;  /**< recv request sequence number - receiver side */
-#endif
     opal_mutex_t matching_lock;   /**< matching lock */
     opal_list_t wild_receives;    /**< queue of unmatched wild (source process not specified) receives */
+    opal_mutex_t proc_lock;
     mca_pml_ob1_comm_proc_t **procs;
     size_t num_procs;
     size_t last_probed;
@@ -71,9 +71,14 @@ static inline mca_pml_ob1_comm_proc_t *mca_pml_ob1_peer_lookup (struct ompi_comm
     mca_pml_ob1_comm_t *pml_comm = (mca_pml_ob1_comm_t *)comm->c_pml_comm;
 
     if (OPAL_UNLIKELY(NULL == pml_comm->procs[rank])) {
-        pml_comm->procs[rank] = OBJ_NEW(mca_pml_ob1_comm_proc_t);
-        pml_comm->procs[rank]->ompi_proc = ompi_comm_peer_lookup (comm, rank);
-        OBJ_RETAIN(pml_comm->procs[rank]->ompi_proc);
+        OPAL_THREAD_LOCK(&pml_comm->proc_lock);
+        if (NULL == pml_comm->procs[rank]) {
+            pml_comm->procs[rank] = OBJ_NEW(mca_pml_ob1_comm_proc_t);
+            pml_comm->procs[rank]->ompi_proc = ompi_comm_peer_lookup (comm, rank);
+            OBJ_RETAIN(pml_comm->procs[rank]->ompi_proc);
+            opal_atomic_wmb ();
+        }
+        OPAL_THREAD_UNLOCK(&pml_comm->proc_lock);
     }
 
     return pml_comm->procs[rank];